long_articles_agentic_supply/src/domains/content_quality/_utils.py

"""内容质量评估工具函数 —— Prompt 构建 + LLM 响应解析"""

import hashlib
import re
import logging
from typing import Dict, List

from src.infra.shared.tools import safe_json_parse

from ._const import ContentQualityConst

logger = logging.getLogger(__name__)


def md5(text: str) -> str:
    return hashlib.md5(text.encode("utf-8")).hexdigest()


# ═══════════════════════════════════════════════════════════════
# Prompt 构建
# ═══════════════════════════════════════════════════════════════


def build_title_quality_prompt(titles: List[Dict]) -> str:
    """构造标题质量评分 prompt，每行一个标题"""
    lines = [t["title"] for t in titles]
    return ContentQualityConst.PROMPT_TITLE_QUALITY.format(
        title_list="\n".join(lines),
    )


def build_category_prompt(titles: List[Dict]) -> str:
    """构造品类识别 prompt，格式: [(id, title), ...] 元组列表"""
    tuples = [(t["id"], t["title"]) for t in titles]
    return ContentQualityConst.PROMPT_TITLE_CATEGORY.format(
        title_list=str(tuples),
    )


# ═══════════════════════════════════════════════════════════════
# LLM 响应解析
# ═══════════════════════════════════════════════════════════════


def parse_title_quality_response(
    raw: str | None,
    batch_titles: List[Dict],
) -> List[Dict]:
    """解析 prompt1 返回值：LLM 按行输出纯数字分数，按顺序对应输入标题

    返回: [{"id": 1, "score": 85}, ...]
    """
    result: List[Dict] = []
    if not raw:
        logger.warning("LLM 标题质量返回为空")
        return result

    # 从文本中提取所有数字
    scores = re.findall(r"\d+", raw)
    if not scores:
        logger.warning("LLM 标题质量未提取到分数: %s", raw[:200])
        return result

    for i, s in enumerate(scores):
        if i >= len(batch_titles):
            break
        score = int(s)
        if 0 <= score <= 100:
            result.append({"id": batch_titles[i]["id"], "score": score})

    return result


def parse_category_response(
    raw: str | None,
    batch_titles: List[Dict],
) -> List[Dict]:
    """解析 prompt2 返回值：{id: category} JSON 对象

    返回: [{"id": 1, "category": "知识科普"}, ...]
    """
    result: List[Dict] = []
    if not raw:
        logger.warning("LLM 品类识别返回为空")
        return result

    parsed = safe_json_parse(raw)
    if not isinstance(parsed, dict):
        logger.warning("LLM 品类识别返回非对象: %s", raw[:200])
        return result

    for t in batch_titles:
        category = parsed.get(str(t["id"]))
        if category:
            result.append({"id": t["id"], "category": str(category)[:32]})

    return result


__all__ = [
    "md5",
    "build_title_quality_prompt",
    "build_category_prompt",
    "parse_title_quality_response",
    "parse_category_response",
]