fix(downloads): shell-safe names, refresh display_name on re-download, GC orphan sweep

Follow-up to user prod testing of 0.51.0:
- re-download now clears job.display_name so the card title + Content-Disposition filename are
  re-derived from the freshly-cleaned title (previously an auto-name from an older un-cleaned
  title — still carrying a "… views · … reactions |" prefix — survived the rebuild).
- on-disk AND served names are now a strict shell-safe slug: only [A-Za-z0-9._-] survive; every
  other char (the "!" the user saw, plus ( ) [ ] & $ ' " ; | ? * …) collapses to "_". The id no
  longer uses [brackets] (glob metacharacters). Shared _slug helper for sanitize + display_filename.
- GC gains a 5th pass: sweep on-disk files/dirs no ready asset owns (old accented channel dirs, a
  poster from a superseded naming scheme left after a re-download), skipping the dot system trees
  and anything newer than an hour so an in-flight download is never touched.
- Tests: slug strips !/$/()[]; rel_path is bracket-free; orphan-sweep keeps owned/system/fresh files.

backend/app/downloads/gc.py

@@ -1,6 +1,6 @@
 """Download retention garbage collector (a scheduler job).
 
-Four passes each run:
+Five 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
@@ -9,11 +9,16 @@ Four passes each run:
      ready assets until back under it.
   4. Orphan reap — delete errored, fileless, unreferenced asset rows (they carry no TTL, so
      passes 1-3 skip them; benign row-count hygiene).
+  5. Orphan FILE sweep — delete on-disk files/dirs no surviving asset owns (cruft left when a
+     re-download renames an asset, e.g. old accented channel dirs), skipping system trees and
+     anything newer than an hour so an in-flight download is never touched.
 
 Runs in the API process (which mounts DOWNLOAD_ROOT). Pure disk/DB work — no YouTube quota.
 """
 import logging
+import time
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -26,6 +31,11 @@ from app.notifications import create_notification
 
 log = logging.getLogger("siftlode.downloads.gc")
 
+# Dot-prefixed top-level trees that are NOT part of the Plex/flat asset layout — the orphan sweep
+# never touches them (user clips, Plex HLS/img caches, in-progress staging, editor storyboards).
+_SYSTEM_TREE = {".edits", ".plex-hls", ".plex-img-cache", ".staging", ".storyboards"}
+_ORPHAN_MIN_AGE_S = 3600  # only sweep files older than this, so an in-flight download is never hit
+
 
 def _holders(db: Session, asset_id: int) -> list[int]:
     return list(
@@ -57,6 +67,53 @@ def _delete_asset(db: Session, asset: MediaAsset, ntype: str) -> None:
     db.delete(asset)  # jobs.asset_id -> NULL via FK ON DELETE SET NULL
 
 
+def _sweep_orphan_files(db: Session, download_root: str) -> int:
+    """Delete media/sidecar files (and now-empty dirs) under DOWNLOAD_ROOT that no `ready` asset
+    owns — cruft left when a re-download changes an asset's on-disk name (e.g. an old accented
+    channel dir, or a poster from a superseded naming scheme). Skips the dot-prefixed system trees
+    and only removes files older than `_ORPHAN_MIN_AGE_S`, so a just-completed or in-flight download
+    is never deleted, and only files NOT owned by a surviving asset."""
+    root = Path(download_root)
+    if not root.is_dir():
+        return 0
+    keep: set[Path] = set()
+    for rel in db.execute(
+        select(MediaAsset.rel_path).where(
+            MediaAsset.status == "ready", MediaAsset.rel_path.is_not(None)
+        )
+    ).scalars():
+        media = (root / rel).resolve()
+        keep |= {media, media.with_suffix(".nfo"), media.with_suffix(".jpg")}
+
+    cutoff = time.time() - _ORPHAN_MIN_AGE_S
+    removed = 0
+    for entry in root.iterdir():
+        if entry.name.startswith("."):  # .edits / .plex-* / .staging / .storyboards + any dotfile
+            continue
+        files = [entry] if entry.is_file() else [p for p in entry.rglob("*") if p.is_file()]
+        for f in files:
+            try:
+                if f.resolve() in keep or f.stat().st_mtime > cutoff:
+                    continue
+                f.unlink()
+                removed += 1
+            except OSError:
+                pass
+        if entry.is_dir():  # prune emptied dirs (deepest first), then the channel dir itself
+            for d in sorted(
+                (p for p in entry.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True
+            ):
+                try:
+                    d.rmdir()
+                except OSError:
+                    pass
+            try:
+                entry.rmdir()
+            except OSError:
+                pass
+    return removed
+
+
 def run_download_gc(db: Session) -> dict:
     now = datetime.now(timezone.utc)
     grace_days = sysconfig.get_int(db, "download_gc_grace_days")
@@ -132,6 +189,12 @@ def run_download_gc(db: Session) -> dict:
     if orphans:
         db.commit()
 
-    result = {"warned": warned, "expired": expired, "evicted": evicted, "reaped": reaped}
+    # 5. Orphan FILE sweep — remove on-disk cruft no surviving asset owns (old accented dirs, stale
+    #    posters). Runs last so the keep-set reflects the assets that survived passes 2-4.
+    swept = _sweep_orphan_files(db, settings.download_root)
+
+    result = {
+        "warned": warned, "expired": expired, "evicted": evicted, "reaped": reaped, "swept": swept
+    }
     log.info("download_gc %s", result)
     return result

backend/app/downloads/storage.py

@@ -18,12 +18,14 @@ 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.
+# Reserve room for the "_id.ext" suffix + parent dirs within the 255-char component limit.
 _MAX_TITLE = 120
 # User-facing (Content-Disposition) names can be longer than an on-disk component.
 _MAX_DISPLAY = 150
-# Filesystem-illegal on Windows/Plex + control chars.
-_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
+# The ONLY characters allowed in a generated name — letters, digits, and `._-`. Everything else
+# (spaces, and every shell/filesystem-unpleasant metacharacter: ! ? * & $ ( ) [ ] { } ' " ; | < >
+# etc.) collapses to a single underscore, so names are safe to handle in any shell or tool.
+_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
 # Unicode categories we drop entirely: emoji/pictographs (So), modifier symbols/skin-tones
 # (Sk), and every control/format/private/surrogate/unassigned char (C*).
 _DROP_CATEGORIES = {"So", "Sk", "Cc", "Cf", "Cs", "Co", "Cn"}
@@ -62,49 +64,33 @@ class MediaMeta:
     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 ASCII path token.
+def _slug(name: str, limit: int, fallback: str) -> str:
+    """Reduce an arbitrary (emoji-laden, clickbait, accented) title to a portable, shell-safe token.
 
-    Steps: NFKC-normalize (fold fancy width/compat forms) → drop emoji/symbols/control chars →
-    ASCII-fold accents (á→a, ő→o, ß→ss) → strip filesystem-illegal chars → collapse whitespace and
-    runs of the same punctuation → join words with underscores → trim junk punctuation from the ends
-    → length-cap. The result is pure ASCII with no spaces or path separators (digits survive)."""
-    text = unicodedata.normalize("NFKC", name or "")
-    text = "".join(
-        ch for ch in text if unicodedata.category(ch) not in _DROP_CATEGORIES
-    )
-    text = _ascii_fold(text)
-    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("_.-·—–|,;:!?ّ ")
-    # Cap by BYTES, not characters: a path component's hard limit is 255 BYTES, and rel_path packs
-    # the (byte-capped) channel + date + id + ext into that same component — so an accented/multi-byte
-    # title capped only by char count could still overflow to ENAMETOOLONG. Truncate on a UTF-8
-    # boundary (decode(..., "ignore") drops a trailing partial char) so we never split a character.
-    if len(text.encode("utf-8")) > limit:
-        text = text.encode("utf-8")[:limit].decode("utf-8", "ignore").rstrip("_.-")
-    return text or "untitled"
-
-
-def display_filename(name: str) -> str:
-    """User-facing download filename (Content-Disposition) base: the same ASCII-slug policy as the
-    on-disk name — emoji/symbols/control + illegal chars dropped, accents folded, whitespace joined
-    with underscores — so a saved file is portable (no spaces, no accents). Only the length cap
-    differs (a served name may be longer than an on-disk path component)."""
+    NFKC-normalize → drop emoji/symbols/control chars → ASCII-fold accents (á→a, ő→o, ß→ss) →
+    replace every run of unsafe characters (spaces + shell/filesystem metacharacters) with a single
+    underscore → collapse separator pile-ups → trim junk from the ends → length-cap. The result is
+    pure ASCII containing only `[A-Za-z0-9._-]` — no spaces, accents, or characters that need
+    quoting in a shell."""
     text = unicodedata.normalize("NFKC", name or "")
     text = "".join(ch for ch in text if unicodedata.category(ch) not in _DROP_CATEGORIES)
     text = _ascii_fold(text)
-    text = _ILLEGAL.sub("", text)
-    text = re.sub(r"([^\w\s])\1{1,}", r"\1", text)
-    text = re.sub(r"\s+", "_", text.strip())
-    text = re.sub(r"_{2,}", "_", text).strip("_.-·—–|,;:!?ّ ")
-    if len(text) > _MAX_DISPLAY:
-        text = text[:_MAX_DISPLAY].rstrip("_.-")
-    return text or "download"
+    text = _UNSAFE.sub("_", text)
+    text = re.sub(r"[-_]{2,}", "_", text).strip("_.-")
+    if len(text) > limit:  # pure ASCII → 1 char == 1 byte, so a char cap is also the byte cap
+        text = text[:limit].rstrip("_.-")
+    return text or fallback
+
+
+def sanitize(name: str, limit: int = _MAX_TITLE) -> str:
+    """On-disk path token: portable, shell-safe, pure ASCII (see `_slug`)."""
+    return _slug(name, limit, "untitled")
+
+
+def display_filename(name: str) -> str:
+    """Content-Disposition filename base — the same shell-safe slug policy as the on-disk name, just
+    with a larger length cap (a served name may be longer than a path component)."""
+    return _slug(name, _MAX_DISPLAY, "download")
 
 
 def rel_path(meta: MediaMeta, ext: str, layout: str = "plex", asset_id: int | None = None) -> str:
@@ -116,14 +102,16 @@ def rel_path(meta: MediaMeta, ext: str, layout: str = "plex", asset_id: int | No
     either removes the shared file. The `_a{id}` suffix (which Plex ignores — the .nfo sidecar carries
     the metadata) keeps them apart. Optional so callers/tests that don't need it keep the plain name."""
     title = sanitize(meta.title)
-    vid = meta.video_id
+    # The id goes into the name bare (no brackets — `[`/`]` are shell glob metacharacters); keep it
+    # shell-safe and bounded in case a URL source has no clean extractor id.
+    vid = _UNSAFE.sub("_", str(meta.video_id))[:40]
     disc = f"_a{asset_id}" if asset_id is not None else ""
     if layout == "flat":
-        return f"{title}_[{vid}]{disc}.{ext}"
+        return f"{title}_{vid}{disc}.{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}]{disc}.{ext}"
+    epname = f"{channel}_-_{d}_-_{title}_{vid}{disc}.{ext}"
     return f"{channel}/Season_{year}/{epname}"
 
 

backend/app/routes/downloads.py

@@ -489,6 +489,10 @@ def redownload_download(
     job.error = None
     job.speed_bps = None
     job.eta_s = None
+    # Re-derive the display name (card title + Content-Disposition filename) from the freshly
+    # cleaned title — otherwise a name auto-set from an older, un-cleaned title (e.g. still carrying
+    # a "1.6M views · … |" social prefix) would survive the rebuild. The worker fills it on completion.
+    job.display_name = None
     job.queue_pos = service.next_queue_pos(db)  # tail of the queue, not its stale original slot
     db.commit()
     return _serialize_job(db, job)