mikkeli
87874fa935
- `album` no longer required; category format changed to Category:Album:<name> - New optional `categories` list for additional arbitrary categories - `wiki.siblings` bill of materials: path+tag required for resolution, label optional - Post-Pandoc [[TAG]] substitution with four-tier fallback (link+label, link, plain label, plain tag) - Sibling map built from all published posts in the same run; unresolved siblings degrade gracefully - SCHEMA.md fully updated; README.md deferred items moved to implemented Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
283 lines
9.3 KiB
Python
283 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Songs pipeline — validates, renders, and publishes song .md files to MediaWiki."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import io
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import frontmatter
|
|
import mwclient
|
|
|
|
|
|
def validate(fm: dict, path: Path) -> None:
|
|
errors = []
|
|
|
|
val = fm.get("title")
|
|
if not val or not str(val).strip():
|
|
errors.append("missing or empty 'title'")
|
|
|
|
if not isinstance(fm.get("wiki", {}).get("publish"), bool):
|
|
errors.append("'wiki.publish' must be a boolean (true/false), not a string")
|
|
|
|
if "release_date" in fm and fm["release_date"]:
|
|
try:
|
|
datetime.strptime(str(fm["release_date"]), "%Y-%m-%d")
|
|
except ValueError:
|
|
errors.append("'release_date' must be YYYY-MM-DD")
|
|
|
|
if "categories" in fm and not isinstance(fm["categories"], list):
|
|
errors.append("'categories' must be a list")
|
|
|
|
for i, sib in enumerate(fm.get("wiki", {}).get("siblings") or []):
|
|
if not isinstance(sib, dict):
|
|
errors.append(f"wiki.siblings[{i}] must be a mapping")
|
|
continue
|
|
sib_path = sib.get("path")
|
|
sib_tag = sib.get("tag")
|
|
if sib_path and sib_tag:
|
|
if not (path.parent / str(sib_path)).resolve().exists():
|
|
errors.append(f"wiki.siblings[{i}]: path '{sib_path}' not found")
|
|
|
|
if errors:
|
|
raise ValueError(f"{path}: " + "; ".join(errors))
|
|
|
|
|
|
def strip_first_h1(text: str) -> str:
|
|
"""Remove the first '# Heading' line and any immediately following blank line."""
|
|
lines = text.split("\n")
|
|
for i, line in enumerate(lines):
|
|
if line.strip().startswith("# "):
|
|
del lines[i]
|
|
if i < len(lines) and lines[i].strip() == "":
|
|
del lines[i]
|
|
break
|
|
return "\n".join(lines)
|
|
|
|
|
|
def markdown_to_wikitext(body: str) -> str:
|
|
result = subprocess.run(
|
|
["pandoc", "-f", "markdown", "-t", "mediawiki"],
|
|
input=body,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"pandoc failed: {result.stderr.strip()}")
|
|
return result.stdout
|
|
|
|
|
|
def build_sibling_map(posts: dict) -> dict[str, str]:
|
|
"""Map abs_path_str → wiki_title for all published posts."""
|
|
return {str(path): str(post.metadata["title"]) for path, post in posts.items()}
|
|
|
|
|
|
def apply_sibling_substitutions(wikitext: str, this_path: Path, fm_wiki: dict, sibling_map: dict[str, str]) -> str:
|
|
"""Replace [[TAG]] placeholders using the declared sibling bill of materials."""
|
|
for sib in (fm_wiki.get("siblings") or []):
|
|
if not isinstance(sib, dict):
|
|
continue
|
|
tag = sib.get("tag")
|
|
path = sib.get("path")
|
|
if not tag or not path:
|
|
continue
|
|
label = (sib.get("label") or "").strip()
|
|
abs_sib = str((this_path.parent / str(path)).resolve())
|
|
title = sibling_map.get(abs_sib)
|
|
if title and label:
|
|
replacement = f"[[{title}|{label}]]"
|
|
elif title:
|
|
replacement = f"[[{title}]]"
|
|
elif label:
|
|
replacement = label
|
|
else:
|
|
replacement = tag
|
|
wikitext = wikitext.replace(f"[[{tag}]]", replacement)
|
|
return wikitext
|
|
|
|
|
|
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}}}}}"
|
|
lrc_line = f"[[Media:{lrc_filename}|Synced lyrics (.lrc)]]\n" if lrc_filename else ""
|
|
|
|
cat_parts = []
|
|
if fm.get("album"):
|
|
cat_parts.append(f"[[Category:Album:{fm['album']}]]")
|
|
for cat in (fm.get("categories") or []):
|
|
cat_parts.append(f"[[Category:{cat}]]")
|
|
categories = "\n".join(cat_parts) + "\n" if cat_parts else ""
|
|
|
|
return f"{banner}\n\n{body_wikitext}{lrc_line}{categories}"
|
|
|
|
|
|
def connect_wiki() -> mwclient.Site:
|
|
from urllib.parse import urlparse
|
|
|
|
api_url = os.environ["WIKI_API_URL"]
|
|
parsed = urlparse(api_url)
|
|
host = parsed.netloc
|
|
# Strip api.php to get the wiki root path (e.g. "/" or "/w/")
|
|
path = parsed.path[: parsed.path.rfind("/") + 1] or "/"
|
|
site = mwclient.Site(host, path=path, scheme=parsed.scheme)
|
|
site.login(os.environ["WIKI_BOT_USER"], os.environ["WIKI_BOT_PASSWORD"])
|
|
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(abs_path: Path, source_dir: Path, post, site: mwclient.Site, source_ref: str, gitea_repo_url: str, sibling_map: dict[str, 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))
|
|
body_wikitext = apply_sibling_substitutions(body_wikitext, abs_path, fm.get("wiki", {}), sibling_map)
|
|
source_url = f"{gitea_repo_url}/src/commit/{source_ref}/{rel}"
|
|
|
|
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):
|
|
return "skipped-manual"
|
|
|
|
if existing == page_content:
|
|
return "noop"
|
|
|
|
ref_short = source_ref[:8] if source_ref else "unknown"
|
|
page.save(page_content, summary=f"Auto-published from {ref_short} (songs pipeline)")
|
|
return "published"
|
|
|
|
|
|
def load_and_validate(files: list[Path], source_dir: Path) -> dict[Path, object]:
|
|
posts = {}
|
|
titles: dict[str, Path] = {}
|
|
|
|
for path in files:
|
|
if not path.exists():
|
|
print(f" warn: {path} not found, skipping", file=sys.stderr)
|
|
continue
|
|
|
|
post = frontmatter.load(str(path))
|
|
|
|
wiki = post.metadata.get("wiki")
|
|
if not isinstance(wiki, dict) or not wiki.get("publish"):
|
|
continue
|
|
|
|
validate(post.metadata, path)
|
|
|
|
title = str(post.metadata["title"])
|
|
if title in titles:
|
|
raise ValueError(
|
|
f"Duplicate wiki title '{title}': {path} and {titles[title]}"
|
|
)
|
|
titles[title] = path
|
|
posts[path] = post
|
|
|
|
return posts
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Publish song .md files to MediaWiki.")
|
|
parser.add_argument("--source-dir", required=True, type=Path)
|
|
parser.add_argument("--files", nargs="*", default=[], help="Relative paths within source-dir")
|
|
parser.add_argument("--all", action="store_true", help="Publish all publishable files")
|
|
args = parser.parse_args()
|
|
|
|
source_dir = args.source_dir.resolve()
|
|
source_ref = os.environ.get("SOURCE_REF", "")
|
|
gitea_repo_url = os.environ.get("GITEA_REPO_URL", "").rstrip("/")
|
|
|
|
if args.all:
|
|
candidates = list(source_dir.rglob("*.md"))
|
|
elif args.files:
|
|
candidates = [source_dir / f for f in args.files]
|
|
else:
|
|
print("Nothing to do: pass --files or --all.")
|
|
return
|
|
|
|
files = [
|
|
p for p in candidates
|
|
if p.exists()
|
|
and p.suffix == ".md"
|
|
and "wip" not in p.relative_to(source_dir).parts
|
|
and p.name != "README.md"
|
|
]
|
|
|
|
if not files:
|
|
print("No publishable candidates after filtering.")
|
|
return
|
|
|
|
print(f"Validating {len(files)} file(s)...")
|
|
posts = load_and_validate(files, source_dir)
|
|
sibling_map = build_sibling_map(posts)
|
|
print("Validation passed.")
|
|
|
|
site = connect_wiki()
|
|
|
|
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(abs_path, source_dir, post, site, source_ref, gitea_repo_url, sibling_map)
|
|
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)
|
|
else:
|
|
print(f" [{outcome}] {rel}")
|
|
|
|
print(
|
|
f"\nDone: {counts['published']} published, {counts['noop']} unchanged, "
|
|
f"{counts['skipped-manual']} skipped (existing manual pages)."
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as e:
|
|
print(f"ERROR: {e}", file=sys.stderr)
|
|
sys.exit(1)
|