SupplyAgent/agents/find_agent/support/video_discovery.py

"""持久化 find_agent 的搜索轨迹、候选证据和分池结果。"""
from __future__ import annotations

import json
import logging
import uuid
from decimal import Decimal
from typing import Any

from supply_infra.video_discovery_gates import (
    build_rule_snapshot,
    normalize_duration_seconds,
    parse_datetime_value,
)
from supply_infra.services.video_discovery_service import (
    RunNotFoundError,
    format_db_error,
    get_video_discovery_service,
)

logger = logging.getLogger(__name__)

_RUN_STATUSES = {"running", "finished", "failed"}
_FINAL_DECISION_BUCKETS = {"primary", "rejected"}


def _json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, default=str)


def _input_error(message: str) -> str:
    return _json({"error": message, "input_error": True})


def _clean_text(value: Any, *, max_length: int | None = None) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None
    return text[:max_length] if max_length else text


def _nonnegative_int(value: Any) -> int | None:
    if value is None or value == "":
        return None
    try:
        return max(0, int(value))
    except (TypeError, ValueError):
        return None


def _optional_decimal(value: Any, places: int) -> Decimal | None:
    if value is None or value == "":
        return None
    try:
        number = Decimal(str(value))
    except (ArithmeticError, TypeError, ValueError):
        return None
    quantum = Decimal(1).scaleb(-places)
    return number.quantize(quantum)


def _optional_ratio_decimal(value: Any, places: int = 6) -> Decimal | None:
    number = _optional_decimal(value, places)
    if number is not None and not Decimal("0") <= number <= Decimal("1"):
        raise ValueError("比例字段必须在 0~1 范围内")
    return number


def _age_side(
    age_normalization: Any,
    side: str,
) -> dict[str, Any]:
    if not isinstance(age_normalization, dict):
        return {}
    value = age_normalization.get(side)
    return value if isinstance(value, dict) else {}


def create_video_discovery_run(
    demand_word: str,
    relevant_points: list[dict[str, Any]],
    seed_video_id: str | None = None,
    seed_video_title: str | None = None,
    demand_grade_id: int | None = None,
    intent_summary: str | None = None,
    run_id: str | None = None,
) -> str:
    """
    创建一次可追踪的视频发现运行。

    若用户消息已提供预创建 run_id，必须原样传入 run_id；工具会复用已有记录，
    不会重复创建。

    Args:
        demand_word: 用户给定需求词；它是输入语义，不强制作为实际搜索词。
        relevant_points: 参考视频中与需求相关的点位对象列表。
        seed_video_id: 兼容旧调用方的可选参考视频 id；调度调用不传。
        seed_video_title: 兼容旧调用方的可选参考视频标题；调度调用不传。
        demand_grade_id: 可选 demand_grade.id。
        intent_summary: Agent 对真正受欢迎内容的初步解释，可稍后更新。
        run_id: 系统预创建的运行 id；传入已存在记录时直接复用。

    Returns:
        JSON，包含后续存储工具必须使用的 run_id。
    """
    service = get_video_discovery_service()
    cleaned_run_id = _clean_text(run_id, max_length=64)
    if cleaned_run_id:
        try:
            existing = service.lookup_run(cleaned_run_id)
            if existing is not None:
                return _json(
                    {
                        "title": "视频发现运行已存在",
                        "run_id": cleaned_run_id,
                        "status": existing["status"],
                        "pre_created": True,
                        "output": f"run_id={cleaned_run_id}",
                    }
                )
        except Exception as exc:
            logger.error("create_video_discovery_run lookup failed: %s", exc, exc_info=True)
            return _json({"error": format_db_error(exc), "title": "查询视频发现运行失败"})

    demand = _clean_text(demand_word, max_length=256)
    if not demand:
        return _input_error("demand_word 不能为空")

    new_run_id = cleaned_run_id or uuid.uuid4().hex
    rule_snapshot = build_rule_snapshot()
    values = {
        "run_id": new_run_id,
        "demand_grade_id": demand_grade_id,
        "demand_word": demand,
        "seed_video_id": _clean_text(seed_video_id, max_length=64),
        "seed_video_title": _clean_text(seed_video_title, max_length=512),
        "relevant_points_json": _json(relevant_points or []),
        "intent_summary": _clean_text(intent_summary),
        "status": "running",
        "rule_version": rule_snapshot["rule_version"],
        "rule_config_json": _json(rule_snapshot),
    }
    try:
        created = service.create_run(values)
        return _json(
            {
                "title": "视频发现运行已创建",
                "run_id": created["run_id"],
                "status": created["status"],
                "output": f"run_id={created['run_id']}",
            }
        )
    except Exception as exc:
        logger.error("create_video_discovery_run failed: %s", exc, exc_info=True)
        return _json({"error": format_db_error(exc), "title": "创建视频发现运行失败"})


