From 317750e4916094a19fc229597b60e069fd75a039 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 2 Jul 2026 23:57:40 +0200 Subject: [PATCH 01/18] =?UTF-8?q?feat(downloads):=20M1=20=E2=80=94=20schem?= =?UTF-8?q?a,=20config,=20worker=20container,=20deps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - models: MediaAsset (shared file cache, unique on source+format_sig), DownloadJob (per-user request), DownloadProfile (presets), DownloadShare (ACL), DownloadQuota - migrations 0036 (5 tables) + 0037 (6 builtin presets) - sysconfig 'downloads' group (per-user defaults, retention/GC, layout, worker concurrency) - config: DOWNLOAD_ROOT / WORKER_ENABLED env + download defaults - deps: yt-dlp in requirements, ffmpeg in Dockerfile runtime stage - worker: dedicated container (python -m app.worker), thin idle entry (loop lands in M2) - localdev compose: worker service + downloads bind-mount; gitignore downloads/ --- .gitignore | 3 + Dockerfile | 6 + .../alembic/versions/0036_download_center.py | 143 ++++++++++++++++ .../0037_download_builtin_profiles.py | 67 ++++++++ backend/app/config.py | 25 +++ backend/app/models.py | 155 ++++++++++++++++++ backend/app/sysconfig.py | 9 + backend/app/worker.py | 50 ++++++ backend/requirements.txt | 4 + docker-compose.localdev.yml | 34 ++++ 10 files changed, 496 insertions(+) create mode 100644 backend/alembic/versions/0036_download_center.py create mode 100644 backend/alembic/versions/0037_download_builtin_profiles.py create mode 100644 backend/app/worker.py diff --git a/.gitignore b/.gitignore index 38f6d4f..1afa1fc 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,9 @@ frontend/dist/ *.sql.gz backups/ +# Download center: locally downloaded media (the DOWNLOAD_ROOT bind mount) +downloads/ + # OS / editor .DS_Store Thumbs.db diff --git a/Dockerfile b/Dockerfile index d5aa7fd..188bd05 100644 --- a/Dockerfile +++ b/Dockerfile @@ -34,6 +34,12 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ WORKDIR /app +# ffmpeg is required by the download center (yt-dlp) to merge separate video+audio streams +# and run postprocessing (audio extract, embed thumbnail/chapters/subs, SponsorBlock). +RUN apt-get update \ + && apt-get install -y --no-install-recommends ffmpeg \ + && 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/app/config.py b/backend/app/config.py index 938cfcb..cb44539 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -131,6 +131,31 @@ 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 + # 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 + # 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/models.py b/backend/app/models.py index 3160514..be145e0 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, @@ -688,6 +689,160 @@ 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" + ) + # 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/sysconfig.py b/backend/app/sysconfig.py index ff396c7..1febd37 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -66,6 +66,15 @@ 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_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/worker.py b/backend/app/worker.py new file mode 100644 index 0000000..1b7564f --- /dev/null +++ b/backend/app/worker.py @@ -0,0 +1,50 @@ +"""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. It +claims `queued` download jobs from the DB with FOR UPDATE SKIP LOCKED and runs them, writing +live progress back to the job row (the frontend polls it — the worker is a separate process and +can't reach the API's in-process WebSocket registry). + +M1: thin, idle entry point (keeps the container healthy). The claim loop + yt-dlp integration +land in M2. +""" +import logging +import signal +import time + +from app.config import settings + +log = logging.getLogger("siftlode.worker") + +_stop = False + + +def _handle_signal(signum, frame): + global _stop + _stop = True + + +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 + + log.info( + "Download worker started (concurrency=%d, root=%s). Claim loop lands in M2.", + settings.download_worker_concurrency, + settings.download_root, + ) + while not _stop: + # M2: claim queued jobs (FOR UPDATE SKIP LOCKED) and download them. + time.sleep(settings.download_poll_seconds) + + log.info("Download worker stopped.") + + +if __name__ == "__main__": + main() diff --git a/backend/requirements.txt b/backend/requirements.txt index 67f14ad..2e293ce 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -13,3 +13,7 @@ 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 (installed at the OS level in +# the Dockerfile) does the merge + postprocessing. yt-dlp is release-often (YouTube changes), +# so keep the floor loose and bump it when extraction breaks. +yt-dlp>=2025.1 diff --git a/docker-compose.localdev.yml b/docker-compose.localdev.yml index 37260aa..2cac4c0 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,37 @@ 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 + security_opt: + - apparmor:unconfined + volumes: + - ./downloads:/downloads restart: unless-stopped volumes: From 1ae43339227d8a3d9a69c9e3f7b20c2962b9450c Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 3 Jul 2026 00:08:18 +0200 Subject: [PATCH 02/18] =?UTF-8?q?feat(downloads):=20M2=20=E2=80=94=20worke?= =?UTF-8?q?r=20download=20loop=20+=20cache/dedup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - downloads/formats.py: profile spec -> format_sig (cache key) + yt-dlp opts (format selector, audio extract, sponsorblock, embed subs/chapters/thumbnail, jpg poster) - downloads/storage.py: Plex tree path ({Channel}/Season {Year}/… [id].ext) + .nfo + poster.jpg sidecars; sanitization; asset-file deletion with empty-dir pruning - downloads/service.py: enqueue + get-or-create shared MediaAsset (race-safe) + instant cache-hit path (ready asset -> job done, ref_count++) - worker.py: N claim-loop threads (FOR UPDATE SKIP LOCKED), asset compare-and-set pending->downloading so identical requests never double-download, throttled progress writes, cooperative pause/cancel via the progress hook, crash-safe orphan recovery Verified on localdev: real download -> Plex tree + .nfo + poster on the host bind-mount, correct metadata/codecs/size; 2nd user = instant cache hit (shared asset, ref_count=2); audio-only MP3 extraction path produces a distinct cached asset. --- backend/app/worker.py | 350 +++++++++++++++++++++++++++++++++++++++--- 1 file changed, 331 insertions(+), 19 deletions(-) diff --git a/backend/app/worker.py b/backend/app/worker.py index 1b7564f..5276e0b 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -1,28 +1,339 @@ """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. It -claims `queued` download jobs from the DB with FOR UPDATE SKIP LOCKED and runs them, writing -live progress back to the job row (the frontend polls it — the worker is a separate process and -can't reach the API's in-process WebSocket registry). +from the API so long yt-dlp downloads + ffmpeg postprocessing never block web requests. -M1: thin, idle entry point (keeps the container healthy). The claim loop + yt-dlp integration -land in M2. +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, service, storage +from app.models import DownloadJob, MediaAsset log = logging.getLogger("siftlode.worker") -_stop = False +_stop = threading.Event() + + +class _Aborted(Exception): + """Raised inside the progress hook when the job was paused/canceled mid-download.""" def _handle_signal(signum, frame): - global _stop - _stop = True + _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: + with SessionLocal() as db: + row = db.execute( + text( + """ + UPDATE download_jobs SET status='running', phase='starting', updated_at=now() + WHERE id = ( + SELECT id FROM download_jobs + WHERE status='queued' + ORDER BY queue_pos, id + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING id + """ + ) + ).fetchone() + db.commit() + return row[0] if row else None + + +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 _make_progress_hook(job_id: int): + state = {"last_db": 0.0, "last_check": 0.0} + + 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 d.get("status") != "downloading": + return + if now - state["last_db"] < 1.0: + return + state["last_db"] = now + 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="downloading", + ) + + 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 + + +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) + + outtmpl = str(staging / "%(id)s.%(ext)s") + opts = formats.build_ydl_opts(spec, outtmpl, _make_progress_hook(job_id)) + url = service.source_url(source_kind, source_ref) + + _set_job(job_id, phase="downloading", progress=0) + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url, download=True) + + 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=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: @@ -34,16 +345,17 @@ def main() -> None: log.info("WORKER_ENABLED is off; worker exiting (nothing to do).") return - log.info( - "Download worker started (concurrency=%d, root=%s). Claim loop lands in M2.", - settings.download_worker_concurrency, - settings.download_root, - ) - while not _stop: - # M2: claim queued jobs (FOR UPDATE SKIP LOCKED) and download them. - time.sleep(settings.download_poll_seconds) - - log.info("Download worker stopped.") + _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__": From 7f9847c2c29d9e70b0b072dc39b2d83b71504da7 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 3 Jul 2026 00:19:22 +0200 Subject: [PATCH 03/18] fix(downloads): add the downloads package (was wrongly gitignored) + robust sanitizer The .gitignore 'downloads/' pattern for the DOWNLOAD_ROOT bind mount also matched the backend/app/downloads source package, so formats.py/storage.py/service.py never got committed with M2. Anchor the ignore to '/downloads/' (repo root only) and add the package. storage.sanitize() now handles arbitrarily messy titles (emoji, ZWJ, fullwidth, clickbait punctuation): NFKC-normalize, drop emoji/symbol/control unicode categories, collapse repeated punctuation, underscore-join words -> space-free paths. Accents (HU/DE) preserved, no ASCII folding. Plex layout uses _-_ separators + Season_{year}. --- .gitignore | 5 +- backend/app/downloads/__init__.py | 0 backend/app/downloads/formats.py | 145 ++++++++++++++++++++++++++++++ backend/app/downloads/service.py | 97 ++++++++++++++++++++ backend/app/downloads/storage.py | 143 +++++++++++++++++++++++++++++ 5 files changed, 388 insertions(+), 2 deletions(-) create mode 100644 backend/app/downloads/__init__.py create mode 100644 backend/app/downloads/formats.py create mode 100644 backend/app/downloads/service.py create mode 100644 backend/app/downloads/storage.py diff --git a/.gitignore b/.gitignore index 1afa1fc..6710e89 100644 --- a/.gitignore +++ b/.gitignore @@ -21,8 +21,9 @@ frontend/dist/ *.sql.gz backups/ -# Download center: locally downloaded media (the DOWNLOAD_ROOT bind mount) -downloads/ +# 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 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..6360ca3 --- /dev/null +++ b/backend/app/downloads/formats.py @@ -0,0 +1,145 @@ +"""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, + "embed_subs": False, + "embed_chapters": True, + "embed_thumbnail": True, + "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) -> dict: + """Assemble yt-dlp options for a single download to `outtmpl` (a full path template).""" + 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": [], + } + + 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/service.py b/backend/app/downloads/service.py new file mode 100644 index 0000000..e832d6b --- /dev/null +++ b/backend/app/downloads/service.py @@ -0,0 +1,97 @@ +"""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 +from app.models import DownloadJob, MediaAsset + + +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 _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).""" + spec = formats.normalize(spec) + sig = formats.format_sig(spec) + asset = get_or_create_asset(db, source_kind, source_ref, sig) + + 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..cf94a28 --- /dev/null +++ b/backend/app/downloads/storage.py @@ -0,0 +1,143 @@ +"""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 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 From 7148335c0934e30ad661b7ec0b56234187544d36 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 3 Jul 2026 00:26:40 +0200 Subject: [PATCH 04/18] =?UTF-8?q?feat(downloads):=20M3=20=E2=80=94=20per-u?= =?UTF-8?q?ser=20quota=20+=20retention=20GC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - downloads/quota.py: per-user limits (DownloadQuota row or sysconfig defaults; unlimited bypass), footprint = distinct ready-asset bytes the user holds, check_enqueue (max_jobs + max_bytes hard at enqueue), at_concurrency_limit (worker-side max_concurrent), usage snapshot - downloads/gc.py: run_download_gc — pre-expiry warning (once, gc_notified flag), TTL deletion (files + row; FK SET NULL orphans holders' jobs -> 'expired'), LRU eviction over a total-cache cap; notifies holders on each event - migration 0038: media_assets.gc_notified - config/sysconfig: download_total_max_bytes (0=unlimited) in the downloads group - service.enqueue enforces quota; worker claim skips users at their concurrency limit - scheduler: download_gc job registered (default 360 min, admin-tunable) Verified on localdev: footprint accounting, max_jobs block, pre-expiry warning (notifies all holders), TTL deletion (asset+files gone, dirs pruned, jobs orphaned, holders notified). --- .../versions/0038_asset_gc_notified.py | 29 ++++ backend/app/config.py | 3 + backend/app/downloads/gc.py | 119 ++++++++++++++++ backend/app/downloads/quota.py | 129 ++++++++++++++++++ backend/app/downloads/service.py | 7 +- backend/app/models.py | 4 + backend/app/scheduler.py | 9 ++ backend/app/sysconfig.py | 1 + backend/app/worker.py | 39 ++++-- 9 files changed, 325 insertions(+), 15 deletions(-) create mode 100644 backend/alembic/versions/0038_asset_gc_notified.py create mode 100644 backend/app/downloads/gc.py create mode 100644 backend/app/downloads/quota.py 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/app/config.py b/backend/app/config.py index cb44539..e521bcd 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -142,6 +142,9 @@ class Settings(BaseSettings): 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 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..fb224cf --- /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 ( + 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 index e832d6b..a446cc2 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -12,7 +12,7 @@ from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session -from app.downloads import formats +from app.downloads import formats, quota from app.models import DownloadJob, MediaAsset @@ -67,7 +67,10 @@ def enqueue( 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).""" + """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) diff --git a/backend/app/models.py b/backend/app/models.py index be145e0..e91d0d8 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -731,6 +731,10 @@ class MediaAsset(Base, TimestampMixin): 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)) 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/sysconfig.py b/backend/app/sysconfig.py index 1febd37..ed6c79b 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -67,6 +67,7 @@ SPECS: tuple[ConfigSpec, ...] = ( # --- 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), diff --git a/backend/app/worker.py b/backend/app/worker.py index 5276e0b..b3a81ee 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -28,7 +28,7 @@ from sqlalchemy import text from app.config import settings from app.db import SessionLocal -from app.downloads import formats, service, storage +from app.downloads import formats, quota, service, storage from app.models import DownloadJob, MediaAsset log = logging.getLogger("siftlode.worker") @@ -83,24 +83,37 @@ def _recover_orphans() -> None: 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: - row = db.execute( + rows = db.execute( text( """ - UPDATE download_jobs SET status='running', phase='starting', updated_at=now() - WHERE id = ( - SELECT id FROM download_jobs - WHERE status='queued' - ORDER BY queue_pos, id - FOR UPDATE SKIP LOCKED - LIMIT 1 - ) - RETURNING id + SELECT id, user_id FROM download_jobs + WHERE status='queued' + ORDER BY queue_pos, id + FOR UPDATE SKIP LOCKED + LIMIT 50 """ ) - ).fetchone() + ).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 row[0] if row else None + return chosen def _claim_asset(asset_id: int) -> bool: From 51158ad9cff52f40627c5c20c235f5215969d389 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 3 Jul 2026 00:34:08 +0200 Subject: [PATCH 05/18] =?UTF-8?q?feat(downloads):=20M4=20=E2=80=94=20REST?= =?UTF-8?q?=20API=20(enqueue,=20manage,=20file-serve,=20sharing,=20admin)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit routes/downloads.py (require_human; admin_router adds admin_user): - profiles: list (builtins + own), create/update/delete custom - enqueue: resolve_source (bare id / watch / youtu.be / shorts URLs -> youtube id; else raw URL for the generic extractor), profile_id or inline spec or builtin fallback; maps quota.QuotaExceeded -> 422 speaking message - list / usage / shared-with-me; rename (display name only); pause/resume/cancel/delete (ref_count bookkeeping) - GET /{id}/file: ownership-or-share check, path-traversal guard, range-aware FileResponse with the user's custom display name (Content-Disposition), bumps last_access - share/unshare by email + a download_shared notification - admin: all-jobs, storage dashboard (totals + per-user footprint), per-user quota GET/PUT/DELETE - cast SUM(bigint) footprints to int for clean numeric JSON - wired both routers in main.py Verified via TestClient: full enqueue->download->206 range fetch->share->access-control (cross-user delete 404); unauth = 401 on user + admin routes; all 15 paths registered. --- backend/app/downloads/quota.py | 2 +- backend/app/main.py | 3 + backend/app/routes/downloads.py | 619 ++++++++++++++++++++++++++++++++ 3 files changed, 623 insertions(+), 1 deletion(-) create mode 100644 backend/app/routes/downloads.py diff --git a/backend/app/downloads/quota.py b/backend/app/downloads/quota.py index fb224cf..b60fcb7 100644 --- a/backend/app/downloads/quota.py +++ b/backend/app/downloads/quota.py @@ -63,7 +63,7 @@ def footprint(db: Session, user_id: int) -> int: ) .distinct() ) - return ( + return int( db.execute( select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where( MediaAsset.id.in_(held_assets), MediaAsset.status == "ready" 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/routes/downloads.py b/backend/app/routes/downloads.py new file mode 100644 index 0000000..1486059 --- /dev/null +++ b/backend/app/routes/downloads.py @@ -0,0 +1,619 @@ +"""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}$") +_BASENAME_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]') + + +# --- 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) + + +@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 + _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"): + job.status = "canceled" + _decrement_ref(db, job) + 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) + _decrement_ref(db, job) + db.delete(job) + db.commit() + return {"deleted": job_id} + + +def _decrement_ref(db: Session, job: DownloadJob) -> None: + """Drop this job's hold on its asset. The asset itself lingers as cache until its TTL / + eviction (so a re-add is still a cache hit); only the reference count changes here.""" + if job.asset_id and job.status not in ("canceled", "error"): + asset = db.get(MediaAsset, job.asset_id) + if asset and asset.ref_count > 0: + asset.ref_count -= 1 + + +# --- file download (range-aware, custom display name) -------------------------------------- + +def _clean_basename(name: str) -> str: + return _BASENAME_ILLEGAL.sub("", name).strip() or "download" + + +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) From 7d1ed24fea06d191f58e6ccc051ed423ca814c12 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 3 Jul 2026 01:52:58 +0200 Subject: [PATCH 06/18] =?UTF-8?q?feat(downloads):=20M5=20=E2=80=94=20Downl?= =?UTF-8?q?oad=20Center=20frontend?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api.ts: download types + methods (profiles, enqueue, list/manage, file url, share, usage, index, admin storage/quota) - format.ts: formatBytes / formatSpeed - i18n: en/hu/de downloads.json (trilingual) - Downloads nav module (Download icon, active-count badge, hidden for demo) placed after Messages; urlState page + App render + Header title - DownloadButton: self-contained per-card/player affordance (demo-hidden, reflects the per-user download index: downloaded / in-queue), opens the preset dialog - DownloadDialog: preset picker + optional display name -> enqueue + 'View' toast - ProfileEditor: manage built-in + custom formats (full spec form incl. SponsorBlock) - DownloadCenter: Queue / Library / Shared / (admin) System tabs, live progress via useLiveQuery, usage bar, pause/resume/cancel/delete, save-to-device, rename, share, admin storage dashboard + per-user quota editor - wired DownloadButton into VideoCard + PlayerModal - lib/nav.ts: decoupled navigator so the download toast can jump to the page tsc + vite build clean. Verified in a headless authed browser: page + tabs render, Library shows a real download with usage bar + actions, no console errors. --- backend/app/routes/downloads.py | 23 + frontend/src/App.tsx | 6 + frontend/src/components/DownloadButton.tsx | 66 +++ frontend/src/components/DownloadCenter.tsx | 511 ++++++++++++++++++++ frontend/src/components/DownloadDialog.tsx | 117 +++++ frontend/src/components/Header.tsx | 4 +- frontend/src/components/NavSidebar.tsx | 13 + frontend/src/components/PlayerModal.tsx | 8 + frontend/src/components/ProfileEditor.tsx | 215 ++++++++ frontend/src/components/VideoCard.tsx | 2 + frontend/src/i18n/locales/de/downloads.json | 136 ++++++ frontend/src/i18n/locales/en/downloads.json | 136 ++++++ frontend/src/i18n/locales/hu/downloads.json | 136 ++++++ frontend/src/lib/api.ts | 144 ++++++ frontend/src/lib/format.ts | 20 + frontend/src/lib/nav.ts | 14 + frontend/src/lib/urlState.ts | 1 + 17 files changed, 1551 insertions(+), 1 deletion(-) create mode 100644 frontend/src/components/DownloadButton.tsx create mode 100644 frontend/src/components/DownloadCenter.tsx create mode 100644 frontend/src/components/DownloadDialog.tsx create mode 100644 frontend/src/components/ProfileEditor.tsx create mode 100644 frontend/src/i18n/locales/de/downloads.json create mode 100644 frontend/src/i18n/locales/en/downloads.json create mode 100644 frontend/src/i18n/locales/hu/downloads.json create mode 100644 frontend/src/lib/nav.ts diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index 1486059..83bb496 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -279,6 +279,29 @@ def my_usage(user: User = Depends(require_human), db: Session = Depends(get_db)) 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) 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" ? (