xiaoxia-saas/apps/api/app/services/asset_segment_tracker.py

"""素材片段级使用记录追踪.

在素材 metadata（assets.classification_result JSON）中持久化已使用的片段时间区间，
供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重。

metadata 中新增字段 ``used_time_ranges``::

    "used_time_ranges": [
        {"start": 12.5, "end": 20.3, "plan_id": "plan-xxx", "created_at": "2026-08-29T12:00:00+00:00"},
        ...
    ]

注意：本模块所有函数都不自行 commit，由调用方控制事务边界
（from-assets 与 replace_all_clips_transactional 同事务；异步任务各自 commit）。
"""

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from typing import Callable

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import AssetModel

logger = logging.getLogger(__name__)

USED_RANGES_KEY = "used_time_ranges"


def _read_ranges(model: AssetModel) -> list[dict]:
    """从 AssetModel 读取 metadata dict（classification_result 列承载的 JSON）."""
    if not model.classification_result:
        return {}
    try:
        return json.loads(model.classification_result)
    except Exception:
        return {}


def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple[float, float]]]:
    """聚合多个素材的历史已用片段区间。

    Args:
        db: SQLAlchemy session
        asset_ids: 素材 ID 列表

    Returns:
        ``{asset_id: [(start, end), ...]}`` 格式，与 ``_calc_random_start_time`` 的
        ``used_segments`` 参数格式一致，可直接传入。
    """
    if not asset_ids:
        return {}
    result: dict[str, list[tuple[float, float]]] = {}
    models = db.query(AssetModel).filter(AssetModel.id.in_(list(set(asset_ids)))).all()
    for model in models:
        meta = _read_ranges(model)
        ranges = meta.get(USED_RANGES_KEY) or []
        segments: list[tuple[float, float]] = []
        for r in ranges:
            try:
                segments.append((float(r["start"]), float(r["end"])))
            except (KeyError, TypeError, ValueError):
                continue
        if segments:
            result[model.id] = segments
    return result


def record_used_segments(
    db: Session,
    asset_id: str,
    start: float,
    end: float,
    plan_id: str,
) -> None:
    """向素材 metadata 追加一条片段使用记录（不 commit）."""
    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
    if model is None:
        logger.warning("[片段追踪] 素材不存在，跳过记录: asset_id=%s", asset_id)
        return
    meta = _read_ranges(model)
    ranges = list(meta.get(USED_RANGES_KEY) or [])
    ranges.append(
        {
            "start": round(float(start), 3),
            "end": round(float(end), 3),
            "plan_id": plan_id,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    meta[USED_RANGES_KEY] = ranges
    model.classification_result = json.dumps(meta, ensure_ascii=False)
    model.updated_at = datetime.now(timezone.utc)


def remove_used_segment(
    db: Session,
    asset_id: str,
    start: float,
    end: float,
    plan_id: str | None = None,
    tolerance: float = 0.5,
) -> bool:
    """删除素材 metadata 中匹配的一条使用记录（不 commit）.

    匹配规则：start/end 与记录值相差不超过 tolerance 秒；plan_id 非空时，
    记录有 plan_id 则需相等，记录缺 plan_id（本功能上线前的旧数据）时按时间匹配。
    Returns:
        是否找到并删除了记录。
    """
    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
    if model is None:
        return False
    meta = _read_ranges(model)
    ranges = list(meta.get(USED_RANGES_KEY) or [])
    remaining: list[dict] = []
    removed = False
    for r in ranges:
        try:
            match = (
                abs(float(r["start"]) - float(start)) <= tolerance and abs(float(r["end"]) - float(end)) <= tolerance
            )
        except (KeyError, TypeError, ValueError):
            remaining.append(r)
            continue
        # plan_id 校验：传入 plan_id 时，记录有 plan_id 则必须相等；
        # 记录本身缺 plan_id（本功能上线前的旧数据）时退化为按时间匹配，
        # 避免旧区间永远删不掉导致素材容量泄漏
        if plan_id is not None and r.get("plan_id") is not None and r.get("plan_id") != plan_id:
            match = False
        if match and not removed:
            removed = True
            continue
        remaining.append(r)
    if removed:
        meta[USED_RANGES_KEY] = remaining
        model.classification_result = json.dumps(meta, ensure_ascii=False)
        model.updated_at = datetime.now(timezone.utc)
    return removed


def reset_used_segments(db: Session, asset_id: str) -> None:
    """清空单个素材的历史片段使用记录（不 commit）.

    单个素材的可用区间被全部占用（轮回一圈）后调用，使后续片段可重新使用整段素材。
    """
    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
    if model is None:
        return
    meta = _read_ranges(model)
    if meta.get(USED_RANGES_KEY):
        meta[USED_RANGES_KEY] = []
        model.classification_result = json.dumps(meta, ensure_ascii=False)
        model.updated_at = datetime.now(timezone.utc)
        logger.info("[片段追踪] 素材区间轮回重置: asset_id=%s", asset_id)


def make_reset_callback(db: Session, used_segments: dict) -> Callable[[str], None]:
    """构造给 _calc_random_start_time 用的 reset 回调.

    回调同时清空持久化 metadata 和内存中的 used_segments，使重试随机能覆盖全素材。
    """

    def _reset(asset_id: str) -> None:
        try:
            reset_used_segments(db, asset_id)
        except Exception:
            logger.warning("[片段追踪] reset 持久化记录失败: asset_id=%s", asset_id, exc_info=True)
        used_segments.pop(asset_id, None)

    return _reset