def _normalize_candidate_update(item: dict[str, Any]) -> dict[str, Any]:
    try:
        candidate_id = int(item.get("candidate_id"))
    except (TypeError, ValueError):
        raise ValueError("candidate_id 必须是整数") from None
    if candidate_id <= 0:
        raise ValueError("candidate_id 必须大于 0")

    content_age = item.get("content_age_evidence")
    account_age = item.get("account_age_evidence")
    age_normalization = item.get("age_normalization")
    content_normalization = _age_side(age_normalization, "content")
    account_normalization = _age_side(age_normalization, "account")
    decision_bucket = (
        _clean_text(item.get("decision_bucket"), max_length=24)
        or ""
    )
    if decision_bucket not in _FINAL_DECISION_BUCKETS:
        raise ValueError(
            "decision_bucket 必须是 primary 或 rejected"
        )

    publish_at = parse_datetime_value(item.get("publish_at"))
    duration = normalize_duration_seconds(
        item.get("duration_seconds")
        if item.get("duration_seconds") not in (None, "")
        else item.get("video_duration")
    )
    content_ratio = item.get(
        "content_50_plus_ratio",
        content_normalization.get("older_ratio"),
    )
    content_tgi = item.get(
        "content_50_plus_tgi",
        content_normalization.get("older_tgi"),
    )
    account_ratio = item.get(
        "account_50_plus_ratio",
        account_normalization.get("older_ratio"),
    )
    account_tgi = item.get(
        "account_50_plus_tgi",
        account_normalization.get("older_tgi"),
    )
    temporal_evidence = item.get("temporal_evidence")
    mapping = {
        "candidate_id": candidate_id,
        "title": _clean_text(item.get("title"), max_length=512),
        "content_link": _clean_text(item.get("content_link"), max_length=1024),
        "author_name": _clean_text(item.get("author_name"), max_length=256),
        "author_sec_uid": _clean_text(item.get("author_sec_uid"), max_length=256),
        "tags_json": item.get("tags") if "tags" in item else None,
        "publish_at": (
            publish_at.replace(tzinfo=None) if publish_at is not None else None
        ),
        "duration_seconds": duration,
        "play_count": _nonnegative_int(item.get("play_count")),
        "like_count": _nonnegative_int(item.get("like_count")),
        "comment_count": _nonnegative_int(item.get("comment_count")),
        "collect_count": _nonnegative_int(item.get("collect_count")),
        "share_count": _nonnegative_int(item.get("share_count")),
        "content_age_evidence_json": (
            _json(content_age) if content_age is not None else None
        ),
        "account_age_evidence_json": (
            _json(account_age) if account_age is not None else None
        ),
        "age_normalization_json": (
            _json(age_normalization) if age_normalization is not None else None
        ),
        "content_50_plus_ratio": _optional_ratio_decimal(content_ratio),
        "content_50_plus_tgi": _optional_decimal(content_tgi, 4),
        "account_50_plus_ratio": _optional_ratio_decimal(account_ratio),
        "account_50_plus_tgi": _optional_decimal(account_tgi, 4),
        "temporal_type": _clean_text(item.get("temporal_type"), max_length=24),
        "temporal_status": _clean_text(item.get("temporal_status"), max_length=16),
        "temporal_evidence_json": (
            _json(temporal_evidence) if temporal_evidence is not None else None
        ),
        "reject_reason_code": _clean_text(
            item.get("reject_reason_code"),
            max_length=64,
        ),
        "relevance_score": _optional_decimal(item.get("relevance_score"), 6),
        "elder_score": _optional_decimal(item.get("elder_score"), 6),
        "share_score": _optional_decimal(item.get("share_score"), 6),
        "value_score": _optional_decimal(item.get("value_score"), 2),
        "decision_reason": _clean_text(item.get("decision_reason")),
        "decision_bucket": decision_bucket,
    }
    return mapping


