diff --git a/apps/api/app/services/asset_segment_tracker.py b/apps/api/app/services/asset_segment_tracker.py index 56c76545f..88ffd07eb 100644 --- a/apps/api/app/services/asset_segment_tracker.py +++ b/apps/api/app/services/asset_segment_tracker.py @@ -60,12 +60,23 @@ def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() -def _read_meta(model: AssetModel) -> dict: - """从 AssetModel 读取 metadata dict(classification_result 列承载的 JSON).""" - if not model.classification_result: +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(model.classification_result) + data = json.loads(raw) if isinstance(raw, str) else raw return data if isinstance(data, dict) else {} except Exception: return {} diff --git a/tests/unit/test_asset_availability.py b/tests/unit/test_asset_availability.py index 40e62f53c..3d2b16cb6 100755 --- a/tests/unit/test_asset_availability.py +++ b/tests/unit/test_asset_availability.py @@ -8,8 +8,8 @@ """ import json -import os import sys +import os from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock @@ -36,12 +36,15 @@ VIDEO_DURATION = 60.0 def _make_asset(duration=VIDEO_DURATION, ranges=None, file_type="video", classification_result=None): - """构造测试用 Asset-like 对象。 + """构造测试用 Asset-like 对象(领域实体形态:metadata 为 dict)。 - ranges: list of dicts(used_time_ranges 条目),会自动写入 classification_result JSON。 + ranges: list of dicts(used_time_ranges 条目),同时写入 metadata dict + (Asset 实体形态,repository 返回)与 classification_result JSON 字符串 + (ORM AssetModel 形态);_read_meta 两种形态都必须能读到。 """ + meta_dict = {"used_time_ranges": ranges} if ranges is not None else {} if classification_result is None and ranges is not None: - classification_result = json.dumps({"used_time_ranges": ranges}) + classification_result = json.dumps(meta_dict) return SimpleNamespace( id="asset-test", project_id="proj-1", @@ -50,7 +53,6 @@ def _make_asset(duration=VIDEO_DURATION, ranges=None, file_type="video", classif storage_key="key/test-asset.mp4", thumbnail_url=None, mime_type="video/mp4" if file_type == "video" else "audio/mpeg", - metadata={}, file_size=1000, duration=duration, width=1080, @@ -64,10 +66,20 @@ def _make_asset(duration=VIDEO_DURATION, ranges=None, file_type="video", classif uploaded_by_user_id="user-1", tag_ids=[], file_type=file_type, + # ORM 形态 classification_result=classification_result, + # 领域实体形态(真实路由 repository 返回的 Asset) + metadata=meta_dict, ) +def _make_orm_style_asset(duration=VIDEO_DURATION, ranges=None): + """ORM AssetModel 形态:只有 classification_result JSON 字符串,无 metadata 属性。""" + a = _make_asset(duration=duration, ranges=ranges) + del a.metadata + return a + + def _range(start, end, use_count=1): return { "start": start, @@ -114,7 +126,9 @@ class TestComputeAssetAvailability: def test_overlapping_ranges_merged(self): """重叠区间合并后计算 used_duration,不重复计时。""" - info = compute_asset_availability(_make_asset(ranges=[_range(0.0, 10.0), _range(5.0, 20.0)])) + info = compute_asset_availability( + _make_asset(ranges=[_range(0.0, 10.0), _range(5.0, 20.0)]) + ) # 合并后 [0,20] → 20s assert info["used_duration"] == pytest.approx(20.0, abs=0.01) assert info["used_ratio"] == pytest.approx(20.0 / 60.0, abs=0.001) @@ -218,7 +232,9 @@ class TestComputeAssetAvailability: def test_broken_classification_json_treated_as_unused(self): """classification_result 是非法 JSON 时按无历史区间处理。""" - info = compute_asset_availability(_make_asset(classification_result="not-json{{{")) + info = compute_asset_availability( + _make_asset(classification_result="not-json{{{") + ) assert info is not None assert info["used_duration"] == 0.0 assert info["usable"] is True @@ -227,6 +243,27 @@ class TestComputeAssetAvailability: """边缘间隙常量为 0.3s(与 MediaKit 冲突检测同口径)。""" assert SEGMENT_EDGE_GAP == 0.3 + def test_domain_entity_metadata_dict_form(self): + """领域实体形态(metadata 为 dict,无 classification_result)也能读到区间。 + + 真实路由 repository 返回 Asset 实体,区间记录在 metadata dict 里 + (repository 与 ORM classification_result JSON 互转)。 + """ + a = _make_asset(duration=30.0, ranges=[_range(0.0, 12.0)]) + del a.classification_result # 实体没有该列 + info = compute_asset_availability(a) + assert info is not None + assert info["used_duration"] == pytest.approx(12.0, abs=0.01) + assert info["usable"] is True + + def test_orm_model_classification_result_form(self): + """ORM AssetModel 形态(只有 classification_result JSON 字符串)正常。""" + a = _make_orm_style_asset(duration=30.0, ranges=[_range(0.0, 12.0)]) + assert not hasattr(a, "metadata") + info = compute_asset_availability(a) + assert info is not None + assert info["used_duration"] == pytest.approx(12.0, abs=0.01) + # ── 路由层辅助:_asset_availability_fields / _to_asset_response ──────────────