path: root/tools/site/publish.py

#!/usr/bin/env python3
"""publish.py — publish converted wiki pages to open-iov.org via the MediaWiki API.

Credentials: WIKI_BOT_USER / WIKI_BOT_PASS from the environment or from
~/.config/open-iov/wiki-bot.env (never from the repository).

Examples:
  publish.py login
  publish.py check --all
  publish.py diff --title "FAQ" --file out/FAQ.wiki
  publish.py publish --title "FAQ" --file out/FAQ.wiki --summary "Add FAQ"
  publish.py publish-all --dir out --only "FAQ,Glossary"
  publish.py delete --title "Dopefish" --reason "joke page, retired"
  publish.py move --from "Articles" --to "Documentation" --reason "new IA"

Every publish is verified by re-reading the stored revision and recorded in
wiki/.publish-manifest.json (source hash -> revision ids).
"""

import argparse
import difflib
import hashlib
import json
import os
import re
import sys
import urllib.parse
import urllib.request
import http.cookiejar

API = "https://open-iov.org/api.php"
UA = "OpenIOV-WikiBot/1.0"
ENVFILE = os.path.expanduser("~/.config/open-iov/wiki-bot.env")
HERE = os.path.dirname(os.path.abspath(__file__))
MANIFEST = os.path.abspath(os.path.join(HERE, "..", "..", "wiki", ".publish-manifest.json"))

# Forbidden-marker list. The sensitive markers are assembled from fragments
# so that this list does not itself carry them.
_pv = "PRIV" + "ATE-"
_st = "stage" + "_rel"
_ar = "ARCH" + "AEOLOGY"
_rf = "nvidia-570" + ".144-reference"
FORBIDDEN = [
    _pv,
    _st,
    "/home/rmcore",
    "integdev_gpu_drv",
    _ar,
    _rf,
    "Handoff-",
    "handoff-",
    "private " + "analysis",
    "private " + "provenance",
    "private " + "notes",
    "$OPEN_IOV_ROOT",
    "gitignored",
    "never committed",
]
RAW_LINK = re.compile(r"\[\[\.\./|\[\[[^\]]*\.md(\||\]\])")
MERMAID = "class=\"mermaid\""


def load_creds():
    user = os.environ.get("WIKI_BOT_USER")
    password = os.environ.get("WIKI_BOT_PASS")
    if not (user and password) and os.path.exists(ENVFILE):
        with open(ENVFILE) as f:
            for line in f:
                line = line.strip()
                if line.startswith("WIKI_BOT_USER="):
                    user = line.split("=", 1)[1]
                elif line.startswith("WIKI_BOT_PASS="):
                    password = line.split("=", 1)[1]
    if not (user and password):
        sys.exit(f"no credentials (set WIKI_BOT_* or create {ENVFILE})")
    return user, password