def batch_update_video_discovery_candidates(
    run_id: str,
    items: list[dict[str, Any]],
) -> str:
    """
    严格按 candidate_id 批量更新候选详情、证据、评分和最终分池。

    只更新 video_discovery_candidate；不会新增候选、修改搜索记录或修改运行状态。

    Args:
        run_id: 发现运行 id。
        items: 候选数组。每项必须包含搜索工具返回的 candidate_id，并直接提供
            decision_bucket；可更新标题、链接、作者、发布时间、时长、互动量、标签、
            双侧年龄证据、画像标准化结果、时间判断、R/E/S/V 和最终分池理由。
            primary 会在数据库事务内强制校验本次运行的 P0 规则快照。

    Returns:
        JSON，包含 updated_count 和按 candidate_id 更新后的候选记录。
    """
    run_text = _clean_text(run_id, max_length=64)
    if not run_text:
        return _input_error("run_id 不能为空")

    rows: list[dict[str, Any]] = []
    errors: list[str] = []
    for index, item in enumerate(items or []):
        if not isinstance(item, dict):
            errors.append(f"[{index}] 不是对象")
            continue
        try:
            rows.append(_normalize_candidate_update(dict(item)))
        except ValueError as exc:
            errors.append(f"[{index}] {exc}")
    if errors:
        return _json(
            {
                "error": "候选更新参数不合法",
                "input_error": True,
                "errors": errors,
            }
        )
    if not rows:
        return _input_error("items 不能为空")

    try:
        updated = get_video_discovery_service().update_candidates(
            run_text,
            rows,
        )
        payload = {
            "title": "候选已更新",
            "run_id": run_text,
            "updated_count": updated["updated_count"],
            "candidates": updated["candidates"],
            "output": f"更新 {updated['updated_count']} 条候选",
        }
        return _json(payload)
    except RunNotFoundError as exc:
        return _input_error(str(exc))
    except ValueError as exc:
        return _input_error(str(exc))
    except Exception as exc:
        logger.error(
            "batch_update_video_discovery_candidates failed: %s",
            exc,
            exc_info=True,
        )
        return _json({"error": format_db_error(exc), "title": "更新候选失败"})


def update_video_discovery_run_status(
    run_id: str,
    status: str,
    intent_summary: str | None = None,
    stop_reason: str | None = None,
) -> str:
    """单独更新 video_discovery_run 的状态、意图摘要和停止原因。"""
    run_text = _clean_text(run_id, max_length=64)
    if not run_text:
        return _input_error("run_id 不能为空")
    if status not in _RUN_STATUSES:
        return _input_error(f"status 必须是: {sorted(_RUN_STATUSES)}")
    try:
        run = get_video_discovery_service().update_run_status(
            run_text,
            status=status,
            intent_summary=_clean_text(intent_summary),
            stop_reason=_clean_text(stop_reason),
        )
        return _json(
            {
                "title": "视频发现运行状态已更新",
                "run": run,
                "output": (
                    f"run_id={run_text}，status={run['status']}，"
                    f"primary_count={run['primary_count']}"
                ),
            }
        )
    except RunNotFoundError as exc:
        return _input_error(str(exc))
    except Exception as exc:
        logger.error(
            "update_video_discovery_run_status failed: %s",
            exc,
            exc_info=True,
        )
        return _json({"error": format_db_error(exc), "title": "更新运行状态失败"})


def query_video_discovery_state(
    run_id: str,
    include_rejected: bool = True,
    limit: int = 100,
) -> str:
    """
    查询一次运行已经保存的搜索轨迹、主推荐与淘汰候选。

    用于长搜索过程恢复状态、检查是否真的翻页和扩词，也用于最终自动保留判断。
    """
    run_text = _clean_text(run_id, max_length=64)
    if not run_text:
        return _json({"error": "run_id 不能为空"})
    try:
        state = get_video_discovery_service().get_full_state(
            run_text,
            include_rejected=include_rejected,
            limit=limit,
        )
        run = state["run"]
        payload = {
            "title": f"视频发现状态: {run_text}",
            "run": run,
            "searches": state["searches"],
            "candidates": state["candidates"],
            "output": (
                f"搜索页 {run['search_count']}；主推荐 {run['primary_count']}"
            ),
        }
        return _json(payload)
    except RunNotFoundError:
        return _json({"error": f"run_id 不存在: {run_text}"})
    except Exception as exc:
        logger.error("query_video_discovery_state failed: %s", exc, exc_info=True)
        return _json({"error": format_db_error(exc), "title": "查询视频发现状态失败"})