"""SmartAssetSelector — 智能素材选择服务. 根据多维度评分从素材库中自动选择最优视频素材， 用于一键生成等需要自动选取素材的场景。 评分维度（加权求和，总分 0-1）： - 质量分（quality_score）：权重 0.5 — 来自人工或AI的质量评分 - 分辨率适配：权重 0.2 — 分辨率越接近 1080p 得分越高 - 时长合理性：权重 0.2 — 3-30 秒区间最佳，过短/过长扣分 - 码率质量：权重 0.1 — 用文件大小/时长估算，码率适中得分高 特性： - 最低质量分门槛：自动过滤低质量素材 - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例） - 兼容全部模式：素材库模式和项目模式都可用 """ from __future__ import annotations import logging from dataclasses import dataclass logger = logging.getLogger(__name__) # ── 评分权重 ────────────────────────────────────────────────────────────────── _WEIGHT_QUALITY = 0.5 _WEIGHT_RESOLUTION = 0.2 _WEIGHT_DURATION = 0.2 _WEIGHT_BITRATE = 0.1 # ── 评分参数 ────────────────────────────────────────────────────────────────── _TARGET_WIDTH = 1920 # 目标分辨率宽度基准 _TARGET_HEIGHT = 1080 # 目标分辨率高度基准 _MIN_QUALITY_SCORE = 30.0 # 最低质量分门槛（低于此值的素材直接排除） _OPTIMAL_DURATION_MIN = 3.0 # 最佳时长区间（秒） _OPTIMAL_DURATION_MAX = 30.0 # ── 多样性分桶 ─────────────────────────────────────────────────────────────── _SHORT_BUCKET_MAX = 5.0 # 短素材：< 5s _MEDIUM_BUCKET_MAX = 15.0 # 中素材：5-15s # 长素材：> 15s @dataclass class SmartSelectResult: """智能选择结果.""" selected_ids: list[str] total_candidates: int filtered_out: int # 被质量门槛过滤的数量 avg_score: float details: list[AssetScoreDetail] @dataclass class AssetScoreDetail: """单个素材的评分详情.""" asset_id: str total_score: float quality_score: float resolution_score: float duration_score: float bitrate_score: float duration: float | None class SmartAssetSelector: """智能素材选择器. 从一组素材中按综合评分选择最优的 N 个， 同时保证时长分布的多样性。 """ def __init__( self, min_quality_score: float = _MIN_QUALITY_SCORE, target_width: int = _TARGET_WIDTH, target_height: int = _TARGET_HEIGHT, ): 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 = 0 for asset in assets: status = getattr(asset, "status", None) status_val = status.value if hasattr(status, "value") else str(status) if status_val != "ready": continue mime_type = getattr(asset, "mime_type", "") or "" if not mime_type.startswith("video"): continue quality = getattr(asset, "quality_score", None) if quality is not None and quality < self.min_quality_score: filtered_out += 1 continue candidates.append(asset) 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 = self._score_asset(asset) 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 = self._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 # ── 内部方法 ────────────────────────────────────────────────────────────── def _score_asset(self, asset) -> AssetScoreDetail: """对单个素材进行多维度评分.""" # 质量分 quality = getattr(asset, "quality_score", None) quality_score = (quality / 100.0) if quality is not None else 0.5 # 分辨率评分：越接近目标分辨率得分越高 width = getattr(asset, "width", None) height = getattr(asset, "height", None) resolution_score = self._score_resolution(width, height) # 时长评分：在最佳区间内得分高，过短过长扣分 duration = getattr(asset, "duration", None) duration_score = self._score_duration(duration) # 码率评分：用 file_size/duration 估算，适中得分高 file_size = getattr(asset, "file_size", 0) or 0 bitrate_score = self._score_bitrate(file_size, duration) # 加权总分 total = ( _WEIGHT_QUALITY * quality_score + _WEIGHT_RESOLUTION * resolution_score + _WEIGHT_DURATION * duration_score + _WEIGHT_BITRATE * bitrate_score ) return AssetScoreDetail( asset_id=asset.id, total_score=round(total, 4), quality_score=round(quality_score, 4), resolution_score=round(resolution_score, 4), duration_score=round(duration_score, 4), bitrate_score=round(bitrate_score, 4), duration=duration, ) def _score_resolution(self, width: int | None, height: int | None) -> float: """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重.""" if width is None or height is None or width <= 0 or height <= 0: return 0.5 # 未知分辨率给中评分 target_pixels = self.target_width * self.target_height actual_pixels = width * height # 计算像素数比例 ratio = actual_pixels / target_pixels if ratio >= 1.0: # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分） return 1.0 else: # 低于目标分辨率：线性衰减，但最低不低于 0.1 # 例如：720p (921600) / 1080p (2073600) = 0.44 → 得分 0.6 score = 0.3 + 0.7 * ratio return max(0.1, min(1.0, score)) def _score_duration(self, duration: float | None) -> float: """时长评分：3-30秒最佳，过短或过长都扣分.""" if duration is None or duration <= 0: return 0.5 # 未知时长给中评分 if _OPTIMAL_DURATION_MIN <= duration <= _OPTIMAL_DURATION_MAX: # 最佳区间：满分 return 1.0 if duration < _OPTIMAL_DURATION_MIN: # 太短：线性衰减，1秒以下给 0.3 ratio = duration / _OPTIMAL_DURATION_MIN return 0.3 + 0.7 * ratio # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2 excess = duration - _OPTIMAL_DURATION_MAX penalty = min(0.8, excess / 10.0 * 0.1) return max(0.2, 1.0 - penalty) def _score_bitrate(self, file_size: int, duration: float | None) -> float: """码率评分：根据文件大小和时长估算码率，适中得分高.""" if not file_size or not duration or duration <= 0: return 0.5 # 未知给中评分 # 估算码率（bps） bitrate = (file_size * 8) / duration # 最佳码率范围：2-8 Mbps optimal_low = 2_000_000 # 2 Mbps optimal_high = 8_000_000 # 8 Mbps if optimal_low <= bitrate <= optimal_high: return 1.0 if bitrate < optimal_low: # 码率太低：线性衰减 ratio = bitrate / optimal_low return 0.3 + 0.7 * ratio # 码率太高（文件太大）：适度扣分 excess = bitrate / optimal_high - 1.0 penalty = min(0.5, excess * 0.2) return max(0.5, 1.0 - penalty) def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]: """多样性选择：按时长分桶，保证每个桶都有素材. 策略： 1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>15s) 2. 每个桶配额 = max(1, count / 3) 3. 先从每桶按配额取最高分的 4. 剩余名额从全局最高分中取（不重复） """ # 分桶 short_bucket = [d for d in scored if d.duration is not None and d.duration < _SHORT_BUCKET_MAX] medium_bucket = [ d for d in scored if d.duration is not None and _SHORT_BUCKET_MAX <= d.duration < _MEDIUM_BUCKET_MAX ] long_bucket = [d for d in scored if d.duration is not None and d.duration >= _MEDIUM_BUCKET_MAX] unknown_bucket = [d for d in scored if d.duration is None] buckets = [short_bucket, medium_bucket, long_bucket] bucket_names = ["short", "medium", "long"] # 每个桶基础配额（至少1个，如果桶非空且需要的话） base_quota = max(1, count // 3) selected: list[AssetScoreDetail] = [] selected_ids: set[str] = set() # 先按配额从每个桶取 for bucket, _name in zip(buckets, bucket_names, strict=False): quota = min(base_quota, len(bucket)) if quota <= 0: continue # 桶内已经按分数排好序了，直接取前 quota 个 for item in bucket[:quota]: if item.asset_id not in selected_ids: selected.append(item) selected_ids.add(item.asset_id) if len(selected) >= count: return selected # 剩余名额：从全局（未被选中的）中按分数高低取 remaining_needed = count - len(selected) if remaining_needed > 0: for item in scored: if item.asset_id not in selected_ids: selected.append(item) selected_ids.add(item.asset_id) if len(selected) >= count: break # 如果还不够（不应该发生），加上未知时长的 if len(selected) < count and unknown_bucket: for item in unknown_bucket: if item.asset_id not in selected_ids: selected.append(item) selected_ids.add(item.asset_id) if len(selected) >= count: break return selected[:count]