- redownload now ALWAYS rebuilds into a fresh private asset instead of ever reusing a shared cached file: the cache sig is salted with job_id + current asset_id, so the fork is distinct from both the dedup pool and the job's own current asset. This resolves the co-held silent no-op (finding #1) — a shared asset is left intact for its other holders while THIS job re-fetches its own copy. Resolving the fork BEFORE releasing the old hold (and the unique salted key) also removes the IntegrityError-rollback-after-file-delete window (finding #2). - format_sig gains an optional `salt` for that fork identity. - titles: don't consume the space after `|`, so a "stats | #hashtags" remainder keeps the leading whitespace _TRAILING_HASHTAGS needs — and fall back to the prefix+hashtag-stripped text, so neither the stats nor the SEO hashtags are resurrected when a title empties (#3). - worker: early-fill asset.title falls back to the video id so a title that cleaned to empty doesn't show a blank card while downloading (#5). - Tests for the salt fork and the hashtag-remainder case. (Finding #4 — quota on re-download — intentionally not changed: a re-download reuses the existing job and re-fetches ~the same bytes, so calling check_enqueue would wrongly reject on the job-count cap while footprint is unchanged.)
100 lines
4.9 KiB
Python
100 lines
4.9 KiB
Python
"""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"):
|
|
0. Strip a leading social engagement prefix ("1.6M views · 66K reactions | …") some sources
|
|
(Facebook) prepend, up to and including the pipe.
|
|
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"}
|
|
# Social engagement prefix some sources (Facebook) prepend, e.g. "1.6M views · 66K reactions | ".
|
|
# Match one-or-more "<count> <label>" tokens (· / , separated) terminated by a pipe, then drop it.
|
|
# Deliberately does NOT consume the space after `|`: leaving it lets a hashtag-only remainder
|
|
# ("… | #foo #bar") keep the leading whitespace `_TRAILING_HASHTAGS` needs to strip it; the normal
|
|
# case's leading space is removed later by the whitespace collapse.
|
|
_ENGAGEMENT_PREFIX = re.compile(
|
|
r"^\s*(?:\d[\d.,]*\s*[KMB]?\s*"
|
|
r"(?:views?|reactions?|likes?|comments?|shares?|plays?|followers?)\s*[·,;]?\s*)+\|",
|
|
re.IGNORECASE,
|
|
)
|
|
_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)
|
|
stripped = _ENGAGEMENT_PREFIX.sub("", t)
|
|
prefix_removed = stripped != t
|
|
t = _TRAILING_HASHTAGS.sub("", stripped)
|
|
base = t # after prefix + trailing-hashtag removal — the fallback if de-shouting empties it
|
|
|
|
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
|
|
# Nothing survived cleaning: fall back to the prefix+hashtag-stripped text (so neither the
|
|
# stats prefix nor the SEO hashtags are resurrected). Only use the untouched raw when no prefix
|
|
# was there to strip — e.g. an all-emoji title whose glyphs were dropped, where returning the
|
|
# original is the lesser evil.
|
|
return t or base.strip() or ("" if prefix_removed else raw)
|