""" Review System API Routes 审核系统 API 路由 提供审核配置、执行、维度和自定义规则的管理接口 """ from fastapi import APIRouter, Depends, HTTPException, status, Query, Body from typing import List, Optional import uuid from datetime import datetime from app.models.review import ( ReviewConfig, ReviewConfigUpdate, ReviewResult, ReviewRequest, ReviewResponse, DimensionInfo, DimensionConfig, DimensionType, CustomRule, CustomRuleCreate, CustomRuleUpdate, ReviewPreset ) from app.models.project import SeriesProject, Episode from app.core.review.review_manager import ReviewManager, get_review_manager from app.db.repositories import project_repo, episode_repo from app.utils.logger import get_logger logger = get_logger(__name__) router = APIRouter(tags=["审核系统"]) # ============================================ # 项目审核配置管理 # ============================================ @router.get("/projects/{project_id}/review-config") async def get_review_config(project_id: str): """ 获取项目的审核配置 Args: project_id: 项目ID Returns: 审核配置（前端格式） """ project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) # 从项目配置中获取审核配置 # 如果还没有配置，返回默认配置 raw_config = getattr(project, 'reviewConfig', None) # 如果配置是字典，需要转换为 ReviewConfig 对象 if isinstance(raw_config, dict): review_config = ReviewConfig(**raw_config) elif raw_config is None: review_config = ReviewConfig() else: # 已经是 ReviewConfig 对象 review_config = raw_config # 转换为前端格式 frontend_config = _convert_to_frontend_config(review_config) return frontend_config def _convert_to_frontend_config(config: ReviewConfig) -> dict: """ 将后端 ReviewConfig 转换为前端格式 Args: config: 后端配置 Returns: 前端格式配置 """ # 定义默认维度配置（前端格式） default_dimensions = { "character_consistency": {"name": "角色一致性", "enabled": True, "strictness": 50, "weight": 0.8}, "plot_coherence": {"name": "剧情连贯性", "enabled": True, "strictness": 50, "weight": 0.9}, "dialogue_quality": {"name": "对话质量", "enabled": True, "strictness": 50, "weight": 0.7}, "pacing": {"name": "节奏控制", "enabled": True, "strictness": 50, "weight": 0.6}, "emotional_depth": {"name": "情感深度", "enabled": True, "strictness": 50, "weight": 0.7}, "thematic_strength": {"name": "主题强度", "enabled": True, "strictness": 50, "weight": 0.8} } # 从后端配置中提取维度设置 dimensions = default_dimensions.copy() for dim_type, dim_config in config.dimension_settings.items(): dim_key = dim_type.value if dim_key in default_dimensions: dimensions[dim_key] = { "name": default_dimensions[dim_key]["name"], "enabled": dim_config.enabled, "strictness": int(dim_config.strictness * 100), # 转换 0-1 为 0-100 "weight": dim_config.weight } else: # 处理未知维度 dimensions[dim_key] = { "name": dim_type.value, "enabled": dim_config.enabled, "strictness": int(dim_config.strictness * 100), "weight": dim_config.weight } # 转换自定义规则 custom_rules = [] for rule in config.custom_rules: dimension_to_category = { DimensionType.character: "character", DimensionType.plot: "plot", DimensionType.dialogue: "dialogue", DimensionType.pacing: "pacing", DimensionType.emotional_depth: "emotion", DimensionType.thematic_strength: "theme", DimensionType.custom: "other" } custom_rules.append({ "id": rule.id, "name": rule.name, "description": rule.description, "triggerCondition": rule.trigger_condition, "severity": rule.severity.value if hasattr(rule.severity, 'value') else rule.severity, "category": dimension_to_category.get(rule.dimension, "other"), "isActive": rule.enabled }) # 根据 overall_strictness 确定 preset strictness_to_preset = { (0.0, 0.6): "draft", (0.6, 0.8): "standard", (0.8, 1.0): "strict" } preset = "custom" for (low, high), p in strictness_to_preset.items(): if low <= config.overall_strictness < high: preset = p break return { "preset": preset, "dimensions": dimensions, "customRules": custom_rules, "createdAt": datetime.now().isoformat(), "updatedAt": datetime.now().isoformat(), # 额外的配置信息 "overall_strictness": int(config.overall_strictness * 100), # 转换为 0-100 "pass_threshold": config.pass_threshold, "auto_fix_enabled": config.auto_fix_enabled, "enabled_review_skills": config.enabled_review_skills, "custom_rules_count": len(config.custom_rules), "enabled_dimensions_count": sum(1 for d in config.dimension_settings.values() if d.enabled), "total_dimensions_count": len(config.dimension_settings) } @router.put("/projects/{project_id}/review-config") async def update_review_config( project_id: str, config_update: ReviewConfigUpdate ): """ 更新项目的审核配置 Args: project_id: 项目ID config_update: 配置更新数据 Returns: 更新后的配置（前端格式） """ project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) try: # 获取现有配置 raw_config = getattr(project, 'reviewConfig', None) # 如果配置是字典，需要转换为 ReviewConfig 对象 if isinstance(raw_config, dict): current_config = ReviewConfig(**raw_config) elif raw_config is None: current_config = ReviewConfig() else: # 已经是 ReviewConfig 对象 current_config = raw_config # 确保 dimension_settings 存在并是 dict 类型 if not hasattr(current_config, 'dimension_settings') or not isinstance(current_config.dimension_settings, dict): current_config.dimension_settings = {} # 获取更新数据 update_data = config_update.dict(exclude_unset=True, exclude_none=True) # 处理前端格式转换 if "dimensions" in update_data and update_data["dimensions"]: # 前端发送的 dimensions 需要转换为后端的 dimension_settings dimension_settings = {} for dim_key, dim_config in update_data["dimensions"].items(): # 转换为 DimensionType try: dim_type = DimensionType(dim_key) except ValueError: # 如果不是有效的 DimensionType，使用 custom dim_type = DimensionType.custom # 转换前端格式到后端格式 # frontend: strictness 0-100, backend: 0-1 strictness = dim_config.get("strictness", 50) / 100.0 dimension_settings[dim_type] = DimensionConfig( enabled=dim_config.get("enabled", True), strictness=strictness, weight=dim_config.get("weight", 0.5), custom_rules=[] ) current_config.dimension_settings = dimension_settings if "customRules" in update_data and update_data["customRules"]: # 前端发送的 customRules 需要转换为后端的 custom_rules custom_rules = [] for rule_data in update_data["customRules"]: # 映射 severity severity = rule_data.get("severity", "medium") # 映射 category 到 dimension category_to_dimension = { "character": DimensionType.character, "plot": DimensionType.plot, "dialogue": DimensionType.dialogue, "pacing": DimensionType.pacing, "emotion": DimensionType.emotional_depth, "theme": DimensionType.thematic_strength, "other": DimensionType.custom } dimension = category_to_dimension.get(rule_data.get("category", "other"), DimensionType.custom) custom_rules.append(CustomRule( id=rule_data.get("id", str(uuid.uuid4())), name=rule_data.get("name", ""), description=rule_data.get("description", ""), trigger_condition=rule_data.get("triggerCondition", ""), dimension=dimension, severity=severity, enabled=rule_data.get("isActive", True), created_at=datetime.now(), updated_at=datetime.now() )) current_config.custom_rules = custom_rules # 处理 preset if "preset" in update_data and update_data["preset"]: # 根据 preset 设置 overall_strictness preset_strictness = { "draft": 0.5, "standard": 0.7, "strict": 0.9, "custom": current_config.overall_strictness } current_config.overall_strictness = preset_strictness.get( update_data["preset"], current_config.overall_strictness ) # 更新其他字段（排除已处理的前端字段） frontend_fields = {"dimensions", "customRules", "preset"} for field, value in update_data.items(): if field not in frontend_fields: setattr(current_config, field, value) # 保存到项目 await project_repo.update(project_id, { "reviewConfig": current_config.dict() }) logger.info(f"更新项目审核配置: {project_id}") # 返回前端格式 return _convert_to_frontend_config(current_config) except Exception as e: logger.error(f"更新审核配置失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"更新配置失败: {str(e)}" ) @router.get("/projects/{project_id}/review-presets", response_model=List[ReviewPreset]) async def get_review_presets(project_id: str = None): """ 获取审核预设列表 Args: project_id: 可选的项目ID，用于返回项目特定的预设 Returns: List[ReviewPreset]: 预设列表 """ # 内置预设 presets = [ ReviewPreset( id="strict", name="严格审核", description="适用于高质量要求，检查所有维度", category="quality", config=ReviewConfig( enabled_review_skills=["consistency_checker"], overall_strictness=0.9, dimension_settings={ DimensionType.consistency: DimensionConfig(enabled=True, strictness=0.9, weight=1.2), DimensionType.quality: DimensionConfig(enabled=True, strictness=0.8, weight=1.0), DimensionType.pacing: DimensionConfig(enabled=True, strictness=0.7, weight=0.8), DimensionType.dialogue: DimensionConfig(enabled=True, strictness=0.8, weight=1.0), DimensionType.character: DimensionConfig(enabled=True, strictness=0.9, weight=1.1), DimensionType.plot: DimensionConfig(enabled=True, strictness=0.9, weight=1.2) }, pass_threshold=85.0 ) ), ReviewPreset( id="balanced", name="平衡审核", description="适用于一般创作需求，平衡质量和效率", category="general", config=ReviewConfig( enabled_review_skills=["consistency_checker"], overall_strictness=0.7, dimension_settings={ DimensionType.consistency: DimensionConfig(enabled=True, strictness=0.7, weight=1.2), DimensionType.quality: DimensionConfig(enabled=True, strictness=0.6, weight=1.0), DimensionType.pacing: DimensionConfig(enabled=False, strictness=0.5, weight=0.8), DimensionType.dialogue: DimensionConfig(enabled=True, strictness=0.6, weight=1.0), DimensionType.character: DimensionConfig(enabled=True, strictness=0.7, weight=1.1), DimensionType.plot: DimensionConfig(enabled=True, strictness=0.7, weight=1.2) }, pass_threshold=75.0 ) ), ReviewPreset( id="quick", name="快速审核", description="只检查核心维度，适用于快速迭代", category="efficiency", config=ReviewConfig( enabled_review_skills=["consistency_checker"], overall_strictness=0.5, dimension_settings={ DimensionType.consistency: DimensionConfig(enabled=True, strictness=0.7, weight=1.5), DimensionType.quality: DimensionConfig(enabled=False, strictness=0.5, weight=1.0), DimensionType.pacing: DimensionConfig(enabled=False, strictness=0.5, weight=0.8), DimensionType.dialogue: DimensionConfig(enabled=False, strictness=0.5, weight=1.0), DimensionType.character: DimensionConfig(enabled=False, strictness=0.5, weight=1.1), DimensionType.plot: DimensionConfig(enabled=True, strictness=0.6, weight=1.5) }, pass_threshold=70.0 ) ), ReviewPreset( id="dialogue_focused", name="对话专项", description="专注于对话质量和人物口吻", category="specialized", config=ReviewConfig( enabled_review_skills=[], overall_strictness=0.8, dimension_settings={ DimensionType.consistency: DimensionConfig(enabled=True, strictness=0.6, weight=0.8), DimensionType.dialogue: DimensionConfig(enabled=True, strictness=0.9, weight=1.5), DimensionType.character: DimensionConfig(enabled=True, strictness=0.8, weight=1.2) }, pass_threshold=75.0 ) ) ] return presets # ============================================ # 项目审核 Skills 管理 # ============================================ @router.get("/projects/{project_id}/review/skills") async def get_project_review_skills(project_id: str): """ 获取项目的审核 Skills 返回所有可用的 Skills（不仅是review类型）以及项目中已启用的审核 Skills Args: project_id: 项目ID Returns: 包含可用 Skills 和已启用 Skills 的响应 """ from app.core.skills.skill_manager import get_skill_manager project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) skill_manager = get_skill_manager() # 获取所有 Skills（不限制分类） all_skills = await skill_manager.list_skills() # 获取项目已启用的审核 Skills raw_config = getattr(project, 'reviewConfig', None) enabled_skill_ids = [] if raw_config and isinstance(raw_config, dict): enabled_skill_ids = raw_config.get('enabled_review_skills', []) # 获取所有有效的 skill IDs valid_skill_ids = {skill_dict.get("id") if hasattr(skill, 'dict') else skill.get("id") for skill in all_skills for skill_dict in [skill.dict() if hasattr(skill, 'dict') else skill]} # 过滤掉不存在的 skill IDs（防止已删除的技能仍在启用列表中） valid_enabled_skill_ids = [sid for sid in enabled_skill_ids if sid in valid_skill_ids] # 如果过滤后发现有不存在的 skill IDs，更新项目配置 if len(valid_enabled_skill_ids) != len(enabled_skill_ids): logger.warning(f"Project {project_id} has invalid enabled_review_skills: " f"removed {set(enabled_skill_ids) - set(valid_enabled_skill_ids)}") # 异步更新项目配置（不阻塞响应） import json from app.models.review import ReviewConfig if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = ReviewConfig() config.enabled_review_skills = valid_enabled_skill_ids await project_repo.update(project_id, { "reviewConfig": json.loads(config.json()) }) # 为每个 Skill 添加启用状态，并转换为前端期望的格式 skills_with_status = [] for skill in all_skills: skill_dict = skill.dict() if hasattr(skill, 'dict') else skill # 转换为前端期望的格式 frontend_skill = { "id": skill_dict.get("id", ""), "name": skill_dict.get("name", ""), "description": skill_dict.get("behavior_guide", "") or skill_dict.get("description", ""), "category": skill_dict.get("category", "other"), "enabled": skill_dict.get("id", "") in valid_enabled_skill_ids, "type": skill_dict.get("type", "user"), "tags": skill_dict.get("tags", []) } skills_with_status.append(frontend_skill) return { "skills": skills_with_status, "enabled_skill_ids": valid_enabled_skill_ids } @router.put("/projects/{project_id}/review/skills/{skill_id}") async def update_project_review_skill( project_id: str, skill_id: str, data: dict = Body(...) ): """ 更新项目审核 Skill 的启用状态 Args: project_id: 项目ID skill_id: Skill ID data: 请求体，包含 enabled 字段 Returns: 更新后的状态 """ from app.models.review import ReviewConfig import json enabled = data.get('enabled', True) project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) # 获取现有配置 raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): current_config = ReviewConfig(**raw_config) else: current_config = ReviewConfig() # 更新启用的 Skills 列表 enabled_skills = set(current_config.enabled_review_skills or []) if enabled: enabled_skills.add(skill_id) else: enabled_skills.discard(skill_id) current_config.enabled_review_skills = list(enabled_skills) # 保存配置 await project_repo.update(project_id, { "reviewConfig": json.loads(current_config.json()) }) logger.info(f"更新项目审核Skill: {project_id} - {skill_id} = {enabled}, enabled_skills: {list(enabled_skills)}") return { "success": True, "skill_id": skill_id, "enabled": enabled, "enabled_skills": list(enabled_skills) } # ============================================ # 审核执行 # ============================================ @router.post("/projects/{project_id}/episodes/{episode_number}/review", response_model=ReviewResponse) async def review_episode( project_id: str, episode_number: int, request: ReviewRequest, review_mgr: ReviewManager = Depends(get_review_manager) ): """ 执行剧集审核 Args: project_id: 项目ID episode_number: 集数 request: 审核请求 review_mgr: 审核管理器（依赖注入） Returns: ReviewResponse: 审核结果 """ # 获取项目 project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) # 获取剧集 episodes = await episode_repo.list_by_project(project_id) episode = None for ep in episodes: if ep.number == episode_number: episode = ep break if not episode: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"剧集不存在: EP{episode_number}" ) # 检查剧集是否有内容 if not episode.content: return ReviewResponse( success=False, message=f"剧集 EP{episode_number} 没有内容，无法审核", episode_id=episode.id ) try: # 获取审核配置 raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = raw_config or ReviewConfig() # 执行审核 logger.info(f"开始审核: 项目 {project_id}, EP{episode_number}") result = await review_mgr.review_episode( project=project, episode=episode, config=config, dimensions=request.dimensions ) # 保存审核结果到剧集 await episode_repo.update(project_id, { "reviewResult": result.dict(), "status": "completed" if result.passed else "needs-review" }) logger.info(f"审核完成: EP{episode_number}, 分数={result.overall_score:.1f}") return ReviewResponse( success=True, review_result=result, message=f"审核完成，总分: {result.overall_score:.1f}", episode_id=episode.id ) except Exception as e: logger.error(f"审核失败: {str(e)}") return ReviewResponse( success=False, message=f"审核失败: {str(e)}", episode_id=episode.id ) @router.get("/projects/{project_id}/episodes/{episode_number}/review-result", response_model=Optional[ReviewResult]) async def get_review_result(project_id: str, episode_number: int): """ 获取剧集的审核结果 Args: project_id: 项目ID episode_number: 集数 Returns: ReviewResult: 审核结果，如果未审核则返回null """ episodes = await episode_repo.list_by_project(project_id) episode = None for ep in episodes: if ep.number == episode_number: episode = ep break if not episode: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"剧集不存在: EP{episode_number}" ) # 返回审核结果（如果有） return getattr(episode, 'reviewResult', None) # ============================================ # 维度管理 # ============================================ @router.get("/dimensions", response_model=List[DimensionInfo]) async def list_dimensions(): """ 列出所有可用的审核维度 Returns: List[DimensionInfo]: 维度信息列表 """ from app.core.review.review_manager import ReviewManager temp_manager = ReviewManager() dimensions = [] for dim_type, dim_info in temp_manager.DIMENSIONS.items(): dimension_info = DimensionInfo( id=dim_type, name=dim_info["name"], description=dim_info["description"], default_strictness=dim_info["default_strictness"], supported_skill_ids=dim_info["skill_ids"] ) dimensions.append(dimension_info) return dimensions @router.get("/dimensions/{dimension_id}", response_model=DimensionInfo) async def get_dimension_info(dimension_id: DimensionType): """ 获取单个维度的详细信息 Args: dimension_id: 维度ID Returns: DimensionInfo: 维度信息 """ from app.core.review.review_manager import ReviewManager temp_manager = ReviewManager() dim_info = temp_manager.DIMENSIONS.get(dimension_id) if not dim_info: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"维度不存在: {dimension_id}" ) return DimensionInfo( id=dimension_id, name=dim_info["name"], description=dim_info["description"], default_strictness=dim_info["default_strictness"], supported_skill_ids=dim_info["skill_ids"] ) # ============================================ # 自定义规则管理 # ============================================ @router.get("/custom-rules", response_model=List[CustomRule]) async def list_custom_rules( project_id: Optional[str] = None, dimension: Optional[DimensionType] = None ): """ 列出自定义规则 Args: project_id: 可选的项目ID，只返回该项目的规则 dimension: 可选的维度筛选 Returns: List[CustomRule]: 自定义规则列表 """ if project_id: # 从项目获取规则 project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = raw_config or ReviewConfig() rules = config.custom_rules else: # 返回空列表（暂时不支持全局规则） rules = [] # 按维度筛选 if dimension: rules = [r for r in rules if r.dimension == dimension] return rules @router.post("/custom-rules", response_model=CustomRule, status_code=status.HTTP_201_CREATED) async def create_custom_rule( rule_data: CustomRuleCreate, project_id: str = Query(..., description="项目ID") ): """ 创建自定义规则 Args: rule_data: 规则创建数据 project_id: 项目ID Returns: CustomRule: 创建的规则 """ project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) try: # 处理 category 到 dimension 的映射 if rule_data.category: # 映射 category 到 dimension category_to_dimension = { "character": DimensionType.character, "plot": DimensionType.plot, "dialogue": DimensionType.dialogue, "pacing": DimensionType.pacing, "emotion": DimensionType.emotional_depth, "theme": DimensionType.thematic_strength, "other": DimensionType.custom } dimension = category_to_dimension.get(rule_data.category, DimensionType.custom) else: dimension = rule_data.dimension or DimensionType.custom # 创建规则 new_rule = CustomRule( id=str(uuid.uuid4()), name=rule_data.name, description=rule_data.description, trigger_condition=rule_data.trigger_condition, dimension=dimension, severity=rule_data.severity, enabled=True ) # 添加到项目配置 raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = raw_config or ReviewConfig() config.custom_rules.append(new_rule) # 保存 await project_repo.update(project_id, { "reviewConfig": config.dict() }) logger.info(f"创建自定义规则: {new_rule.id}") return new_rule except Exception as e: logger.error(f"创建自定义规则失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"创建规则失败: {str(e)}" ) @router.put("/custom-rules/{rule_id}", response_model=CustomRule) async def update_custom_rule( rule_id: str, rule_update: CustomRuleUpdate, project_id: str = Query(..., description="项目ID") ): """ 更新自定义规则 Args: rule_id: 规则ID rule_update: 规则更新数据 project_id: 项目ID Returns: CustomRule: 更新后的规则 """ project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = raw_config or ReviewConfig() # 查找规则 rule = None for r in config.custom_rules: if r.id == rule_id: rule = r break if not rule: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"规则不存在: {rule_id}" ) try: # 更新规则 update_data = rule_update.dict(exclude_unset=True) # 处理 category 到 dimension 的映射 if 'category' in update_data: category_to_dimension = { "character": DimensionType.character, "plot": DimensionType.plot, "dialogue": DimensionType.dialogue, "pacing": DimensionType.pacing, "emotion": DimensionType.emotional_depth, "theme": DimensionType.thematic_strength, "other": DimensionType.custom } update_data['dimension'] = category_to_dimension.get( update_data['category'], rule.dimension or DimensionType.custom ) # 移除 category 字段 del update_data['category'] for field, value in update_data.items(): setattr(rule, field, value) rule.updated_at = datetime.now() # 保存 await project_repo.update(project_id, { "reviewConfig": config.dict() }) logger.info(f"更新自定义规则: {rule_id}") return rule except Exception as e: logger.error(f"更新自定义规则失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"更新规则失败: {str(e)}" ) @router.delete("/custom-rules/{rule_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_custom_rule( rule_id: str, project_id: str = Query(..., description="项目ID") ): """ 删除自定义规则 Args: rule_id: 规则ID project_id: 项目ID Returns: None """ project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = raw_config or ReviewConfig() # 查找并删除规则 original_count = len(config.custom_rules) config.custom_rules = [r for r in config.custom_rules if r.id != rule_id] if len(config.custom_rules) == original_count: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"规则不存在: {rule_id}" ) try: # 保存 await project_repo.update(project_id, { "reviewConfig": config.dict() }) logger.info(f"删除自定义规则: {rule_id}") return None except Exception as e: logger.error(f"删除自定义规则失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"删除规则失败: {str(e)}" ) @router.post("/custom-rules/{rule_id}/test") async def test_custom_rule( rule_id: str, episode_id: str, project_id: str, review_mgr: ReviewManager = Depends(get_review_manager) ): """ 测试自定义规则 Args: rule_id: 规则ID episode_id: 剧集ID project_id: 项目ID review_mgr: 审核管理器 Returns: 测试结果 """ project = await project_repo.get(project_id) if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"项目不存在: {project_id}" ) episode = await episode_repo.get(episode_id) if not episode: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"剧集不存在: {episode_id}" ) raw_config = getattr(project, 'reviewConfig', None) if isinstance(raw_config, dict): config = ReviewConfig(**raw_config) else: config = raw_config or ReviewConfig() # 查找规则 rule = None for r in config.custom_rules: if r.id == rule_id: rule = r break if not rule: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"规则不存在: {rule_id}" ) try: # 测试规则 issues = await review_mgr._check_single_rule( episode=episode, rule=rule, config=config ) return { "rule_id": rule_id, "rule_name": rule.name, "episode_id": episode_id, "violations": len(issues), "issues": [issue.dict() for issue in issues] } except Exception as e: logger.error(f"测试规则失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"测试失败: {str(e)}" ) # ============================================ # 审核配置预览 # ============================================ @router.post("/projects/{project_id}/review/preview") async def preview_review_config( project_id: str, config: ReviewConfigUpdate ): """ 预览审核配置 整合配置+Skills内容，文字形式展示，模拟注入Agent的效果 Args: project_id: 项目ID config: 配置数据 Returns: 预览文本和统计信息 """ try: from app.core.skills.skill_manager import get_skill_manager # 获取Skills详情 skill_manager = get_skill_manager() skills_info = [] if config.enabled_review_skills: for skill_id in config.enabled_review_skills: skill = skill_manager.get_skill_by_id(skill_id) if skill: skills_info.append({ "id": skill.id, "name": skill.name, "description": getattr(skill, 'description', ''), "behavior_guide": skill.behavior_guide }) # 生成预览文本（模拟注入Agent） preview_lines = [] preview_lines.append("# 审核配置预览") preview_lines.append("") preview_lines.append("## 基础配置") preset_text = config.preset if config.preset else "custom" preset_map = { "draft": "草稿模式", "standard": "标准模式", "strict": "严格模式" } preview_lines.append(f"- 预设模式: {preset_map.get(preset_text, preset_text)}") if config.overall_strictness is not None: preview_lines.append(f"- 整体严格度: {int(config.overall_strictness * 100)}%") if config.pass_threshold is not None: preview_lines.append(f"- 通过阈值: {config.pass_threshold}分") if config.auto_fix_enabled is not None: preview_lines.append(f"- 自动修复: {'启用' if config.auto_fix_enabled else '禁用'}") # 维度配置 if config.dimension_settings: preview_lines.append("") preview_lines.append("## 维度配置") for dim_type, dim_config in config.dimension_settings.items(): enabled_text = "启用" if dim_config.enabled else "禁用" preview_lines.append(f"") preview_lines.append(f"### {dim_type.value if hasattr(dim_type, 'value') else dim_type}") preview_lines.append(f"- 状态: {enabled_text}") preview_lines.append(f"- 严格度: {int(dim_config.strictness * 100)}%") preview_lines.append(f"- 权重: {dim_config.weight}") # Skills配置 if skills_info: preview_lines.append("") preview_lines.append("## 启用的Skills") for skill in skills_info: preview_lines.append(f"") preview_lines.append(f"### {skill['name']}") preview_lines.append(f"- ID: `{skill['id']}`") if skill.get('description'): preview_lines.append(f"- 描述: {skill['description']}") preview_lines.append("") preview_lines.append("**行为指南:**") preview_lines.append("```") preview_lines.append(skill['behavior_guide']) preview_lines.append("```") # 自定义规则 custom_rules = config.custom_rules if config.custom_rules else [] if custom_rules: preview_lines.append("") preview_lines.append("## 自定义规则") for rule in custom_rules: preview_lines.append(f"") preview_lines.append(f"### {rule.name}") preview_lines.append(f"- 描述: {rule.description}") preview_lines.append(f"- 维度: {rule.dimension.value if hasattr(rule.dimension, 'value') else rule.dimension}") preview_lines.append(f"- 严重度: {rule.severity.value if hasattr(rule.severity, 'value') else rule.severity}") preview_lines.append(f"- 触发条件: {rule.trigger_condition}") preview_text = "\n".join(preview_lines) return { "preview_text": preview_text, "skills_count": len(skills_info), "rules_count": len(custom_rules), "dimensions_count": len(config.dimension_settings) if config.dimension_settings else 0 } except Exception as e: logger.error(f"预览审核配置失败: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"预览失败: {str(e)}" )