97-documentation/ns_report_publisher/workbench_dashboard.py

from __future__ import annotations

import html
import json
import re
from pathlib import Path
from typing import Any


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except (OSError, UnicodeError):
        return ""


def _read_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError):
        return {}
    return value if isinstance(value, dict) else {}


def _report_href(reports_root: Path, entry: dict[str, Any] | None, source_relative: str) -> str:
    if entry is None:
        return "#"
    report_root = str(entry["reportRoot"]).strip("/")
    candidate = source_relative[:-3] + ".html" if source_relative.endswith(".md") else source_relative
    if (reports_root / report_root / candidate).is_file():
        return f"{report_root}/{candidate}"
    return f"{report_root}/{entry['entryPage']}"


def _markdown_table_rows(text: str) -> list[list[str]]:
    rows: list[list[str]] = []
    lines = text.splitlines()
    for index, line in enumerate(lines):
        if not line.lstrip().startswith("|"):
            continue
        cells = [cell.strip() for cell in line.strip().strip("|").split("|")]
        if not cells:
            continue
        if all(re.fullmatch(r":?-{2,}:?", cell or "-") for cell in cells):
            continue
        if index + 1 < len(lines) and re.match(r"^\s*\|(?:\s*:?-{2,}:?\s*\|)+\s*$", lines[index + 1]):
            continue
        rows.append(cells)
    return rows


def _non_placeholder_rows(text: str) -> list[list[str]]:
    result: list[list[str]] = []
    header_starts = (
        "gap id ", "observation id ", "candidate id ", "bb gegenstand ",
        "bb status ", "arbeitsgegenstand lifecycle-status ",
    )
    for row in _markdown_table_rows(text):
        joined = " ".join(row).strip().lower()
        if not joined or joined.startswith(header_starts):
            continue
        if row[0] in {"–", "-"} or "noch keine" in joined:
            continue
        result.append(row)
    return result


def _extract_bullets_after_heading(text: str, heading: str, limit: int = 8) -> list[str]:
    active = False
    result: list[str] = []
    for line in text.splitlines():
        if line.startswith("## "):
            active = line[3:].strip().casefold() == heading.casefold()
            continue
        if active and line.startswith("- "):
            result.append(line[2:].strip())
            if len(result) >= limit:
                break
    return result


def _first_bold_after_heading(text: str, heading: str) -> str:
    active = False
    for line in text.splitlines():
        if line.startswith("## "):
            active = line[3:].strip().casefold() == heading.casefold()
            continue
        if active:
            match = re.search(r"\*\*(.+?)\*\*", line)
            if match:
                return match.group(1).strip()
    return ""


def _clean_context(line: str) -> str:
    value = re.sub(r"\s+", " ", line.strip())
    return value[:280] + ("…" if len(value) > 280 else "")


def _governance_signals(reports_root: Path, solution: dict[str, Any], source_root: Path) -> dict[str, Any]:
    governance = source_root / "90-governance"
    result: dict[str, Any] = {
        "documents": 0, "todo": 0, "review": 0, "openQuestions": 0,
        "todos": [], "reviews": [], "questions": [],
    }
    if not governance.is_dir():
        return result
    # Only explicit, structured, active work-item markers are collected.
    # Plain mentions such as `[[Review]]` in guidelines, examples or prose are
    # documentation about the process and must not become workbench issues.
    marker_patterns = {
        "todos": re.compile(
            r"\[\[\s*Todo\s+(?=[^]\n]*\bid\s*=)(?=[^]\n]*\bstatus\s*=\s*[\"']?open[\"']?)[^]\n]*\]\]",
            re.IGNORECASE,
        ),
        "reviews": re.compile(
            r"\[\[\s*Review\s+(?=[^]\n]*\bid\s*=)(?=[^]\n]*\bstatus\s*=\s*[\"']?open[\"']?)[^]\n]*\]\]",
            re.IGNORECASE,
        ),
        "questions": re.compile(
            r"\[\[\s*Question\s+(?=[^]\n]*\bid\s*=)(?=[^]\n]*\bstatus\s*=\s*[\"']?open[\"']?)[^]\n]*\]\]",
            re.IGNORECASE,
        ),
    }
    for path in sorted(governance.rglob("*.md")):
        result["documents"] += 1
        relative = path.relative_to(source_root).as_posix()
        href = _report_href(reports_root, solution, relative)
        for line_number, line in enumerate(_read_text(path).splitlines(), start=1):
            # Inline-code examples are ignored even when they contain a
            # syntactically structured marker.
            scan_line = re.sub(r"`[^`]*`", "", line)
            for collection, pattern in marker_patterns.items():
                for match in pattern.finditer(scan_line):
                    result[collection].append({
                        "source": relative, "line": line_number, "text": _clean_context(line),
                        "marker": match.group(0), "href": href,
                    })
    result["todo"] = len(result["todos"])
    result["review"] = len(result["reviews"])
    result["openQuestions"] = len(result["questions"])
    return result


