diff --git a/app/outline_client.py b/app/outline_client.py
index 875f634..8ee8019 100644
--- a/app/outline_client.py
+++ b/app/outline_client.py
@@ -50,10 +50,15 @@ class OutlineClient:
data = self._post("/collections.info", {"id": collection_id_or_slug})
return data["data"]["id"]
+ def get_document(self, document_id_or_slug: str) -> dict:
+ """Lädt die vollständigen Metadaten eines Dokuments per ID oder
+ Friendly-URL-Slug (u.a. für den aktuellen Titel gebraucht)."""
+ data = self._post("/documents.info", {"id": document_id_or_slug})
+ return data["data"]
+
def resolve_document_id(self, document_id_or_slug: str) -> str:
"""Löst eine Dokument-ID oder Friendly-URL-Slug in die echte UUID auf."""
- data = self._post("/documents.info", {"id": document_id_or_slug})
- return data["data"]["id"]
+ return self.get_document(document_id_or_slug)["id"]
def find_document_by_title(
self, title: str, collection_id: str, parent_document_id: str | None = None
@@ -99,6 +104,70 @@ class OutlineClient:
)
return data["data"]
+ def list_documents(self, collection_id: str, parent_document_id: str | None = None) -> list[dict]:
+ """Listet alle Dokumente einer Collection (optional gefiltert auf
+ direkte Kinder von parent_document_id), paginiert."""
+ docs: list[dict] = []
+ offset = 0
+ limit = 100
+ while True:
+ payload: dict = {"collectionId": collection_id, "limit": limit, "offset": offset}
+ if parent_document_id:
+ payload["parentDocumentId"] = parent_document_id
+ data = self._post("/documents.list", payload)
+ batch = data["data"]
+ docs.extend(batch)
+ if len(batch) < limit:
+ break
+ offset += limit
+ return docs
+
+ def list_revisions(self, document_id: str) -> list[dict]:
+ """Listet alle Revisionen (Versionsverlauf) eines Dokuments, paginiert."""
+ revisions: list[dict] = []
+ offset = 0
+ limit = 100
+ while True:
+ data = self._post(
+ "/revisions.list",
+ {"documentId": document_id, "limit": limit, "offset": offset},
+ )
+ batch = data["data"]
+ revisions.extend(batch)
+ if len(batch) < limit:
+ break
+ offset += limit
+ return revisions
+
+ def delete_revision(self, revision_id: str) -> None:
+ self._post("/revisions.delete", {"id": revision_id})
+
+ def purge_document_history(self, document_id: str) -> int:
+ """Löscht ALLE Revisionen eines Dokuments dauerhaft und unwiederbringlich
+ (nicht der aktuelle Inhalt selbst, nur der Versionsverlauf).
+
+ Returns: Anzahl gelöschter Revisionen.
+ """
+ revisions = self.list_revisions(document_id)
+ for revision in revisions:
+ self.delete_revision(revision["id"])
+ return len(revisions)
+
+ def prune_document_history(self, document_id: str, keep: int) -> int:
+ """Behält nur die `keep` neuesten Revisionen eines Dokuments, alle
+ älteren werden dauerhaft gelöscht und sind nicht wiederherstellbar.
+
+ Returns: Anzahl gelöschter Revisionen.
+ """
+ if keep < 0:
+ keep = 0
+ revisions = self.list_revisions(document_id)
+ revisions.sort(key=lambda r: r.get("createdAt") or "", reverse=True)
+ to_delete = revisions[keep:]
+ for revision in to_delete:
+ self.delete_revision(revision["id"])
+ return len(to_delete)
+
def upsert_document(
self,
title: str,
diff --git a/app/sync.py b/app/sync.py
index dd14731..73312ae 100644
--- a/app/sync.py
+++ b/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"
{_escape_mermaid_label(entry['description'])}"
+ if entry.get("vlan_label"):
+ label += f"
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}