319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""Sync-Logik: liest Subnetze + Adressen aus phpipam und schreibt sie als
|
||
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')}"
|
||
desc = subnet.get("description") or ""
|
||
return f"{cidr} – {desc}".rstrip(" –") if desc else cidr
|
||
|
||
|
||
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}")
|
||
lines.append("")
|
||
lines.append(f"- **Beschreibung:** {subnet.get('description') or '–'}")
|
||
if section_name:
|
||
lines.append(f"- **Section:** {section_name}")
|
||
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("")
|
||
|
||
if not addresses:
|
||
lines.append("_Keine belegten Adressen in phpipam hinterlegt._")
|
||
return "\n".join(lines)
|
||
|
||
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} | ".format(
|
||
ip=addr.get("ip", ""),
|
||
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()
|
||
|
||
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_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")
|
||
if section_id not in section_cache:
|
||
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, vlan)
|
||
|
||
if dry_run:
|
||
results.append(
|
||
{
|
||
"subnet_id": subnet["id"],
|
||
"title": title,
|
||
"address_count": len(addresses),
|
||
"action": "dry-run",
|
||
}
|
||
)
|
||
continue
|
||
|
||
doc, created = outline.upsert_document( # type: ignore[union-attr]
|
||
title,
|
||
text,
|
||
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"],
|
||
"title": title,
|
||
"address_count": len(addresses),
|
||
"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}
|