388 lines
13 KiB
Python
388 lines
13 KiB
Python
"""
|
||
正则规则 API 路由
|
||
|
||
提供正则规则的 CRUD 操作和导入导出功能
|
||
"""
|
||
from fastapi import APIRouter, HTTPException, UploadFile, File
|
||
from pydantic import BaseModel
|
||
from typing import List, Optional
|
||
import json
|
||
import logging
|
||
|
||
from services.regex_service import regex_service
|
||
from models.regex_rules import RegexRule, RegexRuleset, RegexScope
|
||
from services.system_settings_service import system_settings_service
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/regex", tags=["regex"])
|
||
|
||
|
||
# ==================== 数据模型 ====================
|
||
|
||
class RuleUpdateRequest(BaseModel):
|
||
"""规则更新请求"""
|
||
rule: RegexRule
|
||
scope: RegexScope
|
||
name: Optional[str] = None # 角色卡名称或预设名称（scope 为 CHARACTER/PRESET 时需要）
|
||
|
||
|
||
class SystemSettingsUpdate(BaseModel):
|
||
"""系统设置更新请求"""
|
||
thinkingTagPrefix: Optional[str] = None
|
||
thinkingTagSuffix: Optional[str] = None
|
||
currentPresetName: Optional[str] = None
|
||
|
||
|
||
# ==================== 规则查询 ====================
|
||
|
||
@router.get("/rules")
|
||
async def get_rules(
|
||
character_name: Optional[str] = None,
|
||
preset_name: Optional[str] = None
|
||
):
|
||
"""
|
||
获取适用的正则规则列表
|
||
|
||
Args:
|
||
character_name: 当前角色卡名称（可选）
|
||
preset_name: 当前预设名称（可选）
|
||
|
||
Returns:
|
||
规则列表
|
||
"""
|
||
try:
|
||
rules = regex_service.get_rules_for_context(character_name, preset_name)
|
||
return {
|
||
"success": True,
|
||
"rules": [rule.dict() for rule in rules],
|
||
"count": len(rules)
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取规则失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/rulesets/global")
|
||
async def get_global_rulesets():
|
||
"""获取所有全局规则集"""
|
||
try:
|
||
rulesets = list(regex_service.global_rulesets.values())
|
||
return {
|
||
"success": True,
|
||
"rulesets": [rs.dict() for rs in rulesets]
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取全局规则集失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/rulesets/character/{character_name}")
|
||
async def get_character_ruleset(character_name: str):
|
||
"""获取指定角色卡的规则集"""
|
||
try:
|
||
if character_name in regex_service.character_rulesets:
|
||
ruleset = regex_service.character_rulesets[character_name]
|
||
return {
|
||
"success": True,
|
||
"ruleset": ruleset.dict()
|
||
}
|
||
else:
|
||
return {
|
||
"success": True,
|
||
"ruleset": None
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取角色规则集失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/rulesets/preset/{preset_name}")
|
||
async def get_preset_ruleset(preset_name: str):
|
||
"""获取指定预设的规则集"""
|
||
try:
|
||
if preset_name in regex_service.preset_rulesets:
|
||
ruleset = regex_service.preset_rulesets[preset_name]
|
||
return {
|
||
"success": True,
|
||
"ruleset": ruleset.dict()
|
||
}
|
||
else:
|
||
return {
|
||
"success": True,
|
||
"ruleset": None
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取预设规则集失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
# ==================== 规则管理 ====================
|
||
|
||
@router.post("/rules")
|
||
async def add_rule(request: RuleUpdateRequest):
|
||
"""添加或更新规则"""
|
||
try:
|
||
# 获取现有的规则集
|
||
existing_ruleset = None
|
||
if request.scope == RegexScope.GLOBAL:
|
||
# 对于全局作用域，查找是否已有同名规则集
|
||
for ruleset_id, ruleset in regex_service.global_rulesets.items():
|
||
if ruleset.name == request.rule.scriptName:
|
||
existing_ruleset = ruleset
|
||
break
|
||
elif request.scope == RegexScope.CHARACTER and request.name:
|
||
if request.name in regex_service.character_rulesets:
|
||
existing_ruleset = regex_service.character_rulesets[request.name]
|
||
elif request.scope == RegexScope.PRESET and request.name:
|
||
if request.name in regex_service.preset_rulesets:
|
||
existing_ruleset = regex_service.preset_rulesets[request.name]
|
||
|
||
if existing_ruleset:
|
||
# 如果已存在同名规则集，则更新其中的规则
|
||
updated_rules = []
|
||
rule_found = False
|
||
for rule in existing_ruleset.rules:
|
||
if rule.id == request.rule.id:
|
||
# 更新现有规则
|
||
updated_rules.append(request.rule)
|
||
rule_found = True
|
||
else:
|
||
# 保留其他规则
|
||
updated_rules.append(rule)
|
||
|
||
if not rule_found:
|
||
# 如果没有找到相同ID的规则，则添加新规则
|
||
updated_rules.append(request.rule)
|
||
|
||
# 更新规则集
|
||
existing_ruleset.rules = updated_rules
|
||
regex_service.save_ruleset(existing_ruleset, request.scope, request.name)
|
||
else:
|
||
# 如果不存在同名规则集，则创建新的规则集
|
||
new_ruleset = RegexRuleset(
|
||
id=request.rule.id,
|
||
name=request.rule.scriptName,
|
||
rules=[request.rule]
|
||
)
|
||
regex_service.save_ruleset(new_ruleset, request.scope, request.name)
|
||
|
||
# 重新加载规则
|
||
regex_service._load_all_rules()
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "规则保存成功"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"保存规则失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.delete("/rules/{rule_id}")
|
||
async def delete_rule(rule_id: str, scope: str = "global", name: Optional[str] = None):
|
||
"""
|
||
删除规则
|
||
|
||
Args:
|
||
rule_id: 规则ID
|
||
scope: 作用域 (global/character/preset)
|
||
name: 角色名或预设名（scope 为 character/preset 时需要）
|
||
"""
|
||
try:
|
||
from models.regex_rules import RegexScope
|
||
|
||
scope_map = {
|
||
"global": RegexScope.GLOBAL,
|
||
"character": RegexScope.CHARACTER,
|
||
"preset": RegexScope.PRESET
|
||
}
|
||
scope_enum = scope_map.get(scope, RegexScope.GLOBAL)
|
||
|
||
# 找到包含该规则的规则集
|
||
if scope_enum == RegexScope.GLOBAL:
|
||
rulesets = regex_service.global_rulesets
|
||
elif scope_enum == RegexScope.CHARACTER:
|
||
if not name:
|
||
raise ValueError("删除角色规则需要提供角色名称")
|
||
rulesets = {name: regex_service.character_rulesets.get(name)} if name in regex_service.character_rulesets else {}
|
||
elif scope_enum == RegexScope.PRESET:
|
||
if not name:
|
||
raise ValueError("删除预设规则需要提供预设名称")
|
||
rulesets = {name: regex_service.preset_rulesets.get(name)} if name in regex_service.preset_rulesets else {}
|
||
|
||
# 查找并删除规则
|
||
deleted = False
|
||
for ruleset_name, ruleset in rulesets.items():
|
||
if not ruleset:
|
||
continue
|
||
|
||
original_count = len(ruleset.rules)
|
||
ruleset.rules = [r for r in ruleset.rules if r.id != rule_id]
|
||
|
||
if len(ruleset.rules) < original_count:
|
||
# 保存更新后的规则集
|
||
regex_service.save_ruleset(ruleset, scope_enum, ruleset_name if scope_enum != RegexScope.GLOBAL else None)
|
||
deleted = True
|
||
break
|
||
|
||
if not deleted:
|
||
return {
|
||
"success": False,
|
||
"message": "未找到指定的规则"
|
||
}
|
||
|
||
# 重新加载规则
|
||
regex_service._load_all_rules()
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "规则已删除"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"删除规则失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
# ==================== 规则导入导出 ====================
|
||
|
||
@router.post("/import")
|
||
async def import_rules(file: UploadFile = File(...)):
|
||
"""
|
||
导入正则规则（支持 SillyTavern 格式）- 文件上传方式
|
||
|
||
可以导入：
|
||
1. 单个规则文件（JSON 数组）
|
||
2. 规则集文件（JSON 对象）
|
||
"""
|
||
try:
|
||
content = await file.read()
|
||
data = json.loads(content.decode('utf-8'))
|
||
|
||
# 判断格式并导入
|
||
if isinstance(data, list):
|
||
# SillyTavern 格式 - 导入为全局规则
|
||
ruleset = regex_service._convert_sillytavern_format(data, file.filename.replace('.json', ''))
|
||
regex_service.save_ruleset(ruleset, RegexScope.GLOBAL)
|
||
elif isinstance(data, dict):
|
||
if 'rules' in data:
|
||
# 规则集格式
|
||
ruleset = RegexRuleset(**data)
|
||
regex_service.save_ruleset(ruleset, RegexScope.GLOBAL)
|
||
else:
|
||
raise ValueError("未知的文件格式")
|
||
else:
|
||
raise ValueError("无效的文件格式")
|
||
|
||
# 重新加载规则
|
||
regex_service._load_all_rules()
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"成功导入规则集: {ruleset.name}"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"导入规则失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.post("/import-from-preset")
|
||
async def import_rules_from_preset(request: dict):
|
||
"""
|
||
从预设导入正则规则 - JSON 数据方式
|
||
|
||
Request Body:
|
||
{
|
||
"rules": [...], // SillyTavern 格式的 regex_scripts 数组
|
||
"scope": "preset", // 作用域：global/character/preset
|
||
"presetName": "预设名称" // 当 scope 为 preset 时需要
|
||
}
|
||
"""
|
||
try:
|
||
rules_data = request.get("rules", [])
|
||
scope_str = request.get("scope", "global")
|
||
preset_name = request.get("presetName")
|
||
|
||
if not rules_data or not isinstance(rules_data, list):
|
||
raise ValueError("无效的规则数据")
|
||
|
||
# 转换作用域字符串为枚举
|
||
scope_map = {
|
||
"global": RegexScope.GLOBAL,
|
||
"character": RegexScope.CHARACTER,
|
||
"preset": RegexScope.PRESET
|
||
}
|
||
scope = scope_map.get(scope_str, RegexScope.GLOBAL)
|
||
|
||
# 转换 SillyTavern 格式
|
||
name = preset_name or "imported_rules"
|
||
ruleset = regex_service._convert_sillytavern_format(rules_data, name, scope)
|
||
|
||
# 保存规则集
|
||
regex_service.save_ruleset(ruleset, scope, name)
|
||
|
||
# 重新加载规则
|
||
regex_service._load_all_rules()
|
||
|
||
logger.info(f"✅ 从预设导入 {len(rules_data)} 条正则规则到 {scope.value}: {name}")
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"成功导入 {len(rules_data)} 条正则规则",
|
||
"rulesetId": ruleset.id
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"从预设导入规则失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/export/global")
|
||
async def export_global_rules():
|
||
"""导出所有全局规则"""
|
||
try:
|
||
all_rulesets = list(regex_service.global_rulesets.values())
|
||
return {
|
||
"success": True,
|
||
"rulesets": [rs.dict() for rs in all_rulesets]
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"导出规则失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
# ==================== 系统设置 ====================
|
||
|
||
@router.get("/settings")
|
||
async def get_system_settings():
|
||
"""获取系统设置"""
|
||
try:
|
||
settings = system_settings_service.settings
|
||
return {
|
||
"success": True,
|
||
"settings": settings.dict()
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取系统设置失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.put("/settings")
|
||
async def update_system_settings(request: SystemSettingsUpdate):
|
||
"""更新系统设置"""
|
||
try:
|
||
if request.thinkingTagPrefix is not None or request.thinkingTagSuffix is not None:
|
||
prefix = request.thinkingTagPrefix or system_settings_service.settings.thinkingTagPrefix
|
||
suffix = request.thinkingTagSuffix or system_settings_service.settings.thinkingTagSuffix
|
||
system_settings_service.update_thinking_tags(prefix, suffix)
|
||
|
||
if request.currentPresetName is not None:
|
||
system_settings_service.update_current_preset(request.currentPresetName)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "系统设置已更新"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"更新系统设置失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|