IBKRTrader/scripts/roadmap_block.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Frischt den erzeugten Roadmap-Block in docs/ROADMAP.md auf und gleicht die Prosa ab.

Der Planungsstand liegt im Deploymentcenter (Projekt ibkrtrader); docs/ROADMAP.md
zeigt zwischen den <!-- dc:roadmap -->-Markern nur eine Kopie davon. Dieses Skript
holt sie neu.

Warum nicht `php cli/roadmap.php --write`: Das setzt voraus, dass man auf der
Maschine des Deploymentcenters sitzt. Der dokumentierte Rueckfallweg der Anleitung
(`curl ... | jq -r .markdown`) setzt jq voraus, das unter Windows in der Regel
fehlt. Dieses Skript braucht nur Python.

    python scripts/roadmap_block.py            # Block auffrischen und abgleichen
    python scripts/roadmap_block.py --check    # nur abgleichen, nichts schreiben

Rueckgabewert 1, wenn der Abgleich Fehler meldet (Warnungen sind erlaubt).

DC_TOKEN kommt aus der Umgebung; fehlt es dort, wird .mcp.json gelesen.
"""
import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request

BASE = os.environ.get("DC_BASE_URL", "https://dc.mhdf.de").rstrip("/")
PROJECT = "ibkrtrader"
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ROADMAP = os.path.join(ROOT, "docs", "ROADMAP.md")
# Nur ein Marker am Zeilenanfang zaehlt. Sonst trifft die Suche den Fliesstext
# weiter oben, der die Marker beim Namen nennt - und loescht den Absatz mitsamt
# allem, was bis zum echten Block dazwischenliegt. Genau das ist beim ersten Lauf
# passiert.
START = re.compile(r"^<!-- dc:roadmap\b[^\n]*-->", re.M)
END = re.compile(r"^<!-- /dc:roadmap -->", re.M)


def token():
    """Token aus der Umgebung, sonst aus .mcp.json (die Datei ist gitignored)."""
    tok = os.environ.get("DC_TOKEN")
    if tok:
        return tok
    mcp = os.path.join(ROOT, ".mcp.json")
    if os.path.exists(mcp):
        with open(mcp, encoding="utf-8") as fh:
            cfg = json.load(fh)
        env = cfg.get("mcpServers", {}).get("deploymentcenter", {}).get("env", {})
        if env.get("DC_TOKEN"):
            return env["DC_TOKEN"]
    sys.exit("DC_TOKEN ist weder gesetzt noch in .mcp.json zu finden.")


def call(method, path, body=None, query=None):
    url = BASE + path
    if query:
        url += "?" + urllib.parse.urlencode(query)
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", "Bearer " + token())
    req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode()
        try:
            return json.loads(raw)
        except ValueError:
            sys.exit("HTTP %d: %s" % (exc.code, raw[:300]))


def read_file():
    with open(ROADMAP, encoding="utf-8") as fh:
        return fh.read()


def refresh(text):
    """Ersetzt den Bereich zwischen den Markern. Fehlt er, wird er angehaengt."""
    answer = call("GET", "/api/bugtracker/v1/roadmap.php",
                  query={"action": "render", "project_slug": PROJECT})
    block = answer.get("markdown")
    if not block:
        sys.exit("Kein Markdown erhalten: " + json.dumps(answer)[:300])

    start = START.search(text)
    end = END.search(text)
    if not start or not end or end.start() < start.start():
        print("Keine Marker gefunden - Block wird angehaengt. Bitte einmal an die "
              "richtige Stelle verschieben; kuenftige Laeufe finden ihn dort wieder.")
        return text.rstrip("\n") + "\n\n" + block + "\n"
    # Leerzeilen um den Block normalisieren, sonst waechst die Datei bei jedem Lauf
    # um eine Zeile und jeder Lauf erzeugt einen Diff, der nichts aussagt.
    return (text[:start.start()].rstrip("\n") + "\n\n"
            + block.strip() + "\n\n"
            + text[end.end():].lstrip("\n"))


def check(text):
    """Gleicht die Prosa gegen den Server ab. Der erzeugte Block kann nicht abweichen."""
    answer = call("POST", "/api/bugtracker/v1/roadmap.php",
                  {"project_slug": PROJECT, "content": text},
                  query={"action": "check"})
    if answer.get("status") == "error":
        sys.exit("Abgleich fehlgeschlagen: " + json.dumps(answer.get("error"))[:300])

    findings = answer.get("findings") or answer.get("befunde") or []
    errors = 0
    for f in findings:
        grad = f.get("severity") or f.get("grad") or "warning"
        code = f.get("code") or f.get("befund") or "?"
        where = f.get("item_id") or f.get("heading") or ""
        print("  [%s] %s %s %s" % (grad, code, where, f.get("message", "")))
        if grad in ("error", "fehler"):
            errors += 1
    if not findings:
        print("  Abgleich ohne Befund.")
    return errors


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--check", action="store_true",
                    help="nur abgleichen, den Block nicht neu holen")
    args = ap.parse_args()

    text = read_file()
    if not args.check:
        new = refresh(text)
        if new != text:
            with open(ROADMAP, "w", encoding="utf-8", newline="\n") as fh:
                fh.write(new)
            print("docs/ROADMAP.md: Block aufgefrischt.")
        else:
            print("docs/ROADMAP.md: Block war bereits auf Stand.")
        text = new

    print("Abgleich:")
    sys.exit(1 if check(text) else 0)


if __name__ == "__main__":
    main()