narty-2027/gen_viz.py

#!/usr/bin/env python3
"""Generuje viz.html (graph viewer) dla tego bundle'a.

Viewer z GoogleCloudPlatform/knowledge-catalog jest samowystarczalny: cały
bundle jest wpiekany do pliku HTML jako `window.BUNDLE` (nodes/edges/bodies).
Nie czyta plików .md w runtime, więc nie potrzebuje manifestu ani serwera —
źródłem danych jest ten skrypt, uruchamiany po każdej zmianie bundle'a.

Logika viewera (templates/viz.html, static/viz.js, static/viz.css) jest
wstawiana dosłownie, bez zmian. Skrypt koryguje wyłącznie warstwę zbierania
danych, w dwóch punktach, w których upstreamowy generator rozmija się ze
specyfikacją OKF:

1. `_extract_links` pomija linki bundle-relative zaczynające się od `/`
   (generator.py:74), czyli dokładnie formę zalecaną przez §5.1. Bundle
   trzymający się zalecenia dostaje graf bez krawędzi. Sam viewer te linki
   rozumie — viz.js:297 rozwiązuje `/...md` na węzeł — więc dokładamy je do
   ekstrakcji krawędzi.
2. `_walk_concepts` pomija tylko `index.md`, przez co zarezerwowany `log.md`
   ląduje w grafie jako koncept typu "Unknown". Wg §3.1 zarezerwowane nazwy
   nie są konceptami.

Dodatkowo skrypt dokłada do grafu **linki zewnętrzne** jako węzły typu
"external" (kolor teal, odróżnialny od szarych konceptów) z krawędzią
koncept → link. Pełny URL trafia do pola `resource` węzła, bo istniejący
panel szczegółów renderuje `resource` jako klikalny odnośnik z
`target="_blank"` (viz.js:171-183) — to daje "klik otwiera URL w nowej
karcie" bez żadnej zmiany w viewerze.

Skąd brane są adresy — cztery niezależne źródła, każde da się wyłączyć
osobno stałą `SOURCES`:

    md-link      [tekst](https://...) w body konceptu
    bare-url     goły https://... w body konceptu
    bare-domain  goła domena w body (np. "alpincard.at"), normalizowana do https://
    frontmatter  pola frontmattera niosące adresy: FRONTMATTER_URL_FIELDS
                 oraz FRONTMATTER_DOMAIN_FIELDS

UWAGA: w obecnym bundle'u body konceptów nie zawiera ani jednego adresu
http(s):// — wszystkie adresy obiektów siedzą we frontmatterze (`www:`,
`gdzie_szukac:`), a w body są tylko gołe domeny ("alpincard.at",
"alpenparks.at"). Same źródła `md-link` i `bare-url` dają dziś 0 węzłów;
są zaimplementowane, bo obsługują przyszłe treści. Źródła `bare-domain`
i `frontmatter` są tym, co faktycznie wyciąga merytoryczne linki bundle'a
(obiekty, skipassy). Żeby zawęzić do samego body — usuń "frontmatter"
z SOURCES.

Pliki zarezerwowane (index.md, log.md) nie są źródłem linków — nie są
konceptami (§3.1) i nie trafiają do grafu.

Wymaga: PyYAML oraz repozytorium knowledge-catalog (domyślnie
/tmp/knowledge-catalog — nadpisz zmienną KNOWLEDGE_CATALOG).

Uruchomienie: python3 gen_viz.py
"""

from __future__ import annotations

import os
import re
import sys
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

import yaml

BUNDLE_ROOT = Path(__file__).resolve().parent
BUNDLE_NAME = "Ski 2027 — Skicircus Saalbach"
CATALOG = Path(os.environ.get("KNOWLEDGE_CATALOG", "/tmp/knowledge-catalog"))

sys.path.insert(0, str(CATALOG / "okf" / "src"))

try:
    from reference_agent.viewer import generator as G
except ModuleNotFoundError as exc:  # pragma: no cover
    sys.exit(
        f"Nie znaleziono viewera ({exc}).\n"
        f"Sklonuj: git clone https://github.com/GoogleCloudPlatform/"
        f"knowledge-catalog {CATALOG}"
    )

RESERVED = {"index.md", "log.md"}

# --- konfiguracja linków zewnętrznych ---------------------------------

SOURCES = ("md-link", "bare-url", "bare-domain", "frontmatter")

EXTERNAL_TYPE = "external"
EXTERNAL_COLOR = "#0d9488"  # teal — koncepty są szare (#94a3b8)
EXTERNAL_SIZE = 22

