# -*- coding: utf-8 -*- """ ETL Gibili / GibShip — extraction des caisses Excel vers un JSON normalisé. Sources : - Caisse SN.xlsx (Sénégal, XOF, 1 feuille par mois) - caisse gibli -Cheikhatou.xlsx (Mauritanie, MRU, 1 feuille par mois) - GibShip_Depenses_Senegal.xlsx (synthèse mensuelle des dépenses, 2026) Sortie : data/gibili_finance.json Usage : python etl/parse_caisse.py """ import openpyxl, re, json, sys, os, datetime, unicodedata from collections import defaultdict sys.stdout.reconfigure(encoding="utf-8") BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) OUT = os.path.join(BASE, "data", "gibili_finance.json") MONTHS_FR = {"janvier":1,"fevrier":2,"février":2,"mars":3,"avril":4,"mai":5,"juin":6,"juillet":7, "aout":8,"août":8,"aoùt":8,"septembre":9,"octobre":10,"novembre":11,"decembre":12,"décembre":12} def norm(s): s = unicodedata.normalize("NFKD", str(s)).encode("ascii","ignore").decode().lower().strip() return re.sub(r"\s+"," ",s) def is_num(v): return isinstance(v,(int,float)) and not isinstance(v,bool) def is_date(v): return isinstance(v,(datetime.datetime,datetime.date)) def is_formula(v): return isinstance(v,str) and v.startswith("=") def is_total(v): """Formule de total (SUM / référence de plage) — à exclure des données ; '=12800+15300' est une donnée.""" return is_formula(v) and (":" in v or re.search(r"[A-Z]{1,3}\d+", v) is not None) # ---------------------------------------------------------------- month detection def month_from_sheet_sn(title): t = norm(title) m = re.match(r"([a-z]+)\s*(\d{4})", t) if not m: return None mo = MONTHS_FR.get(m.group(1)); return (int(m.group(2)), mo) if mo else None def month_from_sheet_mr(title, ws): t = norm(title) m = re.search(r"(\d{2})(\d{2})$", t) # caisse 0926 -> 09/2026 if m: return (2000+int(m.group(2)), int(m.group(1))) m = re.search(r"(\d{1,2})\s*$", t) if not m: return None mo = int(m.group(1)) # année : déduite des dates de la colonne A years = [c.value.year for c in ws["A"] if is_date(c.value)] yr = max(set(years), key=years.count) if years else 2026 return (yr, mo) # ---------------------------------------------------------------- generic helpers def texts_around(ws, r, col, left=2, right=1): out=[] for cc in range(max(1,col-left), col+right+1): v = ws.cell(r,cc).value if isinstance(v,str) and not is_formula(v.strip()) and v.strip(): out.append(v.strip()) return out def date_around(ws, r, col, left=3, right=1): for cc in range(max(1,col-left), col+right+1): v = ws.cell(r,cc).value if is_date(v): return v return None def find_cells(ws, pred, max_row=120, max_col=60): res=[] for row in ws.iter_rows(min_row=1, max_row=max_row, max_col=max_col): for c in row: if pred(c.value): res.append(c) return res def classify_block(headers): h = " ".join(norm(x) for x in headers) if "expedition" in h: return "expedition" if "salair" in h: return "salaire" if "sac n" in h: return "sac" if "motif" in h or "retrait" in h: return "retrait" if "avance" in h or "avence" in h: return "avance" if "prix" in h or "articles" in h or "annulation" in h: return "retour" if "agent" in h: return "avance" return "autre_bloc" # ---------------------------------------------------------------- main-area parsers def parse_main_rows(wsv, wsf, layout, country, year, month): """Parcourt toutes les lignes ; retourne ventes + dépenses journalières.""" sales, ops, reports = [], [], [] cur_day = None L = layout for r in range(1, wsv.max_row+1): a = wsv.cell(r,1).value b = wsv.cell(r,2).value if isinstance(b,str) and norm(b) in ("recette","produit"): # ligne d'en-tête continue if is_date(a): cur_day = a.day if cur_day is None: continue day = min(cur_day, 28) if False else cur_day try: date = datetime.date(year, month, day) except ValueError: date = datetime.date(year, month, 28) # ---- vente amt = wsv.cell(r, L["montant"]).value if is_num(amt) and not is_total(wsf.cell(r, L["montant"]).value) and amt != 0: prod = wsv.cell(r, L["produit"]).value oid = wsv.cell(r, L["id"]).value if prod in (None,"") and isinstance(oid,str) and oid.strip(): # report de caisse du mois précédent saisi dans la colonne ventes (ex. 'caisse 06') reports.append(dict(country=country, month=f"{year}-{month:02d}", block="report", amount=float(amt), label=oid.strip(), date=date.isoformat(), cell=wsv.cell(r, L["montant"]).coordinate)) elif isinstance(prod,str) or is_num(oid): wal = wsv.cell(r, L["wallet"]).value wallet_amt = wal if is_num(wal) else (amt if isinstance(wal,str) and wal.strip() else 0) wallet_flag = bool(wallet_amt) or (isinstance(wal,str) and bool(wal.strip())) txn = wsv.cell(r, L["txn"]).value sales.append(dict(country=country, date=date.isoformat(), month=f"{year}-{month:02d}", product=(prod.strip() if isinstance(prod,str) else ""), amount=float(amt), order_id=int(oid) if is_num(oid) else None, wallet=bool(wallet_flag), wallet_amount=float(wallet_amt or 0), txn=str(txn).strip() if txn not in (None,"") else None)) # ---- dépenses journalières (credit / essence / mecanique / autres / transport) for kind, (lc, ac, extra) in L["ops"].items(): v = wsv.cell(r, ac).value if is_num(v) and v != 0 and not is_total(wsf.cell(r, ac).value): lab = wsv.cell(r, lc).value ex = wsv.cell(r, extra).value if extra else None ops.append(dict(country=country, date=date.isoformat(), month=f"{year}-{month:02d}", kind=kind, label=(str(lab).strip() if lab not in (None,"") else ""), amount=float(v), ref=(int(ex) if is_num(ex) else (str(ex).strip() if ex else None)))) return sales, ops, reports LAYOUT_SN = dict(produit=2, montant=3, id=4, wallet=5, txn=6, ops={"credit":(8,9,None), "autres":(10,11,None), "transport":(12,13,None)}) LAYOUT_MR = dict(produit=2, montant=3, id=4, wallet=5, txn=6, ops={"credit":(7,8,None), "essence":(9,10,11), "mecanique":(12,13,None), "autres":(14,15,None), "transport":(16,17,18)}) LAYOUT_MR_OLD = dict(produit=2, montant=3, id=4, wallet=5, txn=7, ops={"credit":(8,9,None), "essence":(10,11,12), "mecanique":(13,14,None), "autres":(15,16,None), "transport":(17,18,19)}) # ---------------------------------------------------------------- side blocks def parse_block_range(wsv, wsf, col, r1, r2, kind, country, ym): rows=[] for r in range(r1, r2+1): v = wsv.cell(r,col).value if not is_num(v) or is_total(wsf.cell(r,col).value) or v == 0: continue labels = texts_around(wsv, r, col, left=2, right=1) rows.append(dict(country=country, month=ym, block=kind, amount=float(v), label=" / ".join(labels), date=(date_around(wsv,r,col).date().isoformat() if date_around(wsv,r,col) else None), cell=f"{wsv.cell(r,col).coordinate}")) return rows def parse_returns(wsv, wsf, country, ym, layout_mr=False): """Blocs 'Date delivered | Date annulation | Articles | Prix | ID | wave | transaction'.""" res=[] hdrs = find_cells(wsv, lambda v: isinstance(v,str) and norm(v) in ("date delivered","delivary at")) for h in hdrs: c0, r0 = h.column, h.row # bloc précédent ? cherche 'Retour mois precedent' / 'Mois precedents' au-dessus above = " ".join(norm(x) for rr in range(max(1,r0-8), r0) for x in texts_around(wsv, rr, c0+3, 4, 4)) prev = "preced" in above for r in range(r0+1, r0+80): dd, da, art, prix, oid = (wsv.cell(r,c0+k).value for k in range(5)) wal = wsv.cell(r,c0+5).value if art is None and prix is None and dd is None and da is None: # fin de bloc si 2 lignes vides consécutives nxt = [wsv.cell(r+1,c0+k).value for k in range(6)] if all(x is None for x in nxt): break continue if isinstance(art,str) and norm(art) in ("articles","product"): break if isinstance(prix,str) or is_formula(wsf.cell(r,c0+3).value): continue if not (is_num(prix) or is_num(wal)): continue res.append(dict(country=country, month=ym, previous_month=prev, delivered=dd.date().isoformat() if is_date(dd) else None, cancelled=da.date().isoformat() if is_date(da) else None, product=str(art).strip() if art else "", price=float(prix) if is_num(prix) else 0.0, order_id=int(oid) if is_num(oid) else None, wallet_amount=float(wal) if is_num(wal) else 0.0)) return res # ---------------------------------------------------------------- balance formula def parse_balance(wsv, wsf, country, ym, main_cols): """Lit la formule 'Solde Caisse' et retourne les termes + lignes des blocs latéraux.""" cand=None for row in wsf.iter_rows(min_row=1, max_row=8, max_col=60): for c in row: if is_formula(c.value) and (c.value.count("-")+c.value.count("+"))>=3: if cand is None or len(c.value)>len(cand.value): cand=c if cand is None: return None, [] terms=[]; block_rows=[] for sign, ref in re.findall(r"([+-]?)\s*([A-Z]{1,3}\d+)", cand.value.lstrip("=")): sign = sign or "+" cf = wsf[ref].value; cv = wsv[ref].value; col = wsf[ref].column; r = wsf[ref].row rng=None if is_formula(cf): m = re.match(r"=\s*sum\(\$?([A-Z]+)\$?(\d+):\$?([A-Z]+)\$?(\d+)\)", cf, re.I) if m: rng=(openpyxl.utils.column_index_from_string(m.group(1)), int(m.group(2)), int(m.group(4))) hdr=[] rr = rng[1] if rng else r HK = ("avance","avence","montant","montan","salair","expedition","motif","retrait","prix","sac n","agent","articles") for up in range(rr, max(0,rr-8), -1): t = texts_around(wsf, up, col, 3, 1) if not t: continue if up == rr and (is_num(wsv.cell(rr,col).value) or not any(k in norm(x) for x in t for k in HK)): continue # 1re ligne = donnée, pas en-tête hdr=t; break kind=None if rng and rng[0] in main_cols: kind = "main:"+main_cols[rng[0]] elif rng: kind = classify_block(hdr) if kind == "autre_bloc": # colonne 'wave'/'Bankily' d'un bloc Retour (en-tête 'Date delivered' dans la plage ou plus haut, à gauche) for up in range(rng[2], max(0, rng[1]-40), -1): t = " ".join(norm(x) for x in texts_around(wsf, up, col, 7, 1)) if "date delivered" in t or "delivary" in t or "mois preced" in t: kind = "retour"; break else: kind = "scalar" val = cv if is_num(cv) else 0.0 terms.append(dict(sign=sign, ref=ref, value=float(val), kind=kind, header=hdr, range=rng)) if rng and not kind.startswith("main:") and kind!="retour": block_rows += parse_block_range(wsv, wsf, rng[0], rng[1], rng[2], kind, country, ym) return dict(cell=cand.coordinate, formula=cand.value, value=float(wsv[cand.coordinate].value or 0), terms=terms), block_rows # ---------------------------------------------------------------- old MR layout (2025-08..10) def parse_mr_old_side(wsv, wsf, country, ym): rows=[] for h in find_cells(wsv, lambda v: isinstance(v,str) and norm(v).startswith("retrait date"), max_row=10, max_col=70): col = h.column+1 for r in range(h.row+1, h.row+400): v = wsv.cell(r,col).value if v is None and wsv.cell(r,col-1).value is None and wsv.cell(r+1,col).value is None: break if is_num(v) and v!=0 and not is_total(wsf.cell(r,col).value): d = wsv.cell(r,col-1).value; motif = wsv.cell(r,col+1).value rows.append(dict(country=country, month=ym, block="retrait", amount=float(v), label=str(motif).strip() if motif else "", date=d.date().isoformat() if is_date(d) else None, cell=wsv.cell(r,col).coordinate)) for h in find_cells(wsv, lambda v: isinstance(v,str) and norm(v) in ("avence","avance"), max_row=12, max_col=70): col = h.column+1 if not isinstance(wsv.cell(h.row, col).value,str): continue blank=0 for r in range(h.row+1, h.row+30): v = wsv.cell(r,col).value if v is None and wsv.cell(r,col-1).value is None: blank+=1 if blank>=2: break continue blank=0 if is_num(v) and v!=0 and not is_total(wsf.cell(r,col).value): d = wsv.cell(r,col-2).value; who = wsv.cell(r,col-1).value rows.append(dict(country=country, month=ym, block="avance", amount=float(v), label=str(who).strip() if who else "", date=d.date().isoformat() if is_date(d) else None, cell=wsv.cell(r,col).coordinate)) return rows # ---------------------------------------------------------------- catégorisation des blocs CAT_RULES = [ ("report_caisse", r"caiss|restent|reste de|sold caiss|solde caiss|sold negative|place en caisse|mois passer"), ("remise_direction",r"sidaty|sidty|sdaty|moulaye\(devise\)|moulaye5|devis|verssement|versement|bankili|bimbank|masrivi|sedad|^moulaye$|moulaye\+vrais|avance moulaye|augmatation moulaye|agmantation|augmantation"), ("loyer", r"location|sa7b|dar|maison\)|^maison$"), ("logistique_import", r"transit|yango|colis|ravg|cargo|sac livrer|embellage|expedition|livresion|livraison"), ("frais_paiement", r"frais|vrais wave|taxe|taxt|orange mon"), ("utilites", r"courant|sen.?e?au|senau|conne?xion|conextion|conexion|wifi|pointel|gaz|internet"), ("vie_equipe", r"depan|depen|alimentation|medic|preperation|nourriture|tea|water"), ("equipement", r"ecran|telephone|ordinateur|ordenateur|reparation|reperation|casque|moto|mecanique|vidange|kama|carbirah|karbirah"), ("prime", r"motivation|prime|bonus"), ("ecart_caisse", r"perdi|perdu|mankage|manque|argent perd"), ("salaire", r"salair|salaire"), ] def categorize(block, label, amount=0): l = norm(label or "") if block in ("report","scalar"): return "apport_caisse" if ("alimentation" in l or "agmantation" in l) else "report_caisse" if block == "retrait" and not l and amount < 0: return "report_caisse" if block == "salaire": return "salaire" if block == "expedition": return "logistique_import" for cat, rx in CAT_RULES: if re.search(rx, l): return cat if block == "avance": return "avance_personnel" if block == "retrait": return "personnel" if l else "autres" return "autres" # ---------------------------------------------------------------- run def run(): data = dict(generated=datetime.datetime.now().isoformat(timespec="seconds"), sales=[], ops=[], blocks=[], returns=[], months=[], staff_table=[], depenses_synthese=[]) # ===== Sénégal fn = os.path.join(BASE, "Caisse SN.xlsx") wbf = openpyxl.load_workbook(fn); wbv = openpyxl.load_workbook(fn, data_only=True) for wsf in wbf.worksheets: ym = month_from_sheet_sn(wsf.title) if not ym: continue wsv = wbv[wsf.title]; year, month = ym; key=f"{year}-{month:02d}" sales, ops, reports = parse_main_rows(wsv, wsf, LAYOUT_SN, "SN", year, month) bal, brows = parse_balance(wsv, wsf, "SN", key, {3:"ventes",5:"wallet",9:"credit",11:"autres",13:"transport"}) brows += reports rets = parse_returns(wsv, wsf, "SN", key) data["sales"]+=sales; data["ops"]+=ops; data["blocks"]+=brows; data["returns"]+=rets data["months"].append(dict(country="SN", month=key, sheet=wsf.title, balance=bal, n_sales=len(sales), n_ops=len(ops), n_blocks=len(brows), n_returns=len(rets))) print(f"SN {key:8} {wsf.title:16} ventes={len(sales):4} ops={len(ops):4} blocs={len(brows):3} retours={len(rets):3} solde={bal['value'] if bal else None}") # table Salaires (référentiel) ws = wbv["Salaires"] for r in range(2, ws.max_row+1): nom = ws.cell(r,1).value if not nom or norm(nom)=="total": continue data["staff_table"].append(dict(nom=nom, fonction=ws.cell(r,2).value, heures=ws.cell(r,3).value, commandes=ws.cell(r,4).value, gains=ws.cell(r,5).value, avance=ws.cell(r,6).value, base=ws.cell(r,7).value, wave=ws.cell(r,9).value)) # ===== Mauritanie fn = os.path.join(BASE, "caisse gibli -Cheikhatou.xlsx") wbf = openpyxl.load_workbook(fn); wbv = openpyxl.load_workbook(fn, data_only=True) skip = {"retour","bah","dine","issa","khattar","hamdy","hmettou","mouslih","boutar","yessar","hamadi","ndb","mode caisse","caisse 08_monthly"} for wsf in wbf.worksheets: if norm(wsf.title) in skip: continue wsv = wbv[wsf.title] ym = month_from_sheet_mr(wsf.title, wsv) if not ym: continue year, month = ym; key=f"{year}-{month:02d}" old = norm(wsv.cell(2,2).value or "") == "produit" sales, ops, reports = parse_main_rows(wsv, wsf, LAYOUT_MR_OLD if old else LAYOUT_MR, "MR", year, month) if old: bal=None; brows = parse_mr_old_side(wsv, wsf, "MR", key) else: bal, brows = parse_balance(wsv, wsf, "MR", key, {3:"ventes",5:"wallet",8:"credit",10:"essence",13:"mecanique",15:"autres",17:"transport"}) brows += reports rets = parse_returns(wsv, wsf, "MR", key, layout_mr=True) data["sales"]+=sales; data["ops"]+=ops; data["blocks"]+=brows; data["returns"]+=rets data["months"].append(dict(country="MR", month=key, sheet=wsf.title, balance=bal, old_layout=old, n_sales=len(sales), n_ops=len(ops), n_blocks=len(brows), n_returns=len(rets))) print(f"MR {key:8} {wsf.title:16} ventes={len(sales):4} ops={len(ops):4} blocs={len(brows):3} retours={len(rets):3} solde={bal['value'] if bal else None}") # ===== Synthèse dépenses GibShip fn = os.path.join(BASE, "GibShip_Depenses_Senegal.xlsx") ws = openpyxl.load_workbook(fn, data_only=True)["Depenses mensuelles"] country=None; hdr=None for r in range(1, ws.max_row+1): a = ws.cell(r,1).value; d = ws.cell(r,4).value if isinstance(d,str) and norm(d) in ("senegal","mauritanie"): country = "SN" if norm(d)=="senegal" else "MR"; continue if a=="Mois": hdr=[ws.cell(r,c).value for c in range(1,14)]; continue if hdr and isinstance(a,str) and norm(a) in MONTHS_FR and country: vals=[ws.cell(r,c).value for c in range(1,14)] if not any(is_num(v) and v for v in vals[1:9]): continue rec=dict(country=country, month=f"2026-{MONTHS_FR[norm(a)]:02d}") for h,v in zip(hdr[1:], vals[1:]): if h: rec[norm(h)] = v if is_num(v) else (v or None) data["depenses_synthese"].append(rec) for b in data["blocks"]: b["category"] = categorize(b["block"], b["label"], b["amount"]) # termes scalaires de la formule (report du mois précédent, ajustements) for m in data["months"]: if not m["balance"]: continue for t in m["balance"]["terms"]: if t["kind"]=="scalar" and t["value"]: lab = " / ".join(t["header"]) or "ajustement" data["blocks"].append(dict(country=m["country"], month=m["month"], block="scalar", amount=t["value"]*(1 if t["sign"]=="+" else -1), label=lab, date=None, cell=t["ref"], category=categorize("scalar", lab))) os.makedirs(os.path.dirname(OUT), exist_ok=True) with open(OUT,"w",encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, default=str) print("\n->", OUT, f"{os.path.getsize(OUT)/1024:.0f} KB", f"| ventes={len(data['sales'])} ops={len(data['ops'])} blocs={len(data['blocks'])} retours={len(data['returns'])}") if __name__ == "__main__": run() # ---------------------------------------------------------------- export compact pour le tableau de bord def export_dashboard(data, out_js): """Écrit data/gibili_data.js (window.GIBILI_DATA) : tableaux compacts pour la page HTML.""" products = {}; plist = [] def pid(name): key = re.sub(r"^\s*(sn|sng|mr)\s*-\s*", "", name.strip(), flags=re.I).strip() or "(sans nom)" if key not in products: products[key] = len(plist); plist.append(key) return products[key] sales = [[s["country"], s["date"], pid(s["product"]), s["amount"], s["order_id"], round(s["wallet_amount"]) if s["wallet"] else 0] for s in data["sales"]] ops = [[o["country"], o["date"], o["kind"], o["amount"], o["label"], o["ref"]] for o in data["ops"]] blocks= [[b["country"], b["month"], b["block"], b["category"], b["amount"], b["label"], b["date"]] for b in data["blocks"]] rets = [[r["country"], r["month"], int(r["previous_month"]), r["delivered"], r["cancelled"], pid(r["product"]), r["price"], r["order_id"], r["wallet_amount"]] for r in data["returns"]] months= [dict(c=m["country"], m=m["month"], sheet=m["sheet"].strip(), solde=(m["balance"]["value"] if m["balance"] else None), retour_excel=sum(t["value"]*(1 if t["sign"]=="-" else -1) for t in m["balance"]["terms"] if t["kind"]=="retour") if m["balance"] else None, wallet_excel=sum(t["value"] for t in m["balance"]["terms"] if t["kind"]=="main:wallet") if m["balance"] else 0) for m in data["months"]] payload = dict(generated=data["generated"], products=plist, sales=sales, ops=ops, blocks=blocks, returns=rets, months=months, staff=data["staff_table"], synthese=data["depenses_synthese"]) with open(out_js, "w", encoding="utf-8") as f: f.write("window.GIBILI_DATA = "); json.dump(payload, f, ensure_ascii=False, separators=(",",":"), default=str); f.write(";\n") print("->", out_js, f"{os.path.getsize(out_js)/1024:.0f} KB") if __name__ == "__main__": export_dashboard(json.load(open(OUT, encoding="utf-8")), os.path.join(BASE, "data", "gibili_data.js"))