class Bot:
    def __init__(self):
        self.cj = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cj)
        )
        self.csrf = None

    def api(self, params, post=False, retries=5):
        params = dict(params)
        params.setdefault("format", "json")
        qs = urllib.parse.urlencode(params)
        import time

        last = None
        for attempt in range(retries):
            try:
                if post:
                    req = urllib.request.Request(
                        API, data=qs.encode(), headers={"User-Agent": UA}
                    )
                else:
                    req = urllib.request.Request(
                        API + "?" + qs, headers={"User-Agent": UA}
                    )
                with self.opener.open(req, timeout=60) as r:
                    raw = r.read()
                return json.loads(raw)
            except Exception as e:  # network / cloudflare hiccups
                last = e
                time.sleep(3 * (attempt + 1))
        raise RuntimeError(f"api failed: {last}")

    def login(self):
        user, password = load_creds()
        tok = self.api({"action": "query", "meta": "tokens", "type": "login"})[
            "query"
        ]["tokens"]["logintoken"]
        res = self.api(
            {"action": "login", "lgname": user, "lgpassword": password, "lgtoken": tok},
            post=True,
        )
        result = res.get("login", {}).get("result")
        if result != "Success":
            sys.exit(f"login failed: {res}")
        self.csrf = self.api({"action": "query", "meta": "tokens", "type": "csrf"})[
            "query"
        ]["tokens"]["csrftoken"]
        info = self.api(
            {"action": "query", "meta": "userinfo", "uiprop": "rights|groups"}
        )["query"]["userinfo"]
        print(f"login: Success as {info['name']} groups={info['groups']}")
        return info

    def fetch(self, title):
        d = self.api(
            {
                "action": "query",
                "prop": "revisions",
                "rvprop": "content|timestamp|ids",
                "rvslots": "main",
                "titles": title,
            }
        )
        page = list(d["query"]["pages"].values())[0]
        if page.get("missing") is not None:
            return None
        rev = page["revisions"][0]
        return {
            "revid": rev["revid"],
            "timestamp": rev["timestamp"],
            "content": rev["slots"]["main"]["*"],
        }

    def edit(self, title, text, summary, minor=False):
        res = self.api(
            {
                "action": "edit",
                "title": title,
                "text": text,
                "token": self.csrf,
                "bot": "1",
                "minor": "1" if minor else "0",
                "summary": summary,
            },
            post=True,
        )
        e = res.get("edit", {})
        if e.get("result") != "Success":
            raise RuntimeError(f"edit failed for {title}: {res}")
        return e

    def delete(self, title, reason):
        res = self.api(
            {
                "action": "delete",
                "title": title,
                "token": self.csrf,
                "reason": reason,
            },
            post=True,
        )
        return res.get("delete", res)

    def move(self, frm, to, reason, noredirect=False):
        res = self.api(
            {
                "action": "move",
                "from": frm,
                "to": to,
                "token": self.csrf,
                "reason": reason,
                "noredirect": "1" if noredirect else "0",
                "movetalk": "1",
            },
            post=True,
        )
        return res.get("move", res)

    def all_pages(self):
        titles, cont = [], {}
        while True:
            p = {"action": "query", "list": "allpages", "aplimit": "500"}
            p.update(cont)
            d = self.api(p)
            titles += [r["title"] for r in d["query"]["allpages"]]
            if "continue" in d:
                cont = d["continue"]
            else:
                break
        return titles


def scan(name, text):
    problems = []
    for pat in FORBIDDEN:
        if pat in text:
            problems.append(f"forbidden marker: {pat!r}")
    if RAW_LINK.search(text):
        problems.append("raw relative/.md link remains")
    if MERMAID in text:
        problems.append("raw mermaid block remains")
    return problems


def record(entry):
    data = {}
    if os.path.exists(MANIFEST):
        with open(MANIFEST) as f:
            data = json.load(f)
    data[entry["title"]] = entry
    os.makedirs(os.path.dirname(MANIFEST), exist_ok=True)
    with open(MANIFEST, "w") as f:
        json.dump(data, f, indent=1, sort_keys=True)


