vps-health-api/scripts/review-levels.py

#!/usr/bin/env python3
"""Replay the review levels of REVIEW.md on the pull requests that landed (#63).

Usage:
  python3 scripts/review-levels.py [N ...]   (default: every PR of LANDED)
  python3 scripts/review-levels.py --check   (no network)

The first form reads the Forgejo API (token in $FORGEJO_TOKEN, else
~/.forgejo-token; read-only calls) and prints each PR with its level and the
files that set it, then the distribution. --check compares the patterns of
this script with the first column of the Haut table of REVIEW.md and exits 1
if they differ: a PR that changes the Niveaux section updates this script in
the same diff.

Adapted from simpl-resultat/scripts/review-levels.py. The file list comes from
the PR's .diff, never from /pulls/{n}/files: that route answers 500 on a PR
whose base branch is gone, which is every link of a stack merged by
fast-forward. Three PRs (EMPTY_DIFF) have an empty .diff and are read from
the local clone: run the script from the repository. The content conditions of the Haut table are approximated from
the diff: index.js routing and auth, side effects of the shipped modules
(disk writes, the temp-file sweep, body caps, shell commands, outbound calls,
require.main), guard tests (an existing line changed or removed),
package.json (anything but a version), package-lock.json (a package added or
removed, an origin changed), .gitignore (a line removed).
"""
import itertools
import json
import os
import re
import subprocess
import sys
import urllib.request

API = "https://git.lacompagniemaximus.com/api/v1/repos/maximus/vps-health-api"
REVIEW_MD = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "REVIEW.md")

# Every PR whose head reached main, frozen by number. Forgejo flags only part
# of them as merged: a stack merged by fast-forward (#18-#23, #31, #34, #37)
# shows as closed. #32 (a red-CI demo) never landed.
LANDED = [2, 5, 6, 8, 9, 18, 19, 20, 21, 22, 23, 26, 28, 30, 31, 34, 37, 39, 42, 43,
          44, 45, 47, 51, 52, 53, 56, 57, 59, 60, 62]

# Three PRs whose .diff comes back empty: once fast-forwarded into main, their
# merge_base equals their head. Each is one commit, read from the local clone.
EMPTY_DIFF = {
    18: ("39a55fad93414b8e0049724f351710bfe1eb7a77", "79cb813767cfb53282d5fe6f386af3d740bf7f7e"),
    31: ("2b35b78a0beb04dc9ae92c13e42b9ff09fd2d147", "4e7879da6ca5163d37d11b9c5803cf2bed640b59"),
    34: ("4e7879da6ca5163d37d11b9c5803cf2bed640b59", "156ca818ca37441a2166d112251529ac32783afc"),
}

# REVIEW.md, table Haut: rows without a condition.
HIGH_ALWAYS = [
    "agent/pg-backup-pull/**",
    "agent/claude-usage/{limits,push-claude-usage}.js", "agent/push-metrics.js", "agent/**/*.sh",
    "Dockerfile",
    ".forgejo/workflows/**",
    "vitest.config.js", "scripts/boot-from-copy.sh",
    ".claude/settings.json", ".mcp.json",
]
SHIPPED = ["{index,metrics,claude-usage,claude-usage-history,backups}.js"]
GUARDS = ["__tests__/{auth,writes}.test.js"]
LOW = ["**/*.md", "docs/**", "LICENSE", "reports/**"]
NEVER_LOW = ["CLAUDE.md", "REVIEW.md", "agent/README.md", ".claude/**"]

AUTH = re.compile(r"\b(ROUTES|TOKEN_KINDS|resolveRoute|checkAuth|tokenMatches|logRejection|"
                  r"safeLogValue|handler|\w*_TOKEN)\b")
