362 lines
14 KiB
Python
362 lines
14 KiB
Python
"""
|
||
正则规则服务（重构版 - 文件夹结构）
|
||
|
||
负责加载、管理和应用正则替换规则。
|
||
使用文件夹结构组织规则，兼容 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,
|
||
is_for_llm: bool = False, # ✅ 新增：是否发送给LLM
|
||
is_markdown_rendered: bool = False # ✅ 新增：是否已Markdown渲染
|
||
) -> str:
|
||
"""
|
||
根据 placement 应用正则规则
|
||
|
||
Args:
|
||
text: 要处理的文本
|
||
placement: 应用位置（0-5）
|
||
character_name: 当前角色卡名称
|
||
preset_name: 当前预设名称
|
||
message_depth: 消息深度
|
||
is_for_llm: 是否用于发送给 LLM（影响 promptOnly 逻辑）
|
||
is_markdown_rendered: 是否是 Markdown 渲染后的内容（影响 markdownOnly 逻辑）
|
||
|
||
Returns:
|
||
处理后的文本
|
||
"""
|
||
rules = self.get_rules_for_context(character_name, preset_name)
|
||
|
||
result = text
|
||
for rule in rules:
|
||
# ✅ SillyTavern 逻辑：根据 markdownOnly 和 promptOnly 决定是否应用
|
||
# - 双 false：应用到所有场景（包括保存数据、发送LLM、显示）
|
||
# - markdownOnly=true：只应用于 Markdown 渲染（前端显示）
|
||
# - promptOnly=true：只应用于发送给 LLM
|
||
# - 双 true：应用到所有场景（但不修改存储，由调用方决定）
|
||
|
||
# 如果是保存数据的场景（is_for_llm=False 且 is_markdown_rendered=False）
|
||
# 只应用双 false 的规则
|
||
if not is_for_llm and not is_markdown_rendered:
|
||
# 保存数据：只应用双 false 的规则
|
||
if rule.markdownOnly or rule.promptOnly:
|
||
continue
|
||
|
||
# 如果是发送给 LLM 的场景
|
||
elif is_for_llm and not is_markdown_rendered:
|
||
# 不应用 markdownOnly=true 且 promptOnly=false 的规则
|
||
if rule.markdownOnly and not rule.promptOnly:
|
||
continue
|
||
|
||
# 如果是 Markdown 渲染的场景（前端显示）
|
||
elif is_markdown_rendered and not is_for_llm:
|
||
# 不应用 promptOnly=true 且 markdownOnly=false 的规则
|
||
if rule.promptOnly and not rule.markdownOnly:
|
||
continue
|
||
|
||
# 检查此规则是否适用于当前 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()
|