fix: 素材余量四字段兼容 Asset 领域实体（真实接口恒为 None 修复） (#1550)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

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 {}

tests/unit/test_asset_availability.py

@@ -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,
@@ -227,6 +239,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 ──────────────