auto-save 2026-05-18 15:07 (~5)

api/main.py

@@ -238,8 +238,8 @@ JobStatus = Literal[
     "transcribing", "transcribed", "failed",
 ]
 
-KEYFRAME_COUNT = int(os.getenv("KEYFRAME_COUNT", "12"))
-FrameExtractTarget = Literal["transparent_human", "balanced", "subject", "transition", "expression", "motion"]
+KEYFRAME_COUNT = int(os.getenv("KEYFRAME_COUNT", "6"))
+FrameExtractTarget = Literal["random_subject", "transparent_human", "balanced", "subject", "transition", "expression", "motion"]
 FrameExtractMode = Literal["replace", "append"]
 FrameExtractQuality = Literal["auto", "fast", "accurate", "ultra"]
 AnalyzeTask = tuple[str, int, FrameExtractTarget, FrameExtractMode, FrameExtractQuality]
@@ -252,6 +252,7 @@ SceneMode = Literal["remove_subject", "similar", "style"]
 SceneStyle = Literal["source", "premium_product", "clean_studio", "warm_lifestyle", "cinematic"]
 SceneAssetRole = Literal["scene", "first_frame", "last_frame"]
 FRAME_TARGET_LABELS: dict[FrameExtractTarget, str] = {
+    "random_subject": "人物随机",
     "transparent_human": "透明骨架人",
     "balanced": "综合关键帧",
     "subject": "清晰主体",
@@ -1408,7 +1409,10 @@ def _target_score(item: dict, target: FrameExtractTarget) -> float:
     scene = float(item.get("scene_score_n", 0.0))
     motion = float(item.get("motion_n", 0.0))
 
-    if target == "transparent_human":
+    if target == "random_subject":
+        # 人物定向随机抽帧先用中心主体/清晰度形成候选池，再在池内随机取样。
+        score = center * 0.52 + sharp * 0.24 + contrast * 0.14 + color * 0.10
+    elif target == "transparent_human":
         # 当前抽帧阶段走本地算力：优先清晰中心主体、高对比、适度色彩和时间覆盖。
         # 透明骨架人的语义判断留给后续审核/识别，不在抽帧阶段逐帧调用 Vision。
         score = center * 0.45 + sharp * 0.30 + contrast * 0.15 + color * 0.10
@@ -1460,6 +1464,15 @@ def _select_keyframes(candidates: list[dict], n: int, target: FrameExtractTarget
         elif it["score"] > dup["score"]:
             deduped[deduped.index(dup)] = it
 
+    if target == "random_subject":
+        # 人物定向随机：从清晰、中心主体更强的候选池里随机抽，不再按动作峰值排序。
+        ranked = sorted(deduped, key=lambda x: -float(x.get("score", 0.0)))
+        pool_size = min(len(ranked), max(n * 6, n + 8))
+        pool = ranked[:pool_size] if pool_size > 0 else ranked
+        selected = random.sample(pool, k=min(n, len(pool))) if len(pool) > n else list(pool)
+        selected.sort(key=lambda x: x["idx"])
+        return selected
+
     # 时序分桶：把候选时间轴等分 n 段，每段取当前目标下最优的
     total = len(candidates)
     buckets: list[list[dict]] = [[] for _ in range(n)]