feat(sidecar): minimal gallery-dl sidecar find + parse util
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,104 @@
|
|||||||
|
"""Minimal gallery-dl sidecar parsing (one-time filesystem-import aid).
|
||||||
|
|
||||||
|
No per-platform branching: a small common key set with fallbacks; the
|
||||||
|
full JSON is kept in raw so anything unmapped is recoverable later.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class SidecarData:
|
||||||
|
platform: str | None
|
||||||
|
external_post_id: str | None
|
||||||
|
post_url: str | None
|
||||||
|
post_title: str | None
|
||||||
|
description: str | None
|
||||||
|
attachment_count: int | None
|
||||||
|
post_date: datetime | None
|
||||||
|
raw: dict
|
||||||
|
|
||||||
|
|
||||||
|
def find_sidecar(media: Path) -> Path | None:
|
||||||
|
for cand in (media.with_suffix(".json"), Path(str(media) + ".json")):
|
||||||
|
if cand.is_file():
|
||||||
|
return cand
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _first_str(data: dict, keys: tuple[str, ...]) -> str | None:
|
||||||
|
for k in keys:
|
||||||
|
v = data.get(k)
|
||||||
|
if isinstance(v, str) and v.strip():
|
||||||
|
return v.strip()
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_date(v) -> datetime | None:
|
||||||
|
if isinstance(v, bool):
|
||||||
|
return None
|
||||||
|
if isinstance(v, (int, float)):
|
||||||
|
try:
|
||||||
|
return datetime.fromtimestamp(v, tz=UTC)
|
||||||
|
except (OverflowError, OSError, ValueError):
|
||||||
|
return None
|
||||||
|
if isinstance(v, str):
|
||||||
|
s = v.strip()
|
||||||
|
if not s:
|
||||||
|
return None
|
||||||
|
if s.endswith("Z"):
|
||||||
|
s = s[:-1] + "+00:00"
|
||||||
|
try:
|
||||||
|
dt = datetime.fromisoformat(s)
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
|
return dt if dt.tzinfo else dt.replace(tzinfo=UTC)
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def parse_sidecar(data: dict) -> SidecarData:
|
||||||
|
if not isinstance(data, dict):
|
||||||
|
data = {}
|
||||||
|
|
||||||
|
cat = data.get("category")
|
||||||
|
platform = cat if isinstance(cat, str) and cat.strip() else None
|
||||||
|
|
||||||
|
external_post_id = None
|
||||||
|
for k in ("id", "post_id", "index", "message_id"):
|
||||||
|
v = data.get(k)
|
||||||
|
if isinstance(v, bool):
|
||||||
|
continue
|
||||||
|
if isinstance(v, (str, int)) and str(v).strip():
|
||||||
|
external_post_id = str(v)
|
||||||
|
break
|
||||||
|
|
||||||
|
pc = data.get("page_count")
|
||||||
|
if isinstance(pc, bool):
|
||||||
|
attachment_count = None
|
||||||
|
elif isinstance(pc, int) and pc > 0:
|
||||||
|
attachment_count = pc
|
||||||
|
elif isinstance(data.get("images"), list):
|
||||||
|
attachment_count = len(data["images"])
|
||||||
|
else:
|
||||||
|
attachment_count = None
|
||||||
|
|
||||||
|
post_date = None
|
||||||
|
for k in ("published_at", "created_at", "date", "create_date",
|
||||||
|
"timestamp"):
|
||||||
|
if k in data:
|
||||||
|
post_date = _parse_date(data[k])
|
||||||
|
if post_date is not None:
|
||||||
|
break
|
||||||
|
|
||||||
|
return SidecarData(
|
||||||
|
platform=platform,
|
||||||
|
external_post_id=external_post_id,
|
||||||
|
post_url=_first_str(data, ("url", "post_url")),
|
||||||
|
post_title=_first_str(data, ("title",)),
|
||||||
|
description=_first_str(data, ("content", "description", "caption")),
|
||||||
|
attachment_count=attachment_count,
|
||||||
|
post_date=post_date,
|
||||||
|
raw=data,
|
||||||
|
)
|
||||||
@@ -0,0 +1,75 @@
|
|||||||
|
from backend.app.utils.sidecar import SidecarData, find_sidecar, parse_sidecar
|
||||||
|
|
||||||
|
|
||||||
|
def test_find_sidecar_stem_json(tmp_path):
|
||||||
|
media = tmp_path / "photo.jpg"
|
||||||
|
media.write_bytes(b"x")
|
||||||
|
sc = tmp_path / "photo.json"
|
||||||
|
sc.write_text("{}")
|
||||||
|
assert find_sidecar(media) == sc
|
||||||
|
|
||||||
|
|
||||||
|
def test_find_sidecar_full_name_json(tmp_path):
|
||||||
|
media = tmp_path / "photo.jpg"
|
||||||
|
media.write_bytes(b"x")
|
||||||
|
sc = tmp_path / "photo.jpg.json"
|
||||||
|
sc.write_text("{}")
|
||||||
|
assert find_sidecar(media) == sc
|
||||||
|
|
||||||
|
|
||||||
|
def test_find_sidecar_none(tmp_path):
|
||||||
|
media = tmp_path / "photo.jpg"
|
||||||
|
media.write_bytes(b"x")
|
||||||
|
assert find_sidecar(media) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_empty_dict_all_none():
|
||||||
|
sd = parse_sidecar({})
|
||||||
|
assert isinstance(sd, SidecarData)
|
||||||
|
assert (sd.platform, sd.external_post_id, sd.post_url, sd.post_title,
|
||||||
|
sd.description, sd.attachment_count, sd.post_date) == (
|
||||||
|
None, None, None, None, None, None, None)
|
||||||
|
assert sd.raw == {}
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_core_fields_and_id_priority():
|
||||||
|
sd = parse_sidecar({
|
||||||
|
"category": "patreon",
|
||||||
|
"id": 12345, "post_id": 999,
|
||||||
|
"url": "https://patreon.com/posts/12345",
|
||||||
|
"title": " Hello ",
|
||||||
|
"content": "<p>body</p>",
|
||||||
|
"page_count": 4,
|
||||||
|
"published_at": "2023-08-01T04:20:02Z",
|
||||||
|
})
|
||||||
|
assert sd.platform == "patreon"
|
||||||
|
assert sd.external_post_id == "12345" # 'id' wins over 'post_id'
|
||||||
|
assert sd.post_url == "https://patreon.com/posts/12345"
|
||||||
|
assert sd.post_title == "Hello"
|
||||||
|
assert sd.description == "<p>body</p>"
|
||||||
|
assert sd.attachment_count == 4
|
||||||
|
assert sd.post_date.year == 2023 and sd.post_date.tzinfo is not None
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_description_precedence_and_images_count():
|
||||||
|
sd = parse_sidecar({"description": "d", "caption": "c",
|
||||||
|
"images": [1, 2, 3]})
|
||||||
|
assert sd.description == "d" # content>description>caption
|
||||||
|
assert sd.attachment_count == 3 # len(images) fallback
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_date_epoch_and_unparseable_and_naive():
|
||||||
|
assert parse_sidecar({"timestamp": 1690857602}).post_date.tzinfo is not None
|
||||||
|
assert parse_sidecar({"date": "not-a-date"}).post_date is None
|
||||||
|
naive = parse_sidecar({"date": "2023-08-01T00:00:00"}).post_date
|
||||||
|
assert naive is not None and naive.utcoffset().total_seconds() == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ignores_non_str_junk():
|
||||||
|
sd = parse_sidecar({"category": 5, "title": 7, "page_count": "x",
|
||||||
|
"id": True})
|
||||||
|
assert sd.platform is None and sd.post_title is None
|
||||||
|
assert sd.attachment_count is None
|
||||||
|
assert sd.external_post_id is None # bool is not a valid id
|
||||||
|
assert sd.raw == {"category": 5, "title": 7, "page_count": "x",
|
||||||
|
"id": True}
|
||||||
Reference in New Issue
Block a user