"""AI 响应解析纯逻辑模块. 抽离自 ai_service.py 的解析函数，方便单测覆盖，同时保持向后兼容。 包括： - 标题列表解析（JSON/编号/换行/破折号格式） - 语义匹配结果解析（多种JSON格式） - 标题降级生成 - 关键词匹配降级 """ from __future__ import annotations import json import math import random import re from typing import Any # ── 标题解析 ────────────────────────────────────────────────────────────────── def parse_titles_from_response(content: str) -> list[str]: """从模型返回中解析标题列表. 支持多种返回格式： - JSON 数组: ["标题1", "标题2"] - 编号列表: 1. 标题1 / 2. 标题2 - 换行分隔: 标题1\n标题2 - 带破折号: - 标题1 """ if not content: return [] # 尝试解析 JSON try: cleaned = content.strip() if cleaned.startswith("```"): cleaned = cleaned.strip("`") if cleaned.lower().startswith("json"): cleaned = cleaned[4:] cleaned = cleaned.strip() data = json.loads(cleaned) if isinstance(data, list): return [str(item).strip() for item in data if str(item).strip()] if isinstance(data, dict) and "titles" in data: titles = data["titles"] if isinstance(titles, list): return [str(t).strip() for t in titles if str(t).strip()] except (json.JSONDecodeError, ValueError): pass # 尝试按行解析 titles: list[str] = [] for line in content.strip().split("\n"): line = line.strip() if not line: continue # 去掉编号前缀 "1. " "1、" "（1）" line = re.sub(r"^[\d]+[\.、\)）]\s*", "", line) # 去掉破折号前缀 "- " "• " line = re.sub(r"^[-•·]\s*", "", line) # 去掉引号 line = line.strip('"').strip("'").strip("「」") if line and len(line) < 100: # 过滤过长的行 titles.append(line) return titles # ── 语义匹配解析 ──────────────────────────────────────────────────────────── def parse_semantic_match_response( content: str, asset_ids: list[str], ) -> dict[str, float] | None: """从模型返回中解析素材匹配度. 期望格式：JSON 对象 {asset_id: score} 或 {"matches": [{asset_id, score}]} score 范围 0-1，自动截断到 [0, 1]。 """ if not content: return None try: cleaned = content.strip() if cleaned.startswith("```"): cleaned = cleaned.strip("`") if cleaned.lower().startswith("json"): cleaned = cleaned[4:] cleaned = cleaned.strip() data = json.loads(cleaned) result: dict[str, float] = {} # 格式1: {"asset_id1": 0.8, "asset_id2": 0.6} if isinstance(data, dict): if "matches" in data and isinstance(data["matches"], list): # 格式2: {"matches": [{"asset_id": "...", "score": 0.8}]} for item in data["matches"]: if isinstance(item, dict): aid = item.get("asset_id") or item.get("id") score = item.get("score", 0) if aid and isinstance(score, (int, float)): result[str(aid)] = max(0.0, min(1.0, float(score))) else: for key, value in data.items(): if isinstance(value, (int, float)): result[str(key)] = max(0.0, min(1.0, float(value))) # 格式3: [{"asset_id": "...", "score": 0.8}] elif isinstance(data, list): for item in data: if isinstance(item, dict): aid = item.get("asset_id") or item.get("id") score = item.get("score", 0) if aid and isinstance(score, (int, float)): result[str(aid)] = max(0.0, min(1.0, float(score))) # 至少一半素材有评分才算成功 if asset_ids and len(result) >= max(1, len(asset_ids) // 2): return result # 没有 asset_ids 时，只要有结果就返回 if not asset_ids and result: return result except (json.JSONDecodeError, ValueError): pass return None # ── 标题降级生成 ───────────────────────────────────────────────────────────── def generate_titles_fallback( description: str, style_info: dict[str, Any], count: int = 5, ) -> list[str]: """本地降级：基于模板规则生成标题. Args: description: 视频内容描述 style_info: 标题风格配置 {"name": ..., "examples": [...]} count: 生成数量 """ examples = style_info.get("examples", []) # 从描述中提取关键词（取前几个词） keywords = [w for w in description.strip().split() if len(w) > 1][:3] keyword = keywords[0] if keywords else "精彩内容" # 基于模板生成 example_0 = examples[0][:10] + "..." if examples else "必看" example_1 = examples[1] if len(examples) > 1 else "你不知道的事" templates = [ f"「{keyword}」{example_0}", f"{keyword}：{example_1}", f"关于{keyword}，你不知道的3件事", f"{keyword}入门指南，新手必看", f"深度解析：{keyword}背后的秘密", f"{keyword}怎么做？手把手教你", f"干货分享 | {keyword}全攻略", f"建议收藏：{keyword}实用技巧", f"{keyword}避坑指南，别再踩雷了", f"一分钟搞懂{keyword}", ] random.shuffle(templates) return templates[: min(count, len(templates))] # ── 关键词匹配降级 ─────────────────────────────────────────────────────────── def keyword_match_fallback( description: str, assets: list[dict[str, Any]], ) -> list[dict[str, Any]]: """本地降级：基于关键词的简单匹配. 计算描述中的关键词与素材名称/标签/描述的重叠度， 作为匹配度评分。0-1分。 """ # 提取关键词（中文按2字以上片段，英文按单词） desc = description.lower() keywords: set[str] = set() # 英文单词 for word in re.findall(r"[a-zA-Z]{3,}", desc): keywords.add(word) # 中文2-4字片段 for i in range(len(desc)): for j in range(i + 2, min(i + 5, len(desc) + 1)): fragment = desc[i:j] if all("\u4e00" <= c <= "\u9fff" for c in fragment): keywords.add(fragment) if not keywords: # 没有关键词时给所有素材中等分数 results = [] for asset in assets: new_asset = dict(asset) new_asset["match_score"] = 0.5 new_asset["match_reason"] = "fallback_default" results.append(new_asset) return results results = [] for asset in assets: # 组合素材的文本信息：名称 + 标签 + 描述 asset_text_parts = [ str(asset.get("name", "")).lower(), " ".join(str(t) for t in asset.get("tags", [])).lower(), str(asset.get("description", "")).lower(), ] asset_text = " | ".join(asset_text_parts) # 计算匹配度：命中关键词占比 + 稀有关键词加权 hit_count = 0 hit_keywords: list[str] = [] for kw in keywords: if kw in asset_text: hit_count += 1 hit_keywords.append(kw) # 基础匹配度 = 命中关键词数 / 总关键词数（开根号平滑） base_score = math.sqrt(hit_count / len(keywords)) if keywords else 0.5 # 名称命中加分（名称匹配更重要） name = str(asset.get("name", "")).lower() name_hits = sum(1 for kw in hit_keywords if kw in name) name_bonus = min(0.2, name_hits * 0.05) score = min(1.0, base_score * 0.8 + name_bonus) score = round(score, 3) new_asset = dict(asset) new_asset["match_score"] = score new_asset["match_reason"] = "fallback_keyword" results.append(new_asset) # 按匹配度降序 results.sort(key=lambda x: x["match_score"], reverse=True) return results