feat(titles): normalize video titles for display (feed, search, downloads)

Noisy YouTube titles are cleaned for display + storage, raw kept in videos.original_title:
- app/titles.normalize_title: strip emoji/symbols (keep accents), remove trailing SEO hashtag
  clusters (keep numeric #3 episode markers), context-aware de-shout (mostly-ALL-CAPS titles ->
  Title Case with an acronym whitelist + function-word lowercasing; otherwise only long all-caps
  words), collapse repeated punctuation
- applied at enrichment (sync/videos.py) and in the download worker (ad-hoc yt-dlp titles);
  catalog downloads inherit the normalized title automatically
- migration 0039: add original_title, preserve raw, rewrite title (generated search_vector
  regenerates); reversible via original_title

Backfill on localdev: 122115/273417 titles normalized in ~2 min. Verified in the feed + on
real messy samples (emoji/de-shout/hashtags), accents + acronyms (PS5/AI/USA/PC) preserved.

backend/alembic/versions/0039_video_title_normalize.py

@@ -0,0 +1,49 @@
+"""normalize video titles for display; keep the raw one in original_title
+
+Revision ID: 0039_title_normalize
+Revises: 0038_asset_gc_notified
+Create Date: 2026-07-03
+
+Adds videos.original_title (the raw YouTube title) and rewrites videos.title to a normalized,
+display-friendly form (emoji stripped, ALL-CAPS de-shouted, trailing SEO hashtags removed —
+see app.titles). Reversible: original_title preserves the source, and title can be re-derived.
+The generated search_vector column regenerates automatically as each title is updated.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+from app.titles import normalize_title
+
+revision: str = "0039_title_normalize"
+down_revision: Union[str, None] = "0038_asset_gc_notified"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_BATCH = 2000
+
+
+def upgrade() -> None:
+    op.add_column("videos", sa.Column("original_title", sa.Text(), nullable=True))
+    conn = op.get_bind()
+    # Preserve the raw title first (fast, set-based).
+    conn.execute(sa.text("UPDATE videos SET original_title = title WHERE title IS NOT NULL"))
+    # Then rewrite title to the normalized form, batched (each update regenerates its FTS vector).
+    rows = conn.execute(
+        sa.text("SELECT id, original_title FROM videos WHERE original_title IS NOT NULL")
+    ).fetchall()
+    changes = []
+    for vid, raw in rows:
+        norm = normalize_title(raw)
+        if norm != raw:
+            changes.append({"i": vid, "t": norm})
+    stmt = sa.text("UPDATE videos SET title = :t WHERE id = :i")
+    for start in range(0, len(changes), _BATCH):
+        conn.execute(stmt, changes[start : start + _BATCH])
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("UPDATE videos SET title = original_title WHERE original_title IS NOT NULL"))
+    op.drop_column("videos", "original_title")

backend/app/models.py

@@ -241,7 +241,8 @@ class Video(Base, TimestampMixin):
     channel_id: Mapped[str] = mapped_column(
         ForeignKey("channels.id", ondelete="CASCADE"), index=True
     )
-    title: Mapped[str | None] = mapped_column(Text)
+    title: Mapped[str | None] = mapped_column(Text)  # normalized for display (see app.titles)
+    original_title: Mapped[str | None] = mapped_column(Text)  # raw YouTube title, preserved
     description: Mapped[str | None] = mapped_column(Text)
     published_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), index=True

backend/app/sync/videos.py

@@ -11,6 +11,7 @@ from sqlalchemy.orm import Session
 
 from app import progress, sysconfig
 from app.models import Channel, Video
+from app.titles import normalize_title
 from app.youtube.client import YouTubeClient, best_thumbnail
 from app.youtube.rss import fetch_channel_feed
 from app.youtube.shorts import make_client, probe_is_short
@@ -225,7 +226,11 @@ def apply_video_details(video: Video, item: dict) -> None:
     live = item.get("liveStreamingDetails")
     status = item.get("status", {})
 
-    video.title = snippet.get("title") or video.title
+    raw_title = snippet.get("title")
+    if raw_title:
+        # Store the raw title and a normalized one for display (feed/search/downloads).
+        video.original_title = raw_title
+        video.title = normalize_title(raw_title)
     video.description = snippet.get("description")
     if not video.published_at:
         video.published_at = parse_dt(snippet.get("publishedAt"))

backend/app/titles.py

@@ -0,0 +1,80 @@
+"""Video title normalization — clean the noisy YouTube titles for display + storage.
+
+Applied where a video's title is written (enrichment) and where a download's title comes from
+yt-dlp, so the feed, search, channel pages, and the download center all show tidy titles. The
+raw title is preserved in `Video.original_title` so this is reversible / re-derivable.
+
+Rules (user-approved "option b"):
+  1. Drop emoji / pictographs / symbols / control chars (keep accents — HU/DE/…).
+  2. Strip trailing SEO hashtag clusters (word hashtags at the end); a numeric "#3" episode
+     marker survives.
+  3. De-shout ALL-CAPS words, context-aware:
+       - If the title is *mostly shouting* (≥ half the multi-letter words are all-caps), every
+         all-caps word is Title-cased (short shouted words like CAR/WAR/ÉN get fixed too), and
+         the first letter is capitalized. Known acronyms (PS, AI, USA, PC…) and words with
+         digits (PS5, 3D) are kept; function words (to/the/és/az…) are lowercased.
+       - Otherwise only long all-caps words (≥4 letters) are Title-cased, so a lone acronym in
+         an otherwise normal title is left alone.
+  4. Collapse repeated punctuation (!!! → !) and whitespace.
+"""
+import re
+import unicodedata
+
+_DROP_CATEGORIES = {"So", "Sk", "Cc", "Cf", "Cs", "Co", "Cn"}
+_LETTER_RUN = re.compile(r"[^\W\d_]+", re.UNICODE)
+_TRAILING_HASHTAGS = re.compile(r"(?:\s+#[^\s#]*[^\W\d\s_][^\s#]*)+\s*$", re.UNICODE)
+_REPEAT_PUNCT = re.compile(r"([!?.,\-])\1{1,}")
+
+# Short function words → lowercased when the title is de-shouted (proper title-case style).
+_SHORT_LOWER = {
+    "to", "the", "and", "of", "a", "an", "in", "on", "at", "for", "or", "vs", "the",
+    "és", "az", "egy", "meg", "nem", "hogy", "de", "ha", "der", "die", "das", "und", "von",
+}
+# Kept as-is even when everything else is de-shouted (common acronyms; compared lowercased).
+_ACRONYMS = {
+    "ps", "ai", "pc", "tv", "hd", "4k", "8k", "uk", "eu", "us", "usa", "gta", "rpg", "fps",
+    "diy", "vr", "ar", "id", "ok", "faq", "nasa", "fbi", "cia", "ceo", "amd", "pdf", "usb",
+    "gps", "api", "dj", "mc", "suv", "gpu", "cpu", "ssd", "hdmi", "led", "ufo", "dna", "nba",
+    "nfl", "asmr", "pov", "diy", "wtf", "lol", "rtx", "gtx", "ios", "mmo", "vip", "hp",
+}
+
+
+def _titlecase(w: str) -> str:
+    return w[0].upper() + w[1:].lower()
+
+
+def normalize_title(raw: str | None) -> str | None:
+    """Return the cleaned title, or the input unchanged for empty/whitespace."""
+    if not raw or not raw.strip():
+        return raw
+    t = unicodedata.normalize("NFKC", raw)
+    t = "".join(ch for ch in t if unicodedata.category(ch) not in _DROP_CATEGORIES)
+    t = _TRAILING_HASHTAGS.sub("", t)
+
+    multi = [r for r in _LETTER_RUN.findall(t) if len(r) >= 2]
+    caps = [r for r in multi if r.isupper()]
+    shouting = len(caps) >= 2 and len(caps) >= 0.5 * len(multi)
+
+    def repl(m: re.Match) -> str:
+        w = m.group(0)
+        if len(w) < 2 or not w.isupper():
+            return w
+        low = w.lower()
+        if low in _ACRONYMS:
+            return w
+        if low in _SHORT_LOWER:
+            return low
+        if shouting or len(w) >= 4:
+            return _titlecase(w)
+        return w  # short all-caps in a non-shouting title → likely an acronym, keep
+
+    t = _LETTER_RUN.sub(repl, t)
+    t = _REPEAT_PUNCT.sub(r"\1", t)
+    t = re.sub(r"\s+", " ", t).strip()
+
+    if shouting:  # ensure the first letter is capitalized (a leading function word got lowered)
+        for i, ch in enumerate(t):
+            if ch.isalpha():
+                t = t[:i] + ch.upper() + t[i + 1:]
+                break
+    return t or raw

backend/app/worker.py

@@ -30,6 +30,7 @@ from app.config import settings
 from app.db import SessionLocal
 from app.downloads import formats, quota, service, storage
 from app.models import DownloadJob, MediaAsset
+from app.titles import normalize_title
 
 log = logging.getLogger("siftlode.worker")
 
@@ -190,7 +191,7 @@ def _fill_asset_meta_early(asset_id: int, info: dict) -> None:
         asset = db.get(MediaAsset, asset_id)
         if asset is None or asset.title:
             return
-        asset.title = info.get("title")
+        asset.title = normalize_title(info.get("title"))
         asset.uploader = info.get("uploader") or info.get("channel")
         asset.thumbnail_url = info.get("thumbnail")
         asset.duration_s = int(info["duration"]) if info.get("duration") else None
@@ -297,7 +298,7 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe
         ext = produced.suffix.lstrip(".").lower()
         meta = storage.MediaMeta(
             video_id=info.get("id") or source_ref,
-            title=info.get("title") or source_ref,
+            title=normalize_title(info.get("title")) or source_ref,
             uploader=info.get("uploader") or info.get("channel") or "Unknown Channel",
             upload_date=_parse_upload_date(info.get("upload_date")),
             duration_s=int(info["duration"]) if info.get("duration") else None,