EFFECTS = re.compile(
    r"\b(writeFile\w*|writeSync|createWriteStream|appendFile\w*|rename\w*|mkdir\w*|unlink\w*|"
    r"rmSync|rmdir\w*|chmod\w*|chown\w*|copyFile\w*|symlink\w*|truncate\w*|writeJsonAtomic|"
    r"writeHostSnapshot|sweepStaleTempFiles|TEMP_FILE_RE|readBody|\w*MAX_BODY_BYTES|"
    r"child_process|execSync|execFile\w*|spawn\w*|fetch|https?\.(request|get))\b"
    r"|require\.main")


def token():
    return os.environ.get("FORGEJO_TOKEN") or open(os.path.expanduser("~/.forgejo-token")).read().strip()


def get(path, raw=False):
    req = urllib.request.Request(API + path, headers={"Authorization": "token " + token()})
    data = urllib.request.urlopen(req).read()
    return data if raw else json.loads(data)


def expand_braces(pat):
    m = re.search(r"\{([^{}]*)\}", pat)
    if not m:
        return [pat]
    return list(itertools.chain.from_iterable(
        expand_braces(pat[:m.start()] + alt + pat[m.end():]) for alt in m.group(1).split(",")))


def glob_re(pat):
    # '**/' spans zero or more directories, '*' and '?' never cross '/'
    i, out = 0, ""
    while i < len(pat):
        if pat.startswith("**/", i):
            out += "(?:.*/)?"; i += 3; continue
        if pat.startswith("**", i):
            out += ".*"; i += 2; continue
        c = pat[i]
        out += "[^/]*" if c == "*" else "[^/]" if c == "?" else re.escape(c)
        i += 1
    return re.compile("^" + out + "$")


def match(pats, path):
    # '{a,b}' alternates, wildcards allowed inside
    return any(glob_re(p).match(path) for pat in pats for p in expand_braces(pat))


def parse_diff(diff):
    """[{filename, previous_filename, status, block}] from a git diff."""
    files = []
    for block in re.split(r"(?m)^diff --git ", diff)[1:]:
        head = block.split("\n@@", 1)[0]
        m = re.match(r"a/(\S+) b/(\S+)", block)
        old, new = m.group(1), m.group(2)
        status = "changed"
        if re.search(r"(?m)^new file mode", head):
            status = "added"
        elif re.search(r"(?m)^deleted file mode", head):
            status = "deleted"
        elif re.search(r"(?m)^rename from", head) or old != new:
            status = "renamed"
        files.append({"filename": new, "previous_filename": old if old != new else None,
                      "status": status, "block": block})
    return files


def meaningful(text):
    t = text.strip()
    return t != "" and not t.startswith(("#", "//", "*", "/*"))


def changed(block):
    return [l for l in block.split("\n")
            if l[:1] in "+-" and not l.startswith(("+++", "---")) and meaningful(l[1:])]


def touches(block, pattern):
    # a hunk whose enclosing function (the @@ context) or a changed line matches
    for hunk in re.split(r"(?m)^(?=@@)", block)[1:]:
        if pattern.search(hunk.split("\n", 1)[0]) or any(pattern.search(l) for l in changed(hunk)):
            return True
    return False


def package_json_changes(block):
    # any key/value change except "version" and a bump of a package already listed
    before, after = {}, {}
    for l in changed(block):
        m = re.match(r'^([+-])\s*"([^"]+)":\s*(.*?),?\s*$', l)
        if not m:
            return True
        (before if m.group(1) == "-" else after)[m.group(2)] = m.group(3)
    for key in set(before) | set(after):
        if key == "version":
            continue
        if key in before and key in after and all(re.match(r'^"[\^~]?\d', v) for v in (before[key], after[key])):
            continue
        return True
    return False


def origin_of(url):
    m = re.match(r"^([a-z+]+)://([^/]+)", url)
    return m.group(1) + "://" + m.group(2) if m else url


def lock_adds_or_removes(block):
    if any(re.match(r'^[+-]\s+"node_modules/[^"]+": \{', l) for l in block.split("\n")):
        return True
    for hunk in re.split(r"(?m)^@@", block)[1:]:
        hl = hunk.split("\n")
        before, after = set(), set()
        for l in hl:
            m = re.match(r'^([+-])\s+"resolved": "([^"]+)"', l)
            if m:
                (before if m.group(1) == "-" else after).add(origin_of(m.group(2)))
        if after - before:
            return True
        if any(re.match(r'^\+\s+"integrity": ', l) for l in hl) and not any(re.match(r'^\+\s+"version": ', l) for l in hl):
            return True
    return False


