Orchestrator/bahn/wissensdatenbank/scripts/issue_to_source.py

"""Wandelt ein ausgefuelltes 'Neues Wissen'-Issue in einen tools.yaml-Eintrag.

Kern der Automatisierung (Idee): Issue -> yaml-Eintrag -> MR (mit Vorschau).

Aufruf:
    # Issue-Text aus Datei oder STDIN:
    python scripts/issue_to_source.py < issue.md
    glab issue view 42 | python scripts/issue_to_source.py
    # direkt an config/tools.yaml anhaengen:
    python scripts/issue_to_source.py --append < issue.md

Erwartet die Abschnitte der Vorlage (.gitlab/issue_templates/Neues_Wissen.md):
  Tool/Domaene, Link(s), Verarbeitung (Strategie), Sichtbarkeit (Scope),
  Ansprechpartner, Optionen, Hinweise.
"""
from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from src.strategy_detect import detect_strategy, sniff_strategy  # noqa: E402

FIELD_MAP = {
    "tool": ["welches tool", "tool", "domaene", "domäne"],
    "sources": ["quellen", "quelle", "link"],
    "owners": ["verantwortlich", "owners"],
    "contact": ["kontakt", "contact"],
    "options": ["optionen"],
}


def _slug(s: str) -> str:
    s = re.sub(r"[^a-z0-9]+", "-", s.lower())
    return s.strip("-")[:40] or "neues-wissen"


def parse_issue(text: str) -> dict:
    """Zerlegt den Issue-Text in Felder (anhand der ###-Ueberschriften).

    Quellen-Zeilen haben das Format:  URL | Strategie | Scope
    (Strategie/Scope optional). Fuehrendes "- " (Markdown-Liste) wird entfernt.
    """
    sections: dict[str, str] = {}
    cur = None
    buf: list[str] = []
    for line in text.splitlines():
        m = re.match(r"^#{2,4}\s+(.*)", line.strip())
        if m:
            if cur:
                sections[cur] = "\n".join(buf).strip()
            cur = m.group(1).strip().lower()
            buf = []
        elif cur:
            if line.strip().startswith("<!--") or line.strip().startswith("/label"):
                continue
            buf.append(line)
    if cur:
        sections[cur] = "\n".join(buf).strip()

    def find(key: str) -> str:
        for head, val in sections.items():
            if any(alias in head for alias in FIELD_MAP[key]):
                return val.strip()
        return ""

    sources = []
    for raw in find("sources").splitlines():
        ln = raw.strip().lstrip("-").strip()
        if "http" not in ln:
            continue
        parts = [p.strip() for p in ln.split("|")]
        url = parts[0].strip()
        if not url.startswith("http"):
            continue
        strat = parts[1].strip() if len(parts) > 1 and parts[1].strip() else ""
        scope = parts[2].strip() if len(parts) > 2 and parts[2].strip() else ""
        sources.append({"url": url, "strategy": strat, "scope": scope})

    owners = [o.strip() for o in re.split(r"[,\n]", find("owners")) if o.strip()]
    contact = find("contact").splitlines()[0].strip() if find("contact") else ""
    return {
        "tool": find("tool").splitlines()[0].strip() if find("tool") else "",
        "sources": sources,
        "owners": owners,
        "contact": contact,
    }


def build_entry(data: dict, sniff: bool = False) -> str:
    from src.strategy_detect import KNOWN

    tool = data["tool"] or "neues-tool"
    sid = _slug(tool)
    owners = ", ".join(f'"{o}"' for o in data["owners"])
    contact = data.get("contact", "").strip()
    detect = sniff_strategy if sniff else detect_strategy
    lines = [
        f"  - id: {sid}",
        f'    name: "{tool}"',
        f"    domain: {sid}",
        "    scope: mixed",
        f"    owners: [{owners}]",
    ]
    if contact:
        lines.append(f'    contact: "{contact}"')
    lines.append("    sources:")
    unknown = []
    for src in data["sources"] or [{"url": "<URL_FEHLT>", "strategy": "", "scope": ""}]:
        url = src["url"]
        strat = src["strategy"] if src["strategy"] in KNOWN else (detect(url) or "UNBEKANNT")
        scope = src["scope"] or "intern"
        if strat == "UNBEKANNT":
            unknown.append(url)
            lines.append(f"      # !! Keine passende Strategie erkannt -> neue Strategie pruefen: {url}")
        lines += [
            f'      - url: "{url}"',
            f"        strategy: {strat}",
            f'        scope: "{scope}"',
            f'        tags: ["{sid}"]',
        ]
    out = "\n".join(lines) + "\n"
    if unknown:
        out += "\n# WARNUNG: " + str(len(unknown)) + " URL(s) ohne passende Strategie -> ggf. neue Strategie bauen.\n"
    return out


def main() -> None:
    parser = argparse.ArgumentParser(description="Issue -> tools.yaml-Eintrag")
    parser.add_argument("--append", action="store_true", help="an config/tools.yaml anhaengen")
    parser.add_argument("--sniff", action="store_true", help="Strategie per Netzwerk verfeinern (HEAD/GET)")
    parser.add_argument("--config", default="config/tools.yaml")
    args = parser.parse_args()

    text = sys.stdin.read()
    entry = build_entry(parse_issue(text), sniff=args.sniff)

    if args.append:
        with open(args.config, "a", encoding="utf-8") as f:
            f.write("\n" + entry)
        print(f"angehaengt an {args.config}:\n\n{entry}")
    else:
        print(entry)


if __name__ == "__main__":
    main()