"""SmartAssetSelector — 智能素材选择服务. 根据多维度评分从素材库中自动选择最优视频素材, 用于一键生成等需要自动选取素材的场景。 评分维度(加权求和,总分 0-1): - 质量分(quality_score):权重 0.5 — 来自人工或AI的质量评分 - 分辨率适配:权重 0.2 — 分辨率越接近 1080p 得分越高 - 时长合理性:权重 0.2 — 3-30 秒区间最佳,过短/过长扣分 - 码率质量:权重 0.1 — 用文件大小/时长估算,码率适中得分高 特性: - 最低质量分门槛:自动过滤低质量素材 - 时长多样性:保证选出的素材时长分布均匀(短/中/长各占一定比例) - 兼容全部模式:素材库模式和项目模式都可用 纯逻辑部分已抽离到 packages.domain.asset_scoring。 """ from __future__ import annotations import logging from packages.domain.asset_scoring import MEDIUM_BUCKET_MAX as _MEDIUM_BUCKET_MAX # noqa: F401 - re-export for tests from packages.domain.asset_scoring import SHORT_BUCKET_MAX as _SHORT_BUCKET_MAX # noqa: F401 - re-export for tests from packages.domain.asset_scoring import ( AssetScoreDetail, SmartSelectResult, diverse_selection, filter_candidates, score_asset_detail, ) logger = logging.getLogger(__name__) class SmartAssetSelector: """智能素材选择器. 从一组素材中按综合评分选择最优的 N 个, 同时保证时长分布的多样性。 """ def __init__( self, min_quality_score: float = 30.0, target_width: int = 1920, target_height: int = 1080, ): self.min_quality_score = min_quality_score self.target_width = target_width self.target_height = target_height # ── 公开方法 ────────────────────────────────────────────────────────────── def select( self, assets: list, count: int = 0, *, ensure_diversity: bool = True, ) -> SmartSelectResult: """从素材列表中智能选择最优素材. Args: assets: Asset 实体列表(需要有 id/quality_score/width/height/duration/file_size 属性) count: 选取数量,0 表示全部符合条件的 ensure_diversity: 是否保证时长多样性(默认开启) Returns: SmartSelectResult 选择结果 """ # 1. 过滤:只保留 ready 状态的视频素材 + 最低质量分门槛 candidates, filtered_out = filter_candidates(assets, self.min_quality_score) if not candidates: return SmartSelectResult( selected_ids=[], total_candidates=0, filtered_out=filtered_out, avg_score=0.0, details=[], ) # 2. 对每个候选素材评分 scored: list[AssetScoreDetail] = [] for asset in candidates: detail = score_asset_detail( asset_id=asset.id, quality=getattr(asset, "quality_score", None), width=getattr(asset, "width", None), height=getattr(asset, "height", None), duration=getattr(asset, "duration", None), file_size=getattr(asset, "file_size", 0) or 0, target_width=self.target_width, target_height=self.target_height, ) scored.append(detail) # 3. 按总分降序排列 scored.sort(key=lambda d: d.total_score, reverse=True) # 4. 多样性选择(如果需要且数量有限制) if ensure_diversity and count > 0 and len(scored) > count: selected = diverse_selection(scored, count) else: # 无数量限制或不要求多样性,直接按排名取 selected = scored if count <= 0 else scored[:count] avg_score = sum(d.total_score for d in selected) / len(selected) if selected else 0.0 result = SmartSelectResult( selected_ids=[d.asset_id for d in selected], total_candidates=len(candidates), filtered_out=filtered_out, avg_score=avg_score, details=selected, ) logger.info( "智能素材选择完成: 候选=%d, 过滤=%d, 选中=%d, 平均分=%.3f", result.total_candidates, result.filtered_out, len(result.selected_ids), result.avg_score, ) return result # ── 向后兼容:私有方法别名(委托给 asset_scoring 纯函数) ──────────────── def _score_asset(self, asset) -> AssetScoreDetail: """对单个素材进行多维度评分(向后兼容).""" return score_asset_detail( asset_id=asset.id, quality=getattr(asset, "quality_score", None), width=getattr(asset, "width", None), height=getattr(asset, "height", None), duration=getattr(asset, "duration", None), file_size=getattr(asset, "file_size", 0) or 0, target_width=self.target_width, target_height=self.target_height, ) def _score_resolution(self, width: int | None, height: int | None) -> float: """分辨率评分(向后兼容).""" from packages.domain.asset_scoring import score_resolution return score_resolution(width, height, self.target_width, self.target_height) def _score_duration(self, duration: float | None) -> float: """时长评分(向后兼容).""" from packages.domain.asset_scoring import score_duration return score_duration(duration) def _score_bitrate(self, file_size: int, duration: float | None) -> float: """码率评分(向后兼容).""" from packages.domain.asset_scoring import score_bitrate return score_bitrate(file_size, duration) def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]: """多样性选择(向后兼容).""" return diverse_selection(scored, count)