feat: add live douyin video listing route

collector-service/app/douyin_features.py

@@ -1183,6 +1183,60 @@ def register_douyin_routes(app: Any, legacy: Any) -> None:
             "video_summary": video_summary
         }
 
+    def _video_content_type(video: dict[str, Any]) -> str:
+        raw = video.get("raw") if isinstance(video.get("raw"), dict) else {}
+        if raw.get("images") or raw.get("image_infos") or raw.get("is_multi_content"):
+            return "image_text"
+        return "video"
+
+    def _video_performance_score(video: dict[str, Any]) -> float:
+        stats = video.get("stats") if isinstance(video.get("stats"), dict) else {}
+        play = float(stats.get("play") or 0)
+        like = float(stats.get("like") or 0)
+        comment = float(stats.get("comment") or 0)
+        share = float(stats.get("share") or 0)
+        collect = float(stats.get("collect") or 0)
+        score = (
+            min(play / 10000.0, 6.0) * 8.0
+            + min(like / 1000.0, 6.0) * 7.0
+            + min(comment / 200.0, 6.0) * 4.0
+            + min(share / 100.0, 6.0) * 4.0
+            + min(collect / 100.0, 6.0) * 3.0
+        )
+        return round(min(100.0, score), 1)
+
+    def _workspace_video_payload(video: dict[str, Any]) -> dict[str, Any]:
+        tags = video.get("tags") if isinstance(video.get("tags"), list) else []
+        return {
+            "id": video.get("id") or video.get("aweme_id") or "",
+            "aweme_id": video.get("aweme_id") or "",
+            "title": video.get("title") or video.get("description") or "未命名作品",
+            "description": video.get("description") or video.get("title") or "",
+            "share_url": video.get("share_url") or "",
+            "cover_url": video.get("cover_url") or "",
+            "duration_sec": video.get("duration_sec") or 0,
+            "published_at": video.get("published_at") or "",
+            "tags": tags,
+            "stats": video.get("stats") if isinstance(video.get("stats"), dict) else {},
+            "content_type": _video_content_type(video),
+            "score": {
+                "performance_score": _video_performance_score(video)
+            }
+        }
+
+    def _video_sort_key(video: dict[str, Any], sort_by: str) -> tuple[Any, ...]:
+        stats = video.get("stats") if isinstance(video.get("stats"), dict) else {}
+        normalized = (sort_by or "score").strip().lower()
+        if normalized == "latest":
+            return (video.get("published_at") or "", video.get("id") or "")
+        if normalized == "play":
+            return (float(stats.get("play") or 0), video.get("published_at") or "")
+        if normalized == "like":
+            return (float(stats.get("like") or 0), video.get("published_at") or "")
+        if normalized == "comment":
+            return (float(stats.get("comment") or 0), video.get("published_at") or "")
+        return (float(video.get("score", {}).get("performance_score") or 0), video.get("published_at") or "")
+
     def _list_linked_accounts(account_row: dict[str, Any]) -> list[dict[str, Any]]:
         relation_rows = legacy.db.fetch_all(
             """
@@ -2061,6 +2115,92 @@ def register_douyin_routes(app: Any, legacy: Any) -> None:
         account_row = _require_owned_account(account_id, account["id"])
         return _build_workspace_payload(account_row)
 
+    @app.get("/v2/douyin/accounts/{account_id}/videos")
+    def list_douyin_account_videos(
+        account_id: str,
+        limit: int = 200,
+        sort_by: str = "score",
+        scope: str = "all",
+        content_type: str = "all",
+        q: str = "",
+        tag: str = "",
+        account: dict[str, Any] = Depends(legacy.require_approved)
+    ) -> dict[str, Any]:
+        account_row = _require_owned_account(account_id, account["id"])
+        raw_videos = _list_videos(account_row["id"], limit=max(limit, 24))
+        items = [_workspace_video_payload(video) for video in raw_videos]
+        item_map = {item["id"]: item for item in items}
+
+        high_score_threshold = 60.0
+        top_scored_video_ids = [
+            item["id"]
+            for item in sorted(items, key=lambda entry: _video_sort_key(entry, "score"), reverse=True)
+            if float(item.get("score", {}).get("performance_score") or 0) >= high_score_threshold
+        ]
+        if not top_scored_video_ids:
+            top_scored_video_ids = [
+                item["id"]
+                for item in sorted(items, key=lambda entry: _video_sort_key(entry, "score"), reverse=True)[:5]
+            ]
+        latest_video_ids = [
+            item["id"]
+            for item in sorted(items, key=lambda entry: _video_sort_key(entry, "latest"), reverse=True)[:12]
+        ]
+
+        normalized_scope = (scope or "all").strip().lower()
+        if normalized_scope == "top":
+            items = [item_map[video_id] for video_id in top_scored_video_ids if video_id in item_map]
+        elif normalized_scope == "latest":
+            items = [item_map[video_id] for video_id in latest_video_ids if video_id in item_map]
+
+        normalized_content_type = (content_type or "all").strip().lower()
+        if normalized_content_type in {"video", "image_text"}:
+            items = [
+                item for item in items
+                if str(item.get("content_type") or "video").strip().lower() == normalized_content_type
+            ]
+
+        query_text = (q or "").strip().lower()
+        if query_text:
+            items = [
+                item for item in items
+                if query_text in " ".join(
+                    [
+                        str(item.get("title") or ""),
+                        str(item.get("description") or ""),
+                        str(item.get("aweme_id") or ""),
+                        *[str(tag_item) for tag_item in item.get("tags", [])]
+                    ]
+                ).lower()
+            ]
+
+        tag_text = (tag or "").strip().lower()
+        if tag_text:
+            items = [
+                item for item in items
+                if any(tag_text in str(tag_item).lower() for tag_item in item.get("tags", []))
+            ]
+
+        normalized_sort = (sort_by or "score").strip().lower()
+        items.sort(key=lambda item: _video_sort_key(item, normalized_sort), reverse=True)
+        return {
+            "account_id": account_row["id"],
+            "sort_by": normalized_sort,
+            "scope": normalized_scope,
+            "content_type": normalized_content_type,
+            "query": q,
+            "tag": tag,
+            "high_score_threshold": high_score_threshold,
+            "meta": {
+                "source": "fastgpt-live-fallback",
+                "total": len(raw_videos),
+                "filtered": len(items)
+            },
+            "top_scored_video_ids": top_scored_video_ids,
+            "latest_video_ids": latest_video_ids,
+            "items": items[: max(1, min(limit, 1000))]
+        }
+
     @app.get("/v2/douyin/accounts/{account_id}/analysis-reports")
     def list_douyin_analysis_reports(
         account_id: str,