#!/usr/bin/env python3 """ atlas_finance_export.py - Parse Inkomsten xlsx + compute YTD/MRR/concentration, output atlas-finance.json and scp to atlas-01 dashboard. Run manually or via schtask (Atlas-Finance-Export, daily). """ import json import sys import time import subprocess from pathlib import Path from datetime import datetime, date from collections import defaultdict try: import openpyxl except ImportError: print("openpyxl required"); sys.exit(1) XLSX = Path(r"C:\Users\Gebruiker\MEGA\Atlas Corporation\07_Finance\2026\Inkomsten uit onderneming 2026.xlsx") OUT_LOCAL = Path(r"C:\Users\Gebruiker\AppData\Local\Atlas\analyses\atlas-finance.json") SSH = "root@" REMOTE = "/opt/atlas/dashboard/atlas-finance.json" DUTCH_MONTH = { "jan": 1, "januari": 1, "febr": 2, "feb": 2, "februari": 2, "mrt": 3, "maart": 3, "apr": 4, "april": 4, "mei": 5, "juni": 6, "jun": 6, "juli": 7, "jul": 7, "aug": 8, "augustus": 8, "sept": 9, "sep": 9, "september": 9, "okt": 10, "oktober": 10, "nov": 11, "november": 11, "dec": 12, "december": 12, } def parse_xlsx(p): """Read Betaalcontrole sheet — canonical paid-invoices list with bank-verified amounts.""" wb = openpyxl.load_workbook(p, data_only=True, read_only=True) if "Betaalcontrole" not in wb.sheetnames: return [] sheet = wb["Betaalcontrole"] rows = [] headers = None for row in sheet.iter_rows(values_only=True): if not headers: headers = [str(c).strip().lower() if c else "" for c in row] continue if not any(row): continue d = dict(zip(headers, row)) rows.append(d) return rows def to_float(v): if v is None: return 0.0 if isinstance(v, (int, float)): return float(v) s = str(v).replace("€", "").replace(",", ".").replace(" ", "").strip() try: return float(s) except: return 0.0 def to_date(v): if isinstance(v, (datetime, date)): return v if isinstance(v, date) and not isinstance(v, datetime) else v.date() if isinstance(v, str): for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y", "%Y/%m/%d"): try: return datetime.strptime(v.strip(), fmt).date() except: continue return None def normalize(rows): """Normalize Betaalcontrole rows: factuur, maand in rapport, klant, excl btw, incl/bankbedrag, betaalbewijs, actie.""" out = [] for r in rows: # Skip rows without invoice number (= template/blank) factuur = (r.get("factuur") or "").strip() if r.get("factuur") else "" if not factuur or not factuur.startswith("INV"): continue # Skip rows marked as not paid or cancelled actie = (r.get("actie") or "").lower() if actie and ("niet" in actie or "geannul" in actie or "credit" in actie or "open" in actie): continue month_str = (r.get("maand in rapport") or "").strip().lower() month_num = DUTCH_MONTH.get(month_str) if not month_num: continue # Build date as 15th of month (since exact paid date may not be in xlsx — bewijs column has it) from datetime import date as _date d = _date(2026, month_num, 15) client = (r.get("klant") or "?").strip()[:60] # Use "incl/bankbedrag" (actual money received), fall back to excl btw amount = to_float(r.get("incl/bankbedrag") or r.get("excl btw")) if amount <= 0: continue out.append({ "date": d.isoformat(), "month_num": month_num, "factuur": factuur, "client": client, "amount": round(amount, 2), }) return out def aggregate(payments): """Compute YTD, monthly, top clients, MRR estimate.""" today = date.today() ytd = 0 by_month = defaultdict(float) by_client = defaultdict(float) by_client_count = defaultdict(int) last_3mo_by_client = defaultdict(float) for p in payments: d = datetime.fromisoformat(p["date"]).date() if d.year != today.year: continue ytd += p["amount"] by_month[d.strftime("%Y-%m")] += p["amount"] by_client[p["client"]] += p["amount"] by_client_count[p["client"]] += 1 days_ago = (today - d).days if days_ago <= 90: last_3mo_by_client[p["client"]] += p["amount"] sorted_clients = sorted(by_client.items(), key=lambda x: x[1], reverse=True) top_5 = sorted_clients[:5] total_top = sum(a for _, a in top_5) concentration = round(top_5[0][1] / ytd * 100, 1) if top_5 and ytd > 0 else 0 # MRR estimate: clients who paid 2+ times last 3 months mrr_eligible = {k: v for k, v in last_3mo_by_client.items() if by_client_count[k] >= 2} mrr_est = round(sum(mrr_eligible.values()) / 3, 2) if mrr_eligible else 0 months_in_year = today.month avg_per_month = round(ytd / months_in_year, 2) if months_in_year else 0 return { "generated_at": datetime.now().isoformat(timespec="seconds"), "as_of": today.isoformat(), "ytd_eur": round(ytd, 2), "ytd_count": sum(by_client_count.values()), "unique_clients_ytd": len(by_client), "avg_per_month_eur": avg_per_month, "current_month_eur": round(by_month.get(today.strftime("%Y-%m"), 0), 2), "by_month": dict(sorted(by_month.items())), "top_5_clients": [{"name": k, "amount": round(v, 2), "share_pct": round(v/ytd*100, 1) if ytd else 0} for k, v in top_5], "concentration_top_pct": concentration, "mrr_estimate_eur": mrr_est, "mrr_eligible_clients": len(mrr_eligible), } def main(): if not XLSX.exists(): print(f"missing: {XLSX}"); sys.exit(1) rows = parse_xlsx(XLSX) payments = normalize(rows) print(f"parsed {len(rows)} rows, {len(payments)} usable payments") summary = aggregate(payments) OUT_LOCAL.parent.mkdir(parents=True, exist_ok=True) OUT_LOCAL.write_text(json.dumps(summary, indent=2), encoding="utf-8") print(f"wrote {OUT_LOCAL}") print(f" ytd: €{summary['ytd_eur']:.2f} ({summary['ytd_count']} payments)") print(f" unique clients: {summary['unique_clients_ytd']}") print(f" top client: {summary['top_5_clients'][0]['name'] if summary['top_5_clients'] else '?'} ({summary['concentration_top_pct']}%)") print(f" MRR est: €{summary['mrr_estimate_eur']}") # scp to atlas-01 try: subprocess.run(["scp", str(OUT_LOCAL), f"{SSH}:{REMOTE}"], check=True, timeout=20) print(f"pushed to {SSH}:{REMOTE}") except Exception as e: print(f"scp failed: {e}") if __name__ == "__main__": main()