"""素材片段级使用记录追踪与受控复用. 在素材 metadata(assets.classification_result JSON)中持久化已使用的片段时间区间, 供 from-assets 创建片段时避开历史区间,实现跨任务/跨调用的片段去重; 素材可用区间耗尽后进入受控复用:允许有限次数(MAX_RANGE_USE_COUNT)复用最久未用 的历史区间,配合调用方的成片复用占比控制(MAX_REUSE_RATIO = 15%),把任意两条 成片的画面重复率控制在阈值内。 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", "use_count": 1, # 该区间累计被使用次数(复用一次 +1) "last_used_at": "2026-08-29T12:00:00+00:00" # 最近一次使用时间 }, ... ] 注意:本模块所有函数都不自行 commit,由调用方控制事务边界 (from-assets 与 replace_all_clips_transactional 同事务;异步任务各自 commit)。 历史记录永不自动清空(自动轮回重置已下线,reset_used_segments 仅保留给运维/测试)。 """ 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" # ── 受控复用配置常量 ───────────────────────────────────────────────────────── MAX_RANGE_USE_COUNT = 3 """单条历史区间最多被使用次数(含首次),达到后不再参与复用。""" REUSE_RATIO_LIMIT = 0.15 """单条成片中,单个素材的复用片段累计时长 / 该素材在成片中的总时长上限(15%)。 超过则该素材不再分配新片段(调用方在轮询分配时跳过)。""" SEGMENT_EDGE_GAP = 0.3 """冲突判定边缘间隙(秒):历史区间按 [start-gap, end+gap] 扩边后参与冲突检测, 避免两条片段首尾紧贴导致画面观感重复;记录仍存实际值。""" # 判定"新片段与历史区间为同一次使用(复用)"的重叠率阈值: # 重叠时长 / 新区间时长超过该比例视为复用该历史区间(累加 use_count)而非新增记录。 _REUSE_OVERLAP_RATIO = 0.6 def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _read_meta(model) -> dict: """读取素材 metadata dict。 兼容两种对象: - ORM ``AssetModel``:metadata 以 JSON 字符串存在 ``classification_result`` 列; - 领域实体 ``Asset``(路由层 repository 返回):metadata 直接是 dict 属性 (repository 与 classification_result 互转,见 asset_repository.py)。 """ # 领域实体:metadata 已是 dict meta = getattr(model, "metadata", None) if isinstance(meta, dict): return meta raw = getattr(model, "classification_result", None) if not raw: return {} try: data = json.loads(raw) if isinstance(raw, str) else raw return data if isinstance(data, dict) else {} except Exception: return {} def _get_model(db: Session, asset_id: str, for_update: bool = False) -> AssetModel | None: query = db.query(AssetModel).filter(AssetModel.id == asset_id) if for_update: # 行级锁(PostgreSQL SELECT ... FOR UPDATE):序列化同一素材的 # classification_result 读-改-写,避免并发事务丢失使用记录。 # SQLite 不支持时 SQLAlchemy 会忽略该子句(no-op)。 query = query.with_for_update() return query.first() def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple[float, float]]]: """聚合多个素材的历史已用片段区间。 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_meta(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: """记录一次片段使用(不 commit). 若新区间与某条历史区间高度重叠(复用场景,如受控复用回调返回的区间、 MediaKit 挪到历史区间),则累加该记录的 ``use_count`` 并刷新 ``last_used_at``, 不新增记录;否则追加一条新记录(use_count=1)。 """ # 行级锁读取:与并发生成任务互斥,保证区间记录读-改-写一致 model = _get_model(db, asset_id, for_update=True) if model is None: logger.warning("[片段追踪] 素材不存在,跳过记录: asset_id=%s", asset_id) return meta = _read_meta(model) ranges = list(meta.get(USED_RANGES_KEY) or []) new_start = round(float(start), 3) new_end = round(float(end), 3) new_dur = max(new_end - new_start, 1e-6) now = _now_iso() for r in ranges: try: rs, re_ = float(r["start"]), float(r["end"]) except (KeyError, TypeError, ValueError): continue overlap = max(0.0, min(new_end, re_) - max(new_start, rs)) if overlap / new_dur >= _REUSE_OVERLAP_RATIO: # 复用同一条历史区间:累加次数、刷新时间 r["use_count"] = int(r.get("use_count", 1)) + 1 r["last_used_at"] = now r["plan_id"] = plan_id meta[USED_RANGES_KEY] = ranges model.classification_result = json.dumps(meta, ensure_ascii=False) model.updated_at = datetime.now(timezone.utc) return ranges.append( { "start": new_start, "end": new_end, "plan_id": plan_id, "created_at": now, "use_count": 1, "last_used_at": now, } ) 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 = _get_model(db, asset_id) if model is None: return False meta = _read_meta(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 = _get_model(db, asset_id) if model is None: return meta = _read_meta(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) # ── 素材余量/可用性计算(Task H:素材库角标 + smart-match 过滤)────────────── # 判定「是否还有空闲可切区间」时使用的最小片段时长(秒):空闲段长于此值才视为可切 _MIN_FREE_CLIP_DURATION = 3.0 def _merge_intervals(intervals: list[tuple[float, float]]) -> list[tuple[float, float]]: """合并重叠/相接的时间区间,返回升序不重叠区间列表。""" if not intervals: return [] ordered = sorted((float(a), float(b)) for a, b in intervals if b > a) merged: list[tuple[float, float]] = [ordered[0]] for start, end in ordered[1:]: last_start, last_end = merged[-1] if start <= last_end: merged[-1] = (last_start, max(last_end, end)) else: merged.append((start, end)) return merged def _has_free_gap(used: list[tuple[float, float]], total: float, min_free: float = _MIN_FREE_CLIP_DURATION) -> bool: """素材 [0, total] 中是否存在长度 ≥ min_free 的空闲段(考虑边缘间隙)。""" if total <= 0: return False # 历史区间按边缘间隙扩边后判定空闲(与选片冲突检测同一口径) expanded = [(max(0.0, s - SEGMENT_EDGE_GAP), min(total, e + SEGMENT_EDGE_GAP)) for s, e in used] merged = _merge_intervals(expanded) cursor = 0.0 for start, end in merged: if start - cursor >= min_free: return True cursor = max(cursor, end) return total - cursor >= min_free def compute_asset_availability( model: "AssetModel | None", min_free_clip_duration: float = _MIN_FREE_CLIP_DURATION, ) -> dict | None: """计算单个素材的余量与可用性(纯函数,不读写 DB)。 Returns: 视频素材返回 ``{"used_duration", "available_duration", "used_ratio", "usable"}``; 非视频 / 无 model / 无时长信息返回 None(调用方按可用处理,零影响)。 usable=False 条件(与受控复用机制一致): 零重复可切区间已耗尽(不存在 ≥ min_free 的空闲段)且 所有历史区间 use_count 均达 MAX_RANGE_USE_COUNT 上限(无区间可复用)。 """ if model is None: return None file_type = getattr(model, "file_type", None) or getattr(model, "mime_type", "") or "" if file_type != "video" and not str(file_type).startswith("video/"): return None total = float(getattr(model, "duration", 0.0) or 0.0) if total <= 0: return None meta = _read_meta(model) raw_ranges = meta.get(USED_RANGES_KEY) or [] intervals: list[tuple[float, float]] = [] use_counts: list[int] = [] for r in raw_ranges: try: start = float(r["start"]) end = float(r["end"]) except (KeyError, TypeError, ValueError): continue if end <= start: continue intervals.append((start, end)) try: use_counts.append(int(r.get("use_count", 1))) except (TypeError, ValueError): use_counts.append(1) merged = _merge_intervals(intervals) used_duration = round(sum(e - s for s, e in merged), 3) used_duration = min(used_duration, total) available_duration = round(max(total - used_duration, 0.0), 3) used_ratio = round(min(used_duration / total, 1.0), 4) has_free = _has_free_gap(intervals, total, min_free_clip_duration) if has_free: usable = True else: # 空闲段耗尽:仅当存在历史区间且全部达复用上限时才判定不可用; # 无历史区间(理论上不会走到,因为 has_free=True)按可用处理 if not use_counts: usable = True else: usable = any(uc < MAX_RANGE_USE_COUNT for uc in use_counts) return { "used_duration": used_duration, "available_duration": available_duration, "used_ratio": used_ratio, "usable": usable, } def find_reusable_range( db: Session, asset_id: str, clip_duration: float, asset_total: float, *, max_use_count: int = MAX_RANGE_USE_COUNT, ) -> tuple[float, float] | None: """受控复用:在素材历史区间中选一条可复用区间返回 (start, end)。 选择规则: 1. 仅选 ``use_count < max_use_count`` 的历史区间; 2. 优先返回能完整容纳当前 clip_duration(起点后不越素材边界)的最久未用区间; 3. 没有能容纳的,则返回 last_used_at 最老(或缺失 last_used_at 的旧数据优先) 且 use_count 最低的区间起点(可能与其他历史区间重叠,属降级复用); 4. 无任何可复用区间(记录为空或全部达上限)返回 None。 本函数只读不写;复用次数的累加由后续 record_used_segments 完成。 """ model = _get_model(db, asset_id) if model is None: return None meta = _read_meta(model) ranges = [r for r in (meta.get(USED_RANGES_KEY) or []) if int(r.get("use_count", 1)) < max_use_count] if not ranges: return None def _last_used(r: dict) -> str: return str(r.get("last_used_at") or r.get("created_at") or "") max_start = max(0.0, asset_total - clip_duration) # 2. 能完整容纳当前片段的候选:按 last_used_at 升序(最久未用优先) fit = sorted( [r for r in ranges if float(r["start"]) <= max_start + 1e-6], key=_last_used, ) if fit: start = min(float(fit[0]["start"]), max_start) return (start, start + clip_duration) # 3. 降级:最久未用 + use_count 最低的区间起点 fallback = sorted(ranges, key=lambda r: (_last_used(r), int(r.get("use_count", 1))))[0] start = min(float(fallback["start"]), max_start) return (start, start + clip_duration) def make_reuse_callback( db: Session, asset_durations: dict[str, float], reused_tracker: dict[str, float] | None = None, assigned_tracker: dict[str, float] | None = None, ratio_limit: float = REUSE_RATIO_LIMIT, ) -> Callable[[str, float], tuple[float, float] | None]: """构造给 ``_calc_random_start_time`` 用的受控复用回调. Args: db: SQLAlchemy session asset_durations: 素材 ID -> 总时长(回调需要素材总时长做边界约束) reused_tracker: 可选的 ``{asset_id: 累计复用时长}``,回调成功返回复用区间时 会把本次片段时长累加进去,供调用方统计成片复用占比(15% 阈值)。 assigned_tracker: 可选的 ``{asset_id: 已分配片段总时长}``,配合 ratio_limit 在复用前预判:若复用本片段后占比 (reused + clip_duration) / (assigned + clip_duration) 超过 ratio_limit,则拒绝复用、返回 None (保证成片复用占比不超阈值)。 ratio_limit: 单条成片复用时长占比上限,默认 15%。 Returns: 回调函数 ``(asset_id, clip_duration) -> (start, end) | None``。 回调内吞掉 DB 异常返回 None,不影响主生成流程。 """ def _reuse(asset_id: str, clip_duration: float) -> tuple[float, float] | None: try: total = float(asset_durations.get(asset_id, 0.0) or 0.0) if total <= 0: return None # 占比闸门:预判复用本片段后是否超限(仅当调用方提供了 assigned tracker) if assigned_tracker is not None: assigned = float(assigned_tracker.get(asset_id, 0.0) or 0.0) reused_amt = float((reused_tracker or {}).get(asset_id, 0.0) or 0.0) if assigned > 0 and (reused_amt + clip_duration) / (assigned + clip_duration) > ratio_limit: logger.info( "[片段追踪] 复用占比预判超 %.0f%% 阈值,拒绝复用: asset_id=%s " "reused=%.1f assigned=%.1f clip=%.1f", ratio_limit * 100, asset_id, reused_amt, assigned, clip_duration, ) return None result = find_reusable_range(db, asset_id, clip_duration, total) except Exception: logger.warning("[片段追踪] 受控复用查询异常: asset_id=%s", asset_id, exc_info=True) return None if result is not None and reused_tracker is not None: reused_tracker[asset_id] = reused_tracker.get(asset_id, 0.0) + clip_duration return result return _reuse def get_asset_recent_use_counts( db: Session, asset_ids: list[str], recent_video_count: int = 5, ) -> dict[str, int]: """统计每个素材在最近 N 个不同 plan_id 中的使用次数。 遍历素材 metadata 中的 used_time_ranges,统计有多少个不同的 plan_id(去重), 返回 {asset_id: count}。只统计最近 recent_video_count 个不同 plan_id 的使用次数。 Args: db: 数据库会话 asset_ids: 素材 ID 列表 recent_video_count: 统计最近多少个不同 plan_id Returns: {asset_id: 在最近 recent_video_count 个 plan 中的使用次数} """ if not asset_ids: return {} result: dict[str, int] = {} models = db.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all() for model in models: meta = _read_meta(model) ranges = meta.get(USED_RANGES_KEY) or [] if not ranges: result[model.id] = 0 continue # 按 created_at 倒序收集不同 plan_id sorted_ranges = sorted( ranges, key=lambda r: r.get("created_at") or "", reverse=True, ) recent_plan_ids: set[str] = set() for r in sorted_ranges: plan_id = r.get("plan_id") if plan_id: recent_plan_ids.add(plan_id) if len(recent_plan_ids) >= recent_video_count: break result[model.id] = len(recent_plan_ids) # 未找到的素材计为 0 for aid in asset_ids: if aid not in result: result[aid] = 0 return result