def review_level(files):
    triggers = []
    for f in files:
        b = f["block"]
        for path in filter(None, {f["filename"], f["previous_filename"]}):
            if match(HIGH_ALWAYS, path):
                triggers.append(path)
            if path == "index.js" and touches(b, AUTH):
                triggers.append(path + " (routing or auth)")
            if match(SHIPPED, path) and (f["status"] != "changed" or touches(b, EFFECTS)):
                triggers.append(path + " (side effect)")
            if match(GUARDS, path) and [l for l in changed(b) if l.startswith("-")]:
                triggers.append(path + " (existing line changed)")
            if path == "package.json" and package_json_changes(b):
                triggers.append(path)
            if path == "package-lock.json" and lock_adds_or_removes(b):
                triggers.append(path)
            if path == ".gitignore" and [l for l in changed(b) if l.startswith("-")]:
                triggers.append(path + " (line removed)")
            if path == "REVIEW.md":
                triggers.append(path + " (check the « Niveaux » section by hand)")
            parts = path.split("/")
            new_hidden = f["status"] == "added" and any(
                x.startswith(".") for x in parts[:-1]) and parts[0] not in (".forgejo", ".claude")
            if (f["status"] != "changed" and len(parts) == 1 and not path.endswith(".md")) or new_hidden:
                triggers.append(path + " (" + f["status"] + " at top level)")
    paths = [f["filename"] for f in files]
    if triggers:
        return "haut", sorted(set(triggers))
    if all(match(LOW, p) and not match(NEVER_LOW, p) for p in paths):
        return "bas", []
    return "normal", []


CONDITIONAL = {
    "index.js", "{index,metrics,claude-usage,claude-usage-history,backups}.js",
    "__tests__/{auth,writes}.test.js", "package.json", "package-lock.json", ".gitignore",
    "REVIEW.md",
}


def check():
    """Compare this script's patterns with the first column of the Haut table."""
    text = open(REVIEW_MD, encoding="utf-8").read()
    table = text.split("### Haut", 1)[1].split("\n\n", 2)[1]
    documented = set()
    for row in table.split("\n"):
        if not row.startswith("| ") or row.startswith("| Motif") or row.startswith("| fichier à la racine"):
            continue
        documented.update(re.findall(r"`([^`]+)`", row.split(" | ")[0]))
    coded = set(HIGH_ALWAYS) | CONDITIONAL
    for label, extra in (("REVIEW.md only", documented - coded), ("script only", coded - documented)):
        for pattern in sorted(extra):
            print(f"{label}: {pattern}")
    if documented != coded:
        return 1
    print(f"ok: {len(coded)} patterns, identical to the Haut table of REVIEW.md")
    return 0


def main():
    if sys.argv[1:] == ["--check"]:
        sys.exit(check())
    numbers = [int(a) for a in sys.argv[1:]] or LANDED
    rows = []
    for n in numbers:
        if n in EMPTY_DIFF:
            diff = subprocess.run(["git", "diff", *EMPTY_DIFF[n]], capture_output=True, text=True, check=True).stdout
        else:
            diff = get(f"/pulls/{n}.diff", raw=True).decode("utf-8", "replace")
        files = parse_diff(diff)
        if not files:
            sys.exit(f"#{n}: empty diff, the level would read as bas")
        level, triggers = review_level(files)
        rows.append((n, level))
        print(f"#{n}\t{level}\t{'; '.join(triggers)}")
    counts = {lvl: sum(1 for r in rows if r[1] == lvl) for lvl in ("bas", "normal", "haut")}
    print(f"{len(rows)} PRs: {counts} -> haut {100 * counts['haut'] / len(rows):.1f} %")


if __name__ == "__main__":
    main()