Dateien nach "app" hochladen

Anpassungen an der Ausgabe in Outline

app/sync.py

@@ -2,10 +2,14 @@
 ein Outline-Dokument pro Subnetz."""
 from __future__ import annotations
 
+import logging
+
 from .config import settings
 from .outline_client import OutlineClient
 from .phpipam_client import PhpIpamClient
 
+logger = logging.getLogger("phpipam_outline_sync.sync")
+
 
 def _subnet_title(subnet: dict) -> str:
     cidr = f"{subnet.get('subnet')}/{subnet.get('mask')}"
@@ -13,7 +17,32 @@ def _subnet_title(subnet: dict) -> str:
     return f"{cidr} – {desc}".rstrip(" –") if desc else cidr
 
 
-def _subnet_markdown(subnet: dict, addresses: list[dict], section_name: str | None) -> str:
+def _vlan_label(vlan: dict | None) -> str | None:
+    if not vlan or not vlan.get("number"):
+        return None
+    label = str(vlan["number"])
+    if vlan.get("name"):
+        label += f" ({vlan['name']})"
+    return label
+
+
+def _escape_md_cell(text: str | None) -> str:
+    """Macht einen Wert sicher für eine Markdown-Tabellenzelle (Pipe/Zeilenumbrüche)."""
+    if not text:
+        return ""
+    return str(text).replace("|", "\\|").replace("\n", " ").strip()
+
+
+def _escape_mermaid_label(text: str | None) -> str:
+    """Macht einen Wert sicher für ein Mermaid-Knoten-Label (keine Anführungszeichen/Zeilenumbrüche)."""
+    if not text:
+        return ""
+    return str(text).replace('"', "'").replace("\n", " ").strip()
+
+
+def _subnet_markdown(
+    subnet: dict, addresses: list[dict], section_name: str | None, vlan: dict | None
+) -> str:
     cidr = f"{subnet.get('subnet')}/{subnet.get('mask')}"
     lines: list[str] = []
     lines.append(f"# {cidr}")
@@ -21,8 +50,9 @@ def _subnet_markdown(subnet: dict, addresses: list[dict], section_name: str | No
     lines.append(f"- **Beschreibung:** {subnet.get('description') or '–'}")
     if section_name:
         lines.append(f"- **Section:** {section_name}")
-    if subnet.get("vlanId") and subnet.get("vlanId") != "0":
-        lines.append(f"- **VLAN-ID:** {subnet.get('vlanId')}")
+    vlan_label = _vlan_label(vlan)
+    if vlan_label:
+        lines.append(f"- **VLAN:** {vlan_label}")
     if subnet.get("location"):
         lines.append(f"- **Standort:** {subnet.get('location')}")
     lines.append("")
@@ -31,26 +61,96 @@ def _subnet_markdown(subnet: dict, addresses: list[dict], section_name: str | No
         lines.append("_Keine belegten Adressen in phpipam hinterlegt._")
         return "\n".join(lines)
 
-    lines.append("| IP | Hostname | Beschreibung | MAC | Owner |")
-    lines.append("|---|---|---|---|---|")
+    lines.append("| IP | Hostname | Beschreibung | MAC")
+    lines.append("|---|---|---|---|")
     for addr in sorted(addresses, key=lambda a: tuple(int(o) for o in a["ip"].split("."))):
         lines.append(
-            "| {ip} | {hostname} | {desc} | {mac} | {owner} |".format(
+            "| {ip} | {hostname} | {desc} | {mac} | ".format(
                 ip=addr.get("ip", ""),
-                hostname=addr.get("hostname") or "",
-                desc=(addr.get("description") or "").replace("\n", " "),
-                mac=addr.get("mac") or "",
-                owner=addr.get("owner") or "",
+                hostname=_escape_md_cell(addr.get("hostname")),
+                desc=_escape_md_cell(addr.get("description")),
+                mac=_escape_md_cell(addr.get("mac")),
             )
         )
     return "\n".join(lines)
 
 
+def _networks_diagram_mermaid(entries: list[dict]) -> str:
+    """Baut ein Mermaid-Flowchart, gruppiert nach phpipam-Section, mit
+    anklickbaren Knoten, die direkt zur jeweiligen Outline-Seite verlinken."""
+    by_section: dict[str, list[dict]] = {}
+    for entry in entries:
+        section = entry.get("section_name") or "Ohne Section"
+        by_section.setdefault(section, []).append(entry)
+
+    lines: list[str] = ["```mermaid", "graph TD"]
+    click_lines: list[str] = []
+    sec_counter = 0
+    node_counter = 0
+    for section, sec_entries in by_section.items():
+        sec_counter += 1
+        sec_id = f"sec{sec_counter}"
+        lines.append(f'  subgraph {sec_id}["{_escape_mermaid_label(section)}"]')
+        for entry in sec_entries:
+            node_counter += 1
+            node_id = f"n{node_counter}"
+            label = _escape_mermaid_label(entry["cidr"])
+            if entry.get("description"):
+                label += f"<br/>{_escape_mermaid_label(entry['description'])}"
+            if entry.get("vlan_label"):
+                label += f"<br/>VLAN {_escape_mermaid_label(entry['vlan_label'])}"
+            lines.append(f'    {node_id}["{label}"]')
+            if entry.get("url"):
+                click_lines.append(f'  click {node_id} "{entry["url"]}" "Öffnen"')
+        lines.append("  end")
+
+    lines.extend(click_lines)
+    lines.append("```")
+    return "\n".join(lines)
+
+
+def _networks_overview_markdown(entries: list[dict], title: str) -> str:
+    lines: list[str] = [f"# {title}", ""]
+    lines.append(
+        "_Diese Seite wird automatisch vom phpipam-Sync erzeugt. Manuelle "
+        "Änderungen werden beim nächsten Sync überschrieben._"
+    )
+    lines.append("")
+
+    if not entries:
+        lines.append("_Noch keine Netze synchronisiert._")
+        return "\n".join(lines)
+
+    lines.append("## Übersicht")
+    lines.append("")
+    lines.append("| Netz | Beschreibung | VLAN | Adressen | Seite |")
+    lines.append("|---|---|---|---|---|")
+    for entry in sorted(entries, key=lambda e: e["cidr"]):
+        link = f"[Öffnen]({entry['url']})" if entry.get("url") else "–"
+        lines.append(
+            "| {cidr} | {desc} | {vlan} | {count} | {link} |".format(
+                cidr=entry["cidr"],
+                desc=_escape_md_cell(entry.get("description")) or "–",
+                vlan=entry.get("vlan_label") or "–",
+                count=entry["address_count"],
+                link=link,
+            )
+        )
+    lines.append("")
+    lines.append("## Diagramm")
+    lines.append("")
+    lines.append(_networks_diagram_mermaid(entries))
+    return "\n".join(lines)
+
+
 def run_sync(dry_run: bool | None = None) -> list[dict]:
     """Führt den Sync durch. Gibt eine Liste von Ergebnis-Infos je Subnetz zurück."""
     dry_run = settings.sync_dry_run if dry_run is None else dry_run
     results: list[dict] = []
     section_cache: dict[str, str | None] = {}
+    vlan_cache: dict[str, dict | None] = {}
+
+    overview_entries: list[dict] = []
 
     with PhpIpamClient() as ipam:
         subnets = ipam.list_subnets()
@@ -58,11 +158,14 @@ def run_sync(dry_run: bool | None = None) -> list[dict]:
         outline = None if dry_run else OutlineClient()
         collection_id = settings.outline_collection_id
         parent_document_id = settings.outline_parent_document_id
+        parent_document_title = "Netze"
         try:
             if outline is not None:
                 collection_id = outline.resolve_collection_id(settings.outline_collection_id)
                 if parent_document_id:
-                    parent_document_id = outline.resolve_document_id(parent_document_id)
+                    parent_doc = outline.get_document(parent_document_id)
+                    parent_document_id = parent_doc["id"]
+                    parent_document_title = parent_doc.get("title") or parent_document_title
 
             for subnet in subnets:
                 section_id = subnet.get("sectionId")
@@ -70,9 +173,17 @@ def run_sync(dry_run: bool | None = None) -> list[dict]:
                     section_cache[section_id] = ipam.get_section_name(section_id) if section_id else None
                 section_name = section_cache.get(section_id)
 
+                vlan_id = subnet.get("vlanId")
+                vlan = None
+                if vlan_id and vlan_id != "0":
+                    if vlan_id not in vlan_cache:
+                        vlan_cache[vlan_id] = ipam.get_vlan(vlan_id)
+                    vlan = vlan_cache[vlan_id]
+
+                cidr = f"{subnet.get('subnet')}/{subnet.get('mask')}"
                 addresses = ipam.list_addresses(subnet["id"])
                 title = _subnet_title(subnet)
-                text = _subnet_markdown(subnet, addresses, section_name)
+                text = _subnet_markdown(subnet, addresses, section_name, vlan)
 
                 if dry_run:
                     results.append(
@@ -91,6 +202,13 @@ def run_sync(dry_run: bool | None = None) -> list[dict]:
                     collection_id,
                     parent_document_id,
                 )
+
+                revisions_pruned = 0
+                if settings.sync_history_keep and settings.sync_history_keep > 0:
+                    revisions_pruned = outline.prune_document_history(  # type: ignore[union-attr]
+                        doc["id"], settings.sync_history_keep
+                    )
+
                 results.append(
                     {
                         "subnet_id": subnet["id"],
@@ -99,10 +217,102 @@ def run_sync(dry_run: bool | None = None) -> list[dict]:
                         "action": "created" if created else "updated",
                         "outline_document_id": doc.get("id"),
                         "outline_url": doc.get("url"),
+                        "revisions_pruned": revisions_pruned,
                     }
                 )
+
+                overview_entries.append(
+                    {
+                        "cidr": cidr,
+                        "description": subnet.get("description"),
+                        "vlan_label": _vlan_label(vlan),
+                        "section_name": section_name,
+                        "address_count": len(addresses),
+                        "url": doc.get("url"),
+                    }
+                )
+
+            if not dry_run and outline is not None and parent_document_id:
+                try:
+                    overview_text = _networks_overview_markdown(overview_entries, parent_document_title)
+                    outline.update_document(parent_document_id, parent_document_title, overview_text)
+                    if settings.sync_history_keep and settings.sync_history_keep > 0:
+                        outline.prune_document_history(parent_document_id, settings.sync_history_keep)
+                    logger.info("Netze-Übersicht aktualisiert (%d Netze)", len(overview_entries))
+                except Exception:  # noqa: BLE001
+                    logger.exception("Netze-Übersicht konnte nicht aktualisiert werden")
         finally:
             if outline is not None:
                 outline.close()
 
     return results
+
+
+def purge_history_for_synced_documents() -> list[dict]:
+    """Löscht die Revisionshistorie ALLER von diesem Sync erzeugten Dokumente
+    dauerhaft und unwiederbringlich (nur der Versionsverlauf, nicht der
+    aktuelle Inhalt)."""
+    results: list[dict] = []
+    with OutlineClient() as outline:
+        collection_id = outline.resolve_collection_id(settings.outline_collection_id)
+        parent_document_id = (
+            outline.resolve_document_id(settings.outline_parent_document_id)
+            if settings.outline_parent_document_id
+            else None
+        )
+        docs = outline.list_documents(collection_id, parent_document_id)
+        for doc in docs:
+            deleted = outline.purge_document_history(doc["id"])
+            results.append(
+                {
+                    "document_id": doc["id"],
+                    "title": doc.get("title"),
+                    "revisions_deleted": deleted,
+                }
+            )
+    return results
+
+
+def purge_history_for_document(document_id: str) -> dict:
+    """Löscht die Revisionshistorie eines einzelnen Dokuments (ID oder
+    Friendly-URL-Slug) dauerhaft und unwiederbringlich."""
+    with OutlineClient() as outline:
+        real_id = outline.resolve_document_id(document_id)
+        deleted = outline.purge_document_history(real_id)
+    return {"document_id": real_id, "revisions_deleted": deleted}
+
+
+def prune_history_for_synced_documents(keep: int | None = None) -> list[dict]:
+    """Behält pro Sync-Dokument nur die `keep` neuesten Revisionen, alle
+    älteren werden dauerhaft gelöscht. Ohne Angabe wird SYNC_HISTORY_KEEP
+    aus der Config verwendet (Standard 24)."""
+    keep = settings.sync_history_keep if keep is None else keep
+    results: list[dict] = []
+    with OutlineClient() as outline:
+        collection_id = outline.resolve_collection_id(settings.outline_collection_id)
+        parent_document_id = (
+            outline.resolve_document_id(settings.outline_parent_document_id)
+            if settings.outline_parent_document_id
+            else None
+        )
+        docs = outline.list_documents(collection_id, parent_document_id)
+        for doc in docs:
+            deleted = outline.prune_document_history(doc["id"], keep)
+            results.append(
+                {
+                    "document_id": doc["id"],
+                    "title": doc.get("title"),
+                    "revisions_deleted": deleted,
+                }
+            )
+    return results
+
+
+def prune_history_for_document(document_id: str, keep: int | None = None) -> dict:
+    """Behält für ein einzelnes Dokument (ID oder Friendly-URL-Slug) nur die
+    `keep` neuesten Revisionen, alle älteren werden dauerhaft gelöscht."""
+    keep = settings.sync_history_keep if keep is None else keep
+    with OutlineClient() as outline:
+        real_id = outline.resolve_document_id(document_id)
+        deleted = outline.prune_document_history(real_id, keep)
+    return {"document_id": real_id, "revisions_deleted": deleted}