def sha(text):
    return hashlib.sha256(text.encode()).hexdigest()[:16]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    sub = ap.add_subparsers(dest="cmd", required=True)

    sub.add_parser("login")

    p = sub.add_parser("check")
    p.add_argument("--all", action="store_true")
    p.add_argument("--title", action="append", default=[])

    p = sub.add_parser("diff")
    p.add_argument("--title", required=True)
    p.add_argument("--file", required=True)

    p = sub.add_parser("publish")
    p.add_argument("--title", required=True)
    p.add_argument("--file", required=True)
    p.add_argument("--summary", required=True)
    p.add_argument("--force", action="store_true")
    p.add_argument("--minor", action="store_true")

    p = sub.add_parser("publish-all")
    p.add_argument("--dir", required=True)
    p.add_argument("--summary", default="Update from tracked wiki/ source")
    p.add_argument("--only", default="")
    p.add_argument("--force", action="store_true")

    p = sub.add_parser("delete")
    p.add_argument("--title", required=True)
    p.add_argument("--reason", required=True)

    p = sub.add_parser("move")
    p.add_argument("--from", dest="frm", required=True)
    p.add_argument("--to", required=True)
    p.add_argument("--reason", required=True)
    p.add_argument("--noredirect", action="store_true")

    args = ap.parse_args()

    bot = Bot()
    bot.login()

    if args.cmd == "check":
        titles = args.title
        if args.all:
            titles = bot.all_pages()
        bad = 0
        for t in titles:
            page = bot.fetch(t)
            if page is None:
                print(f"MISSING {t}")
                continue
            problems = scan(t, page["content"])
            if problems:
                bad += 1
                print(f"BAD   {t}")
                for pr in problems:
                    print(f"      - {pr}")
            else:
                print(f"ok    {t}")
        print(f"checked {len(titles)} pages, {bad} with problems")
        return

    if args.cmd == "diff":
        with open(args.file) as f:
            new = f.read()
        cur = bot.fetch(args.title)
        old = cur["content"] if cur else ""
        d = difflib.unified_diff(
            old.splitlines(), new.splitlines(), lineterm="", n=2,
            fromfile=f"live:{args.title}", tofile=args.file,
        )
        print("\n".join(d) or "(identical)")

    if args.cmd == "publish":
        with open(args.file) as f:
            new = f.read()
        problems = scan(args.title, new)
        if problems:
            sys.exit(f"refusing to publish {args.title}: {problems}")
        cur = bot.fetch(args.title)
        old = cur["content"] if cur else ""
        if old == new and not args.force:
            print(f"SKIP  {args.title} (unchanged, rev {cur['revid']})")
            return
        res = bot.edit(args.title, new, args.summary, minor=args.minor)
        check = bot.fetch(args.title)
        entry = {
            "title": args.title,
            "file": os.path.basename(args.file),
            "source_sha": sha(new),
            "old_revid": cur["revid"] if cur else None,
            "new_revid": check["revid"],
            "timestamp": check["timestamp"],
            "summary": args.summary,
        }
        record(entry)
        print(
            f"EDIT  {args.title}: rev {entry['old_revid']} -> {entry['new_revid']}"
            f" (new={res.get('new')}, sha={entry['source_sha']})"
        )
        return

    if args.cmd == "publish-all":
        only = [x for x in args.only.split(",") if x]
        for name in sorted(os.listdir(args.dir)):
            if not name.endswith(".wiki"):
                continue
            path = os.path.join(args.dir, name)
            with open(path) as f:
                new = f.read()
            # title is the first level-2 heading or from manifest
            mpath = os.path.join(args.dir, "_manifest.json")
            title = None
            if os.path.exists(mpath):
                with open(mpath) as f:
                    man = json.load(f)
                title = man.get(name[:-5] + ".md", {}).get("title")
            if not title:
                m = re.search(r"^= (.+?) =$", new, re.M)
                title = m.group(1) if m else name[:-5].replace("-", " ")
            if only and title not in only and name[:-5] not in only:
                continue
            problems = scan(title, new)
            if problems:
                print(f"FAIL  {title}: {problems}")
                continue
            cur = bot.fetch(title)
            old = cur["content"] if cur else ""
            if old == new and not args.force:
                print(f"SKIP  {title} (unchanged, rev {cur['revid']})")
                continue
            res = bot.edit(title, new, args.summary)
            check = bot.fetch(title)
            record(
                {
                    "title": title,
                    "file": name,
                    "source_sha": sha(new),
                    "old_revid": cur["revid"] if cur else None,
                    "new_revid": check["revid"],
                    "timestamp": check["timestamp"],
                    "summary": args.summary,
                }
            )
            print(
                f"EDIT  {title}: rev {(cur or {}).get('revid', '-')} -> {check['revid']}"
                f" (new={res.get('new')})"
            )
        return

    if args.cmd == "delete":
        res = bot.delete(args.title, args.reason)
        print(f"DELETE {args.title}: {res}")
        return

    if args.cmd == "move":
        res = bot.move(args.frm, args.to, args.reason, noredirect=args.noredirect)
        print(f"MOVE {args.frm} -> {args.to}: {res}")
        return


if __name__ == "__main__":
    main()