完成请求推送，但组装mes还有问题
This commit is contained in:
332
backend/services/regex_service.py
Normal file
332
backend/services/regex_service.py
Normal file
@@ -0,0 +1,332 @@
|
||||
"""
|
||||
正则规则服务（重构版 - 文件夹结构）
|
||||
|
||||
负责加载、管理和应用正则替换规则。
|
||||
使用文件夹结构组织规则，兼容 SillyTavern 格式。
|
||||
|
||||
文件结构：
|
||||
data/regex/
|
||||
├── global/ # 全局规则
|
||||
│ └── default.json
|
||||
├── characters/ # 角色卡绑定规则
|
||||
│ └── {characterName}.json
|
||||
└── presets/ # 预设绑定规则
|
||||
└── {presetName}.json
|
||||
"""
|
||||
import re
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Dict
|
||||
from uuid import uuid4
|
||||
|
||||
from core.config import settings
|
||||
from models.regex_rules import RegexRule, RegexRuleset, RegexScope, RegexPlacement, SubstituteMode
|
||||
from services.system_settings_service import system_settings_service
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RegexService:
|
||||
"""
|
||||
正则替换规则服务（文件夹结构版）
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.regex_base_path = settings.DATA_PATH / "regex"
|
||||
self.global_path = self.regex_base_path / "global"
|
||||
self.characters_path = self.regex_base_path / "characters"
|
||||
self.presets_path = self.regex_base_path / "presets"
|
||||
|
||||
# 内存缓存
|
||||
self.global_rulesets: Dict[str, RegexRuleset] = {}
|
||||
self.character_rulesets: Dict[str, RegexRuleset] = {} # key: characterName
|
||||
self.preset_rulesets: Dict[str, RegexRuleset] = {} # key: presetName
|
||||
|
||||
self._ensure_directories()
|
||||
self._load_all_rules()
|
||||
|
||||
def _ensure_directories(self):
|
||||
"""确保目录结构存在"""
|
||||
for path in [self.regex_base_path, self.global_path, self.characters_path, self.presets_path]:
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _load_all_rules(self):
|
||||
"""加载所有规则"""
|
||||
self._load_global_rules()
|
||||
self._load_character_rules()
|
||||
self._load_preset_rules()
|
||||
|
||||
def _load_global_rules(self):
|
||||
"""加载全局规则"""
|
||||
if not self.global_path.exists():
|
||||
return
|
||||
|
||||
for json_file in self.global_path.glob("*.json"):
|
||||
try:
|
||||
with open(json_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
if isinstance(data, list):
|
||||
# SillyTavern 格式
|
||||
ruleset = self._convert_sillytavern_format(data, json_file.stem)
|
||||
elif isinstance(data, dict) and 'rules' in data:
|
||||
# 我们的规则集格式
|
||||
ruleset = RegexRuleset(**data)
|
||||
else:
|
||||
logger.warning(f"未知的规则文件格式: {json_file}")
|
||||
continue
|
||||
|
||||
self.global_rulesets[ruleset.id] = ruleset
|
||||
logger.info(f"加载全局规则集: {ruleset.name} ({len(ruleset.rules)} 条规则)")
|
||||
except Exception as e:
|
||||
logger.error(f"加载全局规则失败 {json_file}: {e}")
|
||||
|
||||
def _load_character_rules(self):
|
||||
"""加载角色卡绑定规则"""
|
||||
if not self.characters_path.exists():
|
||||
return
|
||||
|
||||
for json_file in self.characters_path.glob("*.json"):
|
||||
try:
|
||||
character_name = json_file.stem
|
||||
with open(json_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
if isinstance(data, list):
|
||||
ruleset = self._convert_sillytavern_format(data, character_name, RegexScope.CHARACTER)
|
||||
elif isinstance(data, dict) and 'rules' in data:
|
||||
ruleset = RegexRuleset(**data)
|
||||
else:
|
||||
continue
|
||||
|
||||
# 确保所有规则的 scope 正确
|
||||
for rule in ruleset.rules:
|
||||
rule.scope = RegexScope.CHARACTER
|
||||
rule.characterName = character_name
|
||||
|
||||
self.character_rulesets[character_name] = ruleset
|
||||
logger.info(f"加载角色规则: {character_name} ({len(ruleset.rules)} 条规则)")
|
||||
except Exception as e:
|
||||
logger.error(f"加载角色规则失败 {json_file}: {e}")
|
||||
|
||||
def _load_preset_rules(self):
|
||||
"""加载预设绑定规则"""
|
||||
if not self.presets_path.exists():
|
||||
return
|
||||
|
||||
for json_file in self.presets_path.glob("*.json"):
|
||||
try:
|
||||
preset_name = json_file.stem
|
||||
with open(json_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
if isinstance(data, list):
|
||||
ruleset = self._convert_sillytavern_format(data, preset_name, RegexScope.PRESET)
|
||||
elif isinstance(data, dict) and 'rules' in data:
|
||||
ruleset = RegexRuleset(**data)
|
||||
else:
|
||||
continue
|
||||
|
||||
# 确保所有规则的 scope 正确
|
||||
for rule in ruleset.rules:
|
||||
rule.scope = RegexScope.PRESET
|
||||
rule.presetName = preset_name
|
||||
|
||||
self.preset_rulesets[preset_name] = ruleset
|
||||
logger.info(f"加载预设规则: {preset_name} ({len(ruleset.rules)} 条规则)")
|
||||
except Exception as e:
|
||||
logger.error(f"加载预设规则失败 {json_file}: {e}")
|
||||
|
||||
def _convert_sillytavern_format(
|
||||
self,
|
||||
st_rules: List[dict],
|
||||
name: str,
|
||||
scope: RegexScope = RegexScope.GLOBAL
|
||||
) -> RegexRuleset:
|
||||
"""将 SillyTavern 格式转换为内部格式"""
|
||||
rules = []
|
||||
for idx, st_rule in enumerate(st_rules):
|
||||
find_regex = st_rule.get('findRegex', '')
|
||||
pattern, flags = self._parse_st_regex(find_regex)
|
||||
|
||||
# 解析 placement（默认为 AI_OUTPUT）
|
||||
placement_data = st_rule.get('placement', [2])
|
||||
placement = [RegexPlacement(p) for p in placement_data]
|
||||
|
||||
rule = RegexRule(
|
||||
id=str(uuid4()),
|
||||
scriptName=st_rule.get('scriptName', f"{name} 规则 {idx + 1}"),
|
||||
findRegex=pattern,
|
||||
replaceString=st_rule.get('replaceString', ''),
|
||||
trimStrings=st_rule.get('trimStrings', []),
|
||||
placement=placement,
|
||||
substituteRegex=SubstituteMode(st_rule.get('substituteRegex', 0)),
|
||||
markdownOnly=st_rule.get('markdownOnly', False),
|
||||
promptOnly=st_rule.get('promptOnly', False),
|
||||
runOnEdit=st_rule.get('runOnEdit', True),
|
||||
minDepth=st_rule.get('minDepth', 0),
|
||||
maxDepth=st_rule.get('maxDepth'),
|
||||
scope=scope,
|
||||
characterName=name if scope == RegexScope.CHARACTER else None,
|
||||
presetName=name if scope == RegexScope.PRESET else None,
|
||||
disabled=st_rule.get('disabled', False),
|
||||
order=idx
|
||||
)
|
||||
rules.append(rule)
|
||||
|
||||
ruleset = RegexRuleset(
|
||||
id=str(uuid4()),
|
||||
name=f"{name} 规则集",
|
||||
description=f"从 SillyTavern 导入的规则",
|
||||
rules=rules,
|
||||
isSillyTavernFormat=True
|
||||
)
|
||||
|
||||
return ruleset
|
||||
|
||||
def _parse_st_regex(self, st_regex: str) -> tuple[str, str]:
|
||||
"""解析 SillyTavern 的正则表达式格式 /pattern/flags"""
|
||||
if st_regex.startswith('/') and st_regex.count('/') >= 2:
|
||||
parts = st_regex.split('/')
|
||||
pattern = '/'.join(parts[1:-1])
|
||||
flags = parts[-1] if len(parts) > 2 else ''
|
||||
return pattern, flags
|
||||
else:
|
||||
return st_regex, ''
|
||||
|
||||
def apply_rules_by_placement(
|
||||
self,
|
||||
text: str,
|
||||
placement: int,
|
||||
character_name: Optional[str] = None,
|
||||
preset_name: Optional[str] = None,
|
||||
message_depth: int = 0
|
||||
) -> str:
|
||||
"""
|
||||
根据 placement 应用正则规则
|
||||
|
||||
Args:
|
||||
text: 要处理的文本
|
||||
placement: 应用位置（0-5）
|
||||
character_name: 当前角色卡名称
|
||||
preset_name: 当前预设名称
|
||||
message_depth: 消息深度
|
||||
|
||||
Returns:
|
||||
处理后的文本
|
||||
"""
|
||||
rules = self.get_rules_for_context(character_name, preset_name)
|
||||
|
||||
result = text
|
||||
for rule in rules:
|
||||
# 检查此规则是否适用于当前 placement
|
||||
if placement not in [p.value for p in rule.placement]:
|
||||
continue
|
||||
|
||||
# 检查消息深度限制
|
||||
if message_depth < rule.minDepth:
|
||||
continue
|
||||
if rule.maxDepth is not None and message_depth > rule.maxDepth:
|
||||
continue
|
||||
|
||||
# 应用规则
|
||||
result = self._apply_single_rule(result, rule)
|
||||
|
||||
return result
|
||||
|
||||
def _apply_single_rule(self, text: str, rule: RegexRule) -> str:
|
||||
"""应用单条正则规则"""
|
||||
try:
|
||||
flags = 0
|
||||
if 'i' in rule.findRegex:
|
||||
flags |= re.IGNORECASE
|
||||
if 'm' in rule.findRegex:
|
||||
flags |= re.MULTILINE
|
||||
if 's' in rule.findRegex:
|
||||
flags |= re.DOTALL
|
||||
|
||||
pattern = rule.findRegex.replace('i', '').replace('m', '').replace('s', '')
|
||||
|
||||
if rule.substituteRegex == SubstituteMode.REPLACE_FIRST:
|
||||
result = re.sub(pattern, rule.replaceString, text, count=1, flags=flags)
|
||||
else:
|
||||
result = re.sub(pattern, rule.replaceString, text, flags=flags)
|
||||
|
||||
for trim_str in rule.trimStrings:
|
||||
result = result.replace(trim_str, '')
|
||||
|
||||
return result
|
||||
except re.error as e:
|
||||
logger.error(f"正则表达式错误 [{rule.scriptName}]: {e}")
|
||||
return text
|
||||
|
||||
def get_rules_for_context(
|
||||
self,
|
||||
character_name: Optional[str] = None,
|
||||
preset_name: Optional[str] = None
|
||||
) -> List[RegexRule]:
|
||||
"""
|
||||
根据上下文获取适用的规则列表
|
||||
|
||||
优先级：全局规则 + 角色规则 + 预设规则
|
||||
"""
|
||||
applicable_rules = []
|
||||
|
||||
# 1. 加载全局规则
|
||||
for ruleset in self.global_rulesets.values():
|
||||
for rule in ruleset.rules:
|
||||
if not rule.disabled:
|
||||
applicable_rules.append(rule)
|
||||
|
||||
# 2. 加载角色卡规则
|
||||
if character_name and character_name in self.character_rulesets:
|
||||
ruleset = self.character_rulesets[character_name]
|
||||
for rule in ruleset.rules:
|
||||
if not rule.disabled:
|
||||
applicable_rules.append(rule)
|
||||
|
||||
# 3. 加载预设规则
|
||||
if preset_name and preset_name in self.preset_rulesets:
|
||||
ruleset = self.preset_rulesets[preset_name]
|
||||
for rule in ruleset.rules:
|
||||
if not rule.disabled:
|
||||
applicable_rules.append(rule)
|
||||
|
||||
# 按 order 排序
|
||||
applicable_rules.sort(key=lambda r: r.order)
|
||||
|
||||
return applicable_rules
|
||||
|
||||
def save_ruleset(self, ruleset: RegexRuleset, scope: RegexScope, name: Optional[str] = None):
|
||||
"""保存规则集到文件"""
|
||||
if scope == RegexScope.GLOBAL:
|
||||
file_path = self.global_path / f"{ruleset.id}.json"
|
||||
elif scope == RegexScope.CHARACTER:
|
||||
file_path = self.characters_path / f"{name or 'unknown'}.json"
|
||||
elif scope == RegexScope.PRESET:
|
||||
file_path = self.presets_path / f"{name or 'unknown'}.json"
|
||||
else:
|
||||
raise ValueError(f"未知的作用域: {scope}")
|
||||
|
||||
with open(file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(ruleset.dict(), f, ensure_ascii=False, indent=2)
|
||||
|
||||
logger.info(f"保存规则集到: {file_path}")
|
||||
|
||||
def delete_ruleset(self, scope: RegexScope, name: str):
|
||||
"""删除规则集"""
|
||||
if scope == RegexScope.CHARACTER:
|
||||
file_path = self.characters_path / f"{name}.json"
|
||||
elif scope == RegexScope.PRESET:
|
||||
file_path = self.presets_path / f"{name}.json"
|
||||
else:
|
||||
raise ValueError(f"不能删除全局规则集")
|
||||
|
||||
if file_path.exists():
|
||||
file_path.unlink()
|
||||
logger.info(f"删除规则集: {file_path}")
|
||||
|
||||
|
||||
# 全局实例
|
||||
regex_service = RegexService()
|
||||
Reference in New Issue
Block a user