SupplyAgent/api/services/video_discovery.py

"""Demand-first video discovery data for the web workspace."""
from __future__ import annotations

import json
from typing import Any

from supply_infra.db.repositories.demand_feedback_repo import DemandFeedbackRepository
from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
from supply_infra.db.repositories.demand_video_expansion_repo import (
    DemandVideoExpansionRepository,
)
from supply_infra.db.repositories.global_tree_category_repo import (
    GlobalTreeCategoryRepository,
)
from supply_infra.db.repositories.multi_demand_video_detail_repo import (
    MultiDemandVideoDetailRepository,
)
from supply_infra.db.repositories.pipeline_step_run_repo import PipelineStepRunRepository
from supply_infra.db.session import get_session

_POINT_TYPES = {"inspiration", "purpose", "key"}
_SOURCE_VIDEO_GRADES = frozenset({"B", "C", "D"})
_VIDEO_DETAIL_URL_TEMPLATE = "https://admin.piaoquantv.com/cms/post-detail/{vid}/detail"


def _parse_video_ids(raw: Any) -> list[str]:
    """Parse JSON array or comma-separated text into ordered unique vid list."""
    if raw is None:
        return []

    items: list[Any]
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return []
        try:
            parsed = json.loads(text)
        except (TypeError, ValueError):
            parsed = None
        if isinstance(parsed, list):
            items = parsed
        elif parsed is not None:
            items = [parsed]
        else:
            items = [part.strip() for part in text.split(",")]
    elif isinstance(raw, (list, tuple)):
        items = list(raw)
    else:
        return []

    result: list[str] = []
    seen: set[str] = set()
    for item in items:
        vid = str(item).strip() if item is not None else ""
        if vid and vid not in seen:
            seen.add(vid)
            result.append(vid)
    return result