def build_summary(reports_root: Path, workspace: dict[str, Any] | None, solution: dict[str, Any]) -> dict[str, Any]:
    result: dict[str, Any] = {
        "available": False,
        "currentState": [],
        "nextStep": "",
        "gaps": [],
        "observations": [],
        "decisions": [],
        "buildingBlocks": [],
        "links": {},
        "governance": {},
        "publicationFindings": [],
    }
    solution_source = Path(str(solution.get("sourceRoot", "")))
    result["governance"] = _governance_signals(reports_root, solution, solution_source)
    result["links"]["governance"] = _report_href(reports_root, solution, "90-governance/README.md")
    if workspace is None:
        return result

    source_root = Path(str(workspace.get("sourceRoot", "")))
    base = source_root / "working-documents" / "projects" / "architecture-realization"
    if not base.is_dir():
        return result

    files = {
        "next": base / "next-step.md",
        "gaps": base / "architecture-gap-register.md",
        "observations": base / "architecture-observation-register.md",
        "decisions": base / "architecture-decision-candidates.md",
        "bb": base / "bb-analysis-register.md",
        "dashboard": base / "ENGINEERING_START_HERE.md",
    }
    result["available"] = True
    next_text = _read_text(files["next"])
    result["currentState"] = _extract_bullets_after_heading(next_text, "Aktueller Arbeitsstand")
    result["nextStep"] = (
        _first_bold_after_heading(next_text, "Nächster empfohlener Schritt")
        or _first_bold_after_heading(next_text, "Nächster empfohlener Building Block")
    )
    result["gaps"] = _non_placeholder_rows(_read_text(files["gaps"]))
    result["observations"] = _non_placeholder_rows(_read_text(files["observations"]))
    result["decisions"] = _non_placeholder_rows(_read_text(files["decisions"]))
    result["buildingBlocks"] = _non_placeholder_rows(_read_text(files["bb"]))
    prefix = "working-documents/projects/architecture-realization/"
    for key, path in files.items():
        result["links"][key] = _report_href(reports_root, workspace, prefix + path.name)
    return result


