#!/usr/bin/env python3 """ Sync INCRÉMENTAL F → ERPNext — couche OPÉRATIONNELLE adresses + services. Crée les enregistrements MANQUANTS (diff complet par clé legacy, petites tables → léger) : • delivery → Service Location (LOC-{legacy_delivery_id zfill10}) — mapping migrate_locations.py • service (status=1) → Service Subscription (SUB-{legacy_service_id zfill10}) — mapping import_services_and_enrich_customers.py NE crée PAS le doctype standard `Subscription` (moteur de facturation récurrente) : F est autoritaire pour la facturation (scheduler ERPNext en pause) → couche billing gérée à la bascule, comme GL/PLE pour les factures. Ici = miroir OPÉRATIONNEL seulement (ce que les équipes voient). Ordre : Service Location d'ABORD (un service requiert sa location + le client). Idempotent (ON CONFLICT). DRY-RUN par défaut ; n'écrit RIEN sans APPLY=1. Options env : LIMIT (lot), PG_HOST. Lancer : docker cp sync_services_incremental.py erpnext-backend-1:/tmp/sync_svc.py docker exec -e APPLY=0 -e PG_HOST=db erpnext-backend-1 /home/frappe/frappe-bench/env/bin/python /tmp/sync_svc.py docker exec -e APPLY=1 -e PG_HOST=db erpnext-backend-1 /home/frappe/frappe-bench/env/bin/python /tmp/sync_svc.py """ import os import re import uuid import pymysql import psycopg2 from datetime import datetime, timezone from html import unescape LEGACY = {"host": os.environ.get("LEGACY_HOST", "10.100.80.100"), "user": "facturation", "password": os.environ.get("LEGACY_PW", "VD67owoj"), "database": "gestionclient", "connect_timeout": 30, "read_timeout": 600} PG = {"host": os.environ.get("PG_HOST", "db"), "port": 5432, "user": "postgres", "password": os.environ.get("PG_PW", "123"), "dbname": os.environ.get("PG_DB", "_eb65bdc0c4b1b2d6")} ADMIN = "Administrator" APPLY = os.environ.get("APPLY", "0") == "1" LIMIT = int(os.environ.get("LIMIT", "0")) # ── mappings (identiques à la migration) ── PROD_CAT_MAP = {4: "Internet", 32: "Internet", 8: "Internet", 26: "Internet", 29: "Internet", 7: "Internet", 23: "Internet", 17: "Internet", 16: "Internet", 21: "Internet", 33: "IPTV", 34: "IPTV", 9: "VoIP", 15: "Hébergement", 11: "Hébergement", 30: "Hébergement", 10: "Autre", 13: "Autre"} RECUR_MAP = {1: "Mensuel", 2: "Mensuel", 3: "Trimestriel", 4: "Annuel"} def now(): return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f") def uid(p=""): return p + uuid.uuid4().hex[:10] def clean(v): return unescape(str(v)).strip() if v else "" def loc_name(i): return "LOC-{:010d}".format(int(i)) def sub_name(i): return "SUB-{:010d}".format(int(i)) def ts_date(t, dflt=None): if not t or int(t) <= 0: return dflt try: return datetime.fromtimestamp(int(t), tz=timezone.utc).strftime("%Y-%m-%d") except Exception: return dflt def log(m): print(m, flush=True) def normalize_postal(pc): # "j0s1b0" → "J0S 1B0" s = re.sub(r"\s+", "", clean(pc)).upper() return s[:3] + " " + s[3:6] if len(s) >= 6 else clean(pc) def clean_addr(raw, postal=None): # strip postal qui a fui dans address1 + espaces a = clean(raw) if postal: pc = re.sub(r"\s+", "", postal).upper() a = re.sub(r"\s*" + re.escape(pc[:3]) + r"\s*" + re.escape(pc[3:6]) + r"\s*$", "", a, flags=re.I) a = re.sub(r"\s*" + re.escape(pc) + r"\s*$", "", a, flags=re.I) return re.sub(r"\s+", " ", a).strip() def svc_status(s, d, resiliated=False): # règle canonique (= legacy-sync.svcStatus) : F service → statut SS if resiliated: return "Annulé" # FIX 2026-07-06 : compte F résilié (status 3/4/5) → l'abo ne peut PAS rester Actif. # F ne cascade PAS la résiliation du compte vers service.status (cf feedback_ghost_active_subscriptions). if int(s or 0) == 1: return "Actif" # CORRIGE 2026-06-13: F status=1 = ACTIF (date_suspended = historique, ignore) return "Annulé" def main(): ts = now() log("=== Sync incrémental adresses + services (OPÉRATIONNEL, sans Subscription billing) ===") log(" mode = {}".format("APPLY" if APPLY else "DRY-RUN")) pg = psycopg2.connect(**PG); pgc = pg.cursor() mc = pymysql.connect(**LEGACY); cur = mc.cursor(pymysql.cursors.DictCursor) # maps communs pgc.execute('SELECT legacy_account_id, name FROM "tabCustomer" WHERE legacy_account_id>0') cust_map = {int(r[0]): r[1] for r in pgc.fetchall()} cur.execute("SELECT id, account_id FROM delivery"); del_acct = {r["id"]: r["account_id"] for r in cur.fetchall()} # FIX 2026-07-06 — comptes résiliés en F (status 3/4/5) : leurs services restent souvent status=1 # (F ne cascade pas) → sans ce garde-fou, Phase B/D les (re)mettent 'Actif' = abonnements fantômes. cur.execute("SELECT id FROM account WHERE status IN (3,4,5)") resil_acct = set(r["id"] for r in cur.fetchall()) log(" Comptes F résiliés (status 3/4/5) : {}".format(len(resil_acct))) # ═══ PHASE A — Service Location (delivery) ═══ cur.execute("SELECT id FROM delivery") f_del = [r["id"] for r in cur.fetchall()] pgc.execute('SELECT legacy_delivery_id FROM "tabService Location" WHERE legacy_delivery_id>0') erp_loc = set(int(r[0]) for r in pgc.fetchall()) miss_loc = sorted(i for i in f_del if i not in erp_loc) if LIMIT: miss_loc = miss_loc[:LIMIT] log(" Adresses : F={} ERP={} → {} à créer".format(len(f_del), len(erp_loc), len(miss_loc))) loc_ok = loc_skip = 0; loc_samples = [] if miss_loc: for s in range(0, len(miss_loc), 2000): batch = miss_loc[s:s+2000] cur.execute("SELECT id, account_id, name, address1, city, state, zip FROM delivery WHERE id IN ({})".format(",".join(["%s"]*len(batch))), batch) for d in cur.fetchall(): postal = normalize_postal(d.get("zip")) addr = clean_addr(d.get("address1"), postal) city = clean_addr(d.get("city")) lname = (clean(d.get("name")) or ("{}, {}".format(addr, city) if addr else "Location-{}".format(d["id"])))[:140] cust = cust_map.get(d["account_id"]) if len(loc_samples) < 6: loc_samples.append({"loc": loc_name(d["id"]), "addr": addr, "city": city, "customer": cust}) if not APPLY: loc_ok += 1; continue try: pgc.execute(""" INSERT INTO "tabService Location" ( name, creation, modified, modified_by, owner, docstatus, idx, customer, location_name, status, address_line, city, postal_code, province, legacy_delivery_id ) VALUES (%s,%s,%s,%s,%s,0,0, %s,%s,'Active',%s,%s,%s,%s,%s) ON CONFLICT (name) DO NOTHING """, (loc_name(d["id"]), ts, ts, ADMIN, ADMIN, cust, lname, addr, city, postal, clean(d.get("state")), d["id"])) loc_ok += 1 except Exception as e: loc_skip += 1; pg.rollback() if loc_skip <= 5: log(" ERR loc#{} -> {}".format(d["id"], str(e)[:120])) continue if APPLY: pg.commit() # recharge le map location pour la phase B pgc.execute('SELECT legacy_delivery_id, name FROM "tabService Location" WHERE legacy_delivery_id>0') loc_map = {int(r[0]): r[1] for r in pgc.fetchall()} # ═══ PHASE B — Service Subscription (service status=1) ═══ cur.execute("SELECT id FROM service WHERE status=1") f_svc = [r["id"] for r in cur.fetchall()] pgc.execute('SELECT legacy_service_id FROM "tabService Subscription" WHERE legacy_service_id>0') erp_sub = set(int(r[0]) for r in pgc.fetchall()) miss_svc = sorted(i for i in f_svc if i not in erp_sub) if LIMIT: miss_svc = miss_svc[:LIMIT] log(" Services : F actifs={} ERP={} → {} à créer".format(len(f_svc), len(erp_sub), len(miss_svc))) # product names FR cur.execute("SELECT product_id, name AS prod_name FROM product_translate WHERE language_id='francais'") prod_names = {r["product_id"]: r["prod_name"] for r in cur.fetchall()} sub_ok = sub_noloc = sub_skip = 0; sub_samples = [] if miss_svc: for s in range(0, len(miss_svc), 2000): batch = miss_svc[s:s+2000] cur.execute("""SELECT s.id, s.delivery_id, s.product_id, s.payment_recurrence, s.hijack, s.hijack_price, s.hijack_download_speed, s.hijack_upload_speed, s.date_orig, s.date_end_contract, s.radius_user, s.radius_pwd, p.sku, p.price, p.download_speed, p.upload_speed, p.category AS prod_cat FROM service s LEFT JOIN product p ON s.product_id=p.id WHERE s.id IN ({})""".format(",".join(["%s"]*len(batch))), batch) for svc in cur.fetchall(): sl = loc_map.get(svc["delivery_id"]) if svc["delivery_id"] else None acct_id = del_acct.get(svc["delivery_id"]) if svc["delivery_id"] else None cust = cust_map.get(acct_id) if acct_id else None if not sl or not cust: sub_noloc += 1; continue # location + client requis (idem migration) new_status = "Annulé" if acct_id in resil_acct else "Actif" # FIX 2026-07-06 : jamais Actif sur compte résilié cat = PROD_CAT_MAP.get(svc["prod_cat"] or 0, "Autre") plan = unescape(prod_names.get(svc["product_id"], svc.get("sku") or "Unknown"))[:140] if svc["hijack"] and svc["hijack_download_speed"]: sd, su = int(svc["hijack_download_speed"]) // 1024, int(svc["hijack_upload_speed"] or 0) // 1024 elif svc["download_speed"]: sd, su = int(svc["download_speed"]) // 1024, int(svc["upload_speed"] or 0) // 1024 else: sd = su = 0 price = float(svc["hijack_price"] or 0) if svc["hijack"] else float(svc["price"] or 0) cycle = RECUR_MAP.get(svc["payment_recurrence"], "Mensuel") start = ts_date(svc["date_orig"], "2020-01-01"); end = ts_date(svc["date_end_contract"]) if len(sub_samples) < 6: sub_samples.append({"sub": sub_name(svc["id"]), "cat": cat, "plan": plan, "price": price, "cycle": cycle, "loc": sl}) if not APPLY: sub_ok += 1; continue try: pgc.execute(""" INSERT INTO "tabService Subscription" ( name, creation, modified, modified_by, owner, docstatus, idx, customer, service_location, status, service_category, plan_name, speed_down, speed_up, monthly_price, billing_cycle, start_date, end_date, legacy_service_id, radius_user, radius_password, product_sku ) VALUES (%s,%s,%s,%s,%s,0,0, %s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s, %s,%s,%s,%s) ON CONFLICT (name) DO NOTHING """, (sub_name(svc["id"]), ts, ts, ADMIN, ADMIN, cust, sl, new_status, cat, plan, sd, su, price, cycle, start, end, svc["id"], clean(svc.get("radius_user")), clean(svc.get("radius_pwd")), clean(svc.get("sku")))) sub_ok += 1 except Exception as e: sub_skip += 1; pg.rollback() if sub_skip <= 5: log(" ERR svc#{} -> {}".format(svc["id"], str(e)[:120])) continue if APPLY: pg.commit() # ═══ PHASE C — rafraîchir le PRIX des SS existantes (F autoritaire ; opérationnel) ═══ # Les prix F (rabais/crédits/forfaits) bougent après la migration → le monthly_price figé # de la SS diverge. On le réaligne sur F. (Statut Actif/Suspendu/Annulé = legacy-sync.syncServices.) cur.execute("SELECT s.id, IF(s.hijack=1, COALESCE(s.hijack_price,0), COALESCE(p.price,0)) AS fprice FROM service s LEFT JOIN product p ON p.id=s.product_id WHERE s.status=1") fprice = {r["id"]: round(float(r["fprice"]), 2) for r in cur.fetchall()} pgc.execute('SELECT legacy_service_id, monthly_price FROM "tabService Subscription" WHERE legacy_service_id>0') to_upd = [] for lid, mp in pgc.fetchall(): f = fprice.get(int(lid)) if f is None: continue if abs(round(float(mp or 0), 2) - f) > 0.011: to_upd.append((f, int(lid))) price_fixed = 0 log(" Prix SS divergents vs F : {}".format(len(to_upd))) if APPLY and to_upd: for f, lid in to_upd: pgc.execute('UPDATE "tabService Subscription" SET monthly_price=%s, modified=%s WHERE legacy_service_id=%s AND ROUND(monthly_price::numeric,2)<>%s', (f, ts, lid, f)) pg.commit(); price_fixed = len(to_upd) # ═══ PHASE D — rafraîchir le STATUT des SS (Actif/Suspendu/Annulé) depuis F ═══ # Reflète résiliations (status 1→0 = Annulé) + suspensions (date_suspended>0 = Suspendu) + réactivations. # C'est la SOURCE du statut client (disabled dérivé des services — cf. feedback_customer_status_policy). cur.execute("SELECT s.id, s.status, s.date_suspended, d.account_id FROM service s LEFT JOIN delivery d ON d.id=s.delivery_id") want = {r["id"]: svc_status(r["status"], r["date_suspended"], r["account_id"] in resil_acct) for r in cur.fetchall()} pgc.execute('SELECT legacy_service_id, status FROM "tabService Subscription" WHERE legacy_service_id>0') st_upd = []; trans = {} for lid, st in pgc.fetchall(): w = want.get(int(lid)) if w and w != st: st_upd.append((w, int(lid))); k = (st or "∅") + "→" + w; trans[k] = trans.get(k, 0) + 1 status_fixed = 0 log(" Statut SS divergents vs F : {} {}".format(len(st_upd), trans)) if APPLY and st_upd: for w, lid in st_upd: pgc.execute('UPDATE "tabService Subscription" SET status=%s, modified=%s WHERE legacy_service_id=%s AND status<>%s', (w, ts, lid, w)) pg.commit(); status_fixed = len(st_upd) mc.close(); pg.close() log("") log("=" * 60) log("Statut SS réalignés sur F : {} {}".format(status_fixed if APPLY else len(st_upd), "corrigés" if APPLY else "à corriger (dry-run)")) log("Prix SS réalignés sur F : {} {}".format(price_fixed if APPLY else len(to_upd), "corrigés" if APPLY else "à corriger (dry-run)")) log("Adresses : {} {} | {} err".format(loc_ok, "créées" if APPLY else "à créer (dry-run)", loc_skip)) log(" échantillon: {}".format(loc_samples)) log("Services : {} {} | {} sans location/client | {} err".format(sub_ok, "créés" if APPLY else "à créer (dry-run)", sub_noloc, sub_skip)) log(" échantillon: {}".format(sub_samples)) log("=" * 60) if __name__ == "__main__": main()