Compare commits

...

1 Commits

Author SHA1 Message Date
CI Bot 99cbff8abb 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)。
2026-08-30 08:02:27 +08:00
2 changed files with 59 additions and 11 deletions
+15 -4
View File
@@ -60,12 +60,23 @@ def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _read_meta(model: AssetModel) -> dict:
"""从 AssetModel 读取 metadata dictclassification_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 {}
+44 -7
View File
@@ -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 dictsused_time_ranges 条目),会自动写入 classification_result JSON。
ranges: list of dictsused_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 ──────────────