Add SANEF contacts import script
This commit is contained in:
parent
612f3c7cea
commit
faa267c3ff
94
tools/import_sanef_contacts.py
Normal file
94
tools/import_sanef_contacts.py
Normal file
@ -0,0 +1,94 @@
|
||||
"""Import SANEF contacts CSV (iTop Contact export) -> table contacts.
|
||||
|
||||
Usage:
|
||||
python tools/import_sanef_contacts.py <csv_path> [--replace]
|
||||
"""
|
||||
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 main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("csv_path")
|
||||
parser.add_argument("--replace", action="store_true",
|
||||
help="Vider la table contacts avant import (uniquement les itop_id non-null)")
|
||||
args = parser.parse_args()
|
||||
|
||||
engine = create_engine(DATABASE_URL)
|
||||
print(f"[INFO] DB: {DATABASE_URL.split('@')[-1]}")
|
||||
|
||||
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")
|
||||
|
||||
if args.replace:
|
||||
n = conn.execute(text("DELETE FROM contacts WHERE itop_id IS NOT NULL")).rowcount
|
||||
print(f"[OK] {n} contacts iTop supprimes")
|
||||
|
||||
inserted = 0
|
||||
updated = 0
|
||||
skipped = 0
|
||||
|
||||
for r in rows:
|
||||
name = (r.get("Nom") or "").strip()
|
||||
email = (r.get("Email") or "").strip().lower()
|
||||
state = (r.get("Etat") or "").strip().lower()
|
||||
phone = (r.get("Telephone fixe") or r.get("Téléphone fixe") or "").strip() or None
|
||||
team = (r.get("Organisation->Nom organisation") or "").strip()[:100] or None
|
||||
function = (r.get("Fonction") or "").strip()[:200] or None
|
||||
itop_id = (r.get("id (Clef primaire)") or "").strip()
|
||||
|
||||
if not name:
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
# email required - generate placeholder if missing
|
||||
if not email:
|
||||
slug = name.lower().replace(" ", "-").replace("'", "").replace(".", "")[:50]
|
||||
email = f"{slug}@no-email.sanef.local"
|
||||
|
||||
is_active = state in ("actif", "active")
|
||||
itop_id_int = int(itop_id) if itop_id.isdigit() else None
|
||||
|
||||
try:
|
||||
existing = conn.execute(text(
|
||||
"SELECT id FROM contacts WHERE email=:e OR itop_id=:iid"
|
||||
), {"e": email, "iid": itop_id_int}).fetchone()
|
||||
|
||||
if existing:
|
||||
conn.execute(text("""
|
||||
UPDATE contacts SET
|
||||
name=:name, is_active=:active, telephone=:phone,
|
||||
team=:team, function=:function, itop_id=:iid,
|
||||
updated_at=now()
|
||||
WHERE id=:id
|
||||
"""), {"name": name, "active": is_active, "phone": phone,
|
||||
"team": team, "function": function, "iid": itop_id_int,
|
||||
"id": existing.id})
|
||||
updated += 1
|
||||
else:
|
||||
conn.execute(text("""
|
||||
INSERT INTO contacts (name, email, role, is_active, telephone, team, function, itop_id)
|
||||
VALUES (:name, :email, 'autre', :active, :phone, :team, :function, :iid)
|
||||
"""), {"name": name, "email": email, "active": is_active,
|
||||
"phone": phone, "team": team, "function": function,
|
||||
"iid": itop_id_int})
|
||||
inserted += 1
|
||||
except Exception as e:
|
||||
print(f" [ERR] {name} ({email}): {str(e)[:150]}")
|
||||
skipped += 1
|
||||
|
||||
conn.close()
|
||||
print(f"[DONE] Crees: {inserted} | Mis a jour: {updated} | Ignores: {skipped}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Reference in New Issue
Block a user