diff --git a/pipelines/songs/publish.py b/pipelines/songs/publish.py index 926b8ad..49069f9 100644 --- a/pipelines/songs/publish.py +++ b/pipelines/songs/publish.py @@ -1,7 +1,9 @@ #!/usr/bin/env python3 -"""Songs pipeline renderer — converts source .md files to MediaWiki pages.""" +"""Songs pipeline — validates, renders, and publishes song .md files to MediaWiki.""" import argparse +import hashlib +import io import os import subprocess import sys @@ -57,10 +59,11 @@ def markdown_to_wikitext(body: str) -> str: return result.stdout -def build_wikitext(fm: dict, body_wikitext: str, source_url: str, source_ref: str) -> str: +def build_wikitext(fm: dict, body_wikitext: str, source_url: str, source_ref: str, lrc_filename: str | None = None) -> str: banner = f"{{{{Auto-generated|source={source_url}|commit={source_ref}}}}}" category = f"[[Category:{fm['album']}]]" - return f"{banner}\n\n{body_wikitext}\n{category}\n" + lrc_line = f"[[Media:{lrc_filename}|Synced lyrics (.lrc)]]\n" if lrc_filename else "" + return f"{banner}\n\n{body_wikitext}{lrc_line}{category}\n" def connect_wiki() -> mwclient.Site: @@ -76,17 +79,57 @@ def connect_wiki() -> mwclient.Site: return site +def upload_lrc(lrc_path: Path, title: str, site: mwclient.Site) -> bool: + """Upload LRC file to wiki if content has changed. Returns True if uploaded.""" + if not lrc_path.exists(): + print(f" warn: LRC file not found: {lrc_path}", file=sys.stderr) + return False + + content = lrc_path.read_bytes() + try: + content.decode("utf-8") + except UnicodeDecodeError: + print(f" warn: {lrc_path} is not valid UTF-8 text, skipping LRC upload", file=sys.stderr) + return False + + wiki_filename = f"{title}.lrc" + local_sha1 = hashlib.sha1(content).hexdigest() + + if site.images[wiki_filename].imageinfo.get("sha1") == local_sha1: + return False + + site.upload( + file=io.BytesIO(content), + filename=wiki_filename, + description=f"Synced lyrics for [[{title}]] (auto-published)", + ignore=True, + ) + return True + + AUTO_BANNER_PREFIX = "{{Auto-generated" -def publish_one(rel: Path, post, site: mwclient.Site, source_ref: str, gitea_repo_url: str) -> str: +def publish_one(abs_path: Path, source_dir: Path, post, site: mwclient.Site, source_ref: str, gitea_repo_url: str) -> str: """Returns 'published', 'noop', or 'skipped-manual'.""" + rel = abs_path.relative_to(source_dir) fm = post.metadata + title = str(fm["title"]) + body_wikitext = markdown_to_wikitext(strip_first_h1(post.content)) source_url = f"{gitea_repo_url}/src/commit/{source_ref}/{rel}" - page_content = build_wikitext(fm, body_wikitext, source_url, source_ref) - page = site.pages[fm["title"]] + lrc_filename = None + lrc_rel = fm.get("lrc") + if lrc_rel: + lrc_uploaded = upload_lrc(abs_path.parent / str(lrc_rel), title, site) + lrc_filename = f"{title}.lrc" + if lrc_uploaded: + print(f" [lrc-uploaded] {lrc_filename}") + + page_content = build_wikitext(fm, body_wikitext, source_url, source_ref, lrc_filename) + + page = site.pages[title] existing = page.text() if existing and not existing.startswith(AUTO_BANNER_PREFIX): @@ -168,7 +211,7 @@ def main() -> None: counts = {"published": 0, "noop": 0, "skipped-manual": 0} for abs_path, post in posts.items(): rel = abs_path.relative_to(source_dir) - outcome = publish_one(rel, post, site, source_ref, gitea_repo_url) + outcome = publish_one(abs_path, source_dir, post, site, source_ref, gitea_repo_url) counts[outcome] += 1 if outcome == "skipped-manual": print(f" [skipped-manual] {rel} ← page exists without auto-gen banner; delete or add banner to hand over to bot", file=sys.stderr)