def collect_published_findings(reports_root: Path, link_inventory: dict[str, Any]) -> list[dict[str, Any]]:
    """Derive one work list from published reports; no new source of truth."""
    registry = _read_json(reports_root / "repository-reports.json")
    result: list[dict[str, Any]] = []
    link_report = "link-validation/SE-0109-published-link-inventory.md"
    for finding in link_inventory.get("findings", []):
        if finding.get("severity") != "ERROR":
            continue
        source = str(finding.get("sourceReference", "Veröffentlichte Linkprüfung"))
        candidate = source.removeprefix("published:").split("#", 1)[0].lstrip("/")
        href = candidate if candidate and (reports_root / candidate).is_file() else link_report
        result.append({
            "category": "Link", "severity": "error", "repository": candidate.split("/", 1)[0] if "/" in candidate else "Veröffentlichung",
            "title": str(finding.get("errorClass", "LINK_FINDING")), "source": source,
            "target": str(finding.get("expectedTarget", finding.get("target", ""))), "href": href,
            "explanation": str(finding.get("correctionHint", "Aktuelles Linkziel prüfen.")),
            "decision": "Pfad, Ziel oder bewusst historischen Evidenzstatus prüfen.",
        })
    for entry in registry.get("repositories", []):
        report_root = str(entry.get("reportRoot", "")).strip("/")
        if not report_root:
            continue
        index_path = reports_root / report_root / "report-index.json"
        if not index_path.is_file():
            continue
        index = _read_json(index_path)
        entry_page = f"{report_root}/{entry.get('entryPage', 'index.html')}"
        repository = str(entry.get("displayName", entry.get("repositoryId", report_root)))
        for kind, label, severity, decision in (
            ("unresolvedArtifactReferences", "Semantische Referenz", "error", "Ziel-ID oder bewussten historischen Status festlegen."),
            ("ambiguousArtifactReferences", "Mehrdeutige Referenz", "error", "Eindeutige Zielidentität oder Repository-Qualifizierung festlegen."),
        ):
            for raw in index.get(kind, []):
                source = str(raw).split(" -> ", 1)[0]
                page = source[:-3] + ".html" if source.endswith(".md") else source
                href = f"{report_root}/{page}" if (reports_root / report_root / page).is_file() else entry_page
                result.append({
                    "category": "Referenz", "severity": severity, "repository": repository,
                    "title": label, "source": str(raw), "target": "", "href": href,
                    "explanation": "Die Referenz konnte beim Publish nicht eindeutig als Wissenselement materialisiert werden.",
                    "decision": decision,
                })
        graph = index.get("contextGraph", {})
        isolated = graph.get("isolatedDocumentIds", []) if isinstance(graph, dict) else []
        for identifier in isolated:
            result.append({
                "category": "Beziehung", "severity": "review", "repository": repository,
                "title": "Isoliertes Wissenselement", "source": str(identifier), "target": "", "href": f"{report_root}/context-relationship-gaps.html",
                "explanation": "Das Wissenselement hat keine explizite eingehende oder ausgehende Beziehung.",
                "decision": "Mindestens eine fachlich begründete Beziehung ergänzen oder den bewusst vorläufigen Zustand bestätigen.",
            })
        ledger = index.get("relationshipLedger", {})
        if isinstance(ledger, dict):
            for key in ledger.get("duplicateMarkdownKeys", []):
                result.append({"category": "Ledger", "severity": "error", "repository": repository,
                               "title": "Doppelte Markdown-Beziehung", "source": str(key), "target": "",
                               "href": f"{report_root}/relationship-ledger-findings.html",
                               "explanation": "Der gleiche kanonische Beziehungsschlüssel wurde mehrfach deklariert.",
                               "decision": "Eine führende Deklaration beibehalten und die übrigen entfernen."})
            for key in list(ledger.get("orphanLedgerKeys", [])) + list(ledger.get("duplicateLedgerKeys", [])):
                result.append({"category": "Ledger", "severity": "error", "repository": repository,
                               "title": "Ledger-Konflikt", "source": str(key), "target": "",
                               "href": f"{report_root}/relationship-ledger-findings.html",
                               "explanation": "Der Ledger widerspricht der führenden Markdown-Beziehung oder enthält zwei Wahrheiten.",
                               "decision": "Ledger-Eintrag gezielt mit der Markdown-Beziehung abgleichen."})
    return sorted(result, key=lambda item: (item["severity"] != "error", item["category"], item["repository"], item["source"]))


def _render_row_list(rows: list[list[str]], empty_text: str, max_rows: int = 6) -> str:
    if not rows:
        return f'<p class="empty-state">{html.escape(empty_text)}</p>'
    items: list[str] = []
    for row in rows[:max_rows]:
        title = row[0]
        detail = " · ".join(cell for cell in row[1:4] if cell and cell not in {"–", "-"})
        suffix = " — " + html.escape(detail) if detail else ""
        items.append(f'<li><strong>{html.escape(title)}</strong>{suffix}</li>')
    return '<ul class="work-list">' + "".join(items) + "</ul>"


