fix(segments): 余量计算兼容 Asset 领域实体(metadata dict)
staging 真实接口验收发现:路由 repository 返回的是 Asset 领域实体而非 ORM AssetModel,区间记录在 metadata dict(repository 与 classification_result JSON 互转);_read_meta 只读 classification_result 列导致 AttributeError, 余量计算全部走异常兜底,四字段在真实接口里恒为 None。 _read_meta 双形态兼容:实体 metadata 为 dict 时直接用;否则读 classification_result JSON 字符串(ORM 形态)。 测试:补充实体形态/ORM 形态对比用例(13913 passed)。
This commit is contained in:
@@ -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 {}
|
||||
|
||||
@@ -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 ──────────────
|
||||
|
||||
|
||||
Reference in New Issue
Block a user