diff --git a/.gitignore b/.gitignore index 38f6d4f..6710e89 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,10 @@ frontend/dist/ *.sql.gz backups/ +# Download center: locally downloaded media (the DOWNLOAD_ROOT bind mount at the repo root). +# Anchored with a leading slash so it does NOT also ignore the backend/app/downloads package. +/downloads/ + # OS / editor .DS_Store Thumbs.db diff --git a/Dockerfile b/Dockerfile index d5aa7fd..08f9839 100644 --- a/Dockerfile +++ b/Dockerfile @@ -34,6 +34,18 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ WORKDIR /app +# ffmpeg: merge separate video+audio streams + postprocessing. Deno: the JavaScript runtime +# yt-dlp uses to solve YouTube's "n" signature challenge for the web player clients (required +# alongside the PO token, else only storyboards are offered). yt-dlp auto-detects deno on PATH. +ARG DENO_VERSION=v2.9.1 +RUN apt-get update \ + && apt-get install -y --no-install-recommends ffmpeg unzip curl ca-certificates \ + && curl -fsSL "https://github.com/denoland/deno/releases/download/${DENO_VERSION}/deno-x86_64-unknown-linux-gnu.zip" -o /tmp/deno.zip \ + && unzip -o /tmp/deno.zip -d /usr/local/bin \ + && chmod +x /usr/local/bin/deno \ + && rm /tmp/deno.zip \ + && rm -rf /var/lib/apt/lists/* + COPY backend/requirements.txt . RUN pip install --upgrade pip && pip install -r requirements.txt diff --git a/backend/alembic/versions/0036_download_center.py b/backend/alembic/versions/0036_download_center.py new file mode 100644 index 0000000..b9f0677 --- /dev/null +++ b/backend/alembic/versions/0036_download_center.py @@ -0,0 +1,143 @@ +"""download center core schema (yt-dlp) + +Revision ID: 0036_download_center +Revises: 0035_saved_views +Create Date: 2026-07-02 + +The download center's data layer. `media_assets` is the shared file cache (one row per unique +source+format signature); `download_jobs` is the per-user request over it (worker-claimed via +FOR UPDATE SKIP LOCKED, live progress written back for polling); `download_profiles` are the +reusable format presets (builtins seeded in the next migration); `download_shares` grant a file +to another user; `download_quotas` hold admin-set per-user limits (default via sysconfig). +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0036_download_center" +down_revision: Union[str, None] = "0035_saved_views" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "media_assets", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("source_kind", sa.String(length=16), nullable=False), + sa.Column("source_ref", sa.String(length=512), nullable=False), + sa.Column("format_sig", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=16), server_default="pending", nullable=False), + sa.Column("rel_path", sa.Text(), nullable=True), + sa.Column("storyboard_path", sa.Text(), nullable=True), + sa.Column("size_bytes", sa.BigInteger(), nullable=True), + sa.Column("duration_s", sa.Integer(), nullable=True), + sa.Column("width", sa.Integer(), nullable=True), + sa.Column("height", sa.Integer(), nullable=True), + sa.Column("container", sa.String(length=16), nullable=True), + sa.Column("vcodec", sa.String(length=32), nullable=True), + sa.Column("acodec", sa.String(length=32), nullable=True), + sa.Column("nfo_written", sa.Boolean(), server_default="false", nullable=False), + sa.Column("title", sa.Text(), nullable=True), + sa.Column("uploader", sa.String(length=255), nullable=True), + sa.Column("upload_date", sa.Date(), nullable=True), + sa.Column("thumbnail_url", sa.Text(), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("ref_count", sa.Integer(), server_default="0", nullable=False), + sa.Column("last_access_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("source_kind", "source_ref", "format_sig", name="uq_media_asset_cache_key"), + ) + op.create_index("ix_media_assets_status", "media_assets", ["status"]) + + op.create_table( + "download_profiles", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("user_id", sa.Integer(), nullable=True), + sa.Column("name", sa.String(length=80), nullable=False), + sa.Column("spec", sa.JSON(), nullable=False), + sa.Column("is_builtin", sa.Boolean(), server_default="false", nullable=False), + sa.Column("sort_order", sa.Integer(), server_default="0", nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_download_profiles_user_id", "download_profiles", ["user_id"]) + + op.create_table( + "download_jobs", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("user_id", sa.Integer(), nullable=False), + sa.Column("asset_id", sa.Integer(), nullable=True), + sa.Column("source_kind", sa.String(length=16), nullable=False), + sa.Column("source_ref", sa.String(length=512), nullable=False), + sa.Column("profile_id", sa.Integer(), nullable=True), + sa.Column("profile_snapshot", sa.JSON(), nullable=False), + sa.Column("display_name", sa.String(length=255), nullable=True), + sa.Column("status", sa.String(length=16), server_default="queued", nullable=False), + sa.Column("progress", sa.Integer(), server_default="0", nullable=False), + sa.Column("speed_bps", sa.BigInteger(), nullable=True), + sa.Column("eta_s", sa.Integer(), nullable=True), + sa.Column("phase", sa.String(length=32), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("queue_pos", sa.Integer(), server_default="0", nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["asset_id"], ["media_assets.id"], ondelete="SET NULL"), + sa.ForeignKeyConstraint(["profile_id"], ["download_profiles.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_download_jobs_user_id", "download_jobs", ["user_id"]) + op.create_index("ix_download_jobs_asset_id", "download_jobs", ["asset_id"]) + op.create_index("ix_download_jobs_status", "download_jobs", ["status"]) + op.create_index("ix_download_jobs_claim", "download_jobs", ["status", "queue_pos"]) + + op.create_table( + "download_shares", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("job_id", sa.Integer(), nullable=False), + sa.Column("from_user_id", sa.Integer(), nullable=False), + sa.Column("to_user_id", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["job_id"], ["download_jobs.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["from_user_id"], ["users.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["to_user_id"], ["users.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("job_id", "to_user_id", name="uq_download_share"), + ) + op.create_index("ix_download_shares_job_id", "download_shares", ["job_id"]) + op.create_index("ix_download_shares_from_user_id", "download_shares", ["from_user_id"]) + op.create_index("ix_download_shares_to_user_id", "download_shares", ["to_user_id"]) + + op.create_table( + "download_quotas", + sa.Column("user_id", sa.Integer(), nullable=False), + sa.Column("max_bytes", sa.BigInteger(), nullable=False), + sa.Column("max_concurrent", sa.Integer(), nullable=False), + sa.Column("max_jobs", sa.Integer(), nullable=False), + sa.Column("unlimited", sa.Boolean(), server_default="false", nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("user_id"), + ) + + +def downgrade() -> None: + op.drop_table("download_quotas") + op.drop_index("ix_download_shares_to_user_id", table_name="download_shares") + op.drop_index("ix_download_shares_from_user_id", table_name="download_shares") + op.drop_index("ix_download_shares_job_id", table_name="download_shares") + op.drop_table("download_shares") + op.drop_index("ix_download_jobs_claim", table_name="download_jobs") + op.drop_index("ix_download_jobs_status", table_name="download_jobs") + op.drop_index("ix_download_jobs_asset_id", table_name="download_jobs") + op.drop_index("ix_download_jobs_user_id", table_name="download_jobs") + op.drop_table("download_jobs") + op.drop_index("ix_download_profiles_user_id", table_name="download_profiles") + op.drop_table("download_profiles") + op.drop_index("ix_media_assets_status", table_name="media_assets") + op.drop_table("media_assets") diff --git a/backend/alembic/versions/0037_download_builtin_profiles.py b/backend/alembic/versions/0037_download_builtin_profiles.py new file mode 100644 index 0000000..d8742a0 --- /dev/null +++ b/backend/alembic/versions/0037_download_builtin_profiles.py @@ -0,0 +1,67 @@ +"""seed builtin download profiles (presets) + +Revision ID: 0037_dl_builtin_profiles +Revises: 0036_download_center +Create Date: 2026-07-02 + +Seeds the shipped download presets (user_id NULL, is_builtin=true). `spec` is the format +definition consumed by app.downloads.formats: mode ("av" audio+video / "v" video-only / +"a" audio-only), max_height (null = best), container, audio_format, vcodec preference, and +the embed_*/sponsorblock postprocessor flags. Names are English (the UI default); the frontend +localizes builtin labels by a stable key derived from the name. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0037_dl_builtin_profiles" +down_revision: Union[str, None] = "0036_download_center" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _spec(mode, height=None, container=None, audio_format=None, vcodec=None): + return { + "mode": mode, + "max_height": height, + "container": container, + "audio_format": audio_format, + "vcodec": vcodec, + "embed_subs": False, + "embed_chapters": True, + "embed_thumbnail": True, + "sponsorblock": False, + } + + +_BUILTINS = [ + ("Best (MP4)", _spec("av", None, "mp4"), 10), + ("1080p (MP4)", _spec("av", 1080, "mp4"), 20), + ("720p (MP4)", _spec("av", 720, "mp4"), 30), + ("Audio (M4A)", _spec("a", None, None, "m4a"), 40), + ("Audio (MP3)", _spec("a", None, None, "mp3"), 50), + ("Video only (MP4)", _spec("v", None, "mp4"), 60), +] + + +def upgrade() -> None: + profiles = sa.table( + "download_profiles", + sa.column("user_id", sa.Integer), + sa.column("name", sa.String), + sa.column("spec", sa.JSON), + sa.column("is_builtin", sa.Boolean), + sa.column("sort_order", sa.Integer), + ) + op.bulk_insert( + profiles, + [ + {"user_id": None, "name": name, "spec": spec, "is_builtin": True, "sort_order": order} + for name, spec, order in _BUILTINS + ], + ) + + +def downgrade() -> None: + op.execute("DELETE FROM download_profiles WHERE is_builtin = true") diff --git a/backend/alembic/versions/0038_asset_gc_notified.py b/backend/alembic/versions/0038_asset_gc_notified.py new file mode 100644 index 0000000..19b317f --- /dev/null +++ b/backend/alembic/versions/0038_asset_gc_notified.py @@ -0,0 +1,29 @@ +"""media_assets.gc_notified flag for one-shot pre-expiry warnings + +Revision ID: 0038_asset_gc_notified +Revises: 0037_dl_builtin_profiles +Create Date: 2026-07-03 + +The download GC warns each holder once, a grace window before an asset's TTL expiry; this flag +records that so subsequent GC runs don't re-notify until the asset is deleted. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0038_asset_gc_notified" +down_revision: Union[str, None] = "0037_dl_builtin_profiles" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "media_assets", + sa.Column("gc_notified", sa.Boolean(), server_default="false", nullable=False), + ) + + +def downgrade() -> None: + op.drop_column("media_assets", "gc_notified") diff --git a/backend/alembic/versions/0039_video_title_normalize.py b/backend/alembic/versions/0039_video_title_normalize.py new file mode 100644 index 0000000..c8351c3 --- /dev/null +++ b/backend/alembic/versions/0039_video_title_normalize.py @@ -0,0 +1,49 @@ +"""normalize video titles for display; keep the raw one in original_title + +Revision ID: 0039_title_normalize +Revises: 0038_asset_gc_notified +Create Date: 2026-07-03 + +Adds videos.original_title (the raw YouTube title) and rewrites videos.title to a normalized, +display-friendly form (emoji stripped, ALL-CAPS de-shouted, trailing SEO hashtags removed — +see app.titles). Reversible: original_title preserves the source, and title can be re-derived. +The generated search_vector column regenerates automatically as each title is updated. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +from app.titles import normalize_title + +revision: str = "0039_title_normalize" +down_revision: Union[str, None] = "0038_asset_gc_notified" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_BATCH = 2000 + + +def upgrade() -> None: + op.add_column("videos", sa.Column("original_title", sa.Text(), nullable=True)) + conn = op.get_bind() + # Preserve the raw title first (fast, set-based). + conn.execute(sa.text("UPDATE videos SET original_title = title WHERE title IS NOT NULL")) + # Then rewrite title to the normalized form, batched (each update regenerates its FTS vector). + rows = conn.execute( + sa.text("SELECT id, original_title FROM videos WHERE original_title IS NOT NULL") + ).fetchall() + changes = [] + for vid, raw in rows: + norm = normalize_title(raw) + if norm != raw: + changes.append({"i": vid, "t": norm}) + stmt = sa.text("UPDATE videos SET title = :t WHERE id = :i") + for start in range(0, len(changes), _BATCH): + conn.execute(stmt, changes[start : start + _BATCH]) + + +def downgrade() -> None: + conn = op.get_bind() + conn.execute(sa.text("UPDATE videos SET title = original_title WHERE original_title IS NOT NULL")) + op.drop_column("videos", "original_title") diff --git a/backend/alembic/versions/0040_reset_builtin_profiles.py b/backend/alembic/versions/0040_reset_builtin_profiles.py new file mode 100644 index 0000000..ed3de02 --- /dev/null +++ b/backend/alembic/versions/0040_reset_builtin_profiles.py @@ -0,0 +1,72 @@ +"""reset builtin download profiles: no embedding, 1080p default + +Revision ID: 0040_reset_profiles +Revises: 0039_title_normalize +Create Date: 2026-07-03 + +Redefines the shipped presets: embedding (thumbnail/chapters/subs) is OFF (it rewrites the whole +file — expensive on large videos — and the Plex .nfo/poster sidecars already carry the metadata), +and the default is now 1080p (Best stays available but no longer first). Re-seeds the builtin +rows so both fresh installs (0037 then this) and existing DBs converge on the same set. Existing +jobs are unaffected (they snapshot their spec; profile_id just SET NULLs if it pointed at a +replaced row). +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0040_reset_profiles" +down_revision: Union[str, None] = "0039_title_normalize" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _spec(mode, height=None, container=None, audio_format=None): + return { + "mode": mode, + "max_height": height, + "container": container, + "audio_format": audio_format, + "vcodec": None, + "embed_subs": False, + "embed_chapters": False, + "embed_thumbnail": False, + "sponsorblock": False, + } + + +# Default first (1080p), then Best, then the rest. +_BUILTINS = [ + ("1080p (MP4)", _spec("av", 1080, "mp4"), 10), + ("Best (MP4)", _spec("av", None, "mp4"), 20), + ("720p (MP4)", _spec("av", 720, "mp4"), 30), + ("Audio (M4A)", _spec("a", None, None, "m4a"), 40), + ("Audio (MP3)", _spec("a", None, None, "mp3"), 50), + ("Video only (MP4)", _spec("v", None, "mp4"), 60), +] + +_profiles = sa.table( + "download_profiles", + sa.column("user_id", sa.Integer), + sa.column("name", sa.String), + sa.column("spec", sa.JSON), + sa.column("is_builtin", sa.Boolean), + sa.column("sort_order", sa.Integer), +) + + +def upgrade() -> None: + op.execute("DELETE FROM download_profiles WHERE is_builtin = true") + op.bulk_insert( + _profiles, + [ + {"user_id": None, "name": n, "spec": s, "is_builtin": True, "sort_order": o} + for n, s, o in _BUILTINS + ], + ) + + +def downgrade() -> None: + # No-op: the previous builtin set was seeded by 0037; not worth reconstructing. + pass diff --git a/backend/app/auth.py b/backend/app/auth.py index c654dca..f7f6041 100644 --- a/backend/app/auth.py +++ b/backend/app/auth.py @@ -683,7 +683,10 @@ def resolved_user_id(request: Request) -> tuple[int | None, bool]: """ default_id = request.session.get("user_id") wallet = request.session.get("account_ids") or [] - hdr = request.headers.get(ACTIVE_ACCOUNT_HEADER) + # Header for normal XHR; ?account= for contexts that can't set headers (WebSocket, and a + # plain file download). Both are wallet-gated below, so neither can impersonate an + # account that never signed into this browser. + hdr = request.headers.get(ACTIVE_ACCOUNT_HEADER) or request.query_params.get("account") if hdr: try: hid = int(hdr) diff --git a/backend/app/config.py b/backend/app/config.py index 938cfcb..224b11a 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -131,6 +131,57 @@ class Settings(BaseSettings): # ("was_live") are real watchable content and stay visible. feed_default_hidden_live: str = "live,upcoming" + # --- Download center (yt-dlp) --- + # Root directory (inside the container) for downloaded media — a Docker volume/bind mount. + # Env-only (a mount path can't live in the DB). The worker lays a Plex-style tree under it. + download_root: str = "/downloads" + # Whether THIS process runs the download worker loop. On for the dedicated worker + # container, off for the API (which keeps the scheduler). Mirrors scheduler_enabled. + worker_enabled: bool = False + # Base URL of the bgutil PO-token provider sidecar. When set, the download worker asks it + # for YouTube Proof-of-Origin tokens (bypasses bot-detection + unlocks high-quality + # formats). Empty disables it (yt-dlp still works, but may hit "confirm you're not a bot"). + download_pot_base_url: str = "http://bgutil-pot:4416" + # yt-dlp player clients (comma-separated). PRIMARY = android_vr: high-quality DASH, no PO + # token / JS runtime needed, reliable for normal use. On a bot/DRM/format failure (YouTube's + # per-client responses are flaky, and android_vr can get bot-flagged under heavy load) the + # worker retries with the FALLBACK set — the POT-capable web clients (web needs the token + + # Deno) plus android for audio. Both admin-tunable when YouTube shifts what works. + download_player_clients: str = "android_vr" + download_player_clients_fallback: str = "web_safari,web,android" + + @staticmethod + def _clients(raw: str) -> list[str]: + return [c.strip() for c in raw.split(",") if c.strip()] + + @property + def player_client_list(self) -> list[str]: + return self._clients(self.download_player_clients) + + @property + def player_client_fallback_list(self) -> list[str]: + return self._clients(self.download_player_clients_fallback) + # How many downloads the worker runs concurrently (claimed via FOR UPDATE SKIP LOCKED). + download_worker_concurrency: int = 2 + # How often (empty pending queue) the worker polls for a new job, in seconds. + download_poll_seconds: int = 5 + # Total cache size cap across ALL users (bytes). When >0 and exceeded, the GC evicts the + # least-recently-accessed ready assets until under it. 0 = unlimited (disk is the limit). + download_total_max_bytes: int = 0 + # Default per-user quota (admin-tunable; a per-user download_quotas row overrides). + download_default_max_bytes: int = 5_368_709_120 # 5 GiB + download_default_max_concurrent: int = 2 + download_default_max_jobs: int = 50 + # Retention: an asset expires this many days after its last access; the GC job then + # deletes it (once no job references it) after an additional grace window. + download_retention_days: int = 30 + download_gc_grace_days: int = 2 + download_gc_minutes: int = 360 + # On-disk layout: "plex" (Channel/Season YYYY/… [id].ext + .nfo/poster) or "flat". + download_layout: str = "plex" + # Default SponsorBlock (skip sponsor segments) for new custom profiles. + sponsorblock_default: bool = False + @property def allowed_email_set(self) -> set[str]: return {e.strip().lower() for e in self.allowed_emails.split(",") if e.strip()} diff --git a/backend/app/downloads/__init__.py b/backend/app/downloads/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/downloads/formats.py b/backend/app/downloads/formats.py new file mode 100644 index 0000000..0fb927f --- /dev/null +++ b/backend/app/downloads/formats.py @@ -0,0 +1,179 @@ +"""Format profiles → yt-dlp options, and the cache signature. + +A profile `spec` (stored on DownloadProfile / snapshotted on DownloadJob) is the single source +of truth for what file gets produced. `format_sig` hashes exactly the fields that change the +OUTPUT BYTES, so two users picking equivalent settings share one cached MediaAsset. +`build_ydl_opts` turns a spec into a yt-dlp options dict (format selector + postprocessors). + +spec shape (all keys optional; defaults applied by `normalize`): + mode "av" (audio+video) | "v" (video-only) | "a" (audio-only) + max_height int | null (null = best) + container "mp4" | "mkv" | "webm" | null (merge/remux target for av/v) + audio_format "m4a" | "mp3" | "opus" | null (extract target for mode "a") + vcodec "h264" | "vp9" | "av1" | null (preference, best-effort) + embed_subs bool embed_chapters bool embed_thumbnail bool sponsorblock bool +""" +import hashlib +import json + +# Fields that affect the produced file — and therefore the cache identity. +_SIG_KEYS = ( + "mode", + "max_height", + "container", + "audio_format", + "vcodec", + "embed_subs", + "embed_chapters", + "embed_thumbnail", + "sponsorblock", +) + +_DEFAULTS = { + "mode": "av", + "max_height": None, + "container": "mp4", + "audio_format": "m4a", + "vcodec": None, + # Embedding is OFF by default: it rewrites the whole file (doubles I/O — a 10 GB video + # gets a 10 GB temp copy), and the Plex `.nfo` + poster sidecars already carry the + # metadata/thumbnail. Users who want self-contained files opt in via a custom profile. + "embed_subs": False, + "embed_chapters": False, + "embed_thumbnail": False, + "sponsorblock": False, +} + +_SPONSORBLOCK_CATEGORIES = ["sponsor", "selfpromo", "interaction"] +_VCODEC_SORT = {"h264": "vcodec:h264", "vp9": "vcodec:vp9", "av1": "vcodec:av01"} + + +def normalize(spec: dict) -> dict: + """Fill defaults + coerce types so signature and yt-dlp opts are deterministic.""" + s = dict(_DEFAULTS) + for k in _DEFAULTS: + if spec.get(k) is not None: + s[k] = spec[k] + if s["mode"] not in ("av", "v", "a"): + s["mode"] = "av" + if s["max_height"] is not None: + s["max_height"] = int(s["max_height"]) + for b in ("embed_subs", "embed_chapters", "embed_thumbnail", "sponsorblock"): + s[b] = bool(s[b]) + # Audio-only files can't embed subtitles/thumbnails as video; keep the flags meaningful. + if s["mode"] == "a": + s["embed_subs"] = False + return s + + +def format_sig(spec: dict) -> str: + """Stable short hash of the output-affecting spec — the MediaAsset cache key component.""" + s = normalize(spec) + payload = json.dumps({k: s[k] for k in _SIG_KEYS}, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode()).hexdigest()[:24] + + +def target_ext(spec: dict) -> str: + """Best-effort final extension for the produced file (also used to name the staging output).""" + s = normalize(spec) + if s["mode"] == "a": + return s["audio_format"] or "m4a" + return s["container"] or "mp4" + + +def build_ydl_opts( + spec: dict, + outtmpl: str, + progress_hook, + pot_base_url: str | None = None, + player_clients: list[str] | None = None, +) -> dict: + """Assemble yt-dlp options for a single download to `outtmpl` (a full path template). + + `pot_base_url` points the bgutil PO-token provider plugin at the sidecar server; + `player_clients` picks which YouTube clients to use (POT-capable ones like web use the token).""" + s = normalize(spec) + h = s["max_height"] + opts: dict = { + "outtmpl": outtmpl, + "noplaylist": True, + "quiet": True, + "no_warnings": True, + "noprogress": True, + "writethumbnail": True, # kept for the poster.jpg sidecar (and embed, if requested) + "progress_hooks": [progress_hook], + "postprocessors": [], + # Force IPv4: containers often have no working IPv6 route, but googlevideo CDN hosts + # advertise AAAA records, so ffmpeg/the downloader would try IPv6 and fail with + # "[Errno -5] No address associated with hostname". Binding to the IPv4 any-address + # makes every connection IPv4-only. (yt-dlp's --force-ipv4 does the same thing.) + "source_address": "0.0.0.0", + # Self-heal transient network blips instead of failing the whole job. + "retries": 10, + "fragment_retries": 10, + "extractor_retries": 3, + "retry_sleep": {"http": 5, "fragment": 5, "extractor": 5}, + "socket_timeout": 30, + } + + # Extractor args: which YouTube clients to use + where the bgutil PO-token sidecar lives + # (== --extractor-args "youtube:player_client=…;youtubepot-bgutilhttp:base_url=…"). Values + # are lists in the Python API. + ea: dict = {} + if player_clients: + ea["youtube"] = {"player_client": player_clients} + if pot_base_url: + ea["youtubepot-bgutilhttp"] = {"base_url": [pot_base_url]} + if ea: + opts["extractor_args"] = ea + + if s["mode"] == "a": + opts["format"] = "bestaudio/best" + opts["postprocessors"].append( + { + "key": "FFmpegExtractAudio", + "preferredcodec": s["audio_format"] or "m4a", + "preferredquality": "0", + } + ) + else: + hcap = f"[height<=?{h}]" if h else "" + if s["mode"] == "v": + opts["format"] = f"bestvideo{hcap}/best{hcap}" + else: # av + opts["format"] = f"bestvideo{hcap}+bestaudio/best{hcap}" + if s["container"]: + opts["merge_output_format"] = s["container"] + sort = [] + if h: + sort.append(f"res:{h}") + if s["vcodec"] in _VCODEC_SORT: + sort.append(_VCODEC_SORT[s["vcodec"]]) + if sort: + opts["format_sort"] = sort + + # SponsorBlock first (marks segments), then strip them from the media + chapters. + if s["sponsorblock"]: + opts["postprocessors"].append( + {"key": "SponsorBlock", "categories": _SPONSORBLOCK_CATEGORIES, "when": "after_filter"} + ) + opts["postprocessors"].append( + {"key": "ModifyChapters", "remove_sponsor_segments": _SPONSORBLOCK_CATEGORIES} + ) + if s["embed_subs"] and s["mode"] != "a": + opts["writesubtitles"] = True + opts["subtitleslangs"] = ["en.*", "hu.*", "de.*"] + opts["postprocessors"].append({"key": "FFmpegEmbedSubtitle"}) + if s["embed_chapters"]: + opts["postprocessors"].append( + {"key": "FFmpegMetadata", "add_chapters": True, "add_metadata": True} + ) + # Normalize the written thumbnail to jpg so we always have a `.jpg` poster sidecar. + # `already_have_thumbnail` keeps that file on disk after embedding (default deletes it). + opts["postprocessors"].append({"key": "FFmpegThumbnailsConvertor", "format": "jpg"}) + if s["embed_thumbnail"]: + opts["postprocessors"].append( + {"key": "EmbedThumbnail", "already_have_thumbnail": True} + ) + + return opts diff --git a/backend/app/downloads/gc.py b/backend/app/downloads/gc.py new file mode 100644 index 0000000..570d130 --- /dev/null +++ b/backend/app/downloads/gc.py @@ -0,0 +1,119 @@ +"""Download retention garbage collector (a scheduler job). + +Three passes each run: + 1. Pre-expiry warning — for ready assets expiring within the grace window, notify each holder + once (gc_notified flag) so a download isn't silently deleted. + 2. Expiry deletion — delete assets past their TTL (files + row). The FK SET NULL nulls the + holders' jobs' asset_id, so those jobs surface as "expired" (re-downloadable) in the UI. + 3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed + ready assets until back under it. + +Runs in the API process (which mounts DOWNLOAD_ROOT). Pure disk/DB work — no YouTube quota. +""" +import logging +from datetime import datetime, timedelta, timezone + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app import sysconfig +from app.config import settings +from app.downloads import storage +from app.models import DownloadJob, MediaAsset +from app.notifications import create_notification + +log = logging.getLogger("siftlode.downloads.gc") + + +def _holders(db: Session, asset_id: int) -> list[int]: + return list( + db.execute( + select(DownloadJob.user_id) + .where(DownloadJob.asset_id == asset_id) + .distinct() + ).scalars() + ) + + +def _notify_holders(db: Session, asset: MediaAsset, ntype: str) -> None: + title = asset.title or "download" + for uid in _holders(db, asset.id): + create_notification( + db, + user_id=uid, + type=ntype, + title=title, # English fallback; the client renders a translated message from data + data={"kind": ntype, "title": asset.title, "video_id": asset.source_ref}, + commit=False, + ) + + +def _delete_asset(db: Session, asset: MediaAsset, ntype: str) -> None: + if asset.rel_path: + storage.delete_asset_files(settings.download_root, asset.rel_path) + _notify_holders(db, asset, ntype) + db.delete(asset) # jobs.asset_id -> NULL via FK ON DELETE SET NULL + + +def run_download_gc(db: Session) -> dict: + now = datetime.now(timezone.utc) + grace_days = sysconfig.get_int(db, "download_gc_grace_days") + warned = expired = evicted = 0 + + # 1. Pre-expiry warnings. + warn_before = now + timedelta(days=grace_days) + soon = db.execute( + select(MediaAsset).where( + MediaAsset.status == "ready", + MediaAsset.gc_notified.is_(False), + MediaAsset.expires_at.is_not(None), + MediaAsset.expires_at > now, + MediaAsset.expires_at <= warn_before, + ) + ).scalars().all() + for asset in soon: + _notify_holders(db, asset, "download_expiring") + asset.gc_notified = True + warned += 1 + if soon: + db.commit() + + # 2. Expiry deletion (ready assets past TTL, and any errored leftovers with a TTL). + dead = db.execute( + select(MediaAsset).where( + MediaAsset.status.in_(("ready", "error")), + MediaAsset.expires_at.is_not(None), + MediaAsset.expires_at <= now, + ) + ).scalars().all() + for asset in dead: + _delete_asset(db, asset, "download_expired") + expired += 1 + if dead: + db.commit() + + # 3. LRU eviction over the total cache cap. + cap = sysconfig.get_int(db, "download_total_max_bytes") + if cap > 0: + total = db.execute( + select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where( + MediaAsset.status == "ready" + ) + ).scalar() or 0 + if total > cap: + candidates = db.execute( + select(MediaAsset) + .where(MediaAsset.status == "ready") + .order_by(MediaAsset.last_access_at.asc().nulls_first()) + ).scalars().all() + for asset in candidates: + if total <= cap: + break + total -= asset.size_bytes or 0 + _delete_asset(db, asset, "download_evicted") + evicted += 1 + db.commit() + + result = {"warned": warned, "expired": expired, "evicted": evicted} + log.info("download_gc %s", result) + return result diff --git a/backend/app/downloads/quota.py b/backend/app/downloads/quota.py new file mode 100644 index 0000000..b60fcb7 --- /dev/null +++ b/backend/app/downloads/quota.py @@ -0,0 +1,129 @@ +"""Per-user download limits + footprint accounting. + +A user's *footprint* is the total size of the ready assets their library references (cache hits +count — it's "how much disk you're responsible for", not "how much you personally downloaded"). +Limits come from a per-user DownloadQuota row if the admin set one, else the sysconfig defaults. +`unlimited` (e.g. the admin's own account) bypasses the byte cap. + +Enforcement split: max_jobs + max_bytes are checked at enqueue (they bound holdings); per-user +max_concurrent is enforced by the worker at claim time (it bounds simultaneous downloads). +""" +from dataclasses import dataclass + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app import sysconfig +from app.models import DownloadJob, DownloadQuota, MediaAsset + +# Jobs a user is considered to be "holding" (occupy a job slot / footprint). Terminal failures +# (error, canceled) don't count. +_HOLDING = ("queued", "running", "paused", "done") +_RUNNING = ("queued", "running") + + +@dataclass +class QuotaLimits: + max_bytes: int + max_concurrent: int + max_jobs: int + unlimited: bool + + +class QuotaExceeded(Exception): + """Raised by check_enqueue; `reason` is one of max_jobs|max_bytes (i18n key on the client).""" + + def __init__(self, reason: str, limit: int, current: int): + self.reason = reason + self.limit = limit + self.current = current + super().__init__(f"download quota exceeded: {reason} ({current}/{limit})") + + +def resolve(db: Session, user_id: int) -> QuotaLimits: + row = db.get(DownloadQuota, user_id) + if row is not None: + return QuotaLimits(row.max_bytes, row.max_concurrent, row.max_jobs, row.unlimited) + return QuotaLimits( + max_bytes=sysconfig.get_int(db, "download_default_max_bytes"), + max_concurrent=sysconfig.get_int(db, "download_default_max_concurrent"), + max_jobs=sysconfig.get_int(db, "download_default_max_jobs"), + unlimited=False, + ) + + +def footprint(db: Session, user_id: int) -> int: + """Total bytes of the distinct ready assets this user's active jobs reference.""" + held_assets = ( + select(DownloadJob.asset_id) + .where( + DownloadJob.user_id == user_id, + DownloadJob.status.in_(_HOLDING), + DownloadJob.asset_id.is_not(None), + ) + .distinct() + ) + return int( + db.execute( + select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where( + MediaAsset.id.in_(held_assets), MediaAsset.status == "ready" + ) + ).scalar() + or 0 + ) + + +def _count(db: Session, user_id: int, states) -> int: + return ( + db.execute( + select(func.count()) + .select_from(DownloadJob) + .where(DownloadJob.user_id == user_id, DownloadJob.status.in_(states)) + ).scalar() + or 0 + ) + + +def active_jobs(db: Session, user_id: int) -> int: + return _count(db, user_id, _HOLDING) + + +def running_jobs(db: Session, user_id: int) -> int: + return _count(db, user_id, _RUNNING) + + +def at_concurrency_limit(db: Session, user_id: int) -> bool: + """Worker-side gate: is this user already downloading their max? Unlimited users never are.""" + lim = resolve(db, user_id) + if lim.unlimited: + return False + running = _count(db, user_id, ("running",)) + return running >= lim.max_concurrent + + +def check_enqueue(db: Session, user_id: int) -> None: + """Raise QuotaExceeded if the user can't take another job (holdings caps). Concurrency is + enforced later by the worker, so a big queue is allowed; it just drains max_concurrent-wide.""" + lim = resolve(db, user_id) + if lim.unlimited: + return + n = active_jobs(db, user_id) + if n >= lim.max_jobs: + raise QuotaExceeded("max_jobs", lim.max_jobs, n) + used = footprint(db, user_id) + if used >= lim.max_bytes: + raise QuotaExceeded("max_bytes", lim.max_bytes, used) + + +def usage(db: Session, user_id: int) -> dict: + """Compact usage snapshot for the UI (footprint + counts vs limits).""" + lim = resolve(db, user_id) + return { + "footprint_bytes": footprint(db, user_id), + "active_jobs": active_jobs(db, user_id), + "running_jobs": running_jobs(db, user_id), + "max_bytes": lim.max_bytes, + "max_concurrent": lim.max_concurrent, + "max_jobs": lim.max_jobs, + "unlimited": lim.unlimited, + } diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py new file mode 100644 index 0000000..ad2f6bd --- /dev/null +++ b/backend/app/downloads/service.py @@ -0,0 +1,123 @@ +"""Enqueue + cache resolution for the download center (the per-user layer over MediaAsset). + +`enqueue` is the single entry point used by the API: it resolves (or creates) the shared +MediaAsset for the requested source+format, links a per-user DownloadJob to it, and — on a +cache hit (asset already ready) — marks the job done instantly with no re-download. The worker +handles everything else. State transitions used by the routes (pause/resume/cancel/delete) and +the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping. +""" +from datetime import datetime, timezone + +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.downloads import formats, quota +from app.models import Channel, DownloadJob, MediaAsset, Video + + +def source_url(source_kind: str, source_ref: str) -> str: + """The URL yt-dlp should fetch. YouTube stores the 11-char id; ad-hoc stores the URL.""" + if source_kind == "youtube": + return f"https://www.youtube.com/watch?v={source_ref}" + return source_ref + + +def get_or_create_asset( + db: Session, source_kind: str, source_ref: str, format_sig: str +) -> MediaAsset: + """Fetch the cache row for this (source, format) or create a fresh `pending` one. + + Races on the unique cache key are resolved by re-selecting after an IntegrityError, so two + simultaneous enqueues of the same video+format converge on one shared asset.""" + stmt = select(MediaAsset).where( + MediaAsset.source_kind == source_kind, + MediaAsset.source_ref == source_ref, + MediaAsset.format_sig == format_sig, + ) + asset = db.execute(stmt).scalar_one_or_none() + if asset is not None: + return asset + asset = MediaAsset( + source_kind=source_kind, + source_ref=source_ref, + format_sig=format_sig, + status="pending", + ref_count=0, + ) + db.add(asset) + try: + db.flush() + except IntegrityError: + db.rollback() + asset = db.execute(stmt).scalar_one() + return asset + + +def populate_from_catalog(db: Session, asset: MediaAsset) -> bool: + """Fill an asset's display metadata (title/uploader/thumbnail/date/duration) from our own + catalog when the source is a known YouTube video — so a queued/downloading row shows the + real title + thumbnail immediately instead of the bare id. 0 network cost. Returns True if + it found a catalog match.""" + if asset.source_kind != "youtube" or asset.title: + return False + video = db.get(Video, asset.source_ref) + if video is None: + return False + asset.title = video.title + asset.thumbnail_url = video.thumbnail_url + asset.duration_s = video.duration_seconds + if video.published_at: + asset.upload_date = video.published_at.date() + channel = db.get(Channel, video.channel_id) + if channel is not None: + asset.uploader = channel.title + return True + + +def _next_queue_pos(db: Session) -> int: + return (db.execute(select(func.coalesce(func.max(DownloadJob.queue_pos), 0))).scalar() or 0) + 1 + + +def enqueue( + db: Session, + user_id: int, + source_kind: str, + source_ref: str, + spec: dict, + profile_id: int | None = None, + display_name: str | None = None, +) -> DownloadJob: + """Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit). + + Raises quota.QuotaExceeded if the user is at their job-count or footprint cap.""" + quota.check_enqueue(db, user_id) + spec = formats.normalize(spec) + sig = formats.format_sig(spec) + asset = get_or_create_asset(db, source_kind, source_ref, sig) + # Show a real title + thumbnail the moment it's queued (feed downloads are catalog videos). + populate_from_catalog(db, asset) + + job = DownloadJob( + user_id=user_id, + asset_id=asset.id, + source_kind=source_kind, + source_ref=source_ref, + profile_id=profile_id, + profile_snapshot=spec, + display_name=display_name or (asset.title if asset.status == "ready" else None), + status="queued", + queue_pos=_next_queue_pos(db), + ) + db.add(job) + asset.ref_count = (asset.ref_count or 0) + 1 + + # Cache hit: the file already exists — no worker round-trip needed. + if asset.status == "ready": + job.status = "done" + job.progress = 100 + asset.last_access_at = datetime.now(timezone.utc) + + db.commit() + db.refresh(job) + return job diff --git a/backend/app/downloads/storage.py b/backend/app/downloads/storage.py new file mode 100644 index 0000000..1c4f166 --- /dev/null +++ b/backend/app/downloads/storage.py @@ -0,0 +1,154 @@ +"""On-disk layout for downloaded media — Plex-friendly tree + .nfo/poster sidecars. + +The physical filename is system-decided and bound to the source id (never renamed; the user's +custom name is display-only, applied at local-download time). Layout is admin-configurable: + + plex: {Channel}/Season {Year}/{Channel} - {YYYY-MM-DD} - {title} [{id}].{ext} + flat: {title} [{id}].{ext} + +Plus a Kodi/Plex-readable `.nfo` (episodedetails) and `.jpg` thumbnail, so +a Plex library using a metadata agent picks up rich info. Channel = "show", video = episode, +date-based — the community-standard mapping for YouTube-in-Plex. +""" +import re +import shutil +import unicodedata +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from xml.sax.saxutils import escape + +# Reserve room for the " [id].ext" suffix + parent dirs within the 255-byte component limit. +_MAX_TITLE = 120 +# Filesystem-illegal on Windows/Plex + control chars. +_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]') +# Unicode categories we drop entirely: emoji/pictographs (So), modifier symbols/skin-tones +# (Sk), and every control/format/private/surrogate/unassigned char (C*). Letters (incl. +# accented á/ö/ü/ß), digits, and ordinary punctuation are KEPT — Siftlode is trilingual, so +# ASCII-folding would mangle Hungarian/German titles. +_DROP_CATEGORIES = {"So", "Sk", "Cc", "Cf", "Cs", "Co", "Cn"} + + +@dataclass +class MediaMeta: + video_id: str + title: str + uploader: str + upload_date: date | None + duration_s: int | None = None + description: str | None = None + thumbnail_url: str | None = None + + +def sanitize(name: str, limit: int = _MAX_TITLE) -> str: + """Turn an arbitrary (emoji-laden, clickbait) title into a clean, space-free path token. + + Steps: NFKC-normalize (fold fancy width/compat forms) → drop emoji/symbols/control chars → + strip filesystem-illegal chars → collapse whitespace and runs of the same punctuation → + join words with underscores → trim junk punctuation from the ends → length-cap. Accented + letters and digits survive; the result never contains spaces or path separators.""" + text = unicodedata.normalize("NFKC", name or "") + text = "".join( + ch for ch in text if unicodedata.category(ch) not in _DROP_CATEGORIES + ) + text = _ILLEGAL.sub("", text) + # Collapse repeated punctuation ("!!!", "???", "-----") to a single mark. + text = re.sub(r"([^\w\s])\1{1,}", r"\1", text) + # Whitespace → single underscore (no spaces in generated names). + text = re.sub(r"\s+", "_", text.strip()) + # Tidy separator pile-ups and trim junk from the ends. + text = re.sub(r"_{2,}", "_", text).strip("_.-·—–|,;:!?ّ ") + if len(text) > limit: + text = text[:limit].rstrip("_.-") + return text or "untitled" + + +def display_filename(name: str) -> str: + """User-facing download filename (Content-Disposition): strip emoji/symbols/control + + filesystem-illegal chars, but KEEP spaces, accents, and ordinary punctuation — unlike the + underscore-joined on-disk name, this is the readable name the user saves to their device.""" + text = unicodedata.normalize("NFKC", name or "") + text = "".join(ch for ch in text if unicodedata.category(ch) not in _DROP_CATEGORIES) + text = _ILLEGAL.sub("", text) + text = re.sub(r"\s+", " ", text).strip(" .") + return text or "download" + + +def rel_path(meta: MediaMeta, ext: str, layout: str = "plex") -> str: + """Path of the media file relative to DOWNLOAD_ROOT (forward slashes).""" + title = sanitize(meta.title) + vid = meta.video_id + if layout == "flat": + return f"{title}_[{vid}].{ext}" + channel = sanitize(meta.uploader or "Unknown_Channel", 80) + year = meta.upload_date.year if meta.upload_date else 0 + d = meta.upload_date.isoformat() if meta.upload_date else "0000-00-00" + epname = f"{channel}_-_{d}_-_{title}_[{vid}].{ext}" + return f"{channel}/Season_{year}/{epname}" + + +def abs_path(download_root: str, rel: str) -> Path: + return Path(download_root) / rel + + +def place_file(staging_file: Path, download_root: str, rel: str) -> None: + """Move a completed staging file into its final location under DOWNLOAD_ROOT.""" + dest = abs_path(download_root, rel) + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(staging_file), str(dest)) + + +def _nfo_xml(meta: MediaMeta) -> str: + aired = meta.upload_date.isoformat() if meta.upload_date else "" + year = meta.upload_date.year if meta.upload_date else "" + parts = [ + '', + "", + f" {escape(meta.title or '')}", + f" {escape(meta.uploader or '')}", + f" {escape(meta.uploader or '')}", + f" {aired}", + f" {aired}", + f" {year}", + f" {escape(meta.description or '')}", + f" {escape(meta.video_id)}", + ] + if meta.duration_s: + parts.append(f" {round(meta.duration_s / 60)}") + parts.append("") + return "\n".join(parts) + + +def write_sidecars( + download_root: str, rel: str, meta: MediaMeta, thumb_src: Path | None +) -> bool: + """Write `.nfo` + `.jpg` next to the media file. Returns True if written.""" + media = abs_path(download_root, rel) + base = media.with_suffix("") + try: + base.with_suffix(".nfo").write_text(_nfo_xml(meta), encoding="utf-8") + if thumb_src and thumb_src.exists(): + shutil.copyfile(str(thumb_src), str(base.with_suffix(".jpg"))) + return True + except OSError: + return False + + +def delete_asset_files(download_root: str, rel: str) -> None: + """Remove the media file and its sidecars; prune now-empty parent dirs (best effort).""" + media = abs_path(download_root, rel) + base = media.with_suffix("") + for p in (media, base.with_suffix(".nfo"), base.with_suffix(".jpg")): + try: + p.unlink() + except OSError: + pass + # Prune empty Season/Channel dirs up to (but not including) the root. + root = Path(download_root).resolve() + parent = media.parent + while parent != root and root in parent.parents: + try: + parent.rmdir() # only succeeds if empty + except OSError: + break + parent = parent.parent diff --git a/backend/app/main.py b/backend/app/main.py index 7115870..d6329ca 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -31,6 +31,7 @@ from app.routes import ( admin, channels, config as config_routes, + downloads, feed, health, me, @@ -122,6 +123,8 @@ app.include_router(messages.router) app.include_router(channels.router) app.include_router(playlists.router) app.include_router(saved_views.router) +app.include_router(downloads.router) +app.include_router(downloads.admin_router) app.include_router(admin.router) app.include_router(config_routes.router) app.include_router(scheduler_routes.router) diff --git a/backend/app/models.py b/backend/app/models.py index 3160514..bfe6282 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -8,6 +8,7 @@ from sqlalchemy import ( DateTime, Float, ForeignKey, + Index, Integer, JSON, String, @@ -240,7 +241,8 @@ class Video(Base, TimestampMixin): channel_id: Mapped[str] = mapped_column( ForeignKey("channels.id", ondelete="CASCADE"), index=True ) - title: Mapped[str | None] = mapped_column(Text) + title: Mapped[str | None] = mapped_column(Text) # normalized for display (see app.titles) + original_title: Mapped[str | None] = mapped_column(Text) # raw YouTube title, preserved description: Mapped[str | None] = mapped_column(Text) published_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), index=True @@ -688,6 +690,164 @@ class Message(Base): ) +class MediaAsset(Base, TimestampMixin): + """A physical downloaded media file — the SHARED cache layer of the download center. + + Mirrors the app's "shared catalog, private per-user state" model: one row per unique + ``(source_kind, source_ref, format_sig)`` is the global cache. When a second user requests + the same video with a format that hashes to the same `format_sig`, their `download_job` + links to this ready asset instantly — no re-download, no cost. Only the SOURCE download is + shared; per-user edited derivatives (phase 2) are never cached here. + + `rel_path` is relative to `settings.download_root` (the mount), so the physical filename is + system-decided and bound to the source id — never renamed (the user's custom name is display + metadata on `download_job`, applied only when downloading to their own machine). The naming + fields (`title`/`uploader`/`upload_date`) are denormalized so the Plex-tree path can be built + uniformly for both catalog videos and ad-hoc URLs. `ref_count` (active jobs pointing here) + plus `last_access_at`/`expires_at` drive retention + LRU eviction in the GC job.""" + + __tablename__ = "media_assets" + __table_args__ = ( + UniqueConstraint( + "source_kind", "source_ref", "format_sig", name="uq_media_asset_cache_key" + ), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + source_kind: Mapped[str] = mapped_column(String(16)) # "youtube" | "url" + source_ref: Mapped[str] = mapped_column(String(512)) # video id, or full URL for ad-hoc + format_sig: Mapped[str] = mapped_column(String(64)) # hash of the output-affecting spec + status: Mapped[str] = mapped_column( + String(16), default="pending", server_default="pending", index=True + ) # "pending" | "ready" | "error" + rel_path: Mapped[str | None] = mapped_column(Text) # relative to download_root + storyboard_path: Mapped[str | None] = mapped_column(Text) # phase-2 filmstrip mosaic + size_bytes: Mapped[int | None] = mapped_column(BigInteger) + duration_s: Mapped[int | None] = mapped_column(Integer) + width: Mapped[int | None] = mapped_column(Integer) + height: Mapped[int | None] = mapped_column(Integer) + container: Mapped[str | None] = mapped_column(String(16)) + vcodec: Mapped[str | None] = mapped_column(String(32)) + acodec: Mapped[str | None] = mapped_column(String(32)) + nfo_written: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) + # Set once the GC has warned holders of the upcoming expiry, so it doesn't warn every run. + gc_notified: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) + # Denormalized naming/metadata for the Plex tree + ad-hoc display. + title: Mapped[str | None] = mapped_column(Text) + uploader: Mapped[str | None] = mapped_column(String(255)) + upload_date: Mapped[date | None] = mapped_column(Date) + thumbnail_url: Mapped[str | None] = mapped_column(Text) + error: Mapped[str | None] = mapped_column(Text) + ref_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + last_access_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + +class DownloadProfile(Base, TimestampMixin): + """A named download preset — a reusable format profile. + + `user_id` NULL + `is_builtin` marks the shipped presets (Best MP4, 1080p, audio-only, …); + a user's saved custom profiles carry their `user_id`. `spec` is the format definition + (mode av/v/a, max_height, container, audio_format, codec prefs, embed subs/chapters/ + thumbnail, sponsorblock) that `downloads.formats` turns into a yt-dlp selector + a cache + `format_sig`.""" + + __tablename__ = "download_profiles" + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int | None] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), index=True + ) # NULL for a builtin preset + name: Mapped[str] = mapped_column(String(80)) + spec: Mapped[dict] = mapped_column(JSON) + is_builtin: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) + sort_order: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + +class DownloadJob(Base, TimestampMixin, UpdatedAtMixin): + """A user's request to download one source — the per-user layer over `MediaAsset`. + + The worker claims `queued` jobs with ``FOR UPDATE SKIP LOCKED`` and writes live + `progress`/`speed_bps`/`eta_s`/`phase` here (the frontend polls via useLiveQuery — the + worker is a separate process and can't reach the in-process WS registry). `profile_snapshot` + freezes the spec at enqueue so later edits/deletion of the profile don't change this job. + `display_name` is the user's own name for the file (display + the Content-Disposition on + local download); the on-disk asset is never renamed. `asset_id` links to the shared file + (SET NULL if the asset is GC'd — the job then shows as expired).""" + + __tablename__ = "download_jobs" + __table_args__ = (Index("ix_download_jobs_claim", "status", "queue_pos"),) + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), index=True + ) + asset_id: Mapped[int | None] = mapped_column( + ForeignKey("media_assets.id", ondelete="SET NULL"), index=True + ) + source_kind: Mapped[str] = mapped_column(String(16)) # "youtube" | "url" + source_ref: Mapped[str] = mapped_column(String(512)) + profile_id: Mapped[int | None] = mapped_column( + ForeignKey("download_profiles.id", ondelete="SET NULL") + ) + profile_snapshot: Mapped[dict] = mapped_column(JSON) + display_name: Mapped[str | None] = mapped_column(String(255)) + status: Mapped[str] = mapped_column( + String(16), default="queued", server_default="queued", index=True + ) # queued | running | paused | done | error | canceled + progress: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + speed_bps: Mapped[int | None] = mapped_column(BigInteger) + eta_s: Mapped[int | None] = mapped_column(Integer) + phase: Mapped[str | None] = mapped_column(String(32)) + error: Mapped[str | None] = mapped_column(Text) + queue_pos: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + +class DownloadShare(Base, TimestampMixin): + """A user→user grant: `to_user` may access `from_user`'s downloaded file (private by + default). Since assets are already deduped/shared on disk, a share is just an ACL row.""" + + __tablename__ = "download_shares" + __table_args__ = ( + UniqueConstraint("job_id", "to_user_id", name="uq_download_share"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + job_id: Mapped[int] = mapped_column( + ForeignKey("download_jobs.id", ondelete="CASCADE"), index=True + ) + from_user_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), index=True + ) + to_user_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), index=True + ) + + +class DownloadQuota(Base, UpdatedAtMixin): + """Per-user download limits (admin-managed). A row exists only when an admin customizes a + user; otherwise the sysconfig `download_default_*` values apply. `unlimited` bypasses the + byte cap entirely (e.g. for the admin's own account).""" + + __tablename__ = "download_quotas" + + user_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), primary_key=True + ) + max_bytes: Mapped[int] = mapped_column(BigInteger) + max_concurrent: Mapped[int] = mapped_column(Integer) + max_jobs: Mapped[int] = mapped_column(Integer) + unlimited: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) + + class MessageKey(Base, TimestampMixin): """A user's end-to-end-encryption key material for direct messaging (phase 2). diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py new file mode 100644 index 0000000..661f0ae --- /dev/null +++ b/backend/app/routes/downloads.py @@ -0,0 +1,659 @@ +"""Download center HTTP API — the per-user surface over the worker/cache/quota layers. + +All routes are behind `require_human` (the shared demo account can't spend server disk / needs a +real identity). Admin routes add `admin_user`. Live job state is polled by the frontend (the +worker is a separate process); this module never downloads — it enqueues, manages job state, +serves finished files (range-aware, with the user's custom display name), and shares them. +""" +import re +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import FileResponse +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.auth import admin_user, require_human +from app.config import settings +from app.db import get_db +from app.downloads import quota, service, storage +from app.models import ( + DownloadJob, + DownloadProfile, + DownloadQuota, + DownloadShare, + MediaAsset, + User, +) + +router = APIRouter(prefix="/api/downloads", tags=["downloads"]) +admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"]) + +_VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$") + + +# --- source + serialization ---------------------------------------------------------------- + +def resolve_source(raw: str) -> tuple[str, str]: + """Map a user-supplied id/URL to (source_kind, source_ref). YouTube is normalized to its + 11-char id (so the cache dedups across watch/youtu.be/shorts URLs); anything else is kept + as a raw URL for yt-dlp's generic extractor.""" + raw = (raw or "").strip() + if not raw: + raise HTTPException(status_code=400, detail="A video link or id is required.") + if _VIDEO_ID.match(raw): + return "youtube", raw + if raw.startswith(("http://", "https://")): + u = urlparse(raw) + host = u.netloc.lower() + if "youtube.com" in host: + if u.path.startswith(("/shorts/", "/embed/", "/live/")): + vid = u.path.split("/")[2] if len(u.path.split("/")) > 2 else "" + if _VIDEO_ID.match(vid): + return "youtube", vid + qs = parse_qs(u.query).get("v", []) + if qs and _VIDEO_ID.match(qs[0]): + return "youtube", qs[0] + if "youtu.be" in host: + vid = u.path.lstrip("/").split("/")[0] + if _VIDEO_ID.match(vid): + return "youtube", vid + return "url", raw + raise HTTPException(status_code=400, detail="That doesn't look like a valid video link.") + + +def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: + ready = asset is not None and asset.status == "ready" and bool(asset.rel_path) + return { + "id": job.id, + "status": job.status, + "progress": job.progress, + "speed_bps": job.speed_bps, + "eta_s": job.eta_s, + "phase": job.phase, + "error": job.error, + "queue_pos": job.queue_pos, + "created_at": job.created_at.isoformat() if job.created_at else None, + "source_kind": job.source_kind, + "source_ref": job.source_ref, + "profile_id": job.profile_id, + "spec": job.profile_snapshot, + "display_name": job.display_name, + "can_download": ready and job.status == "done", + "expired": job.asset_id is None and job.status == "done", + "asset": None + if asset is None + else { + "id": asset.id, + "status": asset.status, + "title": asset.title, + "uploader": asset.uploader, + "upload_date": asset.upload_date.isoformat() if asset.upload_date else None, + "duration_s": asset.duration_s, + "size_bytes": asset.size_bytes, + "width": asset.width, + "height": asset.height, + "container": asset.container, + "thumbnail_url": asset.thumbnail_url, + "expires_at": asset.expires_at.isoformat() if asset.expires_at else None, + }, + } + + +def _assets_for(db: Session, jobs: list[DownloadJob]) -> dict[int, MediaAsset]: + ids = {j.asset_id for j in jobs if j.asset_id} + if not ids: + return {} + rows = db.execute(select(MediaAsset).where(MediaAsset.id.in_(ids))).scalars() + return {a.id: a for a in rows} + + +def _own_job(db: Session, user: User, job_id: int) -> DownloadJob: + job = db.get(DownloadJob, job_id) + if job is None or job.user_id != user.id: + raise HTTPException(status_code=404, detail="Unknown download") + return job + + +# --- profiles ------------------------------------------------------------------------------ + +def _serialize_profile(p: DownloadProfile) -> dict: + return { + "id": p.id, + "name": p.name, + "spec": p.spec, + "is_builtin": p.is_builtin, + "sort_order": p.sort_order, + } + + +@router.get("/profiles") +def list_profiles( + user: User = Depends(require_human), db: Session = Depends(get_db) +) -> list[dict]: + rows = ( + db.execute( + select(DownloadProfile) + .where( + (DownloadProfile.is_builtin.is_(True)) + | (DownloadProfile.user_id == user.id) + ) + .order_by(DownloadProfile.is_builtin.desc(), DownloadProfile.sort_order, DownloadProfile.id) + ) + .scalars() + .all() + ) + return [_serialize_profile(p) for p in rows] + + +@router.post("/profiles") +def create_profile( + payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + name = (payload.get("name") or "").strip() + spec = payload.get("spec") + if not name: + raise HTTPException(status_code=400, detail="A profile name is required.") + if not isinstance(spec, dict): + raise HTTPException(status_code=400, detail="spec must be an object") + maxpos = db.execute( + select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where( + DownloadProfile.user_id == user.id + ) + ).scalar_one() + p = DownloadProfile( + user_id=user.id, name=name[:80], spec=spec, is_builtin=False, sort_order=maxpos + 1 + ) + db.add(p) + db.commit() + return _serialize_profile(p) + + +@router.patch("/profiles/{profile_id}") +def update_profile( + profile_id: int, + payload: dict, + user: User = Depends(require_human), + db: Session = Depends(get_db), +) -> dict: + p = db.get(DownloadProfile, profile_id) + if p is None or p.is_builtin or p.user_id != user.id: + raise HTTPException(status_code=404, detail="Unknown profile") + if "name" in payload: + name = (payload.get("name") or "").strip() + if not name: + raise HTTPException(status_code=400, detail="A profile name is required.") + p.name = name[:80] + if "spec" in payload: + if not isinstance(payload["spec"], dict): + raise HTTPException(status_code=400, detail="spec must be an object") + p.spec = payload["spec"] + db.commit() + return _serialize_profile(p) + + +@router.delete("/profiles/{profile_id}") +def delete_profile( + profile_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + p = db.get(DownloadProfile, profile_id) + if p is None or p.is_builtin or p.user_id != user.id: + raise HTTPException(status_code=404, detail="Unknown profile") + db.delete(p) + db.commit() + return {"deleted": profile_id} + + +def _resolve_spec(db: Session, user: User, payload: dict) -> tuple[dict, int | None]: + """Pick the format spec for an enqueue: an explicit profile_id (builtin or the user's own), + an inline spec, or the first builtin as a fallback.""" + pid = payload.get("profile_id") + if pid is not None: + p = db.get(DownloadProfile, pid) + if p is None or (not p.is_builtin and p.user_id != user.id): + raise HTTPException(status_code=404, detail="Unknown profile") + return p.spec, p.id + if isinstance(payload.get("spec"), dict): + return payload["spec"], None + fallback = db.execute( + select(DownloadProfile) + .where(DownloadProfile.is_builtin.is_(True)) + .order_by(DownloadProfile.sort_order) + .limit(1) + ).scalar_one_or_none() + if fallback is None: + raise HTTPException(status_code=400, detail="No download profile available.") + return fallback.spec, fallback.id + + +# --- enqueue + list + manage --------------------------------------------------------------- + +@router.post("") +def enqueue_download( + payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + source_kind, source_ref = resolve_source(payload.get("source") or "") + spec, profile_id = _resolve_spec(db, user, payload) + display_name = (payload.get("display_name") or "").strip() or None + try: + job = service.enqueue( + db, user.id, source_kind, source_ref, spec, profile_id, display_name + ) + except quota.QuotaExceeded as e: + raise HTTPException(status_code=422, detail=_quota_message(e)) + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + +def _quota_message(e: quota.QuotaExceeded) -> str: + if e.reason == "max_jobs": + return f"You've reached your download limit ({e.limit} items). Remove some first." + if e.reason == "max_bytes": + gb = e.limit / 1_073_741_824 + return f"You've used your download storage quota ({gb:.1f} GB). Free up space first." + return "Download quota exceeded." + + +@router.get("") +def list_downloads( + user: User = Depends(require_human), db: Session = Depends(get_db) +) -> list[dict]: + jobs = ( + db.execute( + select(DownloadJob) + .where(DownloadJob.user_id == user.id) + .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()) + ) + .scalars() + .all() + ) + assets = _assets_for(db, jobs) + return [_serialize(j, assets.get(j.asset_id)) for j in jobs] + + +@router.get("/usage") +def my_usage(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict: + return quota.usage(db, user.id) + + +# Priority when a source has several jobs (e.g. a done one plus a fresh queued one). +_INDEX_RANK = {"done": 4, "running": 3, "paused": 2, "queued": 1} + + +@router.get("/index") +def my_download_index( + user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + """source_ref -> best active status, for the feed's per-card "downloaded / queued" badge. + Small + cheap so the feed can poll it without loading the full job list.""" + rows = db.execute( + select(DownloadJob.source_ref, DownloadJob.status).where( + DownloadJob.user_id == user.id, + DownloadJob.status.in_(("queued", "running", "paused", "done")), + ) + ).all() + out: dict[str, str] = {} + for ref, status in rows: + if _INDEX_RANK.get(status, 0) > _INDEX_RANK.get(out.get(ref, ""), 0): + out[ref] = status + return out + + +@router.get("/shared") +def shared_with_me( + user: User = Depends(require_human), db: Session = Depends(get_db) +) -> list[dict]: + jobs = ( + db.execute( + select(DownloadJob) + .join(DownloadShare, DownloadShare.job_id == DownloadJob.id) + .where(DownloadShare.to_user_id == user.id) + .order_by(DownloadShare.created_at.desc()) + ) + .scalars() + .all() + ) + assets = _assets_for(db, jobs) + out = [] + for j in jobs: + data = _serialize(j, assets.get(j.asset_id)) + data["shared"] = True + out.append(data) + return out + + +@router.patch("/{job_id}") +def rename_download( + job_id: int, + payload: dict, + user: User = Depends(require_human), + db: Session = Depends(get_db), +) -> dict: + job = _own_job(db, user, job_id) + name = (payload.get("display_name") or "").strip() + # Display name only — the on-disk file keeps its system-decided, id-bound name. + job.display_name = name[:255] or None + db.commit() + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + +def _set_status(db: Session, job: DownloadJob, status: str) -> None: + job.status = status + db.commit() + + +@router.post("/{job_id}/pause") +def pause_download( + job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + job = _own_job(db, user, job_id) + if job.status in ("queued", "running"): + _set_status(db, job, "paused") # a running worker aborts cooperatively via the hook + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + +@router.post("/{job_id}/resume") +def resume_download( + job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + job = _own_job(db, user, job_id) + if job.status in ("paused", "error"): + job.progress = 0 + job.error = None + # If the shared asset itself failed, reset it to pending so the worker actually + # re-downloads — otherwise the requeued job would instantly re-inherit the asset's + # stale error (the worker short-circuits a job whose asset is already errored). + if job.asset_id: + asset = db.get(MediaAsset, job.asset_id) + if asset and asset.status == "error": + asset.status = "pending" + asset.error = None + _set_status(db, job, "queued") + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + +@router.post("/{job_id}/cancel") +def cancel_download( + job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + job = _own_job(db, user, job_id) + if job.status not in ("done", "canceled"): + _release_asset(db, job) # release while the job still counts as holding + job.status = "canceled" + db.commit() + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + +@router.delete("/{job_id}") +def delete_download( + job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + job = _own_job(db, user, job_id) + _release_asset(db, job) + db.delete(job) + db.commit() + return {"deleted": job_id} + + +def _release_asset(db: Session, job: DownloadJob) -> None: + """Drop this job's hold on its asset when it leaves a holding state. Once NO job holds the + asset anymore, delete the file + row immediately — a deleted download should free its disk, + and the shared cache only needs to span *overlapping* holders (a later re-add just downloads + again). Idempotent: a job that's already left holding (canceled/error) is a no-op, so delete + after cancel doesn't double-release.""" + if not job.asset_id or job.status not in ("queued", "running", "paused", "done"): + return + asset = db.get(MediaAsset, job.asset_id) + if asset is None: + return + asset.ref_count = max(0, (asset.ref_count or 0) - 1) + if asset.ref_count == 0 and asset.status == "ready": + if asset.rel_path: + storage.delete_asset_files(settings.download_root, asset.rel_path) + db.delete(asset) + + +# --- file download (range-aware, custom display name) -------------------------------------- + +def _clean_basename(name: str) -> str: + # Emoji/symbol-free but keeps spaces + accents (a readable device filename). + return storage.display_filename(name) + + +def _accessible_job(db: Session, user: User, job_id: int) -> DownloadJob: + job = db.get(DownloadJob, job_id) + if job is None: + raise HTTPException(status_code=404, detail="Unknown download") + if job.user_id == user.id: + return job + shared = db.execute( + select(DownloadShare.id).where( + DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id + ) + ).first() + if shared is None: + raise HTTPException(status_code=404, detail="Unknown download") + return job + + +@router.get("/{job_id}/file") +def download_file( + job_id: int, + request: Request, + user: User = Depends(require_human), + db: Session = Depends(get_db), +): + job = _accessible_job(db, user, job_id) + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + if asset is None or asset.status != "ready" or not asset.rel_path: + raise HTTPException(status_code=409, detail="This download isn't ready.") + path = storage.abs_path(settings.download_root, asset.rel_path).resolve() + root = Path(settings.download_root).resolve() + if root not in path.parents or not path.exists(): + raise HTTPException(status_code=404, detail="File is no longer available.") + + ext = asset.container or path.suffix.lstrip(".") + base = _clean_basename(job.display_name or asset.title or asset.source_ref) + if base.lower().endswith(f".{ext.lower()}"): + base = base[: -(len(ext) + 1)] + filename = f"{base}.{ext}" + + asset.last_access_at = datetime.now(timezone.utc) + db.commit() + # Starlette's FileResponse honours the Range header (206 partial content) for seek/resume. + return FileResponse(path, filename=filename) + + +# --- sharing ------------------------------------------------------------------------------- + +@router.post("/{job_id}/share") +def share_download( + job_id: int, + payload: dict, + user: User = Depends(require_human), + db: Session = Depends(get_db), +) -> dict: + job = _own_job(db, user, job_id) + if job.status != "done" or job.asset_id is None: + raise HTTPException(status_code=409, detail="Only a finished download can be shared.") + target = (payload.get("email") or "").strip().lower() + if not target: + raise HTTPException(status_code=400, detail="A recipient email is required.") + recipient = db.execute( + select(User).where(func.lower(User.email) == target) + ).scalar_one_or_none() + if recipient is None: + raise HTTPException(status_code=404, detail="No user with that email.") + if recipient.id == user.id: + raise HTTPException(status_code=400, detail="That's already your download.") + exists = db.execute( + select(DownloadShare.id).where( + DownloadShare.job_id == job_id, DownloadShare.to_user_id == recipient.id + ) + ).first() + if exists is None: + db.add( + DownloadShare(job_id=job_id, from_user_id=user.id, to_user_id=recipient.id) + ) + from app.notifications import create_notification + + create_notification( + db, + user_id=recipient.id, + type="download_shared", + title=(job.display_name or "A download") + " was shared with you", + data={"kind": "download_shared", "from": user.email, "job_id": job_id}, + commit=False, + ) + db.commit() + return {"shared_with": recipient.email} + + +@router.delete("/{job_id}/share/{email}") +def unshare_download( + job_id: int, + email: str, + user: User = Depends(require_human), + db: Session = Depends(get_db), +) -> dict: + job = _own_job(db, user, job_id) + recipient = db.execute( + select(User).where(func.lower(User.email) == email.strip().lower()) + ).scalar_one_or_none() + if recipient is not None: + row = db.execute( + select(DownloadShare).where( + DownloadShare.job_id == job_id, + DownloadShare.to_user_id == recipient.id, + ) + ).scalar_one_or_none() + if row is not None: + db.delete(row) + db.commit() + return {"ok": True} + + +# --- admin --------------------------------------------------------------------------------- + +@admin_router.get("") +def admin_list_downloads( + admin: User = Depends(admin_user), db: Session = Depends(get_db) +) -> list[dict]: + jobs = ( + db.execute( + select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(500) + ) + .scalars() + .all() + ) + assets = _assets_for(db, jobs) + emails = { + u.id: u.email + for u in db.execute( + select(User).where(User.id.in_({j.user_id for j in jobs})) + ).scalars() + } + out = [] + for j in jobs: + data = _serialize(j, assets.get(j.asset_id)) + data["user_email"] = emails.get(j.user_id) + out.append(data) + return out + + +@admin_router.get("/storage") +def admin_storage( + admin: User = Depends(admin_user), db: Session = Depends(get_db) +) -> dict: + ready = db.execute( + select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where( + MediaAsset.status == "ready" + ) + ).one() + total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar() + per_user = db.execute( + select(DownloadJob.user_id, func.count(func.distinct(DownloadJob.asset_id))) + .where(DownloadJob.asset_id.is_not(None)) + .group_by(DownloadJob.user_id) + ).all() + emails = { + u.id: u.email + for u in db.execute( + select(User).where(User.id.in_([uid for uid, _ in per_user])) + ).scalars() + } + return { + "ready_files": ready[0], + "total_bytes": int(ready[1]), + "total_assets": total_assets, + "total_cap_bytes": settings.download_total_max_bytes, + "per_user": [ + {"user_id": uid, "email": emails.get(uid), "footprint_bytes": quota.footprint(db, uid)} + for uid, _ in per_user + ], + } + + +@admin_router.get("/quota/{user_id}") +def admin_get_quota( + user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db) +) -> dict: + lim = quota.resolve(db, user_id) + row = db.get(DownloadQuota, user_id) + return { + "user_id": user_id, + "custom": row is not None, + "max_bytes": lim.max_bytes, + "max_concurrent": lim.max_concurrent, + "max_jobs": lim.max_jobs, + "unlimited": lim.unlimited, + } + + +@admin_router.put("/quota/{user_id}") +def admin_set_quota( + user_id: int, + payload: dict, + admin: User = Depends(admin_user), + db: Session = Depends(get_db), +) -> dict: + if db.get(User, user_id) is None: + raise HTTPException(status_code=404, detail="Unknown user") + cur = quota.resolve(db, user_id) + row = db.get(DownloadQuota, user_id) + if row is None: + row = DownloadQuota( + user_id=user_id, + max_bytes=cur.max_bytes, + max_concurrent=cur.max_concurrent, + max_jobs=cur.max_jobs, + unlimited=cur.unlimited, + ) + db.add(row) + if "max_bytes" in payload: + row.max_bytes = max(0, int(payload["max_bytes"])) + if "max_concurrent" in payload: + row.max_concurrent = max(1, int(payload["max_concurrent"])) + if "max_jobs" in payload: + row.max_jobs = max(1, int(payload["max_jobs"])) + if "unlimited" in payload: + row.unlimited = bool(payload["unlimited"]) + db.commit() + return admin_get_quota(user_id, admin, db) + + +@admin_router.delete("/quota/{user_id}") +def admin_reset_quota( + user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db) +) -> dict: + row = db.get(DownloadQuota, user_id) + if row is not None: + db.delete(row) + db.commit() + return admin_get_quota(user_id, admin, db) diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index cdec9f3..8c0e506 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -12,6 +12,7 @@ from sqlalchemy import select from app import progress, quota from app.config import settings from app.db import SessionLocal +from app.downloads.gc import run_download_gc from app.models import SchedulerSetting from app.notifications import create_notification from app.state import is_sync_paused @@ -59,6 +60,7 @@ JOB_INTERVALS: dict[str, int] = { "maintenance": settings.maintenance_interval_minutes, "demo_reset": settings.demo_reset_minutes, "explore_cleanup": settings.explore_cleanup_minutes, + "download_gc": settings.download_gc_minutes, } # Sane bounds for an admin-set interval (minutes). @@ -276,6 +278,12 @@ def _explore_cleanup_job() -> None: _job("explore_cleanup", purge_ephemeral) +def _download_gc_job() -> None: + # Retention GC for the download center: pre-expiry warnings, TTL deletion, LRU eviction. + # Pure disk/DB work (no YouTube quota). + _job("download_gc", run_download_gc) + + # job_id -> wrapper. The single source of truth for which jobs exist and how to run one, # shared by start_scheduler (recurring registration) and trigger_job (manual "run now"). JOB_FUNCS: dict[str, Callable[[], None]] = { @@ -289,6 +297,7 @@ JOB_FUNCS: dict[str, Callable[[], None]] = { "maintenance": _maintenance_job, "demo_reset": _demo_reset_job, "explore_cleanup": _explore_cleanup_job, + "download_gc": _download_gc_job, } diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index a7a77d9..89d6865 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -11,6 +11,7 @@ from sqlalchemy.orm import Session from app import progress, sysconfig from app.models import Channel, Video +from app.titles import normalize_title from app.youtube.client import YouTubeClient, best_thumbnail from app.youtube.rss import fetch_channel_feed from app.youtube.shorts import make_client, probe_is_short @@ -53,6 +54,11 @@ def _insert_stubs(db: Session, rows: list[dict]) -> int: return 0 seen: dict[str, dict] = {} for row in rows: + # Normalize the title at first insert too (RSS/backfill stubs), so a brand-new video + # never flashes its raw title before enrichment refreshes it. + raw = row.get("title") + if raw: + row = {**row, "original_title": raw, "title": normalize_title(raw)} seen[row["id"]] = row stmt = ( pg_insert(Video) @@ -225,7 +231,11 @@ def apply_video_details(video: Video, item: dict) -> None: live = item.get("liveStreamingDetails") status = item.get("status", {}) - video.title = snippet.get("title") or video.title + raw_title = snippet.get("title") + if raw_title: + # Store the raw title and a normalized one for display (feed/search/downloads). + video.original_title = raw_title + video.title = normalize_title(raw_title) video.description = snippet.get("description") if not video.published_at: video.published_at = parse_dt(snippet.get("publishedAt")) diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index ff396c7..ed6c79b 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -66,6 +66,16 @@ SPECS: tuple[ConfigSpec, ...] = ( ConfigSpec("google_client_secret", "str", "google", "google_client_secret", secret=True), # --- Access / registration --- ConfigSpec("allow_registration", "bool", "access", "allow_registration"), + # --- Download center (yt-dlp) — per-user defaults + retention/GC + layout --- + ConfigSpec("download_total_max_bytes", "int", "downloads", "download_total_max_bytes", min=0), + ConfigSpec("download_default_max_bytes", "int", "downloads", "download_default_max_bytes", min=0), + ConfigSpec("download_default_max_concurrent", "int", "downloads", "download_default_max_concurrent", min=1, max=20), + ConfigSpec("download_default_max_jobs", "int", "downloads", "download_default_max_jobs", min=1, max=100_000), + ConfigSpec("download_retention_days", "int", "downloads", "download_retention_days", min=0, max=3_650), + ConfigSpec("download_gc_grace_days", "int", "downloads", "download_gc_grace_days", min=0, max=3_650), + ConfigSpec("download_layout", "str", "downloads", "download_layout"), + ConfigSpec("download_worker_concurrency", "int", "downloads", "download_worker_concurrency", min=1, max=16), + ConfigSpec("sponsorblock_default", "bool", "downloads", "sponsorblock_default"), ) _BY_KEY: dict[str, ConfigSpec] = {s.key: s for s in SPECS} diff --git a/backend/app/titles.py b/backend/app/titles.py new file mode 100644 index 0000000..e609f58 --- /dev/null +++ b/backend/app/titles.py @@ -0,0 +1,80 @@ +"""Video title normalization — clean the noisy YouTube titles for display + storage. + +Applied where a video's title is written (enrichment) and where a download's title comes from +yt-dlp, so the feed, search, channel pages, and the download center all show tidy titles. The +raw title is preserved in `Video.original_title` so this is reversible / re-derivable. + +Rules (user-approved "option b"): + 1. Drop emoji / pictographs / symbols / control chars (keep accents — HU/DE/…). + 2. Strip trailing SEO hashtag clusters (word hashtags at the end); a numeric "#3" episode + marker survives. + 3. De-shout ALL-CAPS words, context-aware: + - If the title is *mostly shouting* (≥ half the multi-letter words are all-caps), every + all-caps word is Title-cased (short shouted words like CAR/WAR/ÉN get fixed too), and + the first letter is capitalized. Known acronyms (PS, AI, USA, PC…) and words with + digits (PS5, 3D) are kept; function words (to/the/és/az…) are lowercased. + - Otherwise only long all-caps words (≥4 letters) are Title-cased, so a lone acronym in + an otherwise normal title is left alone. + 4. Collapse repeated punctuation (!!! → !) and whitespace. +""" +import re +import unicodedata + +_DROP_CATEGORIES = {"So", "Sk", "Cc", "Cf", "Cs", "Co", "Cn"} +_LETTER_RUN = re.compile(r"[^\W\d_]+", re.UNICODE) +_TRAILING_HASHTAGS = re.compile(r"(?:\s+#[^\s#]*[^\W\d\s_][^\s#]*)+\s*$", re.UNICODE) +_REPEAT_PUNCT = re.compile(r"([!?.,\-])\1{1,}") + +# Short function words → lowercased when the title is de-shouted (proper title-case style). +_SHORT_LOWER = { + "to", "the", "and", "of", "a", "an", "in", "on", "at", "for", "or", "vs", "the", + "és", "az", "egy", "meg", "nem", "hogy", "de", "ha", "der", "die", "das", "und", "von", +} +# Kept as-is even when everything else is de-shouted (common acronyms; compared lowercased). +_ACRONYMS = { + "ps", "ai", "pc", "tv", "hd", "4k", "8k", "uk", "eu", "us", "usa", "gta", "rpg", "fps", + "diy", "vr", "ar", "id", "ok", "faq", "nasa", "fbi", "cia", "ceo", "amd", "pdf", "usb", + "gps", "api", "dj", "mc", "suv", "gpu", "cpu", "ssd", "hdmi", "led", "ufo", "dna", "nba", + "nfl", "asmr", "pov", "diy", "wtf", "lol", "rtx", "gtx", "ios", "mmo", "vip", "hp", +} + + +def _titlecase(w: str) -> str: + return w[0].upper() + w[1:].lower() + + +def normalize_title(raw: str | None) -> str | None: + """Return the cleaned title, or the input unchanged for empty/whitespace.""" + if not raw or not raw.strip(): + return raw + t = unicodedata.normalize("NFKC", raw) + t = "".join(ch for ch in t if unicodedata.category(ch) not in _DROP_CATEGORIES) + t = _TRAILING_HASHTAGS.sub("", t) + + multi = [r for r in _LETTER_RUN.findall(t) if len(r) >= 2] + caps = [r for r in multi if r.isupper()] + shouting = len(caps) >= 2 and len(caps) >= 0.5 * len(multi) + + def repl(m: re.Match) -> str: + w = m.group(0) + if len(w) < 2 or not w.isupper(): + return w + low = w.lower() + if low in _ACRONYMS: + return w + if low in _SHORT_LOWER: + return low + if shouting or len(w) >= 4: + return _titlecase(w) + return w # short all-caps in a non-shouting title → likely an acronym, keep + + t = _LETTER_RUN.sub(repl, t) + t = _REPEAT_PUNCT.sub(r"\1", t) + t = re.sub(r"\s+", " ", t).strip() + + if shouting: # ensure the first letter is capitalized (a leading function word got lowered) + for i, ch in enumerate(t): + if ch.isalpha(): + t = t[:i] + ch.upper() + t[i + 1:] + break + return t or raw diff --git a/backend/app/worker.py b/backend/app/worker.py new file mode 100644 index 0000000..4633e5e --- /dev/null +++ b/backend/app/worker.py @@ -0,0 +1,477 @@ +"""Download worker process entry point. + +Runs in a DEDICATED container (`command: python -m app.worker`, `WORKER_ENABLED=1`), separate +from the API so long yt-dlp downloads + ffmpeg postprocessing never block web requests. + +Design: + * N loop threads (download_worker_concurrency) each claim a `queued` DownloadJob with + FOR UPDATE SKIP LOCKED and process it. + * The shared MediaAsset is the dedup unit. A job whose asset is already `ready` finishes + instantly (no re-download); a fresh asset is claimed pending→downloading by exactly one + worker (a DB compare-and-set), so two jobs for the same video+format never double-download. + * Live progress is written to the job row (short-lived sessions, throttled ~1/s) — the + frontend polls it; the worker can't reach the API's in-process WebSocket registry. + * Pause/cancel are cooperative: the progress hook re-reads the job status and aborts the + yt-dlp download if it's no longer `running`. + * Crash-safe: on startup, orphaned `running` jobs are requeued and `downloading` assets reset. +""" +import logging +import shutil +import signal +import threading +import time +from datetime import date, datetime, timedelta, timezone +from pathlib import Path + +import yt_dlp +from sqlalchemy import text + +from app.config import settings +from app.db import SessionLocal +from app.downloads import formats, quota, service, storage +from app.models import DownloadJob, MediaAsset +from app.titles import normalize_title + +log = logging.getLogger("siftlode.worker") + +_stop = threading.Event() + + +class _Aborted(Exception): + """Raised inside the progress hook when the job was paused/canceled mid-download.""" + + +def _handle_signal(signum, frame): + _stop.set() + + +# --- small DB helpers (short-lived sessions; the download itself holds no transaction) ------- + +def _set_job(job_id: int, **fields) -> None: + with SessionLocal() as db: + job = db.get(DownloadJob, job_id) + if job is None: + return + for k, v in fields.items(): + setattr(job, k, v) + db.commit() + + +def _job_status(job_id: int) -> str | None: + with SessionLocal() as db: + return db.execute( + text("SELECT status FROM download_jobs WHERE id = :id"), {"id": job_id} + ).scalar() + + +def _parse_upload_date(raw) -> date | None: + if not raw: + return None + try: + return datetime.strptime(str(raw), "%Y%m%d").date() + except ValueError: + return None + + +# --- claim + recovery ----------------------------------------------------------------------- + +def _recover_orphans() -> None: + """Reset state left behind by a previous crash so nothing is stuck.""" + with SessionLocal() as db: + db.execute(text("UPDATE download_jobs SET status='queued' WHERE status='running'")) + db.execute(text("UPDATE media_assets SET status='pending' WHERE status='downloading'")) + db.commit() + + +def _claim_job() -> int | None: + """Claim the oldest queued job whose user is under their per-user concurrency limit. + + Fetches a window of locked candidates (FOR UPDATE SKIP LOCKED) and picks the first eligible + one, so one user flooding the queue can't starve others beyond their max_concurrent.""" + with SessionLocal() as db: + rows = db.execute( + text( + """ + SELECT id, user_id FROM download_jobs + WHERE status='queued' + ORDER BY queue_pos, id + FOR UPDATE SKIP LOCKED + LIMIT 50 + """ + ) + ).fetchall() + chosen = next( + (jid for jid, uid in rows if not quota.at_concurrency_limit(db, uid)), None + ) + if chosen is None: + db.commit() # release the row locks; nothing eligible right now + return None + db.execute( + text( + "UPDATE download_jobs SET status='running', phase='starting', updated_at=now() " + "WHERE id = :id" + ), + {"id": chosen}, + ) + db.commit() + return chosen + + +def _claim_asset(asset_id: int) -> bool: + """Compare-and-set pending→downloading. True if THIS worker won the right to download.""" + with SessionLocal() as db: + row = db.execute( + text( + "UPDATE media_assets SET status='downloading' " + "WHERE id = :id AND status='pending' RETURNING id" + ), + {"id": asset_id}, + ).fetchone() + db.commit() + return row is not None + + +# --- processing ----------------------------------------------------------------------------- + +def _process_job(job_id: int) -> None: + with SessionLocal() as db: + job = db.get(DownloadJob, job_id) + if job is None: + return + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + spec = dict(job.profile_snapshot or {}) + source_kind, source_ref = job.source_kind, job.source_ref + asset_id = asset.id if asset else None + asset_status = asset.status if asset else None + + if asset is None: + _set_job(job_id, status="error", error="No media asset", phase=None) + return + + if asset_status == "ready": + _finish_from_cache(job_id, asset_id) + return + if asset_status == "error": + _set_job(job_id, status="error", error=asset.error or "Download failed", phase=None) + return + if asset_status == "downloading": + # Another worker owns this asset; wait and let the job be reclaimed later. + _set_job(job_id, status="queued", phase="waiting") + time.sleep(2) + return + + # asset_status == "pending": try to win the download. + if not _claim_asset(asset_id): + _set_job(job_id, status="queued", phase="waiting") + time.sleep(2) + return + + _download(job_id, asset_id, source_kind, source_ref, spec) + + +def _finish_from_cache(job_id: int, asset_id: int) -> None: + with SessionLocal() as db: + job = db.get(DownloadJob, job_id) + asset = db.get(MediaAsset, asset_id) + if job and asset: + job.status = "done" + job.progress = 100 + job.phase = None + if not job.display_name: + job.display_name = asset.title + asset.last_access_at = datetime.now(timezone.utc) + db.commit() + + +def _fill_asset_meta_early(asset_id: int, info: dict) -> None: + """Populate an asset's display metadata from yt-dlp's info_dict the first time it's seen + (only if not already filled at enqueue from our catalog) — so an ad-hoc URL's row shows a + real title + thumbnail while it downloads, not just the URL. Zero extra network cost.""" + with SessionLocal() as db: + asset = db.get(MediaAsset, asset_id) + if asset is None or asset.title: + return + asset.title = normalize_title(info.get("title")) + asset.uploader = info.get("uploader") or info.get("channel") + asset.thumbnail_url = info.get("thumbnail") + asset.duration_s = int(info["duration"]) if info.get("duration") else None + asset.upload_date = _parse_upload_date(info.get("upload_date")) + db.commit() + + +def _make_progress_hook(job_id: int, asset_id: int): + state = {"last_db": 0.0, "last_check": 0.0, "meta_done": False} + + def hook(d: dict) -> None: + now = time.monotonic() + # Cooperative pause/cancel: if the job is no longer 'running', abort the download. + if now - state["last_check"] >= 2.0: + state["last_check"] = now + if _job_status(job_id) not in ("running", None): + raise _Aborted + if not state["meta_done"] and d.get("info_dict"): + state["meta_done"] = True + try: + _fill_asset_meta_early(asset_id, d["info_dict"]) + except Exception: # noqa: BLE001 — metadata is best-effort, never fail a download + pass + if d.get("status") != "downloading": + return + if now - state["last_db"] < 1.0: + return + state["last_db"] = now + # yt-dlp downloads the video and audio streams SEPARATELY (each 0→100%), then merges. + # Label the phase so the user understands the bar resetting between streams instead of + # seeing a mysterious 95%→5% jump. + info = d.get("info_dict") or {} + vcodec = info.get("vcodec") or "none" + acodec = info.get("acodec") or "none" + if vcodec != "none" and acodec == "none": + phase = "video" + elif acodec != "none" and vcodec == "none": + phase = "audio" + else: + phase = "media" + total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0 + done = d.get("downloaded_bytes") or 0 + pct = int(done * 100 / total) if total else 0 + _set_job( + job_id, + progress=min(pct, 99), + speed_bps=int(d["speed"]) if d.get("speed") else None, + eta_s=int(d["eta"]) if d.get("eta") else None, + phase=phase, + ) + + return hook + + +def _pp_phase(pp: str) -> str: + """Map a yt-dlp postprocessor class name to a user-facing phase label. ffmpeg post-steps + aren't byte-progress operations (no %), so the UI shows the step name + an indeterminate + pulse — this makes the long merge/finalize legible instead of a silent 'Processing'.""" + if "Merg" in pp: + return "merging" + if "ExtractAudio" in pp: + return "audio_extract" + if "Thumbnail" in pp: + return "thumbnail" + if "SponsorBlock" in pp or "ModifyChapters" in pp: + return "sponsorblock" + if "Metadata" in pp: + return "metadata" + return "processing" + + +def _make_pp_hook(job_id: int): + """Postprocessor hook: surface the current ffmpeg step so the bar shows a concrete label + (Merging / Embedding thumbnail / …) instead of freezing at 100% after the streams finish.""" + + def hook(d: dict) -> None: + if d.get("status") in ("started", "processing"): + _set_job(job_id, phase=_pp_phase(d.get("postprocessor") or ""), speed_bps=None, eta_s=None) + + return hook + + +def _staging_dir(asset_id: int) -> Path: + return Path(settings.download_root) / ".staging" / f"asset-{asset_id}" + + +def _find_thumbnail(staging: Path, media: Path) -> Path | None: + for p in staging.iterdir(): + if p != media and p.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp"): + return p + return None + + +# Errors worth retrying with a different player-client set (YouTube per-client flakiness). +_RETRIABLE_MARKERS = ( + "not a bot", + "DRM protected", + "Requested format is not available", + "Only images are available", + "Sign in to confirm", +) + + +def _extract_with_fallback(job_id: int, asset_id: int, url: str, spec: dict, staging: Path): + """Download via yt-dlp, trying the primary player clients then the POT-backed fallback set + on a bot/DRM/format failure. Returns the info dict of the successful attempt.""" + client_sets = [ + c for c in (settings.player_client_list, settings.player_client_fallback_list) if c + ] or [None] + outtmpl = str(staging / "%(id)s.%(ext)s") + last_exc: Exception | None = None + for i, clients in enumerate(client_sets): + opts = formats.build_ydl_opts( + spec, outtmpl, _make_progress_hook(job_id, asset_id), + settings.download_pot_base_url, clients, + ) + opts["postprocessor_hooks"] = [_make_pp_hook(job_id)] + try: + with yt_dlp.YoutubeDL(opts) as ydl: + return ydl.extract_info(url, download=True) + except yt_dlp.utils.DownloadError as exc: + last_exc = exc + retriable = any(m in str(exc) for m in _RETRIABLE_MARKERS) + if i < len(client_sets) - 1 and retriable: + log.info("job %s: clients %s failed (%s); retrying with fallback", job_id, clients, str(exc)[:70]) + for p in list(staging.iterdir()): # drop partials before the retry + try: + p.unlink() + except OSError: + pass + _set_job(job_id, progress=0, phase="downloading") + continue + raise + raise last_exc # unreachable (loop either returns or raises) + + +def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spec: dict) -> None: + staging = _staging_dir(asset_id) + try: + if staging.exists(): + shutil.rmtree(staging, ignore_errors=True) + staging.mkdir(parents=True, exist_ok=True) + + _set_job(job_id, phase="downloading", progress=0) + url = service.source_url(source_kind, source_ref) + info = _extract_with_fallback(job_id, asset_id, url, spec, staging) + + req = (info.get("requested_downloads") or [{}])[0] + produced = Path(req.get("filepath") or "") + if not produced.exists(): + # Fallback: the single media file left in staging (ignore the thumbnail). + cands = [p for p in staging.iterdir() if p.suffix.lower() not in (".jpg", ".jpeg", ".png", ".webp")] + if not cands: + raise RuntimeError("yt-dlp produced no output file") + produced = cands[0] + + ext = produced.suffix.lstrip(".").lower() + meta = storage.MediaMeta( + video_id=info.get("id") or source_ref, + title=normalize_title(info.get("title")) or source_ref, + uploader=info.get("uploader") or info.get("channel") or "Unknown Channel", + upload_date=_parse_upload_date(info.get("upload_date")), + duration_s=int(info["duration"]) if info.get("duration") else None, + description=info.get("description"), + thumbnail_url=info.get("thumbnail"), + ) + + rel = storage.rel_path(meta, ext, settings.download_layout) + thumb = _find_thumbnail(staging, produced) + storage.place_file(produced, settings.download_root, rel) + nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, thumb) + + final = storage.abs_path(settings.download_root, rel) + size = final.stat().st_size if final.exists() else None + expires = datetime.now(timezone.utc) + timedelta(days=settings.download_retention_days) + + with SessionLocal() as db: + asset = db.get(MediaAsset, asset_id) + if asset: + asset.status = "ready" + asset.rel_path = rel + asset.size_bytes = size + asset.duration_s = meta.duration_s + asset.width = int(info["width"]) if info.get("width") else None + asset.height = int(info["height"]) if info.get("height") else None + asset.container = ext + asset.vcodec = info.get("vcodec") if info.get("vcodec") not in (None, "none") else None + asset.acodec = info.get("acodec") if info.get("acodec") not in (None, "none") else None + asset.title = meta.title + asset.uploader = meta.uploader + asset.upload_date = meta.upload_date + asset.thumbnail_url = meta.thumbnail_url + asset.nfo_written = nfo_ok + asset.error = None + asset.last_access_at = datetime.now(timezone.utc) + asset.expires_at = expires + job = db.get(DownloadJob, job_id) + if job: + job.status = "done" + job.progress = 100 + job.phase = None + job.speed_bps = None + job.eta_s = None + if not job.display_name: + job.display_name = meta.title + db.commit() + log.info("job %s done: %s", job_id, rel) + + except _Aborted: + # Paused/canceled mid-download: free the asset for a later retry; keep the job's status + # (the route set it to paused/canceled). ref_count is adjusted by the route. + _reset_asset_pending(asset_id) + log.info("job %s aborted (pause/cancel)", job_id) + except Exception as exc: # noqa: BLE001 — surface any failure to the user, keep worker alive + msg = str(exc)[:500] + with SessionLocal() as db: + asset = db.get(MediaAsset, asset_id) + if asset: + asset.status = "error" + asset.error = msg + job = db.get(DownloadJob, job_id) + if job: + job.status = "error" + job.error = msg + job.phase = None + db.commit() + log.warning("job %s failed: %s", job_id, msg) + finally: + shutil.rmtree(staging, ignore_errors=True) + + +def _reset_asset_pending(asset_id: int) -> None: + with SessionLocal() as db: + asset = db.get(MediaAsset, asset_id) + if asset and asset.status == "downloading": + asset.status = "pending" + db.commit() + + +# --- loop ----------------------------------------------------------------------------------- + +def _worker_loop(idx: int) -> None: + while not _stop.is_set(): + try: + job_id = _claim_job() + except Exception: # noqa: BLE001 + log.exception("claim failed") + _stop.wait(settings.download_poll_seconds) + continue + if job_id is None: + _stop.wait(settings.download_poll_seconds) + continue + try: + _process_job(job_id) + except Exception: # noqa: BLE001 + log.exception("processing job %s crashed", job_id) + _set_job(job_id, status="error", error="Worker error", phase=None) + + +def main() -> None: + logging.basicConfig(level=logging.INFO) + signal.signal(signal.SIGTERM, _handle_signal) + signal.signal(signal.SIGINT, _handle_signal) + + if not settings.worker_enabled: + log.info("WORKER_ENABLED is off; worker exiting (nothing to do).") + return + + _recover_orphans() + n = max(1, settings.download_worker_concurrency) + log.info("Download worker started (concurrency=%d, root=%s).", n, settings.download_root) + threads = [threading.Thread(target=_worker_loop, args=(i,), daemon=True) for i in range(n)] + for t in threads: + t.start() + while not _stop.is_set(): + _stop.wait(1.0) + log.info("Download worker stopping…") + for t in threads: + t.join(timeout=5) + + +if __name__ == "__main__": + main() diff --git a/backend/requirements.txt b/backend/requirements.txt index 67f14ad..2a8cf38 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -13,3 +13,12 @@ py3langid>=0.3,<1.0 itsdangerous>=2.1,<3.0 cryptography>=46.0.7,<47 argon2-cffi>=23.1,<26 +# Download center: yt-dlp does the extraction/download; ffmpeg (OS-level in the Dockerfile) +# does the merge + postprocessing. The [default] extra pulls yt-dlp-ejs (the challenge-solver +# scripts Deno runs for YouTube's n-signature). yt-dlp is release-often (YouTube changes), so +# keep the floor loose and bump it when extraction breaks. +yt-dlp[default]>=2025.5.22 +# PO Token provider plugin: talks to the bgutil sidecar (brainicism/bgutil-ytdlp-pot-provider) +# to mint YouTube Proof-of-Origin tokens on demand, bypassing "confirm you're not a bot" and +# unlocking high-quality formats. Pin to match the sidecar image tag (versions must align). +bgutil-ytdlp-pot-provider==1.3.1 diff --git a/docker-compose.localdev.yml b/docker-compose.localdev.yml index 37260aa..48b06ff 100644 --- a/docker-compose.localdev.yml +++ b/docker-compose.localdev.yml @@ -43,6 +43,9 @@ services: DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD:-siftlode}@db:5432/${POSTGRES_DB:-siftlode} # This instance owns its scheduler now (its own DB, so no double-write concern). SCHEDULER_ENABLED: "true" + # Download center: the API serves finished files (it does NOT run the worker loop). + DOWNLOAD_ROOT: /downloads + WORKER_ENABLED: "false" depends_on: db: condition: service_healthy @@ -51,6 +54,49 @@ services: - apparmor:unconfined ports: - "${APP_PORT:-8080}:8000" + volumes: + # Downloaded media lives on the host so you can inspect the generated Plex tree. + # Gitignored. The worker writes here; the API reads it to serve local downloads. + - ./downloads:/downloads + restart: unless-stopped + + # Dedicated download worker: same image, runs the yt-dlp job loop instead of the API. + # It shares the DB (job queue) and the downloads mount with the API. + worker: + build: + context: . + dockerfile: Dockerfile + args: + APP_VERSION: ${APP_VERSION:-dev} + GIT_SHA: ${GIT_SHA:-unknown} + BUILD_DATE: ${BUILD_DATE:-} + command: ["python", "-m", "app.worker"] + env_file: .env + environment: + DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD:-siftlode}@db:5432/${POSTGRES_DB:-siftlode} + # The worker must not also run the scheduler; the API owns that. + SCHEDULER_ENABLED: "false" + DOWNLOAD_ROOT: /downloads + WORKER_ENABLED: "true" + depends_on: + db: + condition: service_healthy + bgutil-pot: + condition: service_started + security_opt: + - apparmor:unconfined + volumes: + - ./downloads:/downloads + restart: unless-stopped + + # PO-token provider: mints YouTube Proof-of-Origin tokens on demand so the worker bypasses + # bot-detection and gets high-quality formats. The worker reaches it at http://bgutil-pot:4416 + # (DOWNLOAD_POT_BASE_URL). Pin the tag to match the plugin version in requirements.txt. + bgutil-pot: + image: brainicism/bgutil-ytdlp-pot-provider:1.3.1 + init: true + security_opt: + - apparmor:unconfined restart: unless-stopped volumes: diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index e05e41e..b33ee87 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -57,6 +57,8 @@ import AdminUsers, { focusAccessRequestsTab } from "./components/AdminUsers"; import SettingsPanel from "./components/SettingsPanel"; import NotificationsPanel from "./components/NotificationsPanel"; import Messages from "./components/Messages"; +import DownloadCenter from "./components/DownloadCenter"; +import { setNavigator } from "./lib/nav"; import ChatDock from "./components/ChatDock"; import OnboardingWizard from "./components/OnboardingWizard"; import { shouldAutoOpenOnboarding } from "./lib/onboarding"; @@ -279,6 +281,8 @@ export default function App() { } go(); } + // Expose the current setPage to decoupled callers (download toast "View", etc.). + useEffect(() => setNavigator(setPage)); function setSidebarLayout(next: SidebarLayout) { setSidebarLayoutState(next); @@ -728,6 +732,8 @@ export default function App() {
+ ) : page === "downloads" && !meQuery.data!.is_demo ? ( + ) : page === "settings" ? (
+ +
+
{jobTitle(job)}
+
+ {job.asset?.uploader || job.source_ref} +
+
+ + {job.asset?.size_bytes ? · {formatBytes(job.asset.size_bytes)} : null} + {job.asset?.duration_s ? · {formatDuration(job.asset.duration_s)} : null} + {job.error ? · {job.error} : null} +
+ {running && ( +
+ {job.phase && !DOWNLOAD_PHASES.includes(job.phase) ? ( + // ffmpeg post-steps aren't byte-progress — show an indeterminate pulse, no %. +
+
+
+ ) : ( + <> +
+
+
+
+ {job.progress}% + {job.speed_bps ? {formatSpeed(job.speed_bps)} : null} + {job.eta_s ? {formatEta(job.eta_s)} : null} +
+ + )} +
+ )} +
+
{children}
+
+ ); +} + +// --- Rename + Share small modals ------------------------------------------------------------ + +function RenameModal({ job, onClose }: { job: DownloadJob; onClose: () => void }) { + const { t } = useTranslation(); + const qc = useQueryClient(); + const [name, setName] = useState(jobTitle(job)); + const save = useMutation({ + mutationFn: () => api.renameDownload(job.id, name.trim()), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ["downloads"] }); + onClose(); + }, + }); + return ( + + + setName(e.target.value)} className={inputCls} autoFocus /> +

