004c3f8dee
Scoreboard (console de réconciliation F ↔ ERPNext) :
- hub: legacy-sync.js scoreboard() + GET /legacy-sync/scoreboard — comptes bruts
par module (clients/lieux/abos/appareils/tickets/factures/paiements) + métrique
ghost_active_subscriptions. Devices = tabService Equipment (pas tabDevice).
Count F « abos actifs » exclut les comptes résiliés (comparaison pomme-à-pomme).
- ops: carte « Réconciliation par module » sur LegacySyncPage.vue.
Fix fantômes à la SOURCE (scripts/targo-sync/, jusqu'ici hors repo) :
- La vraie sync des services = scripts Python hôte (/opt/targo-sync/, cron horaire),
PAS le hub. svc_status() ignorait le statut du COMPTE → un compte résilié F
(status 3/4/5) dont les services restent status=1 ressortait « Actif » (F ne
cascade pas). Résultat : ~3855 abonnements fantômes recréés à chaque heure.
- sync_services_incremental.py: svc_status() rendu conscient du compte (force
'Annulé' si compte résilié) sur Phase B (création) + Phase D (rafraîchissement).
Rollout: dry-run {Actif→Annulé: 3855} → APPLY=1 → ghost=0 stable.
- Versionne aussi run.sh + sync_invoices_incremental.py + README (snapshot fidèle
de la prod ; ces scripts écrivent en base chaque heure et n'étaient pas suivis).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
234 lines
12 KiB
Python
234 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Sync INCRÉMENTAL des factures F (gestionclient) → ERPNext — MIROIR OPÉRATIONNEL.
|
||
|
||
Insère les factures F absentes d'ERPNext (id > watermark) en REFLÉTANT leur statut F :
|
||
• Sales Invoice + Sales Invoice Item (mêmes maps que import_invoices.py → cohérence
|
||
exacte avec les ~629k déjà importées : client, SKU→compte produit, service_location).
|
||
• docstatus = 1 (submitted), status = Paid/Unpaid/Overdue/Return d'après F,
|
||
outstanding_amount = max(0, total_amt − billed_amt) (F = vérité du solde).
|
||
• SINV-{legacy_invoice_id zfill10} (idempotent : saute si déjà présent).
|
||
• PAS de GL Entry / PLE (ERPNext non autoritaire — miroir opérationnel, scheduler en pause).
|
||
|
||
LÉGER SUR F : ne lit que `WHERE id > watermark` (PK indexée) sur F LIVE. Aucun scan complet.
|
||
DRY-RUN par défaut : n'écrit RIEN sans la variable d'env APPLY=1.
|
||
|
||
Lancer (depuis l'hôte prod qui joint F live 10.100.80.100 + le PG ERPNext) :
|
||
docker cp sync_invoices_incremental.py erpnext-backend-1:/tmp/sync_inv.py
|
||
docker exec -e APPLY=0 erpnext-backend-1 /home/frappe/frappe-bench/env/bin/python /tmp/sync_inv.py # aperçu
|
||
docker exec -e APPLY=1 erpnext-backend-1 /home/frappe/frappe-bench/env/bin/python /tmp/sync_inv.py # écrit
|
||
Options env : SINCE (force le plancher d'id), LIMIT (taille du lot, 0=tout).
|
||
"""
|
||
import os
|
||
import uuid
|
||
import pymysql
|
||
import psycopg2
|
||
from datetime import datetime, timezone
|
||
from html import unescape
|
||
|
||
# F LIVE (pas le miroir périmé) — lecture bornée par id seulement.
|
||
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"
|
||
COMPANY = "TARGO"
|
||
APPLY = os.environ.get("APPLY", "0") == "1"
|
||
SINCE = int(os.environ.get("SINCE", "0"))
|
||
LIMIT = int(os.environ.get("LIMIT", "0")) # 0 = pas de limite
|
||
|
||
def uid(p=""): return p + uuid.uuid4().hex[:10]
|
||
def now(): return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")
|
||
def today(): return datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||
def ts_to_date(t):
|
||
if not t or t <= 0: return None
|
||
try: return datetime.fromtimestamp(int(t), tz=timezone.utc).strftime("%Y-%m-%d")
|
||
except Exception: return None
|
||
def clean(v): return unescape(str(v)).strip() if v else ""
|
||
def r2(v):
|
||
try: return round(float(v or 0), 2)
|
||
except Exception: return 0.0
|
||
def sinv_name(i): return "SINV-{:010d}".format(int(i))
|
||
def log(m): print(m, flush=True)
|
||
|
||
# F invoice → statut ERPNext reflétant F (= mapInvoice côté hub).
|
||
def map_status(total, billed, due, td):
|
||
is_return = total < 0
|
||
owed = 0.0 if is_return else r2(max(0.0, total - billed))
|
||
if is_return: status = "Return"
|
||
elif owed <= 0.005: status = "Paid"
|
||
elif due and due < td: status = "Overdue"
|
||
else: status = "Unpaid"
|
||
return is_return, owed, status
|
||
|
||
def main():
|
||
td = today()
|
||
ts = now()
|
||
log("=== Sync incrémental factures F→ERPNext (MIROIR OPÉRATIONNEL) ===")
|
||
log(" mode = {}".format("APPLY (écriture)" if APPLY else "DRY-RUN (aucune écriture)"))
|
||
|
||
pg = psycopg2.connect(**PG)
|
||
pgc = pg.cursor()
|
||
|
||
# watermark = max legacy_invoice_id déjà dans ERPNext (ou SINCE)
|
||
pgc.execute('SELECT COALESCE(MAX(legacy_invoice_id),0) FROM "tabSales Invoice" WHERE legacy_invoice_id > 0')
|
||
erp_max = int(pgc.fetchone()[0] or 0)
|
||
watermark = SINCE if SINCE > 0 else erp_max
|
||
log(" watermark (id F) = {}".format(watermark))
|
||
|
||
# maps ERPNext (mêmes que import_invoices.py)
|
||
pgc.execute('SELECT legacy_account_id, name, customer_name FROM "tabCustomer" WHERE legacy_account_id > 0')
|
||
cust_map = {r[0]: (r[1], r[2]) for r in pgc.fetchall()}
|
||
pgc.execute('SELECT item_code FROM "tabItem"')
|
||
valid_items = set(r[0] for r in pgc.fetchall())
|
||
pgc.execute('SELECT legacy_delivery_id, name FROM "tabService Location" WHERE legacy_delivery_id IS NOT NULL AND legacy_delivery_id <> 0')
|
||
loc_by_delivery = dict(pgc.fetchall())
|
||
pgc.execute("SELECT name FROM \"tabAccount\" WHERE account_type='Receivable' AND company=%s AND is_group=0 LIMIT 1", (COMPANY,))
|
||
receivable = pgc.fetchone()[0]
|
||
pgc.execute("SELECT account_number, name FROM \"tabAccount\" WHERE root_type IN ('Income','Expense') AND company=%s AND is_group=0 AND account_number <> ''", (COMPANY,))
|
||
gl_by_number = {r[0]: r[1] for r in pgc.fetchall()}
|
||
pgc.execute("SELECT name FROM \"tabAccount\" WHERE root_type='Income' AND company=%s AND is_group=0 LIMIT 1", (COMPANY,))
|
||
income_default = pgc.fetchone()[0]
|
||
log(" maps: {} clients, {} items, {} locations, {} comptes produit".format(len(cust_map), len(valid_items), len(loc_by_delivery), len(gl_by_number)))
|
||
|
||
# ── SYNC PAR NUMÉRO INCRÉMENTIEL, GAP-AWARE & LÉGÈRE ──
|
||
# On ne fait PAS un simple `id > max` (qui ignorerait les TROUS sous le max :
|
||
# factures sautées « sans client » qui se résolvent plus tard, inserts ratés).
|
||
# On diffe les NUMÉROS sur une fenêtre récente (id seulement = ultra-léger),
|
||
# et on ne charge/insère QUE les manquants (nouveaux au-dessus du max + trous comblés).
|
||
WINDOW = int(os.environ.get("WINDOW", "20000")) # re-vérifie les ~20k derniers numéros
|
||
floor = watermark if SINCE > 0 else max(0, watermark - WINDOW)
|
||
mc = pymysql.connect(**LEGACY)
|
||
cur = mc.cursor(pymysql.cursors.DictCursor)
|
||
# 1) numéros F dans la fenêtre (PK indexée, id seul)
|
||
cur.execute("SELECT id FROM invoice WHERE id > %s", (floor,))
|
||
f_ids = [r["id"] for r in cur.fetchall()]
|
||
# 2) numéros déjà dans ERPNext dans la fenêtre
|
||
pgc.execute('SELECT legacy_invoice_id FROM "tabSales Invoice" WHERE legacy_invoice_id > %s', (floor,))
|
||
erp_ids = set(int(r[0]) for r in pgc.fetchall())
|
||
# 3) manquants = à traiter
|
||
todo = sorted(i for i in f_ids if i not in erp_ids)
|
||
if LIMIT > 0: todo = todo[:LIMIT]
|
||
log(" fenêtre id>{} : F={} · ERP={} · à insérer (nouveaux + trous) = {}".format(floor, len(f_ids), len(erp_ids), len(todo)))
|
||
|
||
invoices = []
|
||
items_by_inv = {}
|
||
for s in range(0, len(todo), 5000):
|
||
batch = todo[s:s+5000]
|
||
if not batch: continue
|
||
ph = ",".join(["%s"]*len(batch))
|
||
cur.execute("SELECT id, account_id, date_orig, due_date, total_amt, billed_amt, billing_status FROM invoice WHERE id IN ({}) ORDER BY id".format(ph), batch)
|
||
invoices.extend(cur.fetchall())
|
||
cur.execute("SELECT ii.*, s.delivery_id AS service_delivery_id FROM invoice_item ii LEFT JOIN service s ON s.id=ii.service_id WHERE ii.invoice_id IN ({}) ORDER BY ii.invoice_id, ii.id".format(ph), batch)
|
||
for r in cur.fetchall():
|
||
items_by_inv.setdefault(r["invoice_id"], []).append(r)
|
||
# mapping SKU→compte produit (product → product_cat.num_compte)
|
||
cur.execute("SELECT p.sku, pc.num_compte FROM product p JOIN product_cat pc ON p.category=pc.id WHERE p.sku IS NOT NULL AND pc.num_compte IS NOT NULL")
|
||
sku_to_gl = {}
|
||
for r in cur.fetchall():
|
||
an = str(int(r["num_compte"])) if r["num_compte"] else None
|
||
if an and an in gl_by_number: sku_to_gl[r["sku"]] = gl_by_number[an]
|
||
mc.close()
|
||
existing = set() # `todo` exclut déjà les présents ; ON CONFLICT couvre toute course
|
||
|
||
by_status = {"Paid": 0, "Unpaid": 0, "Overdue": 0, "Return": 0}
|
||
ok = skip = no_cust = item_ok = 0
|
||
samples = []
|
||
|
||
for inv in invoices:
|
||
if inv["id"] in existing: skip += 1; continue
|
||
cd = cust_map.get(inv["account_id"])
|
||
if not cd: no_cust += 1; continue
|
||
cust_name, cust_disp = cd
|
||
total = r2(inv["total_amt"]); billed = r2(inv["billed_amt"])
|
||
posting = ts_to_date(inv["date_orig"]) or td
|
||
due = ts_to_date(inv["due_date"]) or posting
|
||
is_return, owed, status = map_status(total, billed, due, td)
|
||
by_status[status] += 1
|
||
name = sinv_name(inv["id"])
|
||
if len(samples) < 8:
|
||
samples.append({"sinv": name, "status": status, "total": total, "outstanding": owed})
|
||
|
||
if not APPLY:
|
||
ok += 1
|
||
item_ok += len(items_by_inv.get(inv["id"], []))
|
||
continue
|
||
|
||
try:
|
||
pgc.execute("""
|
||
INSERT INTO "tabSales Invoice" (
|
||
name, creation, modified, modified_by, owner, docstatus, idx,
|
||
naming_series, customer, customer_name, company,
|
||
posting_date, due_date, currency, conversion_rate,
|
||
selling_price_list, price_list_currency,
|
||
base_grand_total, grand_total, base_net_total, net_total, base_total, total,
|
||
outstanding_amount, base_rounded_total, rounded_total,
|
||
is_return, is_debit_note, disable_rounded_total,
|
||
debit_to, party_account_currency, status, legacy_invoice_id
|
||
) VALUES (%s,%s,%s,%s,%s,1,0,
|
||
'ACC-SINV-.YYYY.-',%s,%s,%s,
|
||
%s,%s,'CAD',1,
|
||
'Standard Selling','CAD',
|
||
%s,%s,%s,%s,%s,%s,
|
||
%s,%s,%s,
|
||
%s,0,1,
|
||
%s,'CAD',%s,%s)
|
||
ON CONFLICT (name) DO NOTHING
|
||
""", (name, ts, ts, ADMIN, ADMIN,
|
||
cust_name, cust_disp, COMPANY,
|
||
posting, due,
|
||
total, total, total, total, total, total,
|
||
owed, total, total,
|
||
1 if is_return else 0,
|
||
receivable, status, inv["id"]))
|
||
|
||
for j, li in enumerate(items_by_inv.get(inv["id"], [])):
|
||
sku = clean(li.get("sku")) or "MISC"
|
||
qty = float(li.get("quantity") or 1)
|
||
rate = float(li.get("unitary_price") or 0)
|
||
amount = round(qty * rate, 2)
|
||
desc = (clean(li.get("product_name")) or sku)[:140]
|
||
item_code = sku if sku in valid_items else None
|
||
income = sku_to_gl.get(sku, income_default)
|
||
sloc = loc_by_delivery.get(li.get("service_delivery_id"))
|
||
pgc.execute("""
|
||
INSERT INTO "tabSales Invoice Item" (
|
||
name, creation, modified, modified_by, owner, docstatus, idx,
|
||
item_code, item_name, description, qty, rate, amount,
|
||
base_rate, base_amount, base_net_rate, base_net_amount, net_rate, net_amount,
|
||
stock_uom, uom, conversion_factor, income_account, cost_center, service_location,
|
||
parent, parentfield, parenttype
|
||
) VALUES (%s,%s,%s,%s,%s,1,%s,
|
||
%s,%s,%s,%s,%s,%s,
|
||
%s,%s,%s,%s,%s,%s,
|
||
'Nos','Nos',1,%s,'Main - T',%s,
|
||
%s,'items','Sales Invoice')
|
||
ON CONFLICT (name) DO NOTHING
|
||
""", (uid("SII-"), ts, ts, ADMIN, ADMIN, j+1,
|
||
item_code, desc, desc, qty, rate, amount,
|
||
rate, amount, rate, amount, rate, amount,
|
||
income, sloc, name))
|
||
item_ok += 1
|
||
ok += 1
|
||
except Exception as e:
|
||
pg.rollback()
|
||
log(" ERR inv#{} -> {}".format(inv["id"], str(e)[:160]))
|
||
continue
|
||
if ok % 2000 == 0:
|
||
pg.commit(); log(" ... {} insérées".format(ok))
|
||
|
||
if APPLY: pg.commit()
|
||
pg.close()
|
||
log("")
|
||
log("="*60)
|
||
log("Statut cible (reflète F) : {}".format(by_status))
|
||
log("{} : {} factures, {} lignes, {} déjà présentes, {} sans client".format(
|
||
"INSÉRÉES" if APPLY else "À INSÉRER (dry-run)", ok, item_ok, skip, no_cust))
|
||
log("Échantillon : {}".format(samples))
|
||
log("="*60)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|