def render_workbench(summary: dict[str, Any]) -> str:
    if not summary.get("available"):
        return '<p class="empty-state">Die Architecture Workbench ist im veröffentlichten Workspace noch nicht verfügbar.</p>'
    links = summary["links"]
    state_items = "".join(f'<li>{html.escape(item)}</li>' for item in summary["currentState"])
    state_items = state_items or '<li>Kein Arbeitsstand dokumentiert.</li>'
    next_step = html.escape(summary["nextStep"] or "Kein nächster Schritt dokumentiert")
    return f"""
<div class="workbench-grid">
  <article class="workbench-card workbench-primary">
    <h3>Aktueller Arbeitsstand</h3>
    <ul class="work-list">{state_items}</ul>
    <p><strong>Nächster Schwerpunkt:</strong> {next_step}</p>
    <a href="{html.escape(links['next'], quote=True)}">Vollständigen nächsten Schritt öffnen</a>
  </article>
  <article class="workbench-card">
    <h3>Aktive Gaps</h3>
    {_render_row_list(summary['gaps'], 'Keine aktiven Architecture Gaps registriert.')}
    <a href="{html.escape(links['gaps'], quote=True)}">Gap Register öffnen</a>
  </article>
  <article class="workbench-card">
    <h3>Beobachtungen</h3>
    {_render_row_list(summary['observations'], 'Keine aktiven Beobachtungen registriert.')}
    <a href="{html.escape(links['observations'], quote=True)}">Observation Register öffnen</a>
  </article>
  <article class="workbench-card">
    <h3>Entscheidungskandidaten</h3>
    {_render_row_list(summary['decisions'], 'Keine offenen Entscheidungskandidaten registriert.')}
    <a href="{html.escape(links['decisions'], quote=True)}">Decision Candidates öffnen</a>
  </article>
  <article class="workbench-card">
    <h3>Building-Block-Arbeit</h3>
    {_render_row_list(summary['buildingBlocks'], 'Noch keine Building-Block-Arbeit registriert.')}
    <a href="{html.escape(links['bb'], quote=True)}">BB Analysis Register öffnen</a>
  </article>
  <article class="workbench-card">
    <h3>Engineering-Einstieg</h3>
    <p>Methoden, Gates, Validierungsumfang und verbindlicher nächster Ablauf.</p>
    <a href="{html.escape(links['dashboard'], quote=True)}">ENGINEERING_START_HERE öffnen</a>
  </article>
</div>
"""


def render_governance(summary: dict[str, Any]) -> str:
    stats = summary.get("governance", {})
    href = summary.get("links", {}).get("governance", "#")
    return f"""
<div class="governance-summary">
  <div><strong>{int(stats.get('documents', 0))}</strong><span>Governance-Dokumente</span></div>
  <a class="governance-stat" href="workbench-issues.html#todos"><strong>{int(stats.get('todo', 0))}</strong><span>Todo-Marker öffnen</span></a>
  <a class="governance-stat" href="workbench-issues.html#reviews"><strong>{int(stats.get('review', 0))}</strong><span>Review-Marker öffnen</span></a>
  <a class="governance-stat" href="workbench-issues.html#questions"><strong>{int(stats.get('openQuestions', 0))}</strong><span>Offene Fragen öffnen</span></a>
</div>
<p><a href="workbench-issues.html">Gesamte Arbeitsliste öffnen</a> · <a href="{html.escape(href, quote=True)}">Governance-Dokumentation öffnen</a></p>
"""


def _issue_section(section_id: str, title: str, items: list[dict[str, Any]], empty: str) -> str:
    if not items:
        body = f'<p class="empty-state">{html.escape(empty)}</p>'
    else:
        rows = []
        for index, item in enumerate(items, start=1):
            source = html.escape(str(item.get("source", "")))
            line = int(item.get("line", 0))
            text = html.escape(str(item.get("text", "")))
            href = html.escape(str(item.get("href", "#")), quote=True)
            rows.append(
                f'<article class="issue-item" id="{section_id}-{index}">'
                f'<h3><a href="{href}">{source}</a></h3>'
                f'<p class="issue-meta">Zeile {line}</p><p>{text}</p>'
                f'<p><a href="{href}">Dokument öffnen</a></p></article>'
            )
        body = ''.join(rows)
    return f'<section class="issue-section" id="{section_id}"><h2>{html.escape(title)} ({len(items)})</h2>{body}</section>'


