novoyuuparosk-auto-wiki/pipelines/songs/publish.py

#!/usr/bin/env python3
"""Songs pipeline — validates, renders, and publishes song .md files to MediaWiki."""

import argparse
import hashlib
import io
import os
import sys
import urllib.parse
from datetime import datetime
from pathlib import Path

import frontmatter
import mwclient

sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.wiki import (
    AUTO_BANNER_PREFIX,
    connect_wiki,
    extract_md_link_targets,
    markdown_to_wikitext,
    resolve_md_links,
    strip_first_h1,
)


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")

    if errors:
        raise ValueError(f"{path}: " + "; ".join(errors))


def build_title_index(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 build_link_graph(posts: dict) -> dict[Path, set[Path]]:
    """Map each publishable file to the publishable .md files it links to.

    Read from the raw Markdown, so it costs no Pandoc runs. Targets outside the
    publishable set are dropped here; `resolve_md_links` is what refuses them
    loudly at render time.
    """
    graph: dict[Path, set[Path]] = {}
    for path, post in posts.items():
        targets = set()
        for raw in extract_md_link_targets(post.content):
            rel_path = urllib.parse.unquote(raw.partition("#")[0])
            if not rel_path.lower().endswith(".md"):
                continue
            resolved = (path.parent / rel_path).resolve()
            if resolved in posts:
                targets.add(resolved)
        graph[path] = targets
    return graph


def select_publish_set(posts: dict, seeds: set[Path]) -> dict[Path, object]:
    """The changed files, plus the pages their links touch in either direction.

    Publishing is not instant, so only what a push actually affects is written.
    Two extra hops are unavoidable, both one level deep:

    - **outbound**: a page the changed page links to, so the wikilink cannot
      point at a page that was never published (a red link);
    - **inbound**: a page linking *to* a changed page, because if the changed
      page's `title` moved, every link aimed at it is now stale.

    Ordering follows `posts` so a run's output is stable regardless of which
    files the diff reported.
    """
    graph = build_link_graph(posts)
    selected = set(seeds)
    for path, targets in graph.items():
        if path in seeds:
            selected |= targets
        elif targets & seeds:
            selected.add(path)
    return {path: post for path, post in posts.items() if path in selected}


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("[[Category:Songs]]")
        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 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


def render_page(abs_path: Path, source_dir: Path, post, source_ref: str, gitea_repo_url: str, title_index: dict[str, str]) -> str:
    """Render one file to finished wikitext. Pure: touches no wiki, writes nothing.

    Kept separate from the write so that every page in a run is rendered before
    the first one is published. A bad link or a broken column fence then fails
    the run with the wiki untouched, instead of half-updating it.
    """
    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 = resolve_md_links(body_wikitext, abs_path, title_index)
    source_url = f"{gitea_repo_url}/src/commit/{source_ref}/{rel}"

    lrc_filename = f"{title}.lrc" if fm.get("lrc") else None
    return build_wikitext(fm, body_wikitext, source_url, source_ref, lrc_filename)


def publish_one(abs_path: Path, post, page_content: str, site: mwclient.Site, source_ref: str) -> str:
    """Write one pre-rendered page. Returns 'published', 'noop', or 'skipped-manual'."""
    fm = post.metadata
    title = str(fm["title"])

    lrc_rel = fm.get("lrc")
    if lrc_rel and upload_lrc(abs_path.parent / str(lrc_rel), title, site):
        print(f"  [lrc-uploaded] {title}.lrc")

    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 discover_publishable(source_dir: Path) -> dict[Path, object]:
    """Every publishable .md in the repository, validated, keyed by absolute path.

    The whole tree is scanned on every run, never a subset of it, however few
    pages the run then publishes. Two things depend on seeing all of it:

    - **link resolution**: a link to a sibling .md resolves through this set, so
      a partial scan would resolve the same link differently depending on which
      files a push happened to touch, and silently drop it when the target fell
      outside the subset;
    - **title collisions**: two files claiming one wiki page name must fail the
      run wherever they sit in the tree, not only when both land in one push.

    Files under wip/ and README.md are never published. Everything else needs an
    explicit `wiki.publish: true`; a file is never published as a side effect of
    something else linking to it.
    """
    posts: dict[Path, object] = {}
    titles: dict[str, Path] = {}

    for path in sorted(source_dir.rglob("*.md")):
        rel = path.relative_to(source_dir)
        if "wip" in rel.parts or path.name == "README.md":
            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}': {rel} and "
                f"{titles[title].relative_to(source_dir)}"
            )
        titles[title] = path
        posts[path.resolve()] = 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="Changed files (relative to source-dir) to publish; the push diff",
    )
    parser.add_argument("--all", action="store_true", help="Publish every publishable file")
    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("/")

    # The repository is always scanned in full, whatever gets published: link
    # resolution and title-collision checks must see every publishable file, or
    # the same link resolves differently depending on which files a push touched.
    posts = discover_publishable(source_dir)
    if not posts:
        print("No publishable files found.")
        return

    title_index = build_title_index(posts)
    print(f"Scanned repository: {len(posts)} publishable file(s).")

    if args.all:
        selected = posts
    elif args.files:
        seeds = {(source_dir / f).resolve() for f in args.files} & posts.keys()
        if not seeds:
            print("No changed file is publishable. Nothing to publish.")
            return
        selected = select_publish_set(posts, seeds)
    else:
        print("Nothing to do: pass --files or --all.")
        return

    # Render every page before writing any of them, so a bad link fails the run
    # with the wiki untouched rather than half-updated.
    rendered = {
        abs_path: render_page(abs_path, source_dir, post, source_ref, gitea_repo_url, title_index)
        for abs_path, post in selected.items()
    }
    print(f"Rendered {len(rendered)} page(s). Publishing.")

    site = connect_wiki()

    counts = {"published": 0, "noop": 0, "skipped-manual": 0}
    for abs_path, post in selected.items():
        rel = abs_path.relative_to(source_dir)
        outcome = publish_one(abs_path, post, rendered[abs_path], site, source_ref)
        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)