diff --git a/.env.example b/.env.example index 48812f0..70df651 100644 --- a/.env.example +++ b/.env.example @@ -39,6 +39,11 @@ CELERY_PUB_CONCURRENCY=1 # Build with real ASR (faster-whisper) INSTALL_ASR=1 +# Optional: refresh yt-dlp at image build time. +# - latest: install newest release available when this layer is rebuilt +# - or pin a specific version, e.g. 2026.03.03 +YTDLP_VERSION=latest + # Optional: allow browser origins (comma-separated) CORS_ALLOW_ORIGINS=http://localhost:3000,http://127.0.0.1:3000 diff --git a/Dockerfile b/Dockerfile index 591ab7f..9286998 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,6 +12,7 @@ RUN apt-get update \ COPY pyproject.toml README.md ./ ARG INSTALL_ASR=0 +ARG YTDLP_VERSION=latest # Install dependencies in a cache-friendly layer so editing source code doesn't # force re-downloading everything on every docker build. @@ -25,6 +26,14 @@ COPY docker ./docker RUN pip install --no-cache-dir -e . --no-deps +RUN if [ -n "$YTDLP_VERSION" ]; then \ + if [ "$YTDLP_VERSION" = "latest" ]; then \ + pip install --no-cache-dir -U "yt-dlp[default]"; \ + else \ + pip install --no-cache-dir -U "yt-dlp[default]==${YTDLP_VERSION}"; \ + fi; \ + fi + RUN chmod +x /app/docker/entrypoint.sh CMD ["/app/docker/entrypoint.sh"] diff --git a/compose.yml b/compose.yml index 14c2edb..55d3f9b 100644 --- a/compose.yml +++ b/compose.yml @@ -35,6 +35,7 @@ services: context: . args: INSTALL_ASR: ${INSTALL_ASR:-1} + YTDLP_VERSION: ${YTDLP_VERSION:-latest} image: videoroll:prod environment: DATABASE_URL: ${DATABASE_URL} diff --git a/docker-compose.yml b/docker-compose.yml index 14c2edb..55d3f9b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -35,6 +35,7 @@ services: context: . args: INSTALL_ASR: ${INSTALL_ASR:-1} + YTDLP_VERSION: ${YTDLP_VERSION:-latest} image: videoroll:prod environment: DATABASE_URL: ${DATABASE_URL} diff --git a/skills/openclaw-videoroll/SKILL.md b/skills/openclaw-videoroll/SKILL.md new file mode 100644 index 0000000..2037c27 --- /dev/null +++ b/skills/openclaw-videoroll/SKILL.md @@ -0,0 +1,79 @@ +--- +name: openclaw-videoroll +description: Collect YouTube video URLs on an external machine and submit them to VideoRoll through the remote auto-ingest API. Use when OpenClaw needs to find canonical YouTube watch links, deduplicate them, and push each URL to VideoRoll with the configured token so the project starts auto mode immediately. +--- + +# OpenClaw -> VideoRoll + +## Overview + +Use this skill only for one job: get valid YouTube video links and send them to VideoRoll's remote API. + +## Required config + +- `VIDEOROLL_REMOTE_URL` + Full remote endpoint URL. Usually: + `https://your-host/api/remote/auto/youtube` +- `VIDEOROLL_REMOTE_TOKEN` + Token configured in VideoRoll's `Settings · API` +- Optional: `VIDEOROLL_SOURCE_LICENSE` + Default `authorized` +- Optional: `VIDEOROLL_PROOF_URL` + Authorization proof link if needed + +## Workflow + +1. Collect candidate YouTube links from the requested source. +2. Normalize every link to a canonical watch URL: + `https://www.youtube.com/watch?v=` +3. Drop non-video links, duplicates, shorts landing pages without a resolvable video id, and obviously broken URLs. +4. For each final URL, call VideoRoll remote API once. +5. Return the API result for each URL, especially: + `task_id`, `pipeline_job_id`, `deduped`, `source_id` + +## API call + +Send query parameters: + +- `token` +- `url` +- Optional `license` +- Optional `proof_url` + +Example: + +```bash +curl -G "$VIDEOROLL_REMOTE_URL" \ + --data-urlencode "token=$VIDEOROLL_REMOTE_TOKEN" \ + --data-urlencode "url=https://www.youtube.com/watch?v=dQw4w9WgXcQ" \ + --data-urlencode "license=${VIDEOROLL_SOURCE_LICENSE:-authorized}" +``` + +Success response shape: + +```json +{ + "task_id": "...", + "pipeline_job_id": "...", + "deduped": false, + "source_id": "..." +} +``` + +## Rules + +- Only submit actual YouTube video URLs. +- Preserve the original video id exactly. +- Do not invent metadata that the API does not need. +- If the API returns `deduped=true`, report it instead of retrying. +- If token/auth fails, stop and surface the exact error. +- Do not attempt to bypass paywalls, private videos, or platform restrictions. + +## Output format + +When reporting back, list each processed URL with: +- normalized URL +- API status +- `task_id` if created +- `pipeline_job_id` if started +- whether it was deduped diff --git a/skills/openclaw-videoroll/agents/openai.yaml b/skills/openclaw-videoroll/agents/openai.yaml new file mode 100644 index 0000000..370c4d6 --- /dev/null +++ b/skills/openclaw-videoroll/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "OpenClaw VideoRoll" + short_description: "Collect YouTube URLs and push them to VideoRoll" + default_prompt: "Use $openclaw-videoroll to collect YouTube watch URLs on an external machine and submit them to VideoRoll through the remote API." diff --git a/src/videoroll/apps/bilibili_publisher/worker.py b/src/videoroll/apps/bilibili_publisher/worker.py index 6651b97..16647aa 100644 --- a/src/videoroll/apps/bilibili_publisher/worker.py +++ b/src/videoroll/apps/bilibili_publisher/worker.py @@ -2,8 +2,10 @@ from __future__ import annotations import json import logging +import os import random import tempfile +import threading import traceback import uuid from datetime import datetime, timezone @@ -31,6 +33,8 @@ from videoroll.apps.subtitle_service.translate_settings_store import get_transla settings = get_bilibili_publisher_settings() logger = logging.getLogger(__name__) +_DB_READY_LOCK = threading.Lock() +_DB_READY_PID: int | None = None celery_app = Celery("bilibili_publisher", broker=settings.redis_url, backend=settings.redis_url) celery_app.conf.update( task_serializer="json", @@ -47,9 +51,17 @@ def _db() -> Session: def _ensure_db() -> None: - engine = get_engine(settings.database_url) - Base.metadata.create_all(engine) - auto_migrate(settings.database_url) + global _DB_READY_PID + pid = os.getpid() + if _DB_READY_PID == pid: + return + with _DB_READY_LOCK: + if _DB_READY_PID == pid: + return + engine = get_engine(settings.database_url) + Base.metadata.create_all(engine) + auto_migrate(settings.database_url) + _DB_READY_PID = pid def _utcnow() -> datetime: diff --git a/src/videoroll/apps/orchestrator_api/main.py b/src/videoroll/apps/orchestrator_api/main.py index f4b62b1..1c94fa8 100644 --- a/src/videoroll/apps/orchestrator_api/main.py +++ b/src/videoroll/apps/orchestrator_api/main.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import hmac import json +import logging import os import tempfile import threading @@ -18,6 +19,7 @@ from botocore.exceptions import ClientError from fastapi import Depends, FastAPI, File, HTTPException, Query, Request, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response, StreamingResponse +from starlette.concurrency import run_in_threadpool from starlette.middleware.base import BaseHTTPMiddleware from starlette.types import ASGIApp from sqlalchemy import func @@ -26,7 +28,7 @@ from sqlalchemy.orm import Session from videoroll.config import OrchestratorSettings, get_orchestrator_settings from videoroll.db.base import Base from videoroll.db.auto_migrate import auto_migrate -from videoroll.db.models import AppSetting, Asset, AssetKind, PublishJob, Subtitle, SubtitleJob, SubtitleJobStatus, Task, TaskStatus +from videoroll.db.models import AppSetting, Asset, AssetKind, PublishJob, SourceLicense, Subtitle, SubtitleJob, SubtitleJobStatus, Task, TaskStatus from videoroll.db.session import db_session, get_engine, get_sessionmaker from videoroll.storage.s3 import S3Store from videoroll.utils.internal_api_token import internal_api_token @@ -41,6 +43,8 @@ from videoroll.apps.orchestrator_api.schemas import ( ConvertedVideoItem, PublishActionRequest, PublishJobSummary, + RemoteAPISettingsRead, + RemoteAPISettingsUpdate, RemoteJobResponse, RemotePublishResponse, StorageRetentionSettingsRead, @@ -70,6 +74,7 @@ from videoroll.apps.orchestrator_api.admin_auth_store import ( verify_password_hash, ) from videoroll.apps.orchestrator_api.youtube_downloader import ( + YtDlpRuntimeError, download_youtube_video, download_thumbnail_jpg, extract_youtube_metadata, @@ -79,6 +84,14 @@ from videoroll.apps.orchestrator_api.storage_retention_store import ( get_storage_retention_settings, update_storage_retention_settings, ) +from videoroll.apps.orchestrator_api.remote_api_settings_store import ( + REMOTE_API_TOKEN_QUERY_PARAM, + REMOTE_AUTO_YOUTUBE_PATH, + get_remote_api_settings, + remote_api_token_is_configured, + update_remote_api_settings, + verify_remote_api_token, +) from videoroll.apps.subtitle_service.bilibili_tags_store import get_task_bilibili_tags from videoroll.apps.subtitle_service.task_title_store import get_task_display_title_with_s3 from videoroll.apps.youtube_settings_store import ( @@ -89,6 +102,8 @@ from videoroll.apps.youtube_settings_store import ( update_youtube_settings, ) +logger = logging.getLogger(__name__) + def get_settings() -> OrchestratorSettings: return get_orchestrator_settings() @@ -99,9 +114,7 @@ def get_db(settings: OrchestratorSettings = Depends(get_settings)) -> Generator[ def get_s3(settings: OrchestratorSettings = Depends(get_settings)) -> S3Store: - store = S3Store(settings) - store.ensure_bucket() - return store + return S3Store(settings) def _sha256_bytes(data: bytes) -> str: @@ -300,6 +313,228 @@ def _write_s3_json(s3: S3Store, key: str, obj: dict[str, Any]) -> bytes: return payload +def _write_s3_text(s3: S3Store, key: str, text: str) -> bytes: + payload = str(text).encode("utf-8") + s3.put_bytes(payload, key, content_type="text/plain; charset=utf-8") + return payload + + +def _store_task_log_asset( + db: Session, + s3: S3Store, + *, + task_id: uuid.UUID, + log_key: str, + text: str, +) -> Asset: + payload = _write_s3_text(s3, log_key, text) + asset = Asset( + task_id=task_id, + kind=AssetKind.log, + storage_key=log_key, + sha256=_sha256_bytes(payload), + size_bytes=len(payload), + ) + db.add(asset) + db.commit() + db.refresh(asset) + return asset + + +def _build_youtube_download_failure_log( + *, + task_id: uuid.UUID, + url: str, + error_message: str, + hint: str | None, + diagnostics: list[str] | None, +) -> str: + body = "\n".join([str(line or "") for line in (diagnostics or [])]).strip() + sections = [ + body or "videoroll yt-dlp diagnostics unavailable", + "\n".join( + [ + "---- videoroll error summary ----", + f"task_id={task_id}", + f"url={url}", + f"error={error_message}", + ] + ), + ] + hint_text = str(hint or "").strip() + if hint_text: + sections.append(f"---- videoroll hint ----\n{hint_text}") + return "\n\n".join(sections).rstrip() + "\n" + + +def _youtube_download_log_key(task_id: uuid.UUID) -> str: + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return f"log/{task_id}/youtube_download_{stamp}_{uuid.uuid4().hex[:8]}.log" + + +def _extract_metadata_title(raw: bytes) -> str: + if not raw: + return "" + try: + parsed = json.loads(raw.decode("utf-8")) + except Exception: + return "" + info = parsed if isinstance(parsed, dict) else {} + return str(info.get("title") or info.get("fulltitle") or info.get("alt_title") or "").strip() + + +def _task_title_key(task_id: uuid.UUID) -> str: + return f"task.title.{task_id}" + + +def _load_task_display_titles( + db: Session, + task_ids: list[uuid.UUID], + *, + s3: S3Store | None = None, + allow_s3_fallback: bool, +) -> dict[uuid.UUID, str]: + title_map: dict[uuid.UUID, str] = {} + if not task_ids: + return title_map + + rows = db.query(AppSetting).filter(AppSetting.key.in_([_task_title_key(tid) for tid in task_ids])).all() + by_key = {str(r.key): _as_dict(getattr(r, "value_json", None)) for r in rows} + for tid in task_ids: + data = by_key.get(_task_title_key(tid)) or {} + for key in ("translated_title", "source_title"): + value = data.get(key) + if isinstance(value, str) and value.strip(): + title_map[tid] = value.strip() + break + + if not allow_s3_fallback or s3 is None: + return title_map + + missing = [tid for tid in task_ids if tid not in title_map] + if not missing: + return title_map + + assets = ( + db.query(Asset) + .filter(Asset.task_id.in_(missing), Asset.kind == AssetKind.metadata_json) + .order_by(Asset.created_at.desc()) + .all() + ) + picked: dict[uuid.UUID, Asset] = {} + for asset in assets: + if asset.task_id not in picked: + picked[asset.task_id] = asset + for tid, asset in picked.items(): + try: + title = _extract_metadata_title(_read_s3_bytes(s3, asset.storage_key)) + except Exception: + continue + if title: + title_map[tid] = title + return title_map + + +def _stream_upload_to_tempfile( + file_obj: Any, + *, + prefix: str, + suffix: str, +) -> tuple[Path, str, int]: + tmp_path: Path | None = None + sha256 = hashlib.sha256() + size_bytes = 0 + try: + with tempfile.NamedTemporaryFile(prefix=prefix, suffix=suffix, delete=False) as tmp: + tmp_path = Path(tmp.name) + while True: + chunk = file_obj.read(1024 * 1024) + if not chunk: + break + if not isinstance(chunk, (bytes, bytearray)): + raise TypeError("uploaded file stream returned non-bytes content") + sha256.update(chunk) + size_bytes += len(chunk) + tmp.write(chunk) + return tmp_path, sha256.hexdigest(), size_bytes + except Exception: + if tmp_path is not None: + try: + tmp_path.unlink(missing_ok=True) + except Exception: + pass + raise + + +def _safe_unlink(path: Path | None) -> None: + if path is None: + return + try: + path.unlink(missing_ok=True) + except Exception: + pass + + +async def _store_uploaded_task_asset( + *, + task: Task, + file: UploadFile, + s3: S3Store, + db: Session, + temp_prefix: str, + default_suffix: str, + key_prefix: str, + object_name_prefix: str, + asset_kind: AssetKind, + update_task_status: TaskStatus | None = None, +) -> Asset: + suffix = Path(file.filename or "").suffix or default_suffix + tmp_path: Path | None = None + uploaded_key: str | None = None + + try: + await file.seek(0) + tmp_path, sha256, size_bytes = await run_in_threadpool( + _stream_upload_to_tempfile, + file.file, + prefix=temp_prefix, + suffix=suffix, + ) + uploaded_key = f"{key_prefix}/{task.id}/{object_name_prefix}_{sha256[:16]}{suffix}" + await run_in_threadpool(s3.upload_file, tmp_path, uploaded_key, file.content_type or None) + + asset = Asset( + task_id=task.id, + kind=asset_kind, + storage_key=uploaded_key, + sha256=sha256, + size_bytes=size_bytes, + ) + db.add(asset) + if update_task_status is not None: + task.status = update_task_status + db.add(task) + db.commit() + db.refresh(asset) + return asset + except HTTPException: + raise + except Exception as e: + db.rollback() + if uploaded_key: + try: + await run_in_threadpool(s3.delete_object, uploaded_key) + except Exception: + logger.exception("failed to roll back uploaded S3 object", extra={"storage_key": uploaded_key}) + raise HTTPException(status_code=500, detail=f"upload failed: {e}") from e + finally: + await run_in_threadpool(_safe_unlink, tmp_path) + try: + await file.close() + except Exception: + pass + + def _internal_header_token(settings: OrchestratorSettings) -> str: return internal_api_token(settings.s3_secret_access_key) @@ -334,6 +569,8 @@ class _AdminAuthMiddleware(BaseHTTPMiddleware): return await call_next(request) if path.startswith("/auth"): return await call_next(request) + if path.rstrip("/") == REMOTE_AUTO_YOUTUBE_PATH: + return await call_next(request) pw_hash = _get_admin_password_hash(request) if not pw_hash: @@ -422,48 +659,59 @@ def _cleanup_storage_once(settings: OrchestratorSettings) -> dict[str, int]: cutoff = _utcnow() - timedelta(days=ttl_days) store = S3Store(settings) - store.ensure_bucket() deleted_objects = 0 deleted_assets = 0 deleted_subtitles = 0 + failed_keys: set[str] = set() # 1) Delete fully-expired keys (no Asset rows >= cutoff). while True: - rows = ( - db.query(Asset.storage_key) - .group_by(Asset.storage_key) - .having(func.max(Asset.created_at) < cutoff) - .limit(200) - .all() - ) + query = db.query(Asset.storage_key).group_by(Asset.storage_key).having(func.max(Asset.created_at) < cutoff) + if failed_keys: + query = query.filter(~Asset.storage_key.in_(failed_keys)) + rows = query.limit(200).all() keys = [r[0] for r in rows if r and r[0]] if not keys: break + deleted_keys: list[str] = [] for key in keys: try: store.delete_object(key) deleted_objects += 1 + deleted_keys.append(key) except Exception: - # Best-effort: still delete DB rows to honor TTL. - pass + failed_keys.add(key) + logger.exception("failed to delete expired S3 object", extra={"storage_key": key}) - deleted_assets += ( - db.query(Asset) - .filter(Asset.storage_key.in_(keys)) - .delete(synchronize_session=False) - ) - deleted_subtitles += ( - db.query(Subtitle) - .filter(Subtitle.storage_key.in_(keys)) - .delete(synchronize_session=False) + if not deleted_keys: + continue + + deleted_assets += db.query(Asset).filter(Asset.storage_key.in_(deleted_keys)).delete(synchronize_session=False) + deleted_subtitles += db.query(Subtitle).filter(Subtitle.storage_key.in_(deleted_keys)).delete( + synchronize_session=False ) db.commit() - # 2) Delete old duplicate rows for keys that still exist (object kept). - deleted_assets += db.query(Asset).filter(Asset.created_at < cutoff).delete(synchronize_session=False) - deleted_subtitles += db.query(Subtitle).filter(Subtitle.created_at < cutoff).delete(synchronize_session=False) + # 2) Delete old duplicate rows only for keys that still have a newer Asset row. + active_key_subq = ( + db.query(Asset.storage_key.label("storage_key")) + .group_by(Asset.storage_key) + .having(func.max(Asset.created_at) >= cutoff) + .subquery() + ) + active_key_query = db.query(active_key_subq.c.storage_key) + deleted_assets += ( + db.query(Asset) + .filter(Asset.created_at < cutoff, Asset.storage_key.in_(active_key_query)) + .delete(synchronize_session=False) + ) + deleted_subtitles += ( + db.query(Subtitle) + .filter(Subtitle.created_at < cutoff, Subtitle.storage_key.in_(active_key_query)) + .delete(synchronize_session=False) + ) db.commit() return { @@ -481,7 +729,7 @@ def _cleanup_loop() -> None: try: _cleanup_storage_once(settings) except Exception: - pass + logger.exception("storage cleanup loop failed") _cleanup_stop.wait(timeout=max(30, _cleanup_interval_seconds)) @@ -694,22 +942,25 @@ def auth_logout(request: Request, db: Session = Depends(get_db)) -> JSONResponse return resp -@app.post("/auto/youtube", response_model=AutoYouTubeResponse) -def auto_youtube( - payload: AutoYouTubeRequest, - settings: OrchestratorSettings = Depends(get_settings), +def _start_auto_youtube_pipeline( + *, + url: str, + license: SourceLicense, + proof_url: str | None, + settings: OrchestratorSettings, ) -> AutoYouTubeResponse: - url = (payload.url or "").strip() + url = str(url or "").strip() if not url: raise HTTPException(status_code=400, detail="url is required") if not _is_youtube_url(url): raise HTTPException(status_code=400, detail="url is not a valid youtube url") + proof = str(proof_url or "").strip() or None try: with httpx.Client(timeout=30.0, headers=_internal_http_headers(settings)) as client: resp = client.post( f"{settings.youtube_ingest_url}/youtube/ingest", - json={"url": url, "license": payload.license.value, "proof_url": payload.proof_url}, + json={"url": url, "license": license.value, "proof_url": proof}, ) resp.raise_for_status() data = resp.json() @@ -734,6 +985,41 @@ def auto_youtube( ) +@app.post("/auto/youtube", response_model=AutoYouTubeResponse) +def auto_youtube( + payload: AutoYouTubeRequest, + settings: OrchestratorSettings = Depends(get_settings), +) -> AutoYouTubeResponse: + return _start_auto_youtube_pipeline( + url=payload.url, + license=payload.license, + proof_url=payload.proof_url, + settings=settings, + ) + + +@app.get(REMOTE_AUTO_YOUTUBE_PATH, response_model=AutoYouTubeResponse) +@app.post(REMOTE_AUTO_YOUTUBE_PATH, response_model=AutoYouTubeResponse) +def remote_auto_youtube( + url: str | None = Query(default=None), + token: str | None = Query(default=None, alias=REMOTE_API_TOKEN_QUERY_PARAM), + license: SourceLicense = Query(default=SourceLicense.authorized), + proof_url: str | None = Query(default=None), + settings: OrchestratorSettings = Depends(get_settings), + db: Session = Depends(get_db), +) -> AutoYouTubeResponse: + if not remote_api_token_is_configured(db): + raise HTTPException(status_code=403, detail="remote api token is not set") + if not verify_remote_api_token(db, str(token or "")): + raise HTTPException(status_code=401, detail="invalid remote api token") + return _start_auto_youtube_pipeline( + url=str(url or ""), + license=license, + proof_url=proof_url, + settings=settings, + ) + + @app.post("/tasks", response_model=TaskRead) def create_task(payload: TaskCreate, db: Session = Depends(get_db)) -> Task: task = Task( @@ -755,7 +1041,6 @@ def create_task(payload: TaskCreate, db: Session = Depends(get_db)) -> Task: def list_converted_videos( limit: int = Query(default=50, ge=1, le=200), db: Session = Depends(get_db), - s3: S3Store = Depends(get_s3), ) -> list[dict[str, Any]]: # Fetch recent video_final assets and dedupe by task_id (keep latest per task). fetch_n = min(1000, max(limit, 1) * 5) @@ -767,25 +1052,48 @@ def list_converted_videos( .all() ) - items: list[dict[str, Any]] = [] + final_assets: list[Asset] = [] seen: set[uuid.UUID] = set() - for a in assets: - if a.task_id in seen: + for asset in assets: + if asset.task_id in seen: continue - seen.add(a.task_id) - task = db.get(Task, a.task_id) + seen.add(asset.task_id) + final_assets.append(asset) + if len(final_assets) >= limit: + break + + task_ids = [asset.task_id for asset in final_assets] + if not task_ids: + return [] + + task_map = {task.id: task for task in db.query(Task).filter(Task.id.in_(task_ids)).all()} + cover_assets = ( + db.query(Asset) + .filter(Asset.task_id.in_(task_ids), Asset.kind == AssetKind.cover_image) + .order_by(Asset.created_at.desc()) + .all() + ) + cover_by_task: dict[uuid.UUID, Asset] = {} + for asset in cover_assets: + if asset.task_id not in cover_by_task: + cover_by_task[asset.task_id] = asset + + title_map = _load_task_display_titles(db, task_ids, allow_s3_fallback=False) + + items: list[dict[str, Any]] = [] + for asset in final_assets: + task = task_map.get(asset.task_id) if not task: continue - cover_asset = ( - db.query(Asset) - .filter(Asset.task_id == a.task_id, Asset.kind == AssetKind.cover_image) - .order_by(Asset.created_at.desc()) - .first() + display_title = str(title_map.get(asset.task_id) or "").strip() + items.append( + { + "task": task, + "final_asset": asset, + "cover_asset": cover_by_task.get(asset.task_id), + "display_title": display_title or None, + } ) - display_title = get_task_display_title_with_s3(db, str(a.task_id), s3=s3) - items.append({"task": task, "final_asset": a, "cover_asset": cover_asset, "display_title": display_title or None}) - if len(items) >= limit: - break return items @@ -796,55 +1104,13 @@ def list_tasks( db: Session = Depends(get_db), s3: S3Store = Depends(get_s3), ) -> list[dict[str, Any]]: - def _title_key(tid: uuid.UUID) -> str: - return f"task.title.{tid}" - q = db.query(Task).order_by(Task.created_at.desc()) if status is not None: q = q.filter(Task.status == status) tasks = q.limit(limit).all() ids = [t.id for t in tasks] - - # Batch load translated/source titles from AppSetting first, then fall back to YouTube meta title. - title_map: dict[uuid.UUID, str] = {} - if ids: - keys = [_title_key(tid) for tid in ids] - rows = db.query(AppSetting).filter(AppSetting.key.in_(keys)).all() - by_key: dict[str, dict[str, Any]] = {} - for r in rows: - by_key[str(r.key)] = _as_dict(getattr(r, "value_json", None)) - - for tid in ids: - data = by_key.get(_title_key(tid)) or {} - for k in ("translated_title", "source_title"): - v = data.get(k) - if isinstance(v, str) and v.strip(): - title_map[tid] = v.strip() - break - - missing = [tid for tid in ids if tid not in title_map] - if missing: - assets = ( - db.query(Asset) - .filter(Asset.task_id.in_(missing), Asset.kind == AssetKind.metadata_json) - .order_by(Asset.created_at.desc()) - .all() - ) - picked: dict[uuid.UUID, Asset] = {} - for a in assets: - if a.task_id not in picked: - picked[a.task_id] = a - for tid, asset in picked.items(): - try: - raw = _read_s3_bytes(s3, asset.storage_key) - parsed = json.loads(raw.decode("utf-8")) if raw else {} - info = parsed if isinstance(parsed, dict) else {} - except Exception: - continue - title = str(info.get("title") or info.get("fulltitle") or info.get("alt_title") or "").strip() - if title: - title_map[tid] = title + title_map = _load_task_display_titles(db, ids, s3=s3, allow_s3_fallback=True) out: list[dict[str, Any]] = [] for t in tasks: @@ -870,6 +1136,17 @@ def put_storage_settings(payload: StorageRetentionSettingsUpdate, db: Session = return StorageRetentionSettingsRead(**cfg) +@app.get("/settings/api", response_model=RemoteAPISettingsRead) +def get_remote_api_settings_view(db: Session = Depends(get_db)) -> RemoteAPISettingsRead: + return RemoteAPISettingsRead(**get_remote_api_settings(db)) + + +@app.put("/settings/api", response_model=RemoteAPISettingsRead) +def put_remote_api_settings_view(payload: RemoteAPISettingsUpdate, db: Session = Depends(get_db)) -> RemoteAPISettingsRead: + cfg = update_remote_api_settings(db, payload.model_dump(exclude_unset=True)) + return RemoteAPISettingsRead(**cfg) + + @app.get("/settings/youtube", response_model=YouTubeSettingsRead) def get_youtube_settings_view(settings: OrchestratorSettings = Depends(get_settings), db: Session = Depends(get_db)) -> YouTubeSettingsRead: cfg = get_youtube_settings(db, default_proxy=settings.youtube_proxy) @@ -973,39 +1250,18 @@ async def upload_task_video( task = db.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="task not found") - - suffix = Path(file.filename or "").suffix or ".mp4" - - with tempfile.NamedTemporaryFile(prefix="videoroll_", suffix=suffix, delete=False) as tmp: - tmp_path = Path(tmp.name) - while True: - chunk = await file.read(1024 * 1024) - if not chunk: - break - tmp.write(chunk) - - sha256 = sha256_file(tmp_path) - size_bytes = tmp_path.stat().st_size - key = f"raw/{task_id}/video_{sha256[:16]}{suffix}" - s3.upload_file(tmp_path, key) - try: - tmp_path.unlink(missing_ok=True) - except Exception: - pass - - asset = Asset( - task_id=task_id, - kind=AssetKind.video_raw, - storage_key=key, - sha256=sha256, - size_bytes=size_bytes, + return await _store_uploaded_task_asset( + task=task, + file=file, + s3=s3, + db=db, + temp_prefix="videoroll_", + default_suffix=".mp4", + key_prefix="raw", + object_name_prefix="video", + asset_kind=AssetKind.video_raw, + update_task_status=TaskStatus.downloaded, ) - db.add(asset) - task.status = TaskStatus.downloaded - db.add(task) - db.commit() - db.refresh(asset) - return asset @app.post("/tasks/{task_id}/upload/cover", response_model=AssetRead) @@ -1021,37 +1277,17 @@ async def upload_task_cover( if file.content_type and not file.content_type.startswith("image/"): raise HTTPException(status_code=400, detail="cover must be an image") - - suffix = Path(file.filename or "").suffix or ".jpg" - - with tempfile.NamedTemporaryFile(prefix="videoroll_cover_", suffix=suffix, delete=False) as tmp: - tmp_path = Path(tmp.name) - while True: - chunk = await file.read(1024 * 1024) - if not chunk: - break - tmp.write(chunk) - - sha256 = sha256_file(tmp_path) - size_bytes = tmp_path.stat().st_size - key = f"final/{task_id}/cover_{sha256[:16]}{suffix}" - s3.upload_file(tmp_path, key) - try: - tmp_path.unlink(missing_ok=True) - except Exception: - pass - - asset = Asset( - task_id=task_id, - kind=AssetKind.cover_image, - storage_key=key, - sha256=sha256, - size_bytes=size_bytes, + return await _store_uploaded_task_asset( + task=task, + file=file, + s3=s3, + db=db, + temp_prefix="videoroll_cover_", + default_suffix=".jpg", + key_prefix="final", + object_name_prefix="cover", + asset_kind=AssetKind.cover_image, ) - db.add(asset) - db.commit() - db.refresh(asset) - return asset @app.get("/tasks/{task_id}/assets", response_model=list[AssetRead]) @@ -1376,6 +1612,24 @@ def download_youtube( video_path, info, meta = download_youtube_video(url, yt_settings, work_dir=tmp_dir) except Exception as e: hint = _youtube_bot_check_hint(str(e), yt_settings=yt_settings, db=db) + diagnostics = e.diagnostics if isinstance(e, YtDlpRuntimeError) else [] + try: + _store_task_log_asset( + db, + s3, + task_id=task_id, + log_key=_youtube_download_log_key(task_id), + text=_build_youtube_download_failure_log( + task_id=task_id, + url=url, + error_message=str(e), + hint=hint, + diagnostics=diagnostics, + ), + ) + except Exception: + db.rollback() + logger.exception("failed to persist youtube download diagnostics log for task %s", task_id) detail = f"youtube download failed: {e}" if hint: detail = f"{detail}\n\n{hint}" diff --git a/src/videoroll/apps/orchestrator_api/remote_api_settings_store.py b/src/videoroll/apps/orchestrator_api/remote_api_settings_store.py new file mode 100644 index 0000000..b57b06f --- /dev/null +++ b/src/videoroll/apps/orchestrator_api/remote_api_settings_store.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from videoroll.apps.orchestrator_api.admin_auth_store import encode_password_hash, verify_password_hash +from videoroll.db.models import AppSetting + + +REMOTE_API_SETTINGS_KEY = "remote.api" + +REMOTE_AUTO_YOUTUBE_PATH = "/remote/auto/youtube" +REMOTE_API_TOKEN_QUERY_PARAM = "token" +REMOTE_AUTO_URL_QUERY_PARAM = "url" +REMOTE_AUTO_LICENSE_QUERY_PARAM = "license" +REMOTE_AUTO_PROOF_URL_QUERY_PARAM = "proof_url" + +_MIN_TOKEN_LEN = 8 +_MAX_TOKEN_LEN = 512 + + +def _as_dict(v: Any) -> dict[str, Any]: + return v if isinstance(v, dict) else {} + + +def _get_row(db: Session) -> AppSetting: + row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY) + if row: + return row + row = AppSetting(key=REMOTE_API_SETTINGS_KEY, value_json={}) + db.add(row) + try: + db.commit() + db.refresh(row) + except IntegrityError: + db.rollback() + row2 = db.get(AppSetting, REMOTE_API_SETTINGS_KEY) + if row2: + return row2 + raise + return row + + +def _normalize_token(token: Any) -> str: + value = str(token or "").strip() + if not value: + return "" + if len(value) < _MIN_TOKEN_LEN: + raise ValueError(f"token too short (min {_MIN_TOKEN_LEN} chars)") + if len(value) > _MAX_TOKEN_LEN: + raise ValueError(f"token too long (max {_MAX_TOKEN_LEN} chars)") + return value + + +def get_remote_api_settings(db: Session) -> dict[str, Any]: + row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY) + stored = dict(_as_dict(row.value_json)) if row else {} + token_hash = str(stored.get("token_hash") or "").strip() + token_updated_at = str(stored.get("token_updated_at") or "").strip() or None + return { + "token_set": bool(token_hash), + "token_updated_at": token_updated_at, + "endpoint_path": REMOTE_AUTO_YOUTUBE_PATH, + "token_query_param": REMOTE_API_TOKEN_QUERY_PARAM, + "url_query_param": REMOTE_AUTO_URL_QUERY_PARAM, + "license_query_param": REMOTE_AUTO_LICENSE_QUERY_PARAM, + "proof_url_query_param": REMOTE_AUTO_PROOF_URL_QUERY_PARAM, + } + + +def update_remote_api_settings(db: Session, update: dict[str, Any]) -> dict[str, Any]: + row = _get_row(db) + stored = dict(_as_dict(row.value_json)) + + if "token" in update: + token_raw = update.get("token") + if token_raw is not None: + token = _normalize_token(token_raw) + if token: + stored["token_hash"] = encode_password_hash(token) + stored["token_updated_at"] = datetime.now(timezone.utc).isoformat() + else: + stored.pop("token_hash", None) + stored.pop("token_updated_at", None) + + row.value_json = stored + db.add(row) + db.commit() + + return get_remote_api_settings(db) + + +def remote_api_token_is_configured(db: Session) -> bool: + row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY) + stored = dict(_as_dict(row.value_json)) if row else {} + return bool(str(stored.get("token_hash") or "").strip()) + + +def verify_remote_api_token(db: Session, token: str) -> bool: + value = str(token or "").strip() + if not value: + return False + row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY) + stored = dict(_as_dict(row.value_json)) if row else {} + token_hash = str(stored.get("token_hash") or "").strip() + if not token_hash: + return False + return verify_password_hash(value, token_hash) diff --git a/src/videoroll/apps/orchestrator_api/schemas.py b/src/videoroll/apps/orchestrator_api/schemas.py index 370f030..7c18d94 100644 --- a/src/videoroll/apps/orchestrator_api/schemas.py +++ b/src/videoroll/apps/orchestrator_api/schemas.py @@ -188,6 +188,20 @@ class StorageRetentionSettingsUpdate(BaseModel): asset_ttl_days: Optional[int] = None +class RemoteAPISettingsRead(BaseModel): + token_set: bool = False + token_updated_at: Optional[str] = None + endpoint_path: str = "/remote/auto/youtube" + token_query_param: str = "token" + url_query_param: str = "url" + license_query_param: str = "license" + proof_url_query_param: str = "proof_url" + + +class RemoteAPISettingsUpdate(BaseModel): + token: Optional[str] = None + + class YouTubeSettingsRead(BaseModel): proxy: str = "" cookies_set: bool = False diff --git a/src/videoroll/apps/orchestrator_api/youtube_downloader.py b/src/videoroll/apps/orchestrator_api/youtube_downloader.py index 4d96309..e2aad0f 100644 --- a/src/videoroll/apps/orchestrator_api/youtube_downloader.py +++ b/src/videoroll/apps/orchestrator_api/youtube_downloader.py @@ -6,7 +6,7 @@ import subprocess from dataclasses import dataclass from pathlib import Path from typing import Any, Optional -from urllib.parse import urlparse +from urllib.parse import urlparse, urlsplit, urlunsplit import httpx import yt_dlp @@ -25,6 +25,12 @@ class YouTubeMeta: duration: Optional[int] = None +class YtDlpRuntimeError(RuntimeError): + def __init__(self, message: str, *, diagnostics: list[str] | None = None) -> None: + super().__init__(message) + self.diagnostics = diagnostics or [] + + def _run(cmd: list[str]) -> None: subprocess.run(cmd, check=True) @@ -66,10 +72,85 @@ def _extractor_args(settings: OrchestratorSettings) -> dict[str, Any] | None: return parsed -def build_ydl_opts(settings: OrchestratorSettings, *, outtmpl: str | None = None, for_download: bool) -> dict[str, Any]: +def _redact_url(value: str) -> str: + raw = str(value or "").strip() + if not raw: + return "" + try: + parts = urlsplit(raw) + except Exception: + return raw + hostname = parts.hostname or "" + if not hostname: + return raw + auth = "" + if parts.username: + auth = parts.username + if parts.password is not None: + auth += ":***" + auth += "@" + netloc = auth + hostname + if parts.port is not None: + netloc += f":{parts.port}" + return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment)) + + +def _sanitize_diagnostic_line(line: str, *, proxy: str | None) -> str: + text = str(line or "").rstrip() + if proxy: + text = text.replace(proxy, _redact_url(proxy)) + return text + + +def _yt_dlp_diagnostics_header( + *, + url: str, + settings: OrchestratorSettings, + outtmpl: str | None, +) -> list[str]: + extractor_args = _extractor_args(settings) + return [ + "videoroll yt-dlp diagnostics", + f"url={url}", + f"proxy={_redact_url(str(settings.youtube_proxy or '').strip()) or '(none)'}", + f"cookiefile={(str(settings.youtube_cookie_file or '').strip() or '(none)')}", + f"ffmpeg_location={(str(settings.ffmpeg_path or '').strip() or '(default PATH)')}", + f"outtmpl={outtmpl or '(none)'}", + f"extractor_args={json.dumps(extractor_args, ensure_ascii=False, sort_keys=True) if extractor_args else '(none)'}", + "---- yt-dlp verbose ----", + ] + + +class _DiagnosticsLogger: + def __init__(self, diagnostics: list[str], *, proxy: str | None) -> None: + self._diagnostics = diagnostics + self._proxy = str(proxy or "").strip() or None + + def debug(self, msg: str) -> None: + self._diagnostics.append(_sanitize_diagnostic_line(f"[debug] {msg}", proxy=self._proxy)) + + def info(self, msg: str) -> None: + self._diagnostics.append(_sanitize_diagnostic_line(str(msg), proxy=self._proxy)) + + def warning(self, msg: str) -> None: + self._diagnostics.append(_sanitize_diagnostic_line(f"WARNING: {msg}", proxy=self._proxy)) + + def error(self, msg: str) -> None: + self._diagnostics.append(_sanitize_diagnostic_line(f"ERROR: {msg}", proxy=self._proxy)) + + +def build_ydl_opts( + settings: OrchestratorSettings, + *, + outtmpl: str | None = None, + for_download: bool, + logger: Any | None = None, + verbose: bool = False, +) -> dict[str, Any]: opts: dict[str, Any] = { "quiet": True, "no_warnings": True, + "ignoreconfig": True, "noplaylist": True, "retries": 3, "fragment_retries": 3, @@ -103,6 +184,10 @@ def build_ydl_opts(settings: OrchestratorSettings, *, outtmpl: str | None = None if outtmpl: opts["outtmpl"] = outtmpl + if logger is not None: + opts["logger"] = logger + if verbose: + opts["verbose"] = True return opts @@ -284,10 +369,20 @@ def _resolve_downloaded_file(info: dict[str, Any], *, work_dir: Path) -> Path: raise RuntimeError("yt-dlp download succeeded but no output file was found") -def _download_once(url: str, settings: OrchestratorSettings, *, work_dir: Path) -> tuple[Path, dict[str, Any], YouTubeMeta]: +def _download_once( + url: str, + settings: OrchestratorSettings, + *, + work_dir: Path, + diagnostics: list[str] | None = None, +) -> tuple[Path, dict[str, Any], YouTubeMeta]: work_dir.mkdir(parents=True, exist_ok=True) outtmpl = str(work_dir / "%(id)s.%(ext)s") - with yt_dlp.YoutubeDL(build_ydl_opts(settings, outtmpl=outtmpl, for_download=True)) as ydl: + ydl_logger = None + if diagnostics is not None: + diagnostics.extend(_yt_dlp_diagnostics_header(url=url, settings=settings, outtmpl=outtmpl)) + ydl_logger = _DiagnosticsLogger(diagnostics, proxy=settings.youtube_proxy) + with yt_dlp.YoutubeDL(build_ydl_opts(settings, outtmpl=outtmpl, for_download=True, logger=ydl_logger, verbose=diagnostics is not None)) as ydl: info_raw = ydl.extract_info(url, download=True) info = _pick_video_info(info_raw) if not info: @@ -306,9 +401,10 @@ def download_youtube_video(url: str, settings: OrchestratorSettings, *, work_dir - sanitized info dict - meta summary """ + diagnostics: list[str] = [] try: - return _download_once(url, settings, work_dir=work_dir) + return _download_once(url, settings, work_dir=work_dir, diagnostics=diagnostics) except DownloadError as e: - raise RuntimeError(str(e)) from e + raise YtDlpRuntimeError(str(e), diagnostics=diagnostics) from e except Exception as e: - raise RuntimeError(str(e)) from e + raise YtDlpRuntimeError(str(e), diagnostics=diagnostics) from e diff --git a/src/videoroll/apps/subtitle_service/main.py b/src/videoroll/apps/subtitle_service/main.py index 66699d9..da418e8 100644 --- a/src/videoroll/apps/subtitle_service/main.py +++ b/src/videoroll/apps/subtitle_service/main.py @@ -632,8 +632,29 @@ def _read_task_queue(db: Session, *, limit: int) -> TaskQueueRead: ) unlocked = (Task.lock_owner != TASK_QUEUE_LOCK_OWNER) | (Task.lock_until.is_(None)) | (Task.lock_until <= now) - # queued_count: distinct tasks with queued subtitle/render jobs (best-effort, capped). - queued_task_ids: set[uuid.UUID] = set() + orphaned_render_by_task: dict[uuid.UUID, RenderJob] = {} + orphaned_subtitle_by_task: dict[uuid.UUID, SubtitleJob] = {} + for rj in ( + db.query(RenderJob) + .join(Task, Task.id == RenderJob.task_id) + .filter(RenderJob.status == RenderJobStatus.running, unlocked) + .order_by(RenderJob.updated_at.asc(), RenderJob.created_at.asc()) + .limit(5000) + .all() + ): + orphaned_render_by_task.setdefault(rj.task_id, rj) + for sj in ( + db.query(SubtitleJob) + .join(Task, Task.id == SubtitleJob.task_id) + .filter(SubtitleJob.status == SubtitleJobStatus.running, unlocked) + .order_by(SubtitleJob.updated_at.asc(), SubtitleJob.created_at.asc()) + .limit(5000) + .all() + ): + orphaned_subtitle_by_task.setdefault(sj.task_id, sj) + + # queued_count: distinct tasks with queued jobs, plus orphaned running jobs that lost their task lock. + queued_task_ids: set[uuid.UUID] = set(orphaned_render_by_task) | set(orphaned_subtitle_by_task) for tid, in ( db.query(SubtitleJob.task_id) .join(Task, Task.id == SubtitleJob.task_id) @@ -659,33 +680,74 @@ def _read_task_queue(db: Session, *, limit: int) -> TaskQueueRead: queued_items: list[TaskQueueItemRead] = [] seen: set[uuid.UUID] = set() - fetch_n = min(5000, max(50, remaining * 20)) - for sj in ( - db.query(SubtitleJob) - .join(Task, Task.id == SubtitleJob.task_id) - .filter(SubtitleJob.status == SubtitleJobStatus.queued, unlocked) - .order_by(SubtitleJob.created_at.asc()) - .limit(fetch_n) - .all() - ): - if sj.task_id in seen: + for rj in orphaned_render_by_task.values(): + if rj.task_id in seen: continue - seen.add(sj.task_id) + seen.add(rj.task_id) queued_items.append( TaskQueueItemRead( - task_id=sj.task_id, + task_id=rj.task_id, state="queued", - stage="subtitle", - subtitle_job_id=sj.id, - progress=int(sj.progress or 0), - error_message=sj.error_message, - created_at=sj.created_at, - updated_at=sj.updated_at, + stage="recover_render", + render_job_id=rj.id, + subtitle_job_id=rj.subtitle_job_id, + progress=int(rj.progress or 0), + error_message=rj.error_message, + created_at=rj.created_at, + updated_at=rj.updated_at, ) ) if len(queued_items) >= remaining: break + if len(queued_items) < remaining: + for sj in orphaned_subtitle_by_task.values(): + if sj.task_id in seen: + continue + seen.add(sj.task_id) + queued_items.append( + TaskQueueItemRead( + task_id=sj.task_id, + state="queued", + stage="recover_subtitle", + subtitle_job_id=sj.id, + progress=int(sj.progress or 0), + error_message=sj.error_message, + created_at=sj.created_at, + updated_at=sj.updated_at, + ) + ) + if len(queued_items) >= remaining: + break + + fetch_n = min(5000, max(50, remaining * 20)) + if len(queued_items) < remaining: + for sj in ( + db.query(SubtitleJob) + .join(Task, Task.id == SubtitleJob.task_id) + .filter(SubtitleJob.status == SubtitleJobStatus.queued, unlocked) + .order_by(SubtitleJob.created_at.asc()) + .limit(fetch_n) + .all() + ): + if sj.task_id in seen: + continue + seen.add(sj.task_id) + queued_items.append( + TaskQueueItemRead( + task_id=sj.task_id, + state="queued", + stage="subtitle", + subtitle_job_id=sj.id, + progress=int(sj.progress or 0), + error_message=sj.error_message, + created_at=sj.created_at, + updated_at=sj.updated_at, + ) + ) + if len(queued_items) >= remaining: + break + if len(queued_items) < remaining: for rj in ( db.query(RenderJob) diff --git a/src/videoroll/apps/subtitle_service/worker.py b/src/videoroll/apps/subtitle_service/worker.py index 04feec7..6112909 100644 --- a/src/videoroll/apps/subtitle_service/worker.py +++ b/src/videoroll/apps/subtitle_service/worker.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import logging +import os import re import shutil import threading @@ -70,6 +71,8 @@ settings = get_subtitle_settings() _ORCH_INTERNAL_TOKEN = internal_api_token(settings.s3_secret_access_key) _ORCH_INTERNAL_HEADERS = {INTERNAL_TOKEN_HEADER: _ORCH_INTERNAL_TOKEN} if _ORCH_INTERNAL_TOKEN else {} logger = logging.getLogger(__name__) +_DB_READY_LOCK = threading.Lock() +_DB_READY_PID: int | None = None celery_app = Celery("subtitle_service", broker=settings.redis_url, backend=settings.redis_url) celery_app.conf.update( task_serializer="json", @@ -144,9 +147,17 @@ def _db() -> Session: def _ensure_db() -> None: - engine = get_engine(settings.database_url) - Base.metadata.create_all(engine) - auto_migrate(settings.database_url) + global _DB_READY_PID + pid = os.getpid() + if _DB_READY_PID == pid: + return + with _DB_READY_LOCK: + if _DB_READY_PID == pid: + return + engine = get_engine(settings.database_url) + Base.metadata.create_all(engine) + auto_migrate(settings.database_url) + _DB_READY_PID = pid def _now() -> datetime: @@ -173,6 +184,18 @@ def _task_queue_unlock(task: Task) -> None: task.lock_until = None +def _task_queue_join_message(message: str | None, detail: str, *, limit: int = 2000) -> str: + head = str(message or "").strip() + tail = str(detail or "").strip() + if head and tail: + out = f"{head}\n{tail}" + else: + out = head or tail + if len(out) > limit: + out = out[: limit - 1] + "…" + return out + + def _task_queue_lock_settings_row(db: Session) -> None: """ Serialize queue ticks by locking the settings row. @@ -1178,6 +1201,53 @@ def task_queue_tick() -> dict[str, Any]: except Exception: unlocked_expired = 0 + unlocked = or_( + Task.lock_owner != TASK_QUEUE_LOCK_OWNER, + Task.lock_until.is_(None), + Task.lock_until <= now, + ) + + # Recover jobs that were actively running but lost their task slot/heartbeat. + # This is the "ghost running" case: task disappears from the queue while the job row + # still says running forever. + orphaned_subtitle = ( + db.query(SubtitleJob) + .join(Task, Task.id == SubtitleJob.task_id) + .filter(SubtitleJob.status == SubtitleJobStatus.running, unlocked) + .order_by(SubtitleJob.updated_at.asc(), SubtitleJob.created_at.asc()) + .with_for_update(skip_locked=True) + .all() + ) + for j in orphaned_subtitle: + detail = "检测到任务队列锁已丢失：subtitle job 已自动按 resume 方式重新排队。" + req = dict(j.request_json) if isinstance(j.request_json, dict) else {} + req["resume"] = True + j.request_json = req + j.error_message = _task_queue_join_message(j.error_message, detail) + j.status = SubtitleJobStatus.queued + j.progress = 0 + db.add(j) + recovered_subtitle += 1 + + orphaned_render = ( + db.query(RenderJob) + .join(Task, Task.id == RenderJob.task_id) + .filter(RenderJob.status == RenderJobStatus.running, unlocked) + .order_by(RenderJob.updated_at.asc(), RenderJob.created_at.asc()) + .with_for_update(skip_locked=True) + .all() + ) + for j in orphaned_render: + detail = "检测到任务队列锁已丢失：render job 已自动重新排队。" + j.error_message = _task_queue_join_message(j.error_message, detail) + j.retry_count = int(j.retry_count or 0) + 1 + j.status = RenderJobStatus.queued + j.progress = 0 + j.started_at = None + j.finished_at = None + db.add(j) + recovered_render += 1 + # Recover render jobs that were marked as running by the scheduler but never claimed by a worker. cutoff = now - timedelta(seconds=60) # Subtitle jobs: scheduler sets progress=1; worker bumps it to >=2 once claimed. diff --git a/src/videoroll/db/session.py b/src/videoroll/db/session.py index 22550b4..42c4c72 100644 --- a/src/videoroll/db/session.py +++ b/src/videoroll/db/session.py @@ -4,13 +4,25 @@ from functools import lru_cache from typing import Generator from sqlalchemy import create_engine -from sqlalchemy.engine import Engine +from sqlalchemy.engine import Engine, make_url from sqlalchemy.orm import Session, sessionmaker +def _engine_connect_args(database_url: str) -> dict[str, object]: + try: + url = make_url(database_url) + except Exception: + return {} + if url.get_backend_name() == "postgresql" and url.get_driver_name() == "psycopg": + # PgBouncer/transaction-pooling setups can fail with + # "prepared statement already exists" when psycopg auto-prepares queries. + return {"prepare_threshold": None} + return {} + + @lru_cache def get_engine(database_url: str) -> Engine: - return create_engine(database_url, pool_pre_ping=True) + return create_engine(database_url, pool_pre_ping=True, connect_args=_engine_connect_args(database_url)) def get_sessionmaker(database_url: str) -> sessionmaker[Session]: @@ -25,4 +37,3 @@ def db_session(database_url: str) -> Generator[Session, None, None]: yield db finally: db.close() - diff --git a/src/videoroll/storage/s3.py b/src/videoroll/storage/s3.py index e33ca73..47f2d92 100644 --- a/src/videoroll/storage/s3.py +++ b/src/videoroll/storage/s3.py @@ -42,11 +42,31 @@ class S3Store: def bucket(self) -> str: return self._bucket + @staticmethod + def _is_missing_bucket_error(exc: ClientError) -> bool: + err = exc.response.get("Error") or {} + code = str(err.get("Code") or "").strip() + status = int((exc.response.get("ResponseMetadata") or {}).get("HTTPStatusCode") or 0) + return code in {"404", "NoSuchBucket", "NotFound"} or status == 404 + + @staticmethod + def _is_bucket_already_exists_error(exc: ClientError) -> bool: + err = exc.response.get("Error") or {} + code = str(err.get("Code") or "").strip() + return code in {"BucketAlreadyOwnedByYou", "BucketAlreadyExists"} + def ensure_bucket(self) -> None: try: self._client.head_bucket(Bucket=self._bucket) - except ClientError: - self._client.create_bucket(Bucket=self._bucket) + except ClientError as e: + if not self._is_missing_bucket_error(e): + raise + try: + self._client.create_bucket(Bucket=self._bucket) + except ClientError as create_error: + if self._is_bucket_already_exists_error(create_error): + return + raise def upload_file(self, path: Path, key: str, content_type: Optional[str] = None) -> PutResult: if content_type is None: diff --git a/src/web/src/App.tsx b/src/web/src/App.tsx index a6c682c..8211dcf 100644 --- a/src/web/src/App.tsx +++ b/src/web/src/App.tsx @@ -12,6 +12,7 @@ import YouTubeSourcesPage from "./pages/YouTubeSourcesPage"; import SettingsASRPage from "./pages/SettingsASRPage"; import SettingsYouTubePage from "./pages/SettingsYouTubePage"; import SettingsStoragePage from "./pages/SettingsStoragePage"; +import SettingsApiPage from "./pages/SettingsApiPage"; import SettingsTranslatePage from "./pages/SettingsTranslatePage"; import SettingsBilibiliPage from "./pages/SettingsBilibiliPage"; import SettingsAutoPage from "./pages/SettingsAutoPage"; @@ -97,6 +98,7 @@ export default function App() { + @@ -118,6 +120,7 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/src/web/src/lib/http.ts b/src/web/src/lib/http.ts index 4cc1569..7ab19e3 100644 --- a/src/web/src/lib/http.ts +++ b/src/web/src/lib/http.ts @@ -2,7 +2,18 @@ export async function fetchJson(url: string, init?: RequestInit): Promise const resp = await fetch(url, { credentials: "include", ...init }); if (!resp.ok) { const text = await resp.text().catch(() => ""); - throw new Error(`${resp.status} ${resp.statusText}${text ? ` - ${text}` : ""}`); + let detail = text; + if (text) { + try { + const parsed = JSON.parse(text) as { detail?: unknown; message?: unknown }; + if (typeof parsed.detail === "string" && parsed.detail.trim()) { + detail = parsed.detail.trim(); + } else if (typeof parsed.message === "string" && parsed.message.trim()) { + detail = parsed.message.trim(); + } + } catch {} + } + throw new Error(`${resp.status} ${resp.statusText}${detail ? ` - ${detail}` : ""}`); } return (await resp.json()) as T; } diff --git a/src/web/src/pages/SettingsApiPage.tsx b/src/web/src/pages/SettingsApiPage.tsx new file mode 100644 index 0000000..a1124b5 --- /dev/null +++ b/src/web/src/pages/SettingsApiPage.tsx @@ -0,0 +1,193 @@ +import { useEffect, useMemo, useState } from "react"; +import { fetchJson } from "../lib/http"; +import { ORCHESTRATOR_URL } from "../lib/urls"; + +type RemoteApiSettings = { + token_set: boolean; + token_updated_at?: string | null; + endpoint_path: string; + token_query_param: string; + url_query_param: string; + license_query_param: string; + proof_url_query_param: string; +}; + +const DEFAULT_SETTINGS: RemoteApiSettings = { + token_set: false, + token_updated_at: null, + endpoint_path: "/remote/auto/youtube", + token_query_param: "token", + url_query_param: "url", + license_query_param: "license", + proof_url_query_param: "proof_url", +}; + +function orchestratorBaseUrl(): string { + const raw = ORCHESTRATOR_URL.replace(/\/+$/, ""); + if (raw.startsWith("http://") || raw.startsWith("https://")) return raw; + if (typeof window === "undefined") return raw; + return new URL(raw || "/api", window.location.origin).toString().replace(/\/+$/, ""); +} + +export default function SettingsApiPage() { + const [settings, setSettings] = useState(null); + const [tokenInput, setTokenInput] = useState(""); + const [error, setError] = useState(null); + const [busy, setBusy] = useState(false); + + async function refresh() { + setError(null); + try { + const s = await fetchJson(`${ORCHESTRATOR_URL}/settings/api`); + setSettings(s); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : String(e)); + } + } + + useEffect(() => { + refresh(); + }, []); + + const effective = settings ?? DEFAULT_SETTINGS; + const baseUrl = useMemo(() => orchestratorBaseUrl(), []); + const remoteEndpoint = useMemo(() => `${baseUrl}${effective.endpoint_path}`, [baseUrl, effective.endpoint_path]); + const sampleUrl = useMemo(() => { + const qs = new URLSearchParams(); + qs.set(effective.token_query_param, tokenInput.trim() || "YOUR_TOKEN"); + qs.set(effective.url_query_param, "https://www.youtube.com/watch?v=dQw4w9WgXcQ"); + qs.set(effective.license_query_param, "authorized"); + return `${remoteEndpoint}?${qs.toString()}`; + }, [effective.license_query_param, effective.token_query_param, effective.url_query_param, remoteEndpoint, tokenInput]); + const sampleCurl = useMemo( + () => + [ + `curl -G "${remoteEndpoint}" \\`, + ` --data-urlencode "${effective.token_query_param}=${tokenInput.trim() || "YOUR_TOKEN"}" \\`, + ` --data-urlencode "${effective.url_query_param}=https://www.youtube.com/watch?v=dQw4w9WgXcQ" \\`, + ` --data-urlencode "${effective.license_query_param}=authorized"`, + ].join("\n"), + [effective.license_query_param, effective.token_query_param, effective.url_query_param, remoteEndpoint, tokenInput], + ); + + return ( +
+
+
Settings · API
+
配置远程管理 token。外部请求带上 token 和 YouTube `url` 参数后，会直接按 Auto Mode 自动创建任务并开始处理。
+ {error ?
{error}
: null} +
+ +
+
+
当前配置
+ +
+ {!settings ? ( +
加载中…
+ ) : ( +
+
+
remote token
+
{settings.token_set ? "已设置" : "未设置"}
+
+
+
token_updated_at
+
{settings.token_updated_at || "-"}
+
+
+
remote endpoint
+
{remoteEndpoint}
+
+
+ )} +
说明：当前远程入口仅支持 YouTube 视频链接，行为等价于网页里的 “YouTube 自动模式”。
+
+ +
+
保存 token
+
+ +
+ +
+ + +
+ +
提示：保存后不会回显明文 token。请自行保管；需要更换时直接输入新 token 覆盖即可。
+
+ +
+
调用示例
+
+ 必填参数：{effective.token_query_param}、{effective.url_query_param} + 。可选参数：{effective.license_query_param}、{effective.proof_url_query_param}。 +
+
+
示例 URL
+ 
{sampleUrl}
+
+
+
curl
+ 
{sampleCurl}
+
+
+
+ ); +} diff --git a/src/web/src/pages/TaskDetailPage.tsx b/src/web/src/pages/TaskDetailPage.tsx index 396ab84..6d36456 100644 --- a/src/web/src/pages/TaskDetailPage.tsx +++ b/src/web/src/pages/TaskDetailPage.tsx @@ -427,8 +427,13 @@ export default function TaskDetailPage() { const finalAssets = useMemo(() => (assets ?? []).filter((x) => x.kind === "video_final"), [assets]); const coverAssets = useMemo(() => (assets ?? []).filter((x) => x.kind === "cover_image"), [assets]); const logAssets = useMemo(() => (assets ?? []).filter((x) => x.kind === "log"), [assets]); + const youtubeDownloadLogAssets = useMemo(() => logAssets.filter((x) => x.storage_key.includes("/youtube_download_")), [logAssets]); const subtitleLogAssets = useMemo(() => logAssets.filter((x) => x.storage_key.includes("/subtitle_")), [logAssets]); const renderLogAssets = useMemo(() => logAssets.filter((x) => x.storage_key.includes("/render_")), [logAssets]); + const latestYouTubeDownloadLog = useMemo( + () => (youtubeDownloadLogAssets.length ? youtubeDownloadLogAssets[youtubeDownloadLogAssets.length - 1] : null), + [youtubeDownloadLogAssets], + ); const latestSubtitleLog = useMemo( () => (subtitleLogAssets.length ? subtitleLogAssets[subtitleLogAssets.length - 1] : null), [subtitleLogAssets], @@ -541,6 +546,10 @@ export default function TaskDetailPage() { await refresh(); alert(`${opts.resume ? "已继续字幕任务" : "已提交字幕任务"}：${resp.job_id}`); } catch (e: unknown) { + try { + await refresh({ silent: true }); + await loadLogs({ silent: true }); + } catch {} setError(e instanceof Error ? e.message : String(e)); } finally { setBusy(false); @@ -578,6 +587,9 @@ export default function TaskDetailPage() { const maxBytes = 200_000; if (logSelection === "combined") { const parts: Array<{ title: string; asset: Asset }> = []; + if (latestYouTubeDownloadLog) { + parts.push({ title: `YouTube Download · ${latestYouTubeDownloadLog.storage_key}`, asset: latestYouTubeDownloadLog }); + } if (latestSubtitleLog) parts.push({ title: `Subtitle · ${latestSubtitleLog.storage_key}`, asset: latestSubtitleLog }); if (latestRenderLog) parts.push({ title: `Render · ${latestRenderLog.storage_key}`, asset: latestRenderLog }); if (!parts.length) { @@ -605,7 +617,7 @@ export default function TaskDetailPage() { if (!silent) setLogBusy(false); } }, - [taskId, logSelection, latestSubtitleLog, latestRenderLog, selectedLogAsset], + [taskId, logSelection, latestYouTubeDownloadLog, latestSubtitleLog, latestRenderLog, selectedLogAsset], ); useEffect(() => { @@ -679,7 +691,7 @@ export default function TaskDetailPage() { - {error ?
{error}
: null} + {error ?
{error}
: null} {!task ?
加载中…
: null} {task ? (
@@ -748,6 +760,10 @@ export default function TaskDetailPage() { } await refresh(); } catch (e: unknown) { + try { + await refresh({ silent: true }); + await loadLogs({ silent: true }); + } catch {} setError(e instanceof Error ? e.message : String(e)); } finally { setBusy(false); @@ -996,7 +1012,7 @@ export default function TaskDetailPage() {