def render_issue_center(summary: dict[str, Any]) -> str:
    governance = summary.get("governance", {})
    findings = list(summary.get("publicationFindings", []))
    finding_rows = []
    for index, item in enumerate(findings, start=1):
        href = html.escape(str(item.get("href", "#")), quote=True)
        category = html.escape(str(item.get("category", "Befund")))
        repository = html.escape(str(item.get("repository", "")))
        finding_rows.append(
            f'<article class="issue-item issue-{html.escape(str(item.get("severity", "review")), quote=True)}" '
            f'data-issue-category="{category.casefold()}" data-issue-repository="{repository.casefold()}">'
            f'<h3>{html.escape(str(item.get("title", "Befund")))}</h3>'
            f'<p class="issue-meta">{category} · {repository}</p><p><code>{html.escape(str(item.get("source", "")))}</code></p>'
            f'<p>{html.escape(str(item.get("explanation", "")))}</p><p><strong>Nächste Entscheidung:</strong> {html.escape(str(item.get("decision", "")))}</p>'
            f'<p><a href="{href}">Betroffenes Dokument bzw. Befund öffnen</a></p></article>'
        )
    publication = ''.join(finding_rows) or '<p class="empty-state">Keine aktuellen Publikations- oder Beziehungsbefunde.</p>'
    return f"""<!doctype html>
<html lang="de"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
<title>Architecture Workbench – Arbeitslisten</title><link rel="stylesheet" href="platform.css"></head>
<body><main class="platform-shell">
<section class="hero"><p class="eyebrow">Architecture Workbench</p><h1>Arbeitslisten</h1>
<p class="hero-text">Explizite, strukturierte und aktive Todo-, Review- und Question-Marker aus <code>90-governance</code>. Die Markdown-Dokumente bleiben führend.</p>
<a class="vision-link" href="index.html">Zurück zur Startseite</a></section>
<section class="issue-section" id="current-findings"><h2>Aktuelle Problemstellungen ({len(findings)})</h2>
<p>Abgeleitet aus Linkprüfung, Referenzauflösung, Kontextgraph und Ledger. Filter: <button type="button" data-issue-filter="all">Alle</button> <button type="button" data-issue-filter="link">Links</button> <button type="button" data-issue-filter="referenz">Referenzen</button> <button type="button" data-issue-filter="beziehung">Beziehungen</button> <button type="button" data-issue-filter="ledger">Ledger</button></p>
<div data-publication-findings>{publication}</div></section>
{_issue_section('todos', 'Todos', list(governance.get('todos', [])), 'Keine Todo-Marker gefunden.')}
{_issue_section('reviews', 'Reviews', list(governance.get('reviews', [])), 'Keine Review-Marker gefunden.')}
{_issue_section('questions', 'Offene Fragen', list(governance.get('questions', [])), 'Keine Hinweise auf offene Fragen gefunden.')}
</main><script>
document.querySelectorAll('[data-issue-filter]').forEach(button => button.addEventListener('click', () => {{
 const selected=button.dataset.issueFilter; document.querySelectorAll('[data-publication-findings] .issue-item]').forEach(item => {{ item.hidden=selected !== 'all' && item.dataset.issueCategory !== selected; }});
}}));
</script></body></html>"""


def write_summary(path: Path, summary: dict[str, Any]) -> None:
    path.write_text(json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")


def refresh_issue_center(reports_root: Path, link_inventory: dict[str, Any]) -> int:
    """Refresh the derived issue page after the final published-link check."""
    summary_path = reports_root / "workbench-summary.json"
    summary = _read_json(summary_path) if summary_path.is_file() else {"governance": {}}
    findings = collect_published_findings(reports_root, link_inventory)
    summary["publicationFindings"] = findings
    write_summary(summary_path, summary)
    (reports_root / "workbench-issues.html").write_text(render_issue_center(summary), encoding="utf-8")
    return len(findings)