xiaoxia-saas/packages/domain/ai_parsing.py

"""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