{t("downloads.rename.hint")}

+
+ +
+ + ); +} + +function ShareModal({ job, onClose }: { job: DownloadJob; onClose: () => void }) { + const { t } = useTranslation(); + const [email, setEmail] = useState(""); + const share = useMutation({ + mutationFn: () => api.shareDownload(job.id, email.trim()), + onSuccess: (r) => { + notify({ level: "success", message: t("downloads.share.done", { email: r.shared_with }) }); + onClose(); + }, + }); + return ( + + + setEmail(e.target.value)} + placeholder={t("downloads.share.placeholder")} + className={inputCls} + autoFocus + /> +
+ +
+ + ); +} + +// --- Usage bar ------------------------------------------------------------------------------ + +function UsageBar() { + const { t } = useTranslation(); + const usage = useQuery({ queryKey: ["download-usage"], queryFn: api.downloadUsage }); + const u = usage.data; + if (!u) return null; + const pct = u.unlimited || !u.max_bytes ? 0 : Math.min(100, (u.footprint_bytes / u.max_bytes) * 100); + return ( +
+
+ {t("downloads.usage.title")} + + {formatBytes(u.footprint_bytes)} + {u.unlimited ? "" : ` / ${formatBytes(u.max_bytes)}`} ·{" "} + {u.unlimited ? t("downloads.usage.unlimited") : t("downloads.usage.items", { used: u.active_jobs, max: u.max_jobs })} + +
+ {!u.unlimited && ( +
+
90 ? "bg-red-400" : "bg-accent")} style={{ width: `${pct}%` }} /> +
+ )} +
+ ); +} + +// --- Admin system tab ----------------------------------------------------------------------- + +function QuotaModal({ userId, email, onClose }: { userId: number; email: string; onClose: () => void }) { + const { t } = useTranslation(); + const qc = useQueryClient(); + const q = useQuery({ queryKey: ["dl-quota", userId], queryFn: () => api.adminGetDownloadQuota(userId) }); + const [form, setForm] = useState<{ gb: number; jobs: number; conc: number; unlimited: boolean } | null>(null); + const cur = q.data; + const state = form ?? (cur ? { gb: Math.round((cur.max_bytes / GiB) * 10) / 10, jobs: cur.max_jobs, conc: cur.max_concurrent, unlimited: cur.unlimited } : null); + const done = () => { + qc.invalidateQueries({ queryKey: ["dl-quota", userId] }); + qc.invalidateQueries({ queryKey: ["admin-storage"] }); + onClose(); + }; + const save = useMutation({ + mutationFn: () => + api.adminSetDownloadQuota(userId, { + max_bytes: Math.round((state!.gb) * GiB), + max_jobs: state!.jobs, + max_concurrent: state!.conc, + unlimited: state!.unlimited, + }), + onSuccess: done, + }); + const reset = useMutation({ mutationFn: () => api.adminResetDownloadQuota(userId), onSuccess: done }); + if (!state) return null; + const upd = (p: Partial) => setForm({ ...state, ...p }); + return ( + +
+
+ + + +
+ +
+ + +
+
+ + ); +} + +function AdminSystem() { + const { t } = useTranslation(); + const storage = useLiveQuery(["admin-storage"], api.adminDownloadStorage, { intervalMs: 5000 }); + const jobs = useLiveQuery(["admin-downloads"], api.adminDownloads, { intervalMs: 5000 }); + const [quotaFor, setQuotaFor] = useState<{ id: number; email: string } | null>(null); + const s = storage.data; + return ( +
+ {s && ( +
+ {[ + [t("downloads.admin.readyFiles"), String(s.ready_files)], + [t("downloads.admin.totalSize"), formatBytes(s.total_bytes)], + [t("downloads.admin.cap"), s.total_cap_bytes ? formatBytes(s.total_cap_bytes) : t("downloads.admin.noCap")], + ].map(([label, val]) => ( +
+
{label}
+
{val}
+
+ ))} +
+ )} + + {s && s.per_user.length > 0 && ( +
+
{t("downloads.admin.perUser")}
+
+ {s.per_user.map((u) => ( +
+ {u.email} + {formatBytes(u.footprint_bytes)} + +
+ ))} +
+
+ )} + +
{t("downloads.tabs.system")}
+
+ {(jobs.data ?? []).map((j) => ( +
+ {jobTitle(j)} + {j.user_email} + + {j.asset?.size_bytes ? formatBytes(j.asset.size_bytes) : ""} +
+ ))} + {jobs.data && jobs.data.length === 0 &&
{t("downloads.empty.admin")}
} +
+ + {quotaFor && setQuotaFor(null)} />} +
+ ); +} + +// --- Main ----------------------------------------------------------------------------------- + +export default function DownloadCenter({ me }: { me: Me }) { + const { t } = useTranslation(); + const qc = useQueryClient(); + const confirm = useConfirm(); + const [tab, setTab] = usePersistedTab("siftlode.downloadTab", "queue"); + const [addUrl, setAddUrl] = useState(""); + const [addSource, setAddSource] = useState(null); + const [manage, setManage] = useState(false); + const [renameJob, setRenameJob] = useState(null); + const [shareJob, setShareJob] = useState(null); + + const jobsQ = useLiveQuery(["downloads"], api.downloads, { intervalMs: 2000 }); + const sharedQ = useQuery({ queryKey: ["downloads-shared"], queryFn: api.downloadsShared, enabled: tab === "shared" }); + const jobs = jobsQ.data ?? []; + const queue = jobs.filter((j) => ["queued", "running", "paused", "error"].includes(j.status)); + const library = jobs.filter((j) => j.status === "done"); + + const invalidate = () => { + qc.invalidateQueries({ queryKey: ["downloads"] }); + qc.invalidateQueries({ queryKey: ["download-index"] }); + qc.invalidateQueries({ queryKey: ["download-usage"] }); + }; + const act = useMutation({ + mutationFn: ({ id, op }: { id: number; op: "pause" | "resume" | "cancel" | "delete" }) => + op === "pause" ? api.pauseDownload(id) + : op === "resume" ? api.resumeDownload(id) + : op === "cancel" ? api.cancelDownload(id) + : api.deleteDownload(id), + onSuccess: invalidate, + }); + + const confirmThen = async (title: string, body: string, fn: () => void) => { + if (await confirm({ title, message: body, confirmLabel: title, danger: true })) fn(); + }; + + const tabs = [ + { id: "queue", label: t("downloads.tabs.queue"), badge: queue.length }, + { id: "done", label: t("downloads.tabs.done"), badge: library.length }, + { id: "shared", label: t("downloads.tabs.shared") }, + ...(me.role === "admin" ? [{ id: "system", label: t("downloads.tabs.system") }] : []), + ]; + + const saveBtn = (job: DownloadJob) => ( + + + + ); + + return ( +
+
+