FRONTMATTER_URL_FIELDS = ("www",)          # wartości to pełne URL-e
FRONTMATTER_DOMAIN_FIELDS = ("gdzie_szukac",)  # wartości to gołe domeny

# TLD dopuszczone przy rozpoznawaniu gołych domen — świadomie wąska lista,
# żeby nie łapać "viz.html", "index.md" itp.
_TLDS = "at|com|pl|de|eu|net|org|io|info"

_MD_LINK_RE = re.compile(r"\[[^\]]*\]\((https?://[^)\s]+)\)")
_BARE_URL_RE = re.compile(r"https?://[^\s)>\]},;\"']+")
_BARE_DOMAIN_RE = re.compile(
    rf"(?<![\w./@-])((?:[a-z0-9][a-z0-9-]*\.)+(?:{_TLDS}))(?![\w-])"
)

# Domeny techniczne — nie są wiedzą o wyjeździe.
def _technical_reason(url: str) -> str | None:
    parts = urlsplit(url)
    host = parts.netloc.lower().split("@")[-1].split(":")[0]
    path = parts.path.lower()
    if host == "localhost" or host.endswith(".localhost"):
        return "localhost"
    if host.startswith("192.168."):
        return "192.168.*"
    if host.startswith("100.") and host.replace(".", "").isdigit():
        return "100.*"
    if host in {"github.com", "www.github.com"} and path.startswith(
        "/googlecloudplatform"
    ):
        return "github.com/GoogleCloudPlatform"
    return None


def normalize_url(url: str) -> str:
    """Bez fragmentu #, bez trailing slash, schemat i host małymi literami."""
    if "://" not in url:
        url = "https://" + url
    parts = urlsplit(url)
    return urlunsplit(
        (
            parts.scheme.lower(),
            parts.netloc.lower(),
            parts.path.rstrip("/"),
            parts.query,
            "",  # fragment odcięty
        )
    )


def label_for(url: str) -> str:
    """Hostname + skrócona ścieżka."""
    parts = urlsplit(url)
    host = parts.netloc
    path = parts.path
    if not path or path == "/":
        return host
    segment = path.rsplit("/", 1)[-1] or path
    if len(segment) > 24:
        segment = segment[:23] + "…"
    return f"{host}/…/{segment}" if path.count("/") > 1 else f"{host}/{segment}"


def node_id_for(url: str) -> str:
    parts = urlsplit(url)
    return "ext:" + (parts.netloc + parts.path).rstrip("/")


def _frontmatter(path: Path) -> dict:
    text = path.read_text(encoding="utf-8")
    if not text.startswith("---\n"):
        return {}
    end = text.find("\n---", 3)
    if end == -1:
        return {}
    return yaml.safe_load(text[4:end]) or {}


def _raw_urls(body: str, frontmatter: dict) -> list[tuple[str, str]]:
    """Zwraca [(surowy_url, nazwa_źródła)] dla jednego konceptu."""
    found: list[tuple[str, str]] = []
    remaining = body

    if "md-link" in SOURCES:
        for m in _MD_LINK_RE.finditer(body):
            found.append((m.group(1), "md-link"))
    # Adresy złapane jako pełne URL-e wycinamy z tekstu, żeby nie policzyć
    # ich powtórnie jako gołych domen.
    remaining = _MD_LINK_RE.sub(" ", remaining)

    if "bare-url" in SOURCES:
        for m in _BARE_URL_RE.finditer(remaining):
            found.append((m.group(0).rstrip(".,;:"), "bare-url"))
    remaining = _BARE_URL_RE.sub(" ", remaining)

    if "bare-domain" in SOURCES:
        for m in _BARE_DOMAIN_RE.finditer(remaining):
            found.append((m.group(1), "bare-domain"))

    if "frontmatter" in SOURCES:
        for field in FRONTMATTER_URL_FIELDS + FRONTMATTER_DOMAIN_FIELDS:
            value = frontmatter.get(field)
            if not value:
                continue
            values = value if isinstance(value, list) else [value]
            for item in values:
                if isinstance(item, str) and item.strip():
                    found.append((item.strip(), "frontmatter"))

    return found


# Raport dla wypisu na końcu.
REPORT: dict[str, list] = {"kept": [], "filtered": []}


