Admin MPCZ
db74111d58
- Import: responsable_serveur_name → responsable_nom, responsable_domaine_name → referent_nom - Import: environnement iTop synchronise - Export: cree les VMs manquantes dans iTop (classe VirtualMachine, org MPCZ) - Export: hostname comme nom (pas fqdn)
279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""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) avec contacts"""
|
|
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,"
|
|
"organization_name,location_name,business_criticity,cpu,ram,"
|
|
"contacts_list,responsable_serveur_name,responsable_domaine_name,environnement"
|
|
})
|
|
if r.get("code") == 0 and r.get("objects"):
|
|
for key, obj in r["objects"].items():
|
|
f = obj["fields"]
|
|
# Extract contacts (responsable serveur)
|
|
contacts = f.get("contacts_list", [])
|
|
contact_names = []
|
|
contact_emails = []
|
|
for c in contacts:
|
|
cname = c.get("contact_id_friendlyname", "")
|
|
if cname:
|
|
contact_names.append(cname)
|
|
|
|
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("organization_name", ""),
|
|
"location": f.get("location_name", ""),
|
|
"criticity": f.get("business_criticity", ""),
|
|
"cpu": f.get("cpu", ""),
|
|
"ram": f.get("ram", ""),
|
|
"responsable_nom": f.get("responsable_serveur_name", "") or (contact_names[0] if contact_names else ""),
|
|
"responsable_domaine": f.get("responsable_domaine_name", ""),
|
|
"environnement": f.get("environnement", ""),
|
|
})
|
|
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 s.get("responsable_nom"):
|
|
sets.append("responsable_nom = :resp_nom")
|
|
updates["resp_nom"] = s["responsable_nom"]
|
|
if s.get("responsable_domaine"):
|
|
sets.append("referent_nom = :ref_nom")
|
|
updates["ref_nom"] = s["responsable_domaine"]
|
|
if s.get("description"):
|
|
sets.append("commentaire = :comm")
|
|
updates["comm"] = s["description"]
|
|
if s.get("ip"):
|
|
sets.append("fqdn = :fqdn")
|
|
updates["fqdn"] = s["name"]
|
|
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, "created": 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, responsable_nom, commentaire 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.commentaire:
|
|
fields["description"] = srv.commentaire
|
|
elif 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')}")
|
|
|
|
else:
|
|
# Server not in iTop — create as VirtualMachine
|
|
fields = {
|
|
"name": srv.hostname,
|
|
"org_id": "SELECT Organization WHERE name = 'MPCZ'",
|
|
"status": status_map.get(srv.etat, "production"),
|
|
"description": srv.commentaire or f"OS: {srv.os_version or 'N/A'}",
|
|
}
|
|
r = client.create_server("VirtualMachine", fields)
|
|
if r.get("code") == 0:
|
|
stats["created"] += 1
|
|
else:
|
|
stats["errors"].append(f"Create {srv.hostname}: {r.get('message')}")
|
|
|
|
log.info(f"iTop export: {stats}")
|
|
return stats
|