301cc75de9
3 browser-native chat surfaces (atlas-chat / atlas-hud / atlas-command, no build step) + 11 Python automation scripts (cluster, tray, finance, reminders, autonomy loops) + 2 patterns documented (cluster-then-polish, anti-impulse R1-R10) MIT licensed, anonymized from production stack.
181 lines
6.5 KiB
Python
181 lines
6.5 KiB
Python
#!/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@<YOUR_VPS_IP>"
|
|
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()
|