"""Service iTop REST API — synchronisation bidirectionnelle serveurs et contacts""" import logging import requests import json from datetime import datetime from sqlalchemy import text log = logging.getLogger(__name__) class ITopClient: """Client REST iTop v1.3""" def __init__(self, url, user, password): self.url = url.rstrip("/") + "/webservices/rest.php?version=1.3" self.user = user self.password = password def _call(self, operation, **kwargs): data = {"operation": operation, **kwargs} try: r = requests.post(self.url, data={"json_data": json.dumps(data), "auth_user": self.user, "auth_pwd": self.password}, verify=False, timeout=30) result = r.json() if result.get("code", -1) != 0: log.error(f"iTop API error: {result.get('message')}") return result except Exception as e: log.error(f"iTop connection error: {e}") return {"code": -1, "message": str(e)} def get_servers(self): """Recupere tous les serveurs iTop (Server + VirtualMachine)""" servers = [] for cls in ["Server", "VirtualMachine"]: r = self._call("core/get", **{ "class": cls, "key": f"SELECT {cls}", "output_fields": "name,description,status,managementip,osfamily_id_friendlyname," "osversion_id_friendlyname,brand_name,model_name,serialnumber," "org_name,location_name,business_criticity,cpu,ram" }) if r.get("code") == 0 and r.get("objects"): for key, obj in r["objects"].items(): f = obj["fields"] servers.append({ "itop_id": obj["key"], "itop_class": cls, "name": f.get("name", ""), "description": f.get("description", ""), "status": f.get("status", ""), "ip": f.get("managementip", ""), "os_family": f.get("osfamily_id_friendlyname", ""), "os_version": f.get("osversion_id_friendlyname", ""), "brand": f.get("brand_name", ""), "model": f.get("model_name", ""), "serial": f.get("serialnumber", ""), "org": f.get("org_name", ""), "location": f.get("location_name", ""), "criticity": f.get("business_criticity", ""), "cpu": f.get("cpu", ""), "ram": f.get("ram", ""), }) return servers def get_contacts(self): """Recupere tous les contacts iTop""" r = self._call("core/get", **{ "class": "Person", "key": "SELECT Person", "output_fields": "name,first_name,email,phone,org_name,function" }) contacts = [] if r.get("code") == 0 and r.get("objects"): for key, obj in r["objects"].items(): f = obj["fields"] contacts.append({ "itop_id": obj["key"], "name": f.get("name", ""), "first_name": f.get("first_name", ""), "email": f.get("email", ""), "phone": f.get("phone", ""), "org": f.get("org_name", ""), "function": f.get("function", ""), }) return contacts def get_server_contacts(self, server_class, server_id): """Recupere les contacts associes a un serveur""" r = self._call("core/get", **{ "class": server_class, "key": str(server_id), "output_fields": "contacts_list" }) contacts = [] if r.get("code") == 0 and r.get("objects"): for key, obj in r["objects"].items(): for c in obj["fields"].get("contacts_list", []): contacts.append({ "contact_id": c.get("contact_id"), "contact_name": c.get("contact_id_friendlyname", ""), }) return contacts def update_server(self, server_class, server_id, fields): """Met a jour un serveur dans iTop""" return self._call("core/update", **{ "class": server_class, "key": str(server_id), "fields": fields, "comment": "Sync from PatchCenter" }) def create_server(self, server_class, fields): """Cree un serveur dans iTop""" return self._call("core/create", **{ "class": server_class, "fields": fields, "comment": "Created by PatchCenter" }) def sync_from_itop(db, itop_url, itop_user, itop_pass): """Importe les serveurs et contacts depuis iTop vers PatchCenter""" client = ITopClient(itop_url, itop_user, itop_pass) stats = {"servers_updated": 0, "servers_created": 0, "contacts_updated": 0, "contacts_created": 0, "errors": []} # --- Sync contacts --- itop_contacts = client.get_contacts() for c in itop_contacts: fullname = f"{c['first_name']} {c['name']}".strip() existing = db.execute(text( "SELECT id FROM contacts WHERE LOWER(email) = LOWER(:email) OR LOWER(nom) = LOWER(:nom)" ), {"email": c["email"], "nom": fullname}).fetchone() if existing: db.execute(text(""" UPDATE contacts SET nom = :nom, email = :email, telephone = :tel, updated_at = NOW() WHERE id = :id """), {"id": existing.id, "nom": fullname, "email": c["email"], "tel": c["phone"]}) stats["contacts_updated"] += 1 else: try: db.execute(text(""" INSERT INTO contacts (nom, email, telephone) VALUES (:nom, :email, :tel) """), {"nom": fullname, "email": c["email"], "tel": c["phone"]}) stats["contacts_created"] += 1 except Exception as e: stats["errors"].append(f"Contact {fullname}: {e}") # --- Sync servers --- itop_servers = client.get_servers() for s in itop_servers: hostname = s["name"].split(".")[0].lower() existing = db.execute(text( "SELECT id FROM servers WHERE LOWER(hostname) = LOWER(:h) OR LOWER(fqdn) = LOWER(:f)" ), {"h": hostname, "f": s["name"]}).fetchone() itop_status_map = {"production": "en_production", "stock": "stock", "implementation": "en_cours", "obsolete": "decommissionne"} etat = itop_status_map.get(s["status"], "en_production") if existing: updates = {"sid": existing.id} sets = [] if s["os_version"]: sets.append("os_version = :osv") updates["osv"] = s["os_version"] if s["ip"]: sets.append("fqdn = :fqdn") updates["fqdn"] = s["name"] if s["location"]: sets.append("site = :site") updates["site"] = s["location"] if sets: sets.append("updated_at = NOW()") db.execute(text(f"UPDATE servers SET {', '.join(sets)} WHERE id = :sid"), updates) stats["servers_updated"] += 1 else: try: db.execute(text(""" INSERT INTO servers (hostname, fqdn, os_family, os_version, machine_type, tier, etat, ssh_port, ssh_user, ssh_method, site) VALUES (:h, :f, :osf, :osv, :mt, 'a_definir', :etat, 22, 'root', 'ssh_key', :site) """), { "h": hostname, "f": s["name"], "osf": "linux" if "linux" in s["os_family"].lower() else "windows", "osv": s["os_version"], "etat": etat, "site": s["location"], "mt": "vm" if s["itop_class"] == "VirtualMachine" else "physical", }) stats["servers_created"] += 1 except Exception as e: stats["errors"].append(f"Server {s['name']}: {e}") db.commit() log.info(f"iTop sync: {stats}") return stats def sync_to_itop(db, itop_url, itop_user, itop_pass): """Exporte les serveurs PatchCenter vers iTop (mise a jour)""" client = ITopClient(itop_url, itop_user, itop_pass) stats = {"updated": 0, "errors": []} # Get current iTop servers for matching itop_servers = {s["name"].lower(): s for s in client.get_servers()} # Get patchcenter servers rows = db.execute(text("SELECT hostname, fqdn, os_version, os_family, etat, site FROM servers")).fetchall() status_map = {"en_production": "production", "decommissionne": "obsolete", "stock": "stock", "en_cours": "implementation"} for srv in rows: fqdn = (srv.fqdn or "").lower() hostname = (srv.hostname or "").lower() itop_srv = itop_servers.get(fqdn) or itop_servers.get(hostname) if itop_srv: fields = {} if srv.os_version: fields["description"] = f"OS: {srv.os_version}" if srv.etat: fields["status"] = status_map.get(srv.etat, "production") if fields: r = client.update_server(itop_srv["itop_class"], itop_srv["itop_id"], fields) if r.get("code") == 0: stats["updated"] += 1 else: stats["errors"].append(f"{srv.hostname}: {r.get('message')}") log.info(f"iTop export: {stats}") return stats