def _normalize_url(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _serialize_video(
    video_id: str,
    detail: Any | None,
    points: list[dict[str, Any]],
    feedback_count: int = 0,
) -> dict[str, Any]:
    url1 = _normalize_url(detail.url1) if detail else None
    url2 = _normalize_url(detail.url2) if detail else None
    return {
        "vid": video_id,
        "title": detail.title if detail else None,
        "url1": url1,
        "url2": url2,
        "video_detail_url": _VIDEO_DETAIL_URL_TEMPLATE.format(vid=video_id),
        "point_count": len(points),
        "points": points,
        "feedback_summary": {"count": int(feedback_count)},
    }


def _source_video_count(row: Any, expansion_counts: dict[int, int]) -> int:
    grade = str(row.grade or "").upper()
    if grade in _SOURCE_VIDEO_GRADES:
        return len(_parse_video_ids(row.video_list))
    return expansion_counts.get(int(row.id), 0)


def _parse_string_list(raw: Any) -> list[str]:
    """Parse JSON arrays and legacy comma-separated text into a stable string list."""
    if raw is None:
        return []

    values: list[Any]
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return []
        try:
            parsed = json.loads(text)
        except (TypeError, ValueError):
            parsed = None
        if isinstance(parsed, list):
            values = parsed
        elif parsed is not None:
            values = [parsed]
        else:
            values = [part.strip() for part in text.split(",")]
    elif isinstance(raw, (list, tuple)):
        values = list(raw)
    else:
        values = [raw]

    result: list[str] = []
    seen: set[str] = set()
    for value in values:
        text = str(value).strip() if value is not None else ""
        if text and text not in seen:
            seen.add(text)
            result.append(text)
    return result


def _parse_int_list(raw: Any) -> list[int]:
    result: list[int] = []
    for value in _parse_string_list(raw):
        try:
            result.append(int(value))
        except ValueError:
            continue
    return result


def _number(value: Any) -> float | None:
    return float(value) if value is not None else None


def _serialize_demand(
    row: Any,
    category_names: dict[int, str],
    video_count: int,
    feedback_count: int = 0,
) -> dict[str, Any]:
    category_ids = _parse_int_list(row.category_ids)
    return {
        "id": int(row.id),
        "demand_name": row.demand_name,
        "biz_dt": str(row.biz_dt),
        "grade": row.grade,
        "score": _number(row.score),
        "prior_total_score": _number(row.prior_total_score),
        "posterior_rov_avg": _number(row.posterior_rov_avg),
        "posterior_rov_count": int(row.posterior_rov_count or 0),
        "has_posterior": bool(row.has_posterior),
        "category_ids": category_ids,
        "category_names": [
            category_names[category_id]
            for category_id in category_ids
            if category_id in category_names
        ],
        "strategies": _parse_string_list(row.strategies),
        "reason": row.reason,
        "video_count": video_count,
        "feedback_summary": {"count": int(feedback_count)},
    }


def _category_name_map(session: Any, rows: list[Any]) -> dict[int, str]:
    category_ids = sorted(
        {
            category_id
            for row in rows
            for category_id in _parse_int_list(row.category_ids)
        }
    )
    categories = GlobalTreeCategoryRepository(session).list_active_by_ids(category_ids)
    return {
        int(category.id): str(category.name)
        for category in categories
        if category.name
    }


_DEMAND_EXPAND_STEP_KEY = "demand_expand"


def _resolve_video_discovery_biz_dt(session: Any, biz_dt: str | None) -> str | None:
    """默认取最近一次成功完成的 demand_expand 步骤所属业务日。"""
    if biz_dt:
        return str(biz_dt)
    latest_expand_dt = PipelineStepRunRepository(session).get_latest_succeeded_biz_dt(
        _DEMAND_EXPAND_STEP_KEY
    )
    if latest_expand_dt:
        return latest_expand_dt
    return DemandGradeRepository(session).get_latest_biz_dt()


def list_video_discovery_demands(biz_dt: str | None = None) -> dict[str, Any]:
    """Return one demand card per demand_grade row for the selected/latest day."""
    with get_session() as session:
        grade_repo = DemandGradeRepository(session)
        resolved_biz_dt = _resolve_video_discovery_biz_dt(session, biz_dt)
        if not resolved_biz_dt:
            return {"biz_dt": None, "items": []}

        rows = grade_repo.list_by_biz_dt(str(resolved_biz_dt))
        category_names = _category_name_map(session, rows)
        video_counts = DemandVideoExpansionRepository(
            session
        ).count_distinct_videos_by_demand_grade(str(resolved_biz_dt))
        feedback_counts = DemandFeedbackRepository(session).count_demands(
            [int(row.id) for row in rows]
        )
        return {
            "biz_dt": str(resolved_biz_dt),
            "items": [
                _serialize_demand(
                    row,
                    category_names,
                    _source_video_count(row, video_counts),
                    feedback_counts.get(int(row.id), 0),
                )
                for row in rows
            ],
        }


def get_video_discovery_demand(demand_grade_id: int) -> dict[str, Any] | None:
    """
    Resolve the judged video list and hit evidence from demand_video_expansion.

    S/A 使用拓展命中视频；B/C/D 使用 demand_grade.video_list 中的源视频。
    multi_demand_video_detail 补充 title 与解构 url1/url2。
    """
    with get_session() as session:
        grade = DemandGradeRepository(session).get_by_id(demand_grade_id)
        if grade is None:
            return None

        category_names = _category_name_map(session, [grade])
        grade_code = str(grade.grade or "").upper()
        demand_feedback_count, video_feedback_counts, point_feedback_counts = (
            DemandFeedbackRepository(session).count_for_demand(demand_grade_id)
        )

        if grade_code in _SOURCE_VIDEO_GRADES:
            video_ids = _parse_video_ids(grade.video_list)
            details = MultiDemandVideoDetailRepository(session).list_by_vids(video_ids)
            videos = [
                _serialize_video(
                    video_id,
                    details.get(video_id),
                    [],
                    video_feedback_counts.get(video_id, 0),
                )
                for video_id in video_ids
            ]
        else:
            expansions = DemandVideoExpansionRepository(session).list_by_demand_grade(
                str(grade.biz_dt),
                demand_grade_id,
            )

            video_ids: list[str] = []
            points_by_video: dict[str, list[dict[str, Any]]] = {}
            for row in expansions:
                video_id = str(row.video_id).strip()
                if not video_id or row.point_type not in _POINT_TYPES:
                    continue
                if video_id not in points_by_video:
                    video_ids.append(video_id)
                    points_by_video[video_id] = []
                points_by_video[video_id].append(
                    {
                        "id": int(row.id),
                        "point_type": row.point_type,
                        "expanded_text": row.expanded_text,
                        "point_desc": row.point_desc,
                        "reason": row.reason,
                        "feedback_summary": {
                            "count": point_feedback_counts.get(int(row.id), 0)
                        },
                    }
                )

            details = MultiDemandVideoDetailRepository(session).list_by_vids(video_ids)
            videos = [
                _serialize_video(
                    video_id,
                    details.get(video_id),
                    points_by_video[video_id],
                    video_feedback_counts.get(video_id, 0),
                )
                for video_id in video_ids
            ]

        demand = _serialize_demand(
            grade,
            category_names,
            len(videos),
            demand_feedback_count,
        )
        demand["point_count"] = sum(video["point_count"] for video in videos)
        demand["videos"] = videos
        demand["video_source"] = (
            "pool" if grade_code in _SOURCE_VIDEO_GRADES else "expansion"
        )
        return demand