254 lines
16 KiB
Python
254 lines
16 KiB
Python
#!/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, timedelta
|
|
from html import unescape
|
|
|
|
LEGACY = {"host": os.environ.get("LEGACY_HOST", "10.100.80.100"), "user": "facturation",
|
|
"password": os.environ.get("LEGACY_PW", ""), "database": "gestionclient", # secret via env (jamais commité)
|
|
"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, safe_to_annul=False): # règle canonique (= legacy-sync.svcStatus) : F service → statut SS
|
|
if safe_to_annul:
|
|
return "Annulé" # RAFFINÉ 2026-07-06 : fantôme AVÉRÉ seulement (compte résilié + service payant + F ne facture
|
|
# plus depuis >24 mois). Pas de cascade aveugle : on préserve les gratuits (employés) et tout
|
|
# ce que F facture encore (cf feedback_customer_status_policy : statut SERVICE = vérité).
|
|
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 (RAFFINÉ) — comptes résiliés en F (status 3/4/5) : leurs services restent souvent status=1
|
|
# (F ne cascade pas). On N'ANNULE PAS en cascade aveugle (des comptes terminés gardent des services
|
|
# légitimes : employés à internet gratuit, services encore facturés). On annule UNIQUEMENT le fantôme
|
|
# AVÉRÉ = service PAYANT dont F n'a pas émis de facture depuis >24 mois. cf feedback_customer_status_policy.
|
|
cur.execute("SELECT id FROM account WHERE status IN (3,4,5)")
|
|
resil_acct = set(r["id"] for r in cur.fetchall())
|
|
GHOST_CUTOFF = int((datetime.now(timezone.utc) - timedelta(days=730)).timestamp()) # 24 mois
|
|
last_inv = {}
|
|
if resil_acct:
|
|
ph = ",".join(["%s"] * len(resil_acct))
|
|
cur.execute("SELECT account_id, MAX(date_orig) li FROM invoice WHERE account_id IN ({}) GROUP BY account_id".format(ph), list(resil_acct))
|
|
last_inv = {r["account_id"]: r["li"] for r in cur.fetchall()}
|
|
def annul_ghost(acct, price):
|
|
# True = fantôme avéré (sûr à annuler) ; False = préserver (miroir du statut F / gratuit / facturé récemment)
|
|
if acct not in resil_acct: return False
|
|
if float(price or 0) <= 0: return False # gratuit/comp (employé) → jamais annuler
|
|
li = last_inv.get(acct)
|
|
return li is not None and int(li) < GHOST_CUTOFF # dernière facture F > 24 mois → fantôme
|
|
log(" Comptes F résiliés : {} | dont avec facturation connue : {}".format(len(resil_acct), len(last_inv)))
|
|
|
|
# ═══ 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)
|
|
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)
|
|
new_status = "Annulé" if annul_ghost(acct_id, price) else "Actif" # RAFFINÉ 2026-07-06 : fantôme avéré seulement
|
|
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, IF(s.hijack=1, COALESCE(s.hijack_price,0), COALESCE(p.price,0)) AS price FROM service s LEFT JOIN delivery d ON d.id=s.delivery_id LEFT JOIN product p ON p.id=s.product_id")
|
|
want = {r["id"]: svc_status(r["status"], r["date_suspended"], annul_ghost(r["account_id"], r["price"])) 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()
|