{t("downloads.page.title")}

+ +
+

{t("downloads.page.subtitle")}

+ + + + {tab === "queue" && ( + <> +
+ setAddUrl(e.target.value)} + onKeyDown={(e) => e.key === "Enter" && addUrl.trim() && setAddSource(addUrl.trim())} + placeholder={t("downloads.page.addUrl")} + className={inputCls} + /> + +
+
+ {queue.map((job) => ( + + {(job.status === "queued" || job.status === "running") && ( + act.mutate({ id: job.id, op: "pause" })} title={t("downloads.actions.pause")}> + + + )} + {(job.status === "paused" || job.status === "error") && ( + act.mutate({ id: job.id, op: "resume" })} title={t(job.status === "error" ? "downloads.actions.retry" : "downloads.actions.resume")}> + + + )} + confirmThen(t("downloads.confirm.cancelTitle"), t("downloads.confirm.cancelBody"), () => act.mutate({ id: job.id, op: "cancel" }))} + title={t("downloads.actions.cancel")} + danger + > + + + + ))} + {queue.length === 0 &&
{t("downloads.empty.queue")}
} +
+ + )} + + {tab === "done" && ( + <> + +
+ {library.map((job) => ( + + {saveBtn(job)} + setRenameJob(job)} title={t("downloads.actions.rename")}> + + + setShareJob(job)} title={t("downloads.actions.share")}> + + + confirmThen(t("downloads.confirm.deleteTitle"), t("downloads.confirm.deleteBody"), () => act.mutate({ id: job.id, op: "delete" }))} + title={t("downloads.actions.delete")} + danger + > + + + + ))} + {library.length === 0 &&
{t("downloads.empty.done")}
} +
+ + )} + + {tab === "shared" && ( +
+ {(sharedQ.data ?? []).map((job) => ( + + {job.can_download && saveBtn(job)} + + ))} + {sharedQ.data && sharedQ.data.length === 0 && ( +
{t("downloads.empty.shared")}
+ )} +
+ )} + + {tab === "system" && me.role === "admin" && } + + {addSource && ( + { + setAddSource(null); + setAddUrl(""); + }} + /> + )} + {manage && setManage(false)} />} + {renameJob && setRenameJob(null)} />} + {shareJob && setShareJob(null)} />} +
+ ); +} diff --git a/frontend/src/components/DownloadDialog.tsx b/frontend/src/components/DownloadDialog.tsx new file mode 100644 index 0000000..1f27474 --- /dev/null +++ b/frontend/src/components/DownloadDialog.tsx @@ -0,0 +1,117 @@ +import { useState } from "react"; +import { useTranslation } from "react-i18next"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import Modal from "./Modal"; +import ProfileEditor from "./ProfileEditor"; +import { api } from "../lib/api"; +import { notify } from "../lib/notifications"; +import { navigateTo } from "../lib/nav"; + +const inputCls = + "w-full rounded-lg bg-surface border border-border px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-accent/40"; + +// Preset picker shown when the user hits Download on a card / in the player, or adds a URL from +// the Download Center. Enqueues via the API and points the user at the Download Center. +export default function DownloadDialog({ + source, + title, + onClose, +}: { + source: string; + title?: string | null; + onClose: () => void; +}) { + const { t } = useTranslation(); + const qc = useQueryClient(); + const profilesQ = useQuery({ queryKey: ["download-profiles"], queryFn: api.downloadProfiles }); + const [profileId, setProfileId] = useState(null); + const [name, setName] = useState(""); + const [busy, setBusy] = useState(false); + const [manage, setManage] = useState(false); + + const profiles = profilesQ.data ?? []; + const chosen = profileId ?? profiles[0]?.id ?? null; + + const submit = async () => { + if (!chosen || busy) return; + setBusy(true); + try { + await api.enqueueDownload({ + source, + profile_id: chosen, + display_name: name.trim() || undefined, + }); + qc.invalidateQueries({ queryKey: ["downloads"] }); + qc.invalidateQueries({ queryKey: ["download-index"] }); + qc.invalidateQueries({ queryKey: ["download-usage"] }); + notify({ + level: "success", + message: t("downloads.dialog.added"), + action: { label: t("downloads.dialog.view"), onClick: () => navigateTo("downloads") }, + }); + onClose(); + } catch { + // Non-quiet request: the global error dialog already surfaced the reason (e.g. quota). + setBusy(false); + } + }; + + return ( + + {title &&

