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}.
This commit is contained in:
parent
1ae4333922
commit
7f9847c2c2
5 changed files with 388 additions and 2 deletions
143
backend/app/downloads/storage.py
Normal file
143
backend/app/downloads/storage.py
Normal file
|
|
@ -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 `<basename>.nfo` (episodedetails) and `<basename>.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 = [
|
||||
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>',
|
||||
"<episodedetails>",
|
||||
f" <title>{escape(meta.title or '')}</title>",
|
||||
f" <showtitle>{escape(meta.uploader or '')}</showtitle>",
|
||||
f" <studio>{escape(meta.uploader or '')}</studio>",
|
||||
f" <aired>{aired}</aired>",
|
||||
f" <premiered>{aired}</premiered>",
|
||||
f" <year>{year}</year>",
|
||||
f" <plot>{escape(meta.description or '')}</plot>",
|
||||
f" <uniqueid type=\"youtube\" default=\"true\">{escape(meta.video_id)}</uniqueid>",
|
||||
]
|
||||
if meta.duration_s:
|
||||
parts.append(f" <runtime>{round(meta.duration_s / 60)}</runtime>")
|
||||
parts.append("</episodedetails>")
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def write_sidecars(
|
||||
download_root: str, rel: str, meta: MediaMeta, thumb_src: Path | None
|
||||
) -> bool:
|
||||
"""Write `<basename>.nfo` + `<basename>.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
|
||||
Loading…
Add table
Add a link
Reference in a new issue