SupplyAgent/supply_infra/video_points.py

"""视频点位解析与序列化 — decode_result / JSON 字段 ↔ 行记录。"""
from __future__ import annotations

import json
from typing import Any

from supply_infra.db.models.multi_demand_video_point import (
    POINT_TYPE_INSPIRATION,
    POINT_TYPE_KEY,
    POINT_TYPE_PURPOSE,
)

_POINT_KEY = "点"
_POINT_DESC_KEY = "点描述"

# decode_result 中的 key → point_type
DECODE_RESULT_KEY_TO_POINT_TYPE = {
    "灵感点": POINT_TYPE_INSPIRATION,
    "目的点": POINT_TYPE_PURPOSE,
    "关键点": POINT_TYPE_KEY,
}

# 原 JSON 列名 → point_type
JSON_FIELD_TO_POINT_TYPE = {
    "inspiration_points_json": POINT_TYPE_INSPIRATION,
    "purpose_points_json": POINT_TYPE_PURPOSE,
    "key_points_json": POINT_TYPE_KEY,
}

# point_type → 原 JSON 列名（API 兼容）
POINT_TYPE_TO_JSON_FIELD = {v: k for k, v in JSON_FIELD_TO_POINT_TYPE.items()}


def _item_to_row(
    video_id: str, point_type: str, item: dict[str, Any]
) -> dict[str, str | None] | None:
    point = item.get(_POINT_KEY)
    point_desc = item.get(_POINT_DESC_KEY)
    if point is None and point_desc is None:
        return None
    return {
        "video_id": video_id,
        "point_type": point_type,
        "point_data": str(point) if point is not None else None,
        "point_desc": str(point_desc) if point_desc is not None else None,
    }


def points_from_decode_payload(
    video_id: str, payload: dict[str, Any]
) -> list[dict[str, str | None]]:
    """从 decode_result 提取全部点位行。"""
    rows: list[dict[str, str | None]] = []
    for decode_key, point_type in DECODE_RESULT_KEY_TO_POINT_TYPE.items():
        items = payload.get(decode_key)
        if not isinstance(items, list):
            continue
        for item in items:
            if not isinstance(item, dict):
                continue
            row = _item_to_row(video_id, point_type, item)
            if row:
                rows.append(row)
    return rows


def points_from_json_fields(
    video_id: str,
    *,
    inspiration_points_json: str | None = None,
    purpose_points_json: str | None = None,
    key_points_json: str | None = None,
) -> list[dict[str, str | None]]:
    """从 multi_demand_video_detail 的三个 JSON 列提取点位行。"""
    field_values = {
        "inspiration_points_json": inspiration_points_json,
        "purpose_points_json": purpose_points_json,
        "key_points_json": key_points_json,
    }
    rows: list[dict[str, str | None]] = []
    for field, point_type in JSON_FIELD_TO_POINT_TYPE.items():
        text = field_values.get(field)
        if not text:
            continue
        try:
            items = json.loads(text)
        except json.JSONDecodeError:
            continue
        if not isinstance(items, list):
            continue
        for item in items:
            if not isinstance(item, dict):
                continue
            row = _item_to_row(video_id, point_type, item)
            if row:
                rows.append(row)
    return rows


def json_fields_from_point_rows(
    rows: list[dict[str, Any]],
) -> dict[str, str | None]:
    """将点位行重组为三个 JSON 列（保持 API 兼容）。"""
    grouped: dict[str, list[dict[str, str | None]]] = {
        POINT_TYPE_INSPIRATION: [],
        POINT_TYPE_PURPOSE: [],
        POINT_TYPE_KEY: [],
    }
    for row in rows:
        point_type = row.get("point_type")
        if point_type not in grouped:
            continue
        grouped[point_type].append(
            {
                _POINT_KEY: row.get("point_data"),
                _POINT_DESC_KEY: row.get("point_desc"),
            }
        )

    result: dict[str, str | None] = {}
    for point_type, field in POINT_TYPE_TO_JSON_FIELD.items():
        items = grouped[point_type]
        result[field] = (
            json.dumps(items, ensure_ascii=False) if items else None
        )
    return result


def extract_points_json_from_decode(payload: dict[str, Any]) -> dict[str, str | None]:
    """从 decode_result 提取灵感点/目的点/关键点 JSON 列（原逻辑）。"""
    result: dict[str, str | None] = {}
    for decode_key, field in {
        k: POINT_TYPE_TO_JSON_FIELD[v]
        for k, v in DECODE_RESULT_KEY_TO_POINT_TYPE.items()
    }.items():
        items = payload.get(decode_key)
        if not isinstance(items, list):
            result[field] = None
            continue
        points: list[dict[str, Any]] = []
        for item in items:
            if not isinstance(item, dict):
                continue
            point = item.get(_POINT_KEY)
            point_desc = item.get(_POINT_DESC_KEY)
            if point is None and point_desc is None:
                continue
            points.append({_POINT_KEY: point, _POINT_DESC_KEY: point_desc})
        result[field] = json.dumps(points, ensure_ascii=False) if points else None
    return result