{title}

} + + + + + + + setName(e.target.value)} + placeholder={t("downloads.dialog.namePlaceholder")} + className={`${inputCls} mb-5`} + /> + +
+ + +
+ + {manage && ( + { + setManage(false); + profilesQ.refetch(); + }} + /> + )} + + ); +} diff --git a/frontend/src/components/Header.tsx b/frontend/src/components/Header.tsx index 2dc8024..69fa319 100644 --- a/frontend/src/components/Header.tsx +++ b/frontend/src/components/Header.tsx @@ -89,7 +89,9 @@ export default function Header({ ? t("inbox.navLabel") : page === "messages" ? t("messages.navLabel") - : t("header.channelManager")} + : page === "downloads" + ? t("downloads.navLabel") + : t("header.channelManager")}
)} diff --git a/frontend/src/components/NavSidebar.tsx b/frontend/src/components/NavSidebar.tsx index db5dde3..56a635b 100644 --- a/frontend/src/components/NavSidebar.tsx +++ b/frontend/src/components/NavSidebar.tsx @@ -8,6 +8,7 @@ import { Bell, ChevronLeft, ChevronRight, + Download, Home, Info, ListVideo, @@ -147,6 +148,13 @@ export default function NavSidebar({ { intervalMs: 30000, enabled: !me.is_demo } ); const msgUnread = msgUnreadQuery.data?.count ?? 0; + // Download center badge = items still working (queued/running/paused). Reuses the same + // lightweight per-user index the feed cards poll, so no extra request shape. + const dlIndexQuery = useLiveQuery(["download-index"], api.downloadIndex, { + intervalMs: 30000, + enabled: !me.is_demo, + }); + const dlActive = Object.values(dlIndexQuery.data ?? {}).filter((s) => s !== "done").length; type NavItem = { page: Page; icon: typeof Home; label: string; badge?: number }; // User-facing content modules vs. admin/system modules, separated by a divider in the rail. @@ -159,6 +167,11 @@ export default function NavSidebar({ ...(me.is_demo ? [] : [{ page: "messages" as Page, icon: MessageSquare, label: t("messages.navLabel"), badge: msgUnread }]), + // Download center — YouTube → server → your device. Hidden for the shared demo account + // (spends server disk + needs a real identity). + ...(me.is_demo + ? [] + : [{ page: "downloads" as Page, icon: Download, label: t("downloads.navLabel"), badge: dlActive }]), // Per-user sync status + your own API usage (admins get an extra system-wide tab inside). { page: "stats", icon: BarChart3, label: t("header.account.stats") }, ]; diff --git a/frontend/src/components/PlayerModal.tsx b/frontend/src/components/PlayerModal.tsx index 973ada7..f8456d9 100644 --- a/frontend/src/components/PlayerModal.tsx +++ b/frontend/src/components/PlayerModal.tsx @@ -5,6 +5,7 @@ import { useQuery, useQueryClient } from "@tanstack/react-query"; import { AlertTriangle, ArrowLeft, Check, CheckCheck, ExternalLink, SkipBack, SkipForward, Volume2, VolumeX, X } from "lucide-react"; import Avatar from "./Avatar"; import AddToPlaylist from "./AddToPlaylist"; +import DownloadButton from "./DownloadButton"; import { api, type Video } from "../lib/api"; import { channelYouTubeUrl, formatDuration, formatViews, relativeTime } from "../lib/format"; import { renderDescription } from "../lib/descriptionLinks"; @@ -630,6 +631,13 @@ export default function PlayerModal({ className="ml-auto shrink-0 inline-flex items-center justify-center w-9 h-9 rounded-lg text-muted hover:text-fg hover:bg-surface border border-border transition" /> )} + {!navigated && ( + + )} {!navigated && ( + +
+ ))} +
+ + {!formOpen ? ( + + ) : ( +
+
+ + setName(e.target.value)} placeholder={t("downloads.profiles.namePlaceholder")} className={inputCls} /> +
+
+
+ + +
+ {spec.mode !== "a" ? ( +
+ + +
+ ) : ( +
+ + +
+ )} + {spec.mode !== "a" && ( + <> +
+ + +
+
+ + +
+ + )} +
+
+ {spec.mode !== "a" && ( + patch({ embed_subs: v })} /> + )} + patch({ embed_chapters: v })} /> + patch({ embed_thumbnail: v })} /> + patch({ sponsorblock: v })} /> +
+
+ + +
+
+ )} + + ); +} diff --git a/frontend/src/components/VideoCard.tsx b/frontend/src/components/VideoCard.tsx index 4e55b35..fd5a963 100644 --- a/frontend/src/components/VideoCard.tsx +++ b/frontend/src/components/VideoCard.tsx @@ -11,6 +11,7 @@ import { } from "lucide-react"; import Avatar from "./Avatar"; import AddToPlaylist from "./AddToPlaylist"; +import DownloadButton from "./DownloadButton"; import clsx from "clsx"; import type { Video } from "../lib/api"; import { formatDuration, formatViews, relativeTime } from "../lib/format"; @@ -67,6 +68,7 @@ function Actions({ +