Add SANEF physical/hypervisor import script

tools/import_sanef_physical.py

@@ -0,0 +1,150 @@
+"""Import SANEF Serveurs physiques + Hyperviseurs CSV -> table servers (machine_type=physical).
+
+Usage:
+    python tools/import_sanef_physical.py <csv_path> [--hypervisor]
+
+Le script auto-détecte le type via la colonne 'Sous-classe de CI'.
+Marque les hyperviseurs avec is_hypervisor (via patch_owner_details si pas de colonne dédiée).
+Skip les hostnames déjà présents (évite doublons avec les VMs).
+"""
+import csv
+import os
+import argparse
+from sqlalchemy import create_engine, text
+
+DATABASE_URL = os.getenv("DATABASE_URL_DEMO") or os.getenv("DATABASE_URL") \
+    or "postgresql://patchcenter:PatchCenter2026!@localhost:5432/patchcenter_demo"
+
+
+def norm_os_family(famille):
+    f = (famille or "").strip().lower()
+    if "windows" in f:
+        return "windows"
+    if "linux" in f or "oracle" in f or "esxi" in f:
+        return "linux"
+    return None
+
+
+def norm_etat(status, etat):
+    e = (etat or "").strip().lower()
+    if "stock" in e:
+        return "stock"
+    if "implémentation" in e or "implementation" in e:
+        return "implementation"
+    if "obsol" in e:
+        return "obsolete"
+    if "eol" in e:
+        return "eol"
+    return "production"
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("csv_path")
+    args = parser.parse_args()
+
+    engine = create_engine(DATABASE_URL)
+    print(f"[INFO] DB: {DATABASE_URL.split('@')[-1]}")
+    print(f"[INFO] CSV: {args.csv_path}")
+
+    with open(args.csv_path, "r", encoding="utf-8-sig", newline="") as f:
+        reader = csv.DictReader(f)
+        rows = list(reader)
+    print(f"[INFO] {len(rows)} lignes")
+
+    conn = engine.connect().execution_options(isolation_level="AUTOCOMMIT")
+
+    inserted = 0
+    skipped = 0
+    duplicates = 0
+
+    for r in rows:
+        hostname = (r.get("Nom") or r.get("Hostname") or "").strip()
+        if not hostname or hostname in ("(null)",) or hostname.startswith(("...", "#")):
+            skipped += 1
+            continue
+        # Ignore les lignes parasites (chiffres seuls, etc.)
+        if not any(c.isalpha() for c in hostname):
+            skipped += 1
+            continue
+
+        # Skip si hostname déjà présent (déjà importé comme VM)
+        existing = conn.execute(text("SELECT id FROM servers WHERE hostname=:h"),
+                                {"h": hostname}).fetchone()
+        if existing:
+            duplicates += 1
+            continue
+
+        sous_classe = (r.get("Sous-classe de CI") or "").strip()
+        is_hypervisor = "Hyperviseur" in sous_classe or "Serveur" in r.get("Sous-classe de CI", "") and "vir" in hostname.lower()
+
+        os_family = norm_os_family(r.get("Famille OS->Nom"))
+        os_version = (r.get("Version OS->Nom") or "").strip()[:200] or None
+        ip = (r.get("IP") or "").strip() or None
+        domain = (r.get("Domaine") or "").strip()[:50] or None
+        site = (r.get("Lieu->Nom") or "").strip()[:50] or None
+        etat = norm_etat(r.get("Status"), r.get("Etat"))
+        responsable = (r.get("Logiciel->Responsable Domaine DTS") or "").strip() or None
+
+        marque = (r.get("Marque->Nom") or "").strip()
+        modele = (r.get("Modèle->Nom") or r.get("Mod\u00e8le->Nom") or "").strip()
+        underlying = (r.get("Serveur->Nom") or "").strip()
+        cluster = (r.get("vCluster->Nom") or "").strip()
+        desc = (r.get("Description") or "").strip()
+        fonction = (r.get("Fonction") or "").strip()
+
+        details_parts = []
+        if is_hypervisor:
+            details_parts.append("[HYPERVISEUR]")
+        if marque or modele:
+            details_parts.append(f"{marque} {modele}".strip())
+        if underlying:
+            details_parts.append(f"Serveur: {underlying}")
+        if cluster:
+            details_parts.append(f"Cluster: {cluster}")
+        if desc:
+            details_parts.append(desc)
+        if fonction:
+            details_parts.append(fonction)
+        details = " | ".join(details_parts)[:2000] or None
+
+        fqdn = None
+        if domain and "." in domain and domain.lower() not in ("workgroup", "host"):
+            fqdn = f"{hostname}.{domain}"[:255]
+
+        try:
+            conn.execute(text("""
+                INSERT INTO servers
+                (hostname, fqdn, domain_ltd, os_family, os_version, machine_type,
+                 etat, site, responsable_nom, patch_owner_details)
+                VALUES
+                (:hostname, :fqdn, :domain_ltd, :os_family, :os_version, 'physical',
+                 :etat, :site, :responsable_nom, :patch_owner_details)
+            """), {
+                "hostname": hostname, "fqdn": fqdn, "domain_ltd": domain,
+                "os_family": os_family, "os_version": os_version,
+                "etat": etat, "site": site, "responsable_nom": responsable,
+                "patch_owner_details": details,
+            })
+            inserted += 1
+            if ip:
+                sid = conn.execute(text("SELECT id FROM servers WHERE hostname=:h"),
+                                   {"h": hostname}).fetchone()
+                if sid:
+                    try:
+                        conn.execute(text("""
+                            INSERT INTO server_ips (server_id, ip_address, is_primary)
+                            VALUES (:sid, :ip, true)
+                        """), {"sid": sid.id, "ip": ip})
+                    except Exception:
+                        pass
+        except Exception as e:
+            print(f"  [ERR] {hostname}: {str(e)[:150]}")
+            skipped += 1
+
+    conn.close()
+    print(f"[DONE] Inseres: {inserted} | Doublons (deja en base): {duplicates} | Ignores: {skipped}")
+
+
+if __name__ == "__main__":
+    main()