def _add_external(graph: dict, concepts: list) -> None:
    kept: dict[str, dict] = {}   # znormalizowany URL -> {origins, concepts}
    filtered: list[tuple[str, str, str]] = []

    for concept in concepts:
        path = BUNDLE_ROOT / f"{concept.id}.md"
        frontmatter = _frontmatter(path) if path.is_file() else {}
        for raw, origin in _raw_urls(concept.body or "", frontmatter):
            url = normalize_url(raw)
            reason = _technical_reason(url)
            if reason:
                filtered.append((url, reason, concept.id))
                continue
            entry = kept.setdefault(url, {"origins": set(), "concepts": []})
            entry["origins"].add(origin)
            if concept.id not in entry["concepts"]:
                entry["concepts"].append(concept.id)

    concept_ids = {c.id for c in concepts}
    seen_edges = set()
    for url in sorted(kept):
        entry = kept[url]
        node_id = node_id_for(url)
        graph["nodes"].append(
            {
                "data": {
                    "id": node_id,
                    "label": label_for(url),
                    "type": EXTERNAL_TYPE,
                    "description": url,
                    "resource": url,  # panel renderuje to jako link _blank
                    "tags": [],
                    "status": EXTERNAL_TYPE,
                    "generated": {},
                    "verified": [],
                    "stale_after": "",
                    "sources": [],
                    "trust_tier": "unverified",
                    "stale": False,
                    "color": EXTERNAL_COLOR,
                    "size": EXTERNAL_SIZE,
                }
            }
        )
        for source_id in entry["concepts"]:
            if source_id not in concept_ids:
                continue
            key = (source_id, node_id)
            if key in seen_edges:
                continue
            seen_edges.add(key)
            graph["edges"].append(
                {
                    "data": {
                        "id": f"{source_id}__{node_id}",
                        "source": source_id,
                        "target": node_id,
                    }
                }
            )
        REPORT["kept"].append(
            (url, sorted(entry["origins"]), entry["concepts"])
        )

    REPORT["filtered"] = filtered
    if kept:
        graph["types"] = sorted(set(graph["types"]) | {EXTERNAL_TYPE})
        graph["palette"] = {**graph.get("palette", {}), EXTERNAL_TYPE: EXTERNAL_COLOR}


# --- łatki na generator upstreamowy -----------------------------------

_upstream_extract_links = G._extract_links
_upstream_walk_concepts = G._walk_concepts
_upstream_build_graph = G._build_graph


def _extract_links(body: str, doc_dir: Path, bundle_root: Path) -> list[str]:
    """Upstream + linki bundle-relative (§5.1), które upstream odrzuca."""
    out = _upstream_extract_links(body, doc_dir, bundle_root)
    seen = set(out)
    for match in G._LINK_RE.finditer(body):
        target = match.group(1)
        if "://" in target or not target.startswith("/"):
            continue
        rel = target[1:]
        if rel.endswith(".md"):
            rel = rel[:-3]
        if rel and rel not in seen:
            seen.add(rel)
            out.append(rel)
    return out


def _walk_concepts(bundle_root: Path):
    """Upstream, ale bez plików zarezerwowanych (§3.1)."""
    return [
        c
        for c in _upstream_walk_concepts(bundle_root)
        if f"{c.id}.md".rsplit("/", 1)[-1] not in RESERVED
    ]


def _build_graph(concepts: list) -> dict:
    """Upstream + węzły/krawędzie linków zewnętrznych."""
    graph = _upstream_build_graph(concepts)
    _add_external(graph, concepts)
    return graph


G._extract_links = _extract_links
G._walk_concepts = _walk_concepts
G._build_graph = _build_graph


def main() -> int:
    out_path = BUNDLE_ROOT / "viz.html"
    stats = G.generate_visualization(
        BUNDLE_ROOT, out_path, bundle_name=BUNDLE_NAME
    )
    external = len(REPORT["kept"])
    print(f"Zapisano {out_path}")
    print(
        f"  węzły: {stats['concepts']} konceptów + {external} external, "
        f"krawędzie: {stats['edges']}, rozmiar: {stats['bytes'] / 1024:.1f} KiB"
    )
    print()
    print(f"Linki zewnętrzne, które weszły do grafu ({external}):")
    for url, origins, sources in REPORT["kept"]:
        print(f"  + {url}")
        print(f"      źródło: {', '.join(origins)} | z: {', '.join(sources)}")
    print()
    print(f"Odfiltrowane jako techniczne ({len(REPORT['filtered'])}):")
    if REPORT["filtered"]:
        for url, reason, concept in REPORT["filtered"]:
            print(f"  - {url}  [{reason}]  z: {concept}")
    else:
        print("  (żadnych — bundle nie zawiera adresów z domen technicznych)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())