feat: isolate live recorder assets by tenant

collector-service/app/core_main.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 import httpx
 import json
+import mimetypes
 import os
 import re
 import secrets
@@ -13,10 +15,11 @@ import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
-from urllib.parse import urljoin, urlparse
+from urllib.parse import quote, urljoin, urlparse
 
 from fastapi import Body, Depends, FastAPI, File, Form, Header, HTTPException, Query, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, Field
 
@@ -292,6 +295,24 @@ class LiveRecorderEnabledRequest(BaseModel):
     assistant_id: str | None = None
 
 
+class LiveRecorderSourceCreateRequest(BaseModel):
+    project_id: str = ""
+    assistant_id: str = ""
+    platform: str = ""
+    source_url: str
+    title: str = ""
+    quality: str = "原画"
+    enabled: bool = True
+
+
+class LiveRecorderSourceUpdateRequest(BaseModel):
+    project_id: str | None = None
+    assistant_id: str | None = None
+    title: str | None = None
+    quality: str | None = None
+    enabled: bool | None = None
+
+
 class InternalStepRequest(BaseModel):
     job_id: str = ""
     jobId: str = ""
@@ -425,6 +446,461 @@ def content_source_payload(row: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+LIVE_RECORDER_MANAGED_PREFIX = "sfsrc_"
+LIVE_RECORDER_QUALITY_RANKS = {
+    "流畅": 1,
+    "标清": 2,
+    "高清": 3,
+    "超清": 4,
+    "蓝光": 5,
+    "原画": 6,
+}
+
+
+def normalize_live_recorder_quality(value: str | None) -> str:
+    normalized = str(value or "").strip()
+    if normalized in LIVE_RECORDER_QUALITY_RANKS:
+        return normalized
+    return "原画"
+
+
+def live_recorder_remote_name(source_id: str) -> str:
+    return f"{LIVE_RECORDER_MANAGED_PREFIX}{source_id.replace('-', '')[:12]}"
+
+
+def parse_live_recorder_import_text(raw_text: str) -> list[dict[str, Any]]:
+    entries: list[dict[str, Any]] = []
+    for raw_line in str(raw_text or "").replace("\r\n", "\n").replace("\r", "\n").split("\n"):
+        line = raw_line.strip()
+        if not line:
+            continue
+        enabled = not line.startswith("#")
+        if not enabled:
+            line = line.lstrip("#").strip()
+        if not line:
+            continue
+        parts = [part.strip() for part in re.split(r"[,，]", line) if part.strip()]
+        quality = "原画"
+        source_url = ""
+        title = ""
+        if len(parts) == 1:
+            source_url = parts[0]
+        elif len(parts) == 2:
+            if parts[0] in LIVE_RECORDER_QUALITY_RANKS:
+                quality = parts[0]
+                source_url = parts[1]
+            else:
+                source_url = parts[0]
+                title = parts[1]
+        else:
+            if parts[0] in LIVE_RECORDER_QUALITY_RANKS:
+                quality = parts[0]
+                source_url = parts[1]
+                title = " ".join(parts[2:])
+            else:
+                source_url = parts[0]
+                title = " ".join(parts[1:])
+        source_url = source_url.strip()
+        if not source_url:
+            continue
+        entries.append(
+            {
+                "source_url": source_url,
+                "quality": normalize_live_recorder_quality(quality),
+                "title": title.strip(),
+                "enabled": enabled,
+            }
+        )
+    deduped: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for entry in entries:
+        key = entry["source_url"]
+        if key in seen:
+            continue
+        seen.add(key)
+        deduped.append(entry)
+    return deduped
+
+
+def live_recorder_binding_payload(row: dict[str, Any], *, active_recordings: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+    metadata = parse_json_object(row.get("source_metadata_json") or "{}")
+    items = active_recordings or []
+    return {
+        "id": row["binding_id"],
+        "user_id": row["user_id"],
+        "project_id": row.get("project_id", "") or "",
+        "assistant_id": row.get("assistant_id", "") or "",
+        "source_id": row["source_id"],
+        "platform": row.get("platform", "") or "",
+        "source_url": row.get("source_url", "") or "",
+        "remote_name": row.get("remote_name", "") or "",
+        "title": row.get("binding_title") or row.get("source_title") or "",
+        "quality": normalize_live_recorder_quality(row.get("quality")),
+        "enabled": bool(row.get("enabled", 1)),
+        "metadata": metadata,
+        "active_recordings": items,
+        "recording_count": len(items),
+        "created_at": row["created_at"],
+        "updated_at": row["updated_at"],
+    }
+
+
+def live_recorder_binding_rows(user_id: str, project_id: str = "") -> list[dict[str, Any]]:
+    clauses = ["b.user_id = ?"]
+    params: list[Any] = [user_id]
+    if project_id:
+        clauses.append("b.project_id = ?")
+        params.append(project_id)
+    sql = f"""
+        SELECT
+            b.id AS binding_id,
+            b.user_id,
+            b.project_id,
+            b.assistant_id,
+            b.source_id,
+            b.title AS binding_title,
+            b.quality,
+            b.enabled,
+            b.created_at,
+            b.updated_at,
+            s.platform,
+            s.source_url,
+            s.remote_name,
+            s.title AS source_title,
+            s.metadata_json AS source_metadata_json
+        FROM live_recorder_bindings b
+        JOIN live_recorder_sources s ON s.id = b.source_id
+        WHERE {" AND ".join(clauses)}
+        ORDER BY b.updated_at DESC, b.created_at DESC
+    """
+    return db.fetch_all(sql, tuple(params))
+
+
+def load_owned_live_recorder_binding(binding_id: str, user_id: str) -> dict[str, Any]:
+    row = db.fetch_one(
+        """
+        SELECT
+            b.id AS binding_id,
+            b.user_id,
+            b.project_id,
+            b.assistant_id,
+            b.source_id,
+            b.title AS binding_title,
+            b.quality,
+            b.enabled,
+            b.created_at,
+            b.updated_at,
+            s.platform,
+            s.source_url,
+            s.remote_name,
+            s.title AS source_title,
+            s.metadata_json AS source_metadata_json
+        FROM live_recorder_bindings b
+        JOIN live_recorder_sources s ON s.id = b.source_id
+        WHERE b.id = ? AND b.user_id = ?
+        """,
+        (binding_id, user_id),
+    )
+    if not row:
+        raise HTTPException(status_code=404, detail="Live recorder source not found")
+    return row
+
+
+def get_or_create_live_recorder_source(*, platform: str, source_url: str, title: str = "") -> dict[str, Any]:
+    normalized_platform = ensure_domestic_platform(platform or infer_platform_from_url(source_url), allow_blank=False)
+    normalized_url = source_url.strip()
+    existing = db.fetch_one(
+        "SELECT * FROM live_recorder_sources WHERE platform = ? AND source_url = ?",
+        (normalized_platform, normalized_url),
+    )
+    timestamp = utc_now()
+    if existing:
+        if title.strip() and title.strip() != (existing.get("title") or ""):
+            db.execute(
+                "UPDATE live_recorder_sources SET title = ?, updated_at = ? WHERE id = ?",
+                (title.strip(), timestamp, existing["id"]),
+            )
+            existing = db.fetch_one("SELECT * FROM live_recorder_sources WHERE id = ?", (existing["id"],))
+        return existing
+    source_id = make_id("lrsrc")
+    remote_name = live_recorder_remote_name(source_id)
+    db.execute(
+        """
+        INSERT INTO live_recorder_sources (id, platform, source_url, remote_name, title, metadata_json, created_at, updated_at)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            source_id,
+            normalized_platform,
+            normalized_url,
+            remote_name,
+            title.strip(),
+            json.dumps({}, ensure_ascii=False),
+            timestamp,
+            timestamp,
+        ),
+    )
+    return db.fetch_one("SELECT * FROM live_recorder_sources WHERE id = ?", (source_id,))
+
+
+def upsert_live_recorder_binding(
+    *,
+    user_id: str,
+    project_id: str,
+    assistant_id: str = "",
+    platform: str,
+    source_url: str,
+    title: str = "",
+    quality: str = "原画",
+    enabled: bool = True,
+) -> dict[str, Any]:
+    source_row = get_or_create_live_recorder_source(platform=platform, source_url=source_url, title=title)
+    normalized_project_id = project_id.strip() or None
+    normalized_assistant_id = assistant_id.strip() or None
+    existing = db.fetch_one(
+        "SELECT * FROM live_recorder_bindings WHERE user_id = ? AND source_id = ?",
+        (user_id, source_row["id"]),
+    )
+    timestamp = utc_now()
+    normalized_quality = normalize_live_recorder_quality(quality)
+    if existing:
+        db.execute(
+            """
+            UPDATE live_recorder_bindings
+            SET project_id = ?, assistant_id = ?, title = ?, quality = ?, enabled = ?, updated_at = ?
+            WHERE id = ?
+            """,
+            (
+                normalized_project_id if normalized_project_id is not None else existing.get("project_id"),
+                normalized_assistant_id,
+                title.strip(),
+                normalized_quality,
+                1 if enabled else 0,
+                timestamp,
+                existing["id"],
+            ),
+        )
+        return load_owned_live_recorder_binding(existing["id"], user_id)
+    binding_id = make_id("lrbind")
+    db.execute(
+        """
+        INSERT INTO live_recorder_bindings (
+            id, user_id, project_id, assistant_id, source_id, title, quality, enabled, created_at, updated_at
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            binding_id,
+            user_id,
+            normalized_project_id,
+            normalized_assistant_id,
+            source_row["id"],
+            title.strip(),
+            normalized_quality,
+            1 if enabled else 0,
+            timestamp,
+            timestamp,
+        ),
+    )
+    return load_owned_live_recorder_binding(binding_id, user_id)
+
+
+def live_recorder_source_groups() -> dict[str, dict[str, Any]]:
+    rows = db.fetch_all(
+        """
+        SELECT
+            s.id AS source_id,
+            s.platform,
+            s.source_url,
+            s.remote_name,
+            s.title AS source_title,
+            b.id AS binding_id,
+            b.user_id,
+            b.project_id,
+            b.assistant_id,
+            b.title AS binding_title,
+            b.quality,
+            b.enabled
+        FROM live_recorder_sources s
+        LEFT JOIN live_recorder_bindings b ON b.source_id = s.id
+        ORDER BY s.created_at ASC, b.created_at ASC
+        """
+    )
+    grouped: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        source_id = row["source_id"]
+        target = grouped.setdefault(
+            source_id,
+            {
+                "source_id": source_id,
+                "platform": row.get("platform", ""),
+                "source_url": row.get("source_url", ""),
+                "remote_name": row.get("remote_name", ""),
+                "source_title": row.get("source_title", ""),
+                "bindings": [],
+            },
+        )
+        if row.get("binding_id"):
+            target["bindings"].append(
+                {
+                    "binding_id": row["binding_id"],
+                    "user_id": row["user_id"],
+                    "project_id": row.get("project_id", "") or "",
+                    "assistant_id": row.get("assistant_id", "") or "",
+                    "binding_title": row.get("binding_title", "") or "",
+                    "quality": normalize_live_recorder_quality(row.get("quality")),
+                    "enabled": bool(row.get("enabled", 1)),
+                }
+            )
+    return grouped
+
+
+def build_live_recorder_managed_raw() -> tuple[str, dict[str, Any]]:
+    existing = live_recorder_request("GET", "/api/url-config", timeout=15.0)
+    preserved_lines: list[str] = []
+    for line in existing.get("lines") or []:
+        raw = str(line.get("raw") or "")
+        name = str(line.get("name") or "").strip()
+        if line.get("type") == "url" and name.startswith(LIVE_RECORDER_MANAGED_PREFIX):
+            continue
+        preserved_lines.append(raw)
+
+    generated_lines: list[str] = []
+    active_source_count = 0
+    active_binding_count = 0
+    for source in live_recorder_source_groups().values():
+        active_bindings = [binding for binding in source["bindings"] if binding.get("enabled")]
+        if not active_bindings:
+            continue
+        quality = max(
+            (binding.get("quality") or "原画" for binding in active_bindings),
+            key=lambda item: LIVE_RECORDER_QUALITY_RANKS.get(item, 0),
+        )
+        generated_lines.append(f"{quality},{source['source_url']},{source['remote_name']}")
+        active_source_count += 1
+        active_binding_count += len(active_bindings)
+
+    final_lines = [line for line in preserved_lines if str(line).strip()]
+    final_lines.extend(generated_lines)
+    raw = "\n".join(final_lines).strip()
+    if raw:
+        raw += "\n"
+    stats = {
+        "preserved_count": len([line for line in preserved_lines if str(line).strip()]),
+        "managed_source_count": active_source_count,
+        "managed_binding_count": active_binding_count,
+        "generated_count": len(generated_lines),
+    }
+    return raw, stats
+
+
+def sync_live_recorder_remote_config() -> dict[str, Any]:
+    raw, stats = build_live_recorder_managed_raw()
+    payload = live_recorder_request("POST", "/api/url-config", {"raw": raw}, timeout=20.0)
+    return {
+        "ok": True,
+        "stats": stats,
+        "remote": payload,
+    }
+
+
+def live_recorder_runtime_payload() -> dict[str, Any]:
+    payload = live_recorder_request("GET", "/api/status-lite", timeout=8.0)
+    return payload if isinstance(payload, dict) else {"recordings": []}
+
+
+def live_recorder_recordings_payload() -> dict[str, Any]:
+    payload = live_recorder_request("GET", "/api/recordings", timeout=12.0)
+    return payload if isinstance(payload, dict) else {"recordings": []}
+
+
+def live_recorder_downloads_payload() -> dict[str, Any]:
+    payload = live_recorder_request("GET", "/api/downloads", timeout=20.0)
+    return payload if isinstance(payload, dict) else {"files": []}
+
+
+def owned_live_recorder_sources(user_id: str, project_id: str = "") -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]]]:
+    rows = live_recorder_binding_rows(user_id, project_id)
+    mapping = {row["remote_name"]: row for row in rows}
+    return rows, mapping
+
+
+def filter_owned_live_recorder_recordings(user_id: str, project_id: str = "") -> list[dict[str, Any]]:
+    rows, mapping = owned_live_recorder_sources(user_id, project_id)
+    if not rows:
+        return []
+    status_payload = live_recorder_recordings_payload()
+    items = []
+    for recording in status_payload.get("recordings") or []:
+        haystack = " ".join(
+            [
+                str(recording.get("anchor_name") or ""),
+                str(recording.get("record_name") or ""),
+                str(recording.get("save_dir") or ""),
+                str(recording.get("save_file") or ""),
+            ]
+        )
+        matched = None
+        for remote_name, row in mapping.items():
+            if remote_name and remote_name in haystack:
+                matched = row
+                break
+            if str(recording.get("record_url") or "").strip() == str(row.get("source_url") or "").strip():
+                matched = row
+                break
+        if not matched:
+            continue
+        payload = dict(recording)
+        payload["source_id"] = matched["source_id"]
+        payload["binding_id"] = matched["binding_id"]
+        payload["project_id"] = matched.get("project_id", "") or ""
+        payload["title"] = matched.get("binding_title") or matched.get("source_title") or matched.get("remote_name") or ""
+        items.append(payload)
+    return items
+
+
+def encode_live_recorder_file_id(relative_path: str) -> str:
+    return base64.urlsafe_b64encode(relative_path.encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def decode_live_recorder_file_id(file_id: str) -> str:
+    padding = "=" * (-len(file_id) % 4)
+    try:
+        return base64.urlsafe_b64decode((file_id + padding).encode("ascii")).decode("utf-8")
+    except Exception as exc:
+        raise HTTPException(status_code=400, detail=f"Invalid live recorder file id: {exc}")
+
+
+def filter_owned_live_recorder_files(user_id: str, project_id: str = "", limit: int = 200) -> list[dict[str, Any]]:
+    rows, mapping = owned_live_recorder_sources(user_id, project_id)
+    if not rows:
+        return []
+    downloads_payload = live_recorder_downloads_payload()
+    files = []
+    for item in downloads_payload.get("files") or []:
+        relative_path = str(item.get("relative_path") or "")
+        name = str(item.get("name") or "")
+        haystack = f"{relative_path} {name}"
+        matched = None
+        for remote_name, row in mapping.items():
+            if remote_name and remote_name in haystack:
+                matched = row
+                break
+        if not matched:
+            continue
+        payload = dict(item)
+        payload["id"] = encode_live_recorder_file_id(relative_path)
+        payload["source_id"] = matched["source_id"]
+        payload["binding_id"] = matched["binding_id"]
+        payload["project_id"] = matched.get("project_id", "") or ""
+        payload["platform"] = matched.get("platform", "") or ""
+        payload["title"] = matched.get("binding_title") or matched.get("source_title") or matched.get("remote_name") or ""
+        payload["content_url"] = f"/v2/live-recorder/files/{payload['id']}/content"
+        files.append(payload)
+        if len(files) >= limit:
+            break
+    return files
+
+
 def job_event_payload(row: dict[str, Any]) -> dict[str, Any]:
     return {
         "id": row["id"],
@@ -1821,10 +2297,120 @@ def live_recorder_health(account: dict[str, Any] = Depends(require_approved)) ->
     return live_recorder_request("GET", "/api/healthz", timeout=8.0)
 
 
+@app.get("/v2/live-recorder/sources")
+def list_live_recorder_sources(
+    project_id: str | None = Query(default=None),
+    account: dict[str, Any] = Depends(require_approved),
+) -> dict[str, Any]:
+    normalized_project_id = (project_id or "").strip()
+    if normalized_project_id:
+        resolve_target_project(account["id"], normalized_project_id, username=account["username"])
+    active_by_binding: dict[str, list[dict[str, Any]]] = {}
+    for item in filter_owned_live_recorder_recordings(account["id"], normalized_project_id):
+        active_by_binding.setdefault(item["binding_id"], []).append(item)
+    items = [
+        live_recorder_binding_payload(row, active_recordings=active_by_binding.get(row["binding_id"], []))
+        for row in live_recorder_binding_rows(account["id"], normalized_project_id)
+    ]
+    return {"items": items, "count": len(items)}
+
+
+@app.post("/v2/live-recorder/sources")
+def create_live_recorder_source(
+    request: LiveRecorderSourceCreateRequest,
+    account: dict[str, Any] = Depends(require_approved),
+) -> dict[str, Any]:
+    project = resolve_target_project(account["id"], request.project_id or None, username=account["username"])
+    assistant = resolve_target_assistant(account["id"], request.assistant_id or None, project["id"])
+    binding = upsert_live_recorder_binding(
+        user_id=account["id"],
+        project_id=project["id"],
+        assistant_id=(assistant or {}).get("id", ""),
+        platform=request.platform or infer_platform_from_url(request.source_url),
+        source_url=request.source_url,
+        title=request.title,
+        quality=request.quality,
+        enabled=request.enabled,
+    )
+    sync_result = sync_live_recorder_remote_config()
+    return {"item": live_recorder_binding_payload(binding), "sync": sync_result}
+
+
+@app.patch("/v2/live-recorder/sources/{binding_id}")
+def update_live_recorder_source(
+    binding_id: str,
+    request: LiveRecorderSourceUpdateRequest,
+    account: dict[str, Any] = Depends(require_approved),
+) -> dict[str, Any]:
+    current = load_owned_live_recorder_binding(binding_id, account["id"])
+    project_id = current.get("project_id", "") or ""
+    if request.project_id is not None:
+        if request.project_id.strip():
+            project = resolve_target_project(account["id"], request.project_id.strip(), username=account["username"])
+            project_id = project["id"]
+        else:
+            project_id = ""
+    assistant_id = current.get("assistant_id", "") or ""
+    if request.assistant_id is not None:
+        assistant = resolve_target_assistant(account["id"], request.assistant_id or None, project_id)
+        assistant_id = (assistant or {}).get("id", "")
+    db.execute(
+        """
+        UPDATE live_recorder_bindings
+        SET project_id = ?, assistant_id = ?, title = ?, quality = ?, enabled = ?, updated_at = ?
+        WHERE id = ? AND user_id = ?
+        """,
+        (
+            project_id or None,
+            assistant_id or None,
+            request.title.strip() if request.title is not None else current.get("binding_title", ""),
+            normalize_live_recorder_quality(request.quality if request.quality is not None else current.get("quality")),
+            1 if (request.enabled if request.enabled is not None else bool(current.get("enabled", 1))) else 0,
+            utc_now(),
+            binding_id,
+            account["id"],
+        ),
+    )
+    updated = load_owned_live_recorder_binding(binding_id, account["id"])
+    sync_result = sync_live_recorder_remote_config()
+    return {"item": live_recorder_binding_payload(updated), "sync": sync_result}
+
+
+@app.delete("/v2/live-recorder/sources/{binding_id}")
+def delete_live_recorder_source(binding_id: str, account: dict[str, Any] = Depends(require_approved)) -> dict[str, Any]:
+    current = load_owned_live_recorder_binding(binding_id, account["id"])
+    db.execute("DELETE FROM live_recorder_bindings WHERE id = ? AND user_id = ?", (binding_id, account["id"]))
+    remaining = db.fetch_one("SELECT COUNT(*) AS count FROM live_recorder_bindings WHERE source_id = ?", (current["source_id"],))
+    if not remaining or int(remaining.get("count") or 0) <= 0:
+        db.execute("DELETE FROM live_recorder_sources WHERE id = ?", (current["source_id"],))
+    sync_result = sync_live_recorder_remote_config()
+    return {"ok": True, "deleted_id": binding_id, "sync": sync_result}
+
+
 @app.get("/v2/live-recorder/status")
-def live_recorder_status(account: dict[str, Any] = Depends(require_approved)) -> Any:
+def live_recorder_status(
-    _ = account
+    project_id: str | None = Query(default=None),
-    return live_recorder_request("GET", "/api/status-lite", timeout=8.0)
+    account: dict[str, Any] = Depends(require_approved),
+) -> Any:
+    normalized_project_id = (project_id or "").strip()
+    if normalized_project_id:
+        resolve_target_project(account["id"], normalized_project_id, username=account["username"])
+    runtime = live_recorder_runtime_payload()
+    active_items = filter_owned_live_recorder_recordings(account["id"], normalized_project_id)
+    files = filter_owned_live_recorder_files(account["id"], normalized_project_id, limit=12)
+    return {
+        "backend": runtime.get("backend") or {},
+        "running": bool(runtime.get("running")),
+        "pid": runtime.get("pid"),
+        "uptime_seconds": runtime.get("uptime_seconds", 0),
+        "started_at": runtime.get("started_at"),
+        "last_exit_code": runtime.get("last_exit_code"),
+        "managed": bool(runtime.get("managed")),
+        "url_info": runtime.get("url_info") or {},
+        "active_recordings": active_items,
+        "recording_count": len(active_items),
+        "files_preview": files,
+    }
 
 
 @app.post("/v2/live-recorder/url-config/import")
@@ -1832,8 +2418,23 @@ def live_recorder_import_urls(
     request: LiveRecorderImportRequest,
     account: dict[str, Any] = Depends(require_approved),
 ) -> Any:
-    _ = account
+    project = ensure_default_project(account["id"], username=account["username"])
-    return live_recorder_request("POST", "/api/url-config/import", {"raw": request.raw}, timeout=20.0)
+    entries = parse_live_recorder_import_text(request.raw)
+    items = []
+    for entry in entries:
+        platform = infer_platform_from_url(entry["source_url"]) or "kuaishou"
+        binding = upsert_live_recorder_binding(
+            user_id=account["id"],
+            project_id=project["id"],
+            platform=platform,
+            source_url=entry["source_url"],
+            title=entry.get("title") or "",
+            quality=entry.get("quality") or "原画",
+            enabled=bool(entry.get("enabled", True)),
+        )
+        items.append(live_recorder_binding_payload(binding))
+    sync_result = sync_live_recorder_remote_config()
+    return {"ok": True, "items": items, "count": len(items), "sync": sync_result}
 
 
 @app.post("/v2/live-recorder/url-config/set-enabled")
@@ -1841,14 +2442,57 @@ def live_recorder_set_enabled(
     request: LiveRecorderEnabledRequest,
     account: dict[str, Any] = Depends(require_approved),
 ) -> Any:
+    _ = request
     _ = account
-    return live_recorder_request(
+    raise HTTPException(status_code=410, detail="Use /v2/live-recorder/sources/{binding_id} to update tenant-scoped recorder sources")
-        "POST",
+
-        "/api/url-config/set-enabled",
+
-        {"line_no": request.line_no, "enabled": request.enabled},
+@app.get("/v2/live-recorder/files")
-        timeout=20.0,
+def list_live_recorder_files(
+    project_id: str | None = Query(default=None),
+    limit: int = Query(default=100, ge=1, le=500),
+    account: dict[str, Any] = Depends(require_approved),
+) -> dict[str, Any]:
+    normalized_project_id = (project_id or "").strip()
+    if normalized_project_id:
+        resolve_target_project(account["id"], normalized_project_id, username=account["username"])
+    items = filter_owned_live_recorder_files(account["id"], normalized_project_id, limit=limit)
+    return {"items": items, "count": len(items)}
+
+
+@app.get("/v2/live-recorder/files/{file_id}/content")
+def stream_live_recorder_file(
+    file_id: str,
+    project_id: str | None = Query(default=None),
+    account: dict[str, Any] = Depends(require_approved),
+) -> StreamingResponse:
+    normalized_project_id = (project_id or "").strip()
+    relative_path = decode_live_recorder_file_id(file_id)
+    owned_items = filter_owned_live_recorder_files(account["id"], normalized_project_id, limit=500)
+    matched = next((item for item in owned_items if item.get("relative_path") == relative_path), None)
+    if not matched:
+        raise HTTPException(status_code=404, detail="Live recorder file not found")
+    target_url = urljoin(
+        LIVE_RECORDER_BASE_URL if LIVE_RECORDER_BASE_URL.endswith("/") else f"{LIVE_RECORDER_BASE_URL}/",
+        f"downloads/{quote(relative_path.lstrip('/'), safe='/')}",
     )
 
+    def iterator():
+        try:
+            with httpx.stream("GET", target_url, timeout=120.0, follow_redirects=True) as response:
+                response.raise_for_status()
+                for chunk in response.iter_bytes():
+                    if chunk:
+                        yield chunk
+        except httpx.HTTPStatusError as exc:
+            raise HTTPException(status_code=exc.response.status_code, detail="Failed to stream live recorder file")
+        except Exception as exc:
+            raise HTTPException(status_code=502, detail=f"Failed to proxy live recorder file: {exc}")
+
+    media_type = mimetypes.guess_type(relative_path)[0] or "application/octet-stream"
+    headers = {"Content-Disposition": f'inline; filename="{matched.get("name") or "recording.bin"}"'}
+    return StreamingResponse(iterator(), media_type=media_type, headers=headers)
+
 
 @app.post("/v2/live-recorder/recorder/start")
 def live_recorder_start(account: dict[str, Any] = Depends(require_approved)) -> Any:
@@ -1857,7 +2501,7 @@ def live_recorder_start(account: dict[str, Any] = Depends(require_approved)) ->
 
 
 @app.post("/v2/live-recorder/recorder/stop")
-def live_recorder_stop(account: dict[str, Any] = Depends(require_approved)) -> Any:
+def live_recorder_stop(account: dict[str, Any] = Depends(require_super_admin)) -> Any:
     _ = account
     return live_recorder_request("POST", "/api/recorder/stop", timeout=30.0)
 

collector-service/app/database.py

@@ -235,6 +235,36 @@ class Database:
             FOREIGN KEY(assistant_id) REFERENCES assistants(id) ON DELETE SET NULL
         );
 
+        CREATE TABLE IF NOT EXISTS live_recorder_sources (
+            id TEXT PRIMARY KEY,
+            platform TEXT NOT NULL DEFAULT '',
+            source_url TEXT NOT NULL,
+            remote_name TEXT NOT NULL UNIQUE,
+            title TEXT NOT NULL DEFAULT '',
+            metadata_json TEXT NOT NULL DEFAULT '{}',
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            UNIQUE(platform, source_url)
+        );
+
+        CREATE TABLE IF NOT EXISTS live_recorder_bindings (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL,
+            project_id TEXT,
+            assistant_id TEXT,
+            source_id TEXT NOT NULL,
+            title TEXT NOT NULL DEFAULT '',
+            quality TEXT NOT NULL DEFAULT '原画',
+            enabled INTEGER NOT NULL DEFAULT 1,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            UNIQUE(user_id, source_id),
+            FOREIGN KEY(user_id) REFERENCES accounts(id) ON DELETE CASCADE,
+            FOREIGN KEY(project_id) REFERENCES projects(id) ON DELETE SET NULL,
+            FOREIGN KEY(assistant_id) REFERENCES assistants(id) ON DELETE SET NULL,
+            FOREIGN KEY(source_id) REFERENCES live_recorder_sources(id) ON DELETE CASCADE
+        );
+
         CREATE TABLE IF NOT EXISTS job_events (
             id TEXT PRIMARY KEY,
             job_id TEXT NOT NULL,

collector-service/app/main.py

@@ -19,3 +19,4 @@ register_domestic_platform_routes(app, core, platform="xiaohongshu", label="小
 register_domestic_platform_routes(app, core, platform="bilibili", label="哔哩哔哩")
 register_domestic_platform_routes(app, core, platform="kuaishou", label="快手")
 register_domestic_platform_routes(app, core, platform="wechat_video", label="微信视频号")
+app.openapi_schema = None