#!/usr/bin/env python3 """Prüft den Eingangsordner input/ für den Skill sn-einlesen und setzt Einlese-Vermerke. Der Ordner input/ ist nicht versioniert. Dort legen Fachleute Prozessbeschreibungen als .md ab (Rohtext, Stichpunkte oder schon eine Arbeitsschritt-.md). Dieses Skript erledigt nur das Mechanische; die fachliche Einschätzung und die Rückfrage an den Menschen macht der Skill. pruefen (Standard) Jede Datei einordnen: abgelehnt kein .md, keine UTF-8-Datei oder Unterordner – wird nicht bearbeitet neu .md ohne Einlese-Vermerk geändert .md mit Vermerk, aber der Inhalt hat sich seitdem geändert eingelesen .md mit Vermerk, Inhalt unverändert – wird übersprungen Für neue und geänderte Dateien: vorgeschlagener Prozessname, ob die Datei schon eine gültige Arbeitsschritt-.md ist, und ähnliche vorhandene Prozesse (Prozessbibliothek, entwuerfe/). vermerk Trägt nach erfolgreichem Einlesen oben in die .md einen Vermerk mit Zeitstempel ein. Aufrufe (aus dem Workspace-Root): python scripts/einlesen.py # Übersicht python scripts/einlesen.py --json # dasselbe maschinenlesbar (für den Agenten) python scripts/einlesen.py vermerk input/antrag.md --ergebnis neu --ziel entwuerfe/antrag-pruefen.md Der Vermerk ist ein HTML-Kommentar in der ersten Zeile, z. B. Er ist in der Markdown-Ansicht unsichtbar und stört md_to_bpmn.py nicht (steht vor dem ersten Schritt). „inhalt“ ist der Anfang der SHA-256 des Dateiinhalts ohne Vermerke; so fällt eine spätere Änderung auf. Exit-Code: 0 = in Ordnung, 1 = Fehler beim Vermerk (Datei fehlt, kein .md, …), 2 = Aufruffehler. """ from __future__ import annotations import argparse import difflib import hashlib import json import re import sys import unicodedata from datetime import datetime from pathlib import Path from setup_workspace import repos_root ROOT = Path(__file__).resolve().parents[1] EINGANG = ROOT / "input" BIBLIOTHEK = repos_root() / "Prozessbibliothek" # eigenes Repository (ADR 0004), wie in pruefe_prozesse.py ENTWUERFE = ROOT / "entwuerfe" SKILL = ROOT / ".agents" / "skills" / "sn-bpmn" sys.path.insert(0, str(SKILL)) import md_to_bpmn # noqa: E402 (liegt im Skill-Ordner, nicht als Paket installiert) ERGEBNISSE = ("neu", "aktualisiert") VERMERK = re.compile( r"^$" ) # Von Windows oder Werkzeugen angelegte Dateien, die niemand bewusst in input/ legt STILL_IGNORIERT = {"desktop.ini", "thumbs.db"} FUELLWOERTER = { "aber", "alle", "auch", "beim", "bzw", "dann", "dass", "diese", "dieser", "durch", "eine", "einem", "einen", "einer", "eines", "falls", "für", "hier", "immer", "jede", "jeder", "kann", "keine", "mehr", "muss", "nach", "noch", "oder", "ohne", "sich", "sind", "soll", "sowie", "über", "unter", "wenn", "werden", "wird", "zwei", # Wörter der Arbeitsschritt-Syntax sagen nichts über den Inhalt "prozess", "start", "ende", "task", "aufgabe", "gateway", "extern", "hinweis", "sendeaufgabe", "empfangsaufgabe", "zwischenereignis", "nachricht", "beispiel", } SCHWELLE = 0.25 # ab dieser Ähnlichkeit wird ein vorhandener Prozess als Kandidat genannt MAX_KANDIDATEN = 3 # ------------------------------------------------------------------ Vermerk --- def ohne_vermerke(text: str) -> str: return "\n".join(z for z in text.splitlines() if not VERMERK.match(z.strip())) def inhalt_hash(text: str) -> str: """Kurz-Hash des Inhalts ohne Vermerke und ohne Unterschiede bei Zeilenenden/Leerzeilen am Ende.""" kern = ohne_vermerke(text.replace("\r\n", "\n")).strip() return hashlib.sha256(kern.encode("utf-8")).hexdigest()[:12] def letzter_vermerk(text: str) -> dict | None: """Der oberste (= neueste) Vermerk der Datei oder None.""" for zeile in text.splitlines(): treffer = VERMERK.match(zeile.strip()) if treffer: return treffer.groupdict() return None def setze_vermerk(pfad: Path, ergebnis: str, ziel: str, jetzt: datetime | None = None) -> str: """Schreibt den Vermerk als erste Zeile; ältere Vermerke bleiben darunter stehen. Liefert die Zeile.""" if ergebnis not in ERGEBNISSE: raise ValueError(f"Ergebnis '{ergebnis}' unbekannt, erlaubt: {', '.join(ERGEBNISSE)}") # Als Bytes lesen: read_text() würde CRLF stillschweigend zu LF machen und die Datei so umschreiben. text = pfad.read_bytes().decode("utf-8-sig") zeit =(jetzt or datetime.now().astimezone()).strftime("%Y-%m-%d %H:%M:%S %z") zeile = f"" zeilenende = "\r\n" if "\r\n" in text else "\n" pfad.write_text(zeile + zeilenende + text, encoding="utf-8", newline="") return zeile # ------------------------------------------------------------ Einordnung --- def slug(text: str) -> str: """Ordnername nach ADR 0004: Kleinbuchstaben, Ziffern, Bindestriche; Umlaute ausgeschrieben.""" text = text.lower() for alt, neu in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")): text = text.replace(alt, neu) text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode("ascii") return re.sub(r"[^a-z0-9]+", "-", text).strip("-") def titel_von(text: str, ersatz: str) -> str: """Prozessname aus 'Prozess:'-Zeile, sonst erster Überschrift, sonst dem Dateinamen.""" for zeile in ohne_vermerke(text).splitlines(): treffer = re.match(r"^\s*(?:prozess|prozessname)\s*:\s*(.+)$", zeile, re.IGNORECASE) if treffer: return treffer.group(1).strip() for zeile in ohne_vermerke(text).splitlines(): treffer = re.match(r"^\s*#{1,3}\s+(.+)$", zeile) if treffer: return treffer.group(1).strip() return ersatz def woerter(text: str) -> set[str]: return {w for w in re.findall(r"[a-zäöüß]{4,}", text.lower()) if w not in FUELLWOERTER} def ist_arbeitsschritt_md(text: str) -> bool: """True, wenn md_to_bpmn.py die Datei ohne Fehler liest und mindestens einen Schritt findet.""" try: _, _, order, _ = md_to_bpmn.parse_markdown(text) except ValueError: return False return bool(order) def bestand() -> list[dict]: """Vorhandene Prozesse (Prozessbibliothek) und lokale Entwürfe (entwuerfe/) mit Titel und Wörtern.""" eintraege = [] if BIBLIOTHEK.is_dir(): for ordner in sorted(p for p in BIBLIOTHEK.iterdir() if p.is_dir() and not p.name.startswith(".")): md = ordner / f"{ordner.name}.md" text = md.read_text(encoding="utf-8") if md.is_file() else "" readme = ordner / "README.md" titel = titel_von(readme.read_text(encoding="utf-8"), ordner.name) if readme.is_file() else ordner.name eintraege.append({"art": "Prozess", "name": ordner.name, "titel": titel_von(text, titel), "pfad": f"{BIBLIOTHEK.name}/{ordner.name}/", "woerter": woerter(text)}) if ENTWUERFE.is_dir(): for md in sorted(ENTWUERFE.glob("*.md")): if re.match(r"^\d{8}-\d{6}-(issue|kommentar)-", md.name): continue # Gitea-Entwürfe aus scripts/gitea.py text = md.read_text(encoding="utf-8", errors="replace") eintraege.append({"art": "Entwurf (lokal)", "name": md.stem, "titel": titel_von(text, md.stem), "pfad": md.relative_to(ROOT).as_posix(), "woerter": woerter(text)}) return eintraege def aehnlichkeit(titel: str, text: str, kandidat: dict) -> float: """0 … 1: halb Namensähnlichkeit, halb Wortüberschneidung. Gleicher Ordnername ergibt 1.""" eigener = slug(titel) if eigener and eigener in (kandidat["name"], slug(kandidat["titel"])): return 1.0 name = max(difflib.SequenceMatcher(None, eigener, slug(t)).ratio() for t in (kandidat["name"], kandidat["titel"])) a, b = woerter(text), kandidat["woerter"] ueberschneidung = len(a & b) / len(a | b) if a and b else 0.0 return round(0.5 * name + 0.5 * ueberschneidung, 2) def ordne_ein(pfad: Path, vorhanden: list[dict]) -> dict: eintrag = {"datei": pfad.relative_to(ROOT).as_posix()} if pfad.is_dir(): return eintrag | {"status": "abgelehnt", "grund": "Unterordner werden nicht gelesen"} if pfad.suffix.lower() != ".md": return eintrag | {"status": "abgelehnt", "grund": f"Dateityp {pfad.suffix or '(ohne Endung)'} – vorerst nur .md"} try: text = pfad.read_text(encoding="utf-8-sig") except UnicodeDecodeError: return eintrag | {"status": "abgelehnt", "grund": "keine UTF-8-Datei – im Editor als UTF-8 speichern"} if not ohne_vermerke(text).strip(): return eintrag | {"status": "abgelehnt", "grund": "Datei ist leer"} vermerk = letzter_vermerk(text) if vermerk and vermerk["hash"] == inhalt_hash(text): return eintrag | {"status": "eingelesen", "vermerk": vermerk} titel = titel_von(text, pfad.stem) kandidaten = sorted( ({"art": k["art"], "name": k["name"], "titel": k["titel"], "pfad": k["pfad"], "aehnlichkeit": aehnlichkeit(titel, text, k)} for k in vorhanden), key=lambda k: -k["aehnlichkeit"], ) return eintrag | { "status": "geändert" if vermerk else "neu", "vermerk": vermerk, "titel": titel, "vorschlag_name": slug(titel) or slug(pfad.stem), "arbeitsschritt_md": ist_arbeitsschritt_md(ohne_vermerke(text)), "kandidaten": [k for k in kandidaten if k["aehnlichkeit"] >= SCHWELLE][:MAX_KANDIDATEN], } def pruefe_eingang(eingang: Path | None = None) -> dict: """Ordnet alle Einträge in input/ ein; legt den Ordner an, falls er fehlt.""" eingang = eingang or EINGANG angelegt = not eingang.is_dir() eingang.mkdir(exist_ok=True) vorhanden = bestand() dateien = [ordne_ein(p, vorhanden) for p in sorted(eingang.iterdir()) if not p.name.startswith(".") and p.name.lower() not in STILL_IGNORIERT] return { "eingang": eingang.relative_to(ROOT).as_posix() + "/", "angelegt": angelegt, "dateien": dateien, "bestand": [{k: v for k, v in e.items() if k != "woerter"} for e in vorhanden], } # ---------------------------------------------------------------- Ausgabe --- def als_text(bericht: dict) -> str: zeilen = [] if bericht["angelegt"]: zeilen.append(f"Ordner {bericht['eingang']} war nicht vorhanden und wurde angelegt.") if not bericht["dateien"]: zeilen.append(f"{bericht['eingang']} ist leer – nichts einzulesen.") for d in bericht["dateien"]: status = d["status"] if status == "abgelehnt": zeilen.append(f"{'[abgelehnt]':<13}{d['datei']}: {d['grund']}") elif status == "eingelesen": v = d["vermerk"] zeilen.append(f"{'[eingelesen]':<13}{d['datei']}: am {v['zeit']} ({v['ergebnis']} -> {v['ziel']})") else: art = "Arbeitsschritt-.md" if d["arbeitsschritt_md"] else "Rohtext" zeilen.append(f"{'[' + status + ']':<13}{d['datei']}: „{d['titel']}“ ({art}), " f"Vorschlag: {d['vorschlag_name']}") for k in d["kandidaten"]: zeilen.append(f" ähnlich ({k['aehnlichkeit']:.2f}): {k['art']} {k['pfad']} – „{k['titel']}“") if not d["kandidaten"]: zeilen.append(" kein ähnlicher Prozess im Bestand") zaehler = {s: sum(1 for d in bericht["dateien"] if d["status"] == s) for s in ("neu", "geändert", "eingelesen", "abgelehnt")} zeilen.append("\n" + ", ".join(f"{n} {s}" for s, n in zaehler.items()) + f" | Bestand: {len(bericht['bestand'])} Prozess(e)/Entwurf(e)") return "\n".join(zeilen) def main() -> int: for strom in (sys.stdout, sys.stderr): if hasattr(strom, "reconfigure"): strom.reconfigure(encoding="utf-8", errors="replace") ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--json", action="store_true", help="Übersicht als JSON") unter = ap.add_subparsers(dest="befehl") v = unter.add_parser("vermerk", help="Einlese-Vermerk in eine .md schreiben") v.add_argument("datei", help="Datei in input/") v.add_argument("--ergebnis", required=True, choices=ERGEBNISSE) v.add_argument("--ziel", required=True, help="wo das Ergebnis liegt, z. B. entwuerfe/.md oder Prozessbibliothek//") args = ap.parse_args() if args.befehl == "vermerk": pfad = Path(args.datei) if not pfad.is_file() or pfad.suffix.lower() != ".md": print(f"Keine .md-Datei: {pfad}") return 1 print(setze_vermerk(pfad, args.ergebnis, args.ziel)) return 0 bericht = pruefe_eingang() print(json.dumps(bericht, ensure_ascii=False, indent=2) if args.json else als_text(bericht)) return 0 if __name__ == "__main__": raise SystemExit(main())