#!/usr/bin/env python3
"""ZZP Dashboard — JG Consultancy"""
import os, sqlite3, hashlib, secrets, json, time, ssl, urllib.request
from datetime import datetime, timedelta
from functools import wraps
from flask import Flask, request, session, redirect, url_for, render_template_string, flash, g, jsonify
from werkzeug.security import generate_password_hash, check_password_hash
# ── Config ──────────────────────────────────────────────────────────────────
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, 'data')
DB_PATH = os.path.join(DATA_DIR, 'zzp.db')
os.makedirs(DATA_DIR, exist_ok=True)
app = Flask(__name__)
app.config['APPLICATION_ROOT'] = '/zzp'
app.secret_key = secrets.token_hex(32)
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=8)
app.config['SESSION_COOKIE_PATH'] = '/zzp'
# Colors
C = dict(primary='#003366', accent='#3B82F6', success='#00A859', warning='#F59E0B',
text='#1F2937', bg='#F3F4F6', white='#FFFFFF', border='#E5E7EB', danger='#EF4444')
# ── Database ────────────────────────────────────────────────────────────────
def get_db():
if 'db' not in g:
g.db = sqlite3.connect(DB_PATH)
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA journal_mode=WAL")
return g.db
@app.teardown_appcontext
def close_db(exc):
db = g.pop('db', None)
if db: db.close()
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE TABLE IF NOT EXISTS timesheets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
client TEXT NOT NULL,
activity TEXT NOT NULL,
hours REAL NOT NULL,
billable INTEGER DEFAULT 1,
note TEXT DEFAULT '',
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS invoices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
invoice_number TEXT UNIQUE NOT NULL,
client TEXT NOT NULL,
period_year INTEGER,
period_month INTEGER,
invoice_date TEXT,
due_date TEXT,
subtotal REAL DEFAULT 0,
btw REAL DEFAULT 0,
total REAL DEFAULT 0,
status TEXT DEFAULT 'Concept',
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS invoice_lines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
invoice_id INTEGER REFERENCES invoices(id),
description TEXT,
hours REAL,
rate REAL,
amount REAL
);
CREATE TABLE IF NOT EXISTS expenses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
category TEXT NOT NULL,
description TEXT,
amount_excl REAL NOT NULL,
btw REAL DEFAULT 0,
total REAL NOT NULL,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS btw_quarters (
id INTEGER PRIMARY KEY AUTOINCREMENT,
year INTEGER,
quarter INTEGER,
status TEXT DEFAULT 'Nog doen',
filed_at TEXT
);
CREATE TABLE IF NOT EXISTS travel_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
client TEXT DEFAULT '',
from_location TEXT DEFAULT '',
to_location TEXT DEFAULT '',
km REAL NOT NULL,
rate_per_km REAL DEFAULT 0.23,
amount REAL NOT NULL,
transport_type TEXT DEFAULT 'Auto',
purpose TEXT DEFAULT '',
billable INTEGER DEFAULT 1,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
""")
# Seed admin user
pw_hash = generate_password_hash('Admin123!')
c.execute("INSERT OR IGNORE INTO users (username, password_hash) VALUES (?,?)", ('admin', pw_hash))
# Seed settings
defaults = {
'company_name': 'JG Consultancy',
'company_address': '', 'company_kvk': '', 'company_btw': '', 'company_bank': '',
'client_phoenix_metals_address': '', 'client_club_of_engineers_address': '',
'rate_phoenix_metals': '100', 'rate_club_of_engineers': '100', 'rate_overig': '85',
'payment_terms': '30',
'travel_rate': '0.23', 'client_overig_address': ''
}
for k, v in defaults.items():
c.execute("INSERT OR IGNORE INTO settings (key, value) VALUES (?,?)", (k, v))
conn.commit()
conn.close()
# ── Auth ────────────────────────────────────────────────────────────────────
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if 'user' not in session:
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated
# ── Helpers ─────────────────────────────────────────────────────────────────
def get_setting(key, default=''):
r = get_db().execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
return r['value'] if r else default
def set_setting(key, value):
get_db().execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?,?)", (key, value))
get_db().commit()
def fmt_eur(v): return f"€ {v:,.2f}".replace(",","X").replace(".",",").replace("X",".")
def client_prefix(client):
mapping = {'Phoenix Metals': 'PM', 'Club of Engineers': 'COE', 'Overig': 'OVR', 'Overhead': 'OHD'}
return mapping.get(client, 'OVR')
def next_invoice_number(client, year=2026):
prefix = client_prefix(client)
r = get_db().execute(
"SELECT invoice_number FROM invoices WHERE invoice_number LIKE ? ORDER BY id DESC LIMIT 1",
(f"{year}-{prefix}-%",)).fetchone()
if r:
num = int(r['invoice_number'].split('-')[-1]) + 1
else:
num = 1
return f"{year}-{prefix}-{num:03d}"
def calc_btw_reserve():
db = get_db()
year = datetime.now().year
r = db.execute("SELECT COALESCE(SUM(btw),0) as total FROM invoices WHERE status != 'Betaald' AND invoice_date LIKE ?",
(f"{year}%",)).fetchone()
return r['total'] if r else 0
# ── Base Layout ─────────────────────────────────────────────────────────────
LAYOUT = """
{% block title %}ZZP Dashboard{% endblock %}
{% block body %}{% endblock %}
"""
# ── Login Page ──────────────────────────────────────────────────────────────
LOGIN_TMPL = LAYOUT.replace('{% block body %}{% endblock %}', """
{% block body %}
{% endblock %}
""")
# ════════════════════════════════════════════════════════════════════════════
# ROUTES
# ════════════════════════════════════════════════════════════════════════════
@app.route('/zzp/login', methods=['GET','POST'])
def login():
if request.method == 'POST':
r = get_db().execute("SELECT * FROM users WHERE username=?", (request.form['username'],)).fetchone()
if r and check_password_hash(r['password_hash'], request.form['password']):
session.permanent = True
session['user'] = r['username']
return redirect(url_for('dashboard'))
return render_template_string(LOGIN_TMPL, error='Ongeldige inloggegevens')
return render_template_string(LOGIN_TMPL, error=None)
@app.route('/zzp/logout')
def logout():
session.clear()
return redirect(url_for('login'))
# ── Dashboard ───────────────────────────────────────────────────────────────
@app.route('/zzp/')
@login_required
def dashboard():
db = get_db()
year = datetime.now().year
# Hours
r = db.execute("SELECT COALESCE(SUM(hours),0) as h FROM timesheets WHERE date LIKE ?",
(f"{year}%",)).fetchone()
total_hours = r['h']
r = db.execute("SELECT COALESCE(SUM(hours),0) as h FROM timesheets WHERE date LIKE ? AND billable=1",
(f"{year}%",)).fetchone()
billable_hours = r['h']
overhead_hours = total_hours - billable_hours
pct = (total_hours / 1225) * 100 if total_hours else 0
# Invoices
open_inv = db.execute("SELECT COUNT(*) as c FROM invoices WHERE status IN ('Verstuurd','Over tijdig') AND invoice_date LIKE ?",
(f"{year}%",)).fetchone()['c']
# Revenue & expenses
rev = db.execute("SELECT COALESCE(SUM(total),0) as t FROM invoices WHERE status='Betaald' AND invoice_date LIKE ?",
(f"{year}%",)).fetchone()['t']
exp = db.execute("SELECT COALESCE(SUM(total),0) as t FROM expenses WHERE date LIKE ?",
(f"{year}%",)).fetchone()['t']
profit = rev - exp
btw_reserve = calc_btw_reserve()
# Recent
recent = db.execute("SELECT * FROM timesheets ORDER BY date DESC LIMIT 5").fetchall()
# Alerts
alerts = []
overdue = db.execute("SELECT invoice_number, due_date FROM invoices WHERE status='Verstuurd' AND due_date < date('now')").fetchall()
for o in overdue:
alerts.append({'text': f"⚠️ Factuur {o['invoice_number']} is achterstallig ( vervaldatum {o['due_date']})", 'style': 'color:%s' % C['danger']})
# BTW deadline
now = datetime.now()
q = (now.month - 1) // 3 + 1
deadlines = {1: '30 april', 2: '31 juli', 3: '31 oktober', 4: '31 januari'}
btws = db.execute("SELECT * FROM btw_quarters WHERE year=? AND quarter=? AND status='Nog doen'", (year, q)).fetchall()
if btws:
alerts.append({'text': f"📋 BTW Q{q} deadline: {deadlines[q]}", 'style': 'color:%s' % C['warning']})
# Travel costs
travel_total = db.execute("SELECT COALESCE(SUM(amount),0) as t FROM travel_logs WHERE date LIKE ?", (f"{year}%",)).fetchone()['t']
travel_km = db.execute("SELECT COALESCE(SUM(km),0) as k FROM travel_logs WHERE date LIKE ?", (f"{year}%",)).fetchone()['k']
data = dict(total_hours=total_hours, billable_hours=billable_hours, overhead_hours=overhead_hours,
pct=pct, open_invoices=open_inv, profit=profit, btw_reserve=btw_reserve,
travel_total=travel_total, travel_km=travel_km)
return render_template_string(DASHBOARD_TMPL, data=data, recent=recent, alerts=alerts,
fmt=fmt_eur)
# ── Timesheets ──────────────────────────────────────────────────────────────
@app.route('/zzp/timesheets')
@login_required
def timesheets():
db = get_db()
year = datetime.now().year
rows = db.execute("SELECT * FROM timesheets WHERE date LIKE ? ORDER BY date DESC", (f"{year}%",)).fetchall()
r = db.execute("SELECT COALESCE(SUM(hours),0) as h FROM timesheets WHERE date LIKE ?", (f"{year}%",)).fetchone()
total = r['h']
r = db.execute("SELECT COALESCE(SUM(hours),0) as h FROM timesheets WHERE date LIKE ? AND billable=1", (f"{year}%",)).fetchone()
billable = r['h']
months = sorted(set(r['date'][:7] for r in rows), reverse=True) if rows else []
return render_template_string(TIMESHEETS_TMPL, timesheets=rows, total_hours=total,
billable_hours=billable, months=months, msg=request.args.get('msg'))
@app.route('/zzp/timesheets/add', methods=['GET','POST'])
@login_required
def timesheet_add():
if request.method == 'POST':
get_db().execute("INSERT INTO timesheets (date,client,activity,hours,billable,note) VALUES (?,?,?,?,?,?)",
(request.form['date'], request.form['client'], request.form['activity'],
float(request.form['hours']), int(request.form.get('billable',1)), request.form.get('note','')))
get_db().commit()
return redirect(url_for('timesheets', msg='Uren toegevoegd'))
return render_template_string(TIMESHEET_FORM_TMPL, edit=False, ts=None,
today=datetime.now().strftime('%Y-%m-%d'))
@app.route('/zzp/timesheets/edit/', methods=['GET','POST'])
@login_required
def timesheet_edit(tid):
db = get_db()
if request.method == 'POST':
db.execute("UPDATE timesheets SET date=?,client=?,activity=?,hours=?,billable=?,note=? WHERE id=?",
(request.form['date'], request.form['client'], request.form['activity'],
float(request.form['hours']), int(request.form.get('billable',1)),
request.form.get('note',''), tid))
db.commit()
return redirect(url_for('timesheets', msg='Uren bijgewerkt'))
ts = db.execute("SELECT * FROM timesheets WHERE id=?", (tid,)).fetchone()
return render_template_string(TIMESHEET_FORM_TMPL, edit=True, ts=ts, today=ts['date'])
@app.route('/zzp/timesheets/delete/', methods=['POST'])
@login_required
def timesheet_delete(tid):
get_db().execute("DELETE FROM timesheets WHERE id=?", (tid,))
get_db().commit()
return redirect(url_for('timesheets', msg='Uren verwijderd'))
# ── Invoices ────────────────────────────────────────────────────────────────
@app.route('/zzp/invoices')
@login_required
def invoices():
db = get_db()
invs = db.execute("SELECT * FROM invoices ORDER BY id DESC").fetchall()
total_invoiced = db.execute("SELECT COALESCE(SUM(total),0) as t FROM invoices").fetchone()['t']
total_paid = db.execute("SELECT COALESCE(SUM(total),0) as t FROM invoices WHERE status='Betaald'").fetchone()['t']
return render_template_string(INVOICES_TMPL, invoices=invs, fmt=fmt_eur,
total_invoiced=total_invoiced, total_paid=total_paid,
total_open=total_invoiced - total_paid, msg=request.args.get('msg'))
@app.route('/zzp/invoices/generate', methods=['GET','POST'])
@login_required
def invoice_generate():
db = get_db()
if request.method == 'POST':
client = request.form['client']
year = int(request.form['year'])
month = int(request.form['month'])
inv_date = request.form['invoice_date']
prefix = client_prefix(client)
rate_key = f'rate_{client.lower().replace(" ","_")}'
rate = float(get_setting(rate_key, '100'))
# Get billable hours for this period
rows = db.execute("SELECT activity, SUM(hours) as hours FROM timesheets WHERE client=? AND billable=1 AND date LIKE ? GROUP BY activity",
(client, f"{year}-{month:02d}%")).fetchall()
if not rows:
return render_template_string(INVOICE_GENERATE_TMPL, error='Geen facturabele uren gevonden voor deze periode/opdrachtgever.',
year=year, month=month, today=inv_date, preview=None)
inv_num = next_invoice_number(client, year)
terms = int(get_setting('payment_terms','30'))
due_date = (datetime.strptime(inv_date, '%Y-%m-%d') + timedelta(days=terms)).strftime('%Y-%m-%d')
subtotal = sum(r['hours'] * rate for r in rows)
btw = subtotal * 0.21
total = subtotal + btw
db.execute("INSERT INTO invoices (invoice_number,client,period_year,period_month,invoice_date,due_date,subtotal,btw,total) VALUES (?,?,?,?,?,?,?,?,?)",
(inv_num, client, year, month, inv_date, due_date, subtotal, btw, total))
inv_id = db.execute("SELECT last_insert_rowid()").fetchone()[0]
for r in rows:
db.execute("INSERT INTO invoice_lines (invoice_id,description,hours,rate,amount) VALUES (?,?,?,?,?)",
(inv_id, r['activity'], r['hours'], rate, r['hours'] * rate))
db.commit()
return redirect(url_for('invoices', msg=f'Factuur {inv_num} gegenereerd'))
today = datetime.now().strftime('%Y-%m-%d')
return render_template_string(INVOICE_GENERATE_TMPL, error=None,
year=datetime.now().year, month=datetime.now().month,
today=today, preview=None)
@app.route('/zzp/invoices/view/')
@login_required
def invoice_view(iid):
db = get_db()
inv = db.execute("SELECT * FROM invoices WHERE id=?", (iid,)).fetchone()
lines = db.execute("SELECT * FROM invoice_lines WHERE invoice_id=?", (iid,)).fetchall()
client_addr_key = f'client_{inv["client"].lower().replace(" ","_")}_address'
client_addr = get_setting(client_addr_key, '')
return render_template_string(INVOICE_VIEW_TMPL, inv=inv, lines=lines, fmt=fmt_eur,
company_name=get_setting('company_name'),
company_address=get_setting('company_address'),
company_kvk=get_setting('company_kvk'),
company_btw=get_setting('company_btw'),
company_bank=get_setting('company_bank'),
payment_terms=get_setting('payment_terms','30'),
client_address=client_addr)
@app.route('/zzp/invoices/status/', methods=['POST'])
@login_required
def invoice_status(iid):
get_db().execute("UPDATE invoices SET status=? WHERE id=?", (request.form['status'], iid))
get_db().commit()
return redirect(url_for('invoices'))
@app.route('/zzp/invoices/delete/', methods=['POST'])
@login_required
def invoice_delete(iid):
db = get_db()
db.execute("DELETE FROM invoice_lines WHERE invoice_id=?", (iid,))
db.execute("DELETE FROM invoices WHERE id=?", (iid,))
db.commit()
return redirect(url_for('invoices', msg='Factuur verwijderd'))
# ── Expenses ────────────────────────────────────────────────────────────────
EXPENSE_CATEGORIES = ['Boekhouper','Verzekeringen','Reiskosten','Software','Hardware','Opleiding','Kantoor','Overig']
@app.route('/zzp/expenses')
@login_required
def expenses():
db = get_db()
year = datetime.now().year
rows = db.execute("SELECT * FROM expenses WHERE date LIKE ? ORDER BY date DESC", (f"{year}%",)).fetchall()
total_exp = db.execute("SELECT COALESCE(SUM(total),0) as t FROM expenses WHERE date LIKE ?", (f"{year}%",)).fetchone()['t']
total_btw_exp = db.execute("SELECT COALESCE(SUM(btw),0) as t FROM expenses WHERE date LIKE ?", (f"{year}%",)).fetchone()['t']
# Monthly overview
monthly = []
for m in range(1,13):
label = f"{year}-{m:02d}"
rev = db.execute("SELECT COALESCE(SUM(subtotal),0) as t FROM invoices WHERE invoice_date LIKE ?", (f"{label}%",)).fetchone()['t']
exp_m = db.execute("SELECT COALESCE(SUM(total),0) as t FROM expenses WHERE date LIKE ?", (f"{label}%",)).fetchone()['t']
monthly.append({'label': label, 'revenue': rev, 'expenses': exp_m, 'profit': rev - exp_m})
return render_template_string(EXPENSES_TMPL, expenses=rows, total_expenses=total_exp,
total_btw=total_btw_exp, monthly=monthly, year=year,
fmt=fmt_eur, msg=request.args.get('msg'))
@app.route('/zzp/expenses/add', methods=['GET','POST'])
@login_required
def expense_add():
if request.method == 'POST':
excl = float(request.form['amount_excl'])
btw = float(request.form.get('btw','0'))
get_db().execute("INSERT INTO expenses (date,category,description,amount_excl,btw,total) VALUES (?,?,?,?,?,?)",
(request.form['date'], request.form['category'], request.form.get('description',''),
excl, btw, excl + btw))
get_db().commit()
return redirect(url_for('expenses', msg='Kosten toegevoegd'))
return render_template_string(EXPENSE_FORM_TMPL, edit=False, exp=None,
today=datetime.now().strftime('%Y-%m-%d'),
categories=EXPENSE_CATEGORIES)
@app.route('/zzp/expenses/edit/', methods=['GET','POST'])
@login_required
def expense_edit(eid):
db = get_db()
if request.method == 'POST':
excl = float(request.form['amount_excl'])
btw = float(request.form.get('btw','0'))
db.execute("UPDATE expenses SET date=?,category=?,description=?,amount_excl=?,btw=?,total=? WHERE id=?",
(request.form['date'], request.form['category'], request.form.get('description',''),
excl, btw, excl + btw, eid))
db.commit()
return redirect(url_for('expenses', msg='Kosten bijgewerkt'))
exp = db.execute("SELECT * FROM expenses WHERE id=?", (eid,)).fetchone()
return render_template_string(EXPENSE_FORM_TMPL, edit=True, exp=exp, today=exp['date'],
categories=EXPENSE_CATEGORIES)
@app.route('/zzp/expenses/delete/', methods=['POST'])
@login_required
def expense_delete(eid):
get_db().execute("DELETE FROM expenses WHERE id=?", (eid,))
get_db().commit()
return redirect(url_for('expenses', msg='Kosten verwijderd'))
# ── Travel Costs ────────────────────────────────────────────────────────────
@app.route('/zzp/travel')
@login_required
def travel():
db = get_db()
year = datetime.now().year
rows = db.execute("SELECT * FROM travel_logs WHERE date LIKE ? ORDER BY date DESC", (f"{year}%",)).fetchall()
total_km = db.execute("SELECT COALESCE(SUM(km),0) as k FROM travel_logs WHERE date LIKE ?", (f"{year}%",)).fetchone()['k']
total_amount = db.execute("SELECT COALESCE(SUM(amount),0) as t FROM travel_logs WHERE date LIKE ?", (f"{year}%",)).fetchone()['t']
billable_amount = db.execute("SELECT COALESCE(SUM(amount),0) as t FROM travel_logs WHERE date LIKE ? AND billable=1", (f"{year}%",)).fetchone()['t']
months = sorted(set(r['date'][:7] for r in rows), reverse=True) if rows else []
return render_template_string(TRAVEL_TMPL, travels=rows, total_km=total_km,
total_amount=total_amount, billable_amount=billable_amount,
months=months, fmt=fmt_eur, msg=request.args.get('msg'))
@app.route('/zzp/travel/add', methods=['GET','POST'])
@login_required
def travel_add():
if request.method == 'POST':
km = float(request.form['km'])
rate = float(request.form.get('rate_per_km', '0.23'))
amount = round(km * rate, 2)
get_db().execute(
"INSERT INTO travel_logs (date,client,from_location,to_location,km,rate_per_km,amount,transport_type,purpose,billable) VALUES (?,?,?,?,?,?,?,?,?,?)",
(request.form['date'], request.form.get('client',''), request.form.get('from_location',''),
request.form.get('to_location',''), km, rate, amount,
request.form.get('transport_type','Auto'), request.form.get('purpose',''),
int(request.form.get('billable',1))))
get_db().commit()
return redirect(url_for('travel', msg='Rit toegevoegd'))
return render_template_string(TRAVEL_FORM_TMPL, edit=False, t=None,
today=datetime.now().strftime('%Y-%m-%d'),
rate=get_setting('travel_rate','0.23'))
@app.route('/zzp/travel/edit/', methods=['GET','POST'])
@login_required
def travel_edit(tid):
db = get_db()
if request.method == 'POST':
km = float(request.form['km'])
rate = float(request.form.get('rate_per_km', '0.23'))
amount = round(km * rate, 2)
db.execute(
"UPDATE travel_logs SET date=?,client=?,from_location=?,to_location=?,km=?,rate_per_km=?,amount=?,transport_type=?,purpose=?,billable=? WHERE id=?",
(request.form['date'], request.form.get('client',''), request.form.get('from_location',''),
request.form.get('to_location',''), km, rate, amount,
request.form.get('transport_type','Auto'), request.form.get('purpose',''),
int(request.form.get('billable',1)), tid))
db.commit()
return redirect(url_for('travel', msg='Rit bijgewerkt'))
t = db.execute("SELECT * FROM travel_logs WHERE id=?", (tid,)).fetchone()
return render_template_string(TRAVEL_FORM_TMPL, edit=True, t=t, today=t['date'],
rate=get_setting('travel_rate','0.23'))
@app.route('/zzp/travel/delete/', methods=['POST'])
@login_required
def travel_delete(tid):
get_db().execute("DELETE FROM travel_logs WHERE id=?", (tid,))
get_db().commit()
return redirect(url_for('travel', msg='Rit verwijderd'))
# ── BTW ─────────────────────────────────────────────────────────────────────
@app.route('/zzp/btw')
@login_required
def btw():
db = get_db()
year = datetime.now().year
quarters = []
deadlines_map = {1: '30 april', 2: '31 juli', 3: '31 oktober', 4: '31 januari'}
for q in range(1,5):
m_start = (q-1)*3+1
m_end = q*3
patterns = [f"{year}-{m:02d}%" for m in range(m_start, m_end+1)]
btw_in = 0
btw_out = 0
for p in patterns:
r = db.execute("SELECT COALESCE(SUM(btw),0) as t FROM invoices WHERE invoice_date LIKE ?", (p,)).fetchone()
btw_in += r['t']
r = db.execute("SELECT COALESCE(SUM(btw),0) as t FROM expenses WHERE date LIKE ?", (p,)).fetchone()
btw_out += r['t']
r = db.execute("SELECT * FROM btw_quarters WHERE year=? AND quarter=?", (year, q)).fetchone()
status = r['status'] if r else 'Nog doen'
quarters.append({'quarter': q, 'btw_in': btw_in, 'btw_out': btw_out,
'net': btw_in - btw_out, 'deadline': deadlines_map[q], 'status': status})
return render_template_string(BTW_TMPL, quarters=quarters, year=year, fmt=fmt_eur)
@app.route('/zzp/btw/toggle//', methods=['POST'])
@login_required
def btw_toggle(year, quarter):
db = get_db()
r = db.execute("SELECT * FROM btw_quarters WHERE year=? AND quarter=?", (year, quarter)).fetchone()
if r:
new_status = 'Nog doen' if r['status'] == 'Gedaan' else 'Gedaan'
db.execute("UPDATE btw_quarters SET status=?, filed_at=? WHERE year=? AND quarter=?",
(new_status, datetime.now().isoformat() if new_status=='Gedaan' else None, year, quarter))
else:
db.execute("INSERT INTO btw_quarters (year,quarter,status) VALUES (?,?,?)",
(year, quarter, 'Gedaan'))
db.commit()
return redirect(url_for('btw'))
# ── Settings ────────────────────────────────────────────────────────────────
SETTINGS_KEYS = ['company_name','company_address','company_kvk','company_btw','company_bank',
'rate_phoenix_metals','rate_club_of_engineers','rate_overig','payment_terms',
'client_phoenix_metals_address','client_club_of_engineers_address','client_overig_address',
'travel_rate']
@app.route('/zzp/settings', methods=['GET','POST'])
@login_required
def settings_page():
if request.method == 'POST':
for k in SETTINGS_KEYS:
if k in request.form:
set_setting(k, request.form[k])
return render_template_string(SETTINGS_TMPL, msg='Instellingen opgeslagen',
s={k: get_setting(k) for k in SETTINGS_KEYS})
return render_template_string(SETTINGS_TMPL, msg=None,
s={k: get_setting(k) for k in SETTINGS_KEYS})
# ── Run ─────────────────────────────────────────────────────────────────────
init_db()
# ── ZZP Advisor AI ────────────────────────────────────────────────────────────
ZAI_URL = os.environ.get('AI_API_URL', 'https://api.z.ai/api/coding/paas/v4/chat/completions')
ZAI_KEY = os.environ.get('AI_API_KEY', '43a6c7e3d7b240daafae006e8488f674.ivYYRRLySgUwgqVE')
ZAI_MODEL = os.environ.get('AI_MODEL', 'glm-5-turbo')
OR_URL = os.environ.get('OPENROUTER_URL', 'https://openrouter.ai/api/v1/chat/completions')
OR_KEY = os.environ.get('OPENROUTER_API_KEY', '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')
OR_MODEL = os.environ.get('OPENROUTER_MODEL', 'google/gemma-3-27b-it:free')
TAVILY_KEY = os.environ.get('TAVILY_API_KEY', '')
ZZP_SYSTEM = """Je bent een Nederlandse ZZP Adviseur. antwoord ALLEEN in het Nederlands.
Je geeft concreet, feitelijk advies over ondernemerschap in Nederland. Gebruik cijfers, percentages en termijnen. Verwijs naar Belastingdienst, KVK of UWV waar relevant.
Expertise: KVK-inschrijving, BTW (21%/9%/0%), inkomstenbelasting, IZB, MKB-winstvrijstelling, facturatie vereisten, DBA-wetgeving, modelovereenkomsten, pensioen (lijfrente), AOV-verzekering, contracten, auto/fiets fiscaliteit, KIA afschrijving, urencriterium (1.225 uur).
REGELS:
- Antwoord DIRECT met het antwoord. Geen inleiding, geen meta-tekst, geen "Analyze" of "Drafting".
- Gebruik opsommingen met bullet points.
- Als je niet zeker bent, zeg dat dan.
- NOOIT Engels. ALTIJD Nederlands."""
def ai_call(messages, max_retries=2):
"""AI call with retry + fallback — same architecture as HSEQ dashboard."""
import time as _time
providers = [
('zai', ZAI_URL, ZAI_KEY, ZAI_MODEL, True),
('or', OR_URL, OR_KEY, OR_MODEL, False),
]
for pname, url, key, model, use_ctx in providers:
for attempt in range(max_retries):
try:
payload = json.dumps({"model":model,"messages":messages,"stream":False,"temperature":0.7,"max_tokens":4000}).encode('utf-8')
headers = {"Content-Type":"application/json","Authorization":f"Bearer {key}"}
if pname == 'or':
headers["HTTP-Referer"] = "https://mescalinerabbit.shop"
headers["X-Title"] = "ZZP Advisor"
req = urllib.request.Request(url, data=payload, headers=headers)
if use_ctx:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
resp = urllib.request.urlopen(req, context=ctx, timeout=120)
else:
resp = urllib.request.urlopen(req, timeout=60)
return resp
except urllib.error.HTTPError as e:
body = e.read().decode('utf-8', errors='replace')[:200]
print(f'[ZZP AI] {pname} HTTP {e.code} attempt {attempt+1}/{max_retries}: {body}')
if attempt < max_retries - 1:
_time.sleep(2 ** attempt)
else:
break
except Exception as e:
print(f'[ZZP AI] {pname} error attempt {attempt+1}: {e}')
if attempt < max_retries - 1:
_time.sleep(2 ** attempt)
else:
break
return None
def ai_complete(messages):
"""Non-streaming AI call with retry — HSEQ dashboard architecture."""
resp = ai_call(messages)
if resp is None:
return '⚠️ Momenteel zijn alle AI-modellen bezet. Probeer het over een minuut opnieuw.'
data = json.loads(resp.read().decode('utf-8'))
msg = data['choices'][0]['message']
content = msg.get('content', '')
if not content:
content = msg.get('reasoning_content', '')
return content
# ── ZZP Advisor Page ────────────────────────────────────────────────────────
ADVISOR_TMPL = LAYOUT.replace('{% block body %}{% endblock %}', """
{% block body %}
{% endblock %}
""")
@app.route('/zzp/advisor')
@login_required
def zzp_advisor():
return render_template_string(ADVISOR_TMPL)
@app.route('/zzp/api/chat', methods=['POST'])
@login_required
def zzp_chat():
data = request.get_json()
user_msg = data.get('message','').strip()
if not user_msg:
return jsonify({'error':'Leeg bericht'}), 400
messages = [{"role":"system","content":ZZP_SYSTEM}]
# Tavily crosscheck for legal/regulatory questions
tavily_ctx = ''
if TAVILY_KEY and any(w in user_msg.lower() for w in ['wetgeving','belasting','btw','kvk','dba','var','pensioen','aov','controle','facturatie','aftrek','vrijstelling','tarief']):
try:
tp = json.dumps({"query":user_msg+" Nederland 2024 2025","max_results":3,"search_depth":"basic","api_key":TAVILY_KEY}).encode()
tr = urllib.request.Request("https://api.tavily.com/search",data=tp,headers={"Content-Type":"application/json"})
tresp = urllib.request.urlopen(tr,timeout=10)
tdata = json.loads(tresp.read())
if tdata.get('results'):
tavily_ctx = '\n\nActuele informatie:\n'+'\n'.join(f"- {r.get('content','')[:200]}" for r in tdata['results'][:3])
except Exception as e:
print(f'[ZZP Tavily] failed: {e}')
if tavily_ctx:
messages.append({"role":"system","content":"Gebruik deze actuele informatie ter aanvulling:"+tavily_ctx})
messages.append({"role":"user","content":user_msg})
try:
reply = ai_complete(messages)
return jsonify({'reply':reply})
except Exception as e:
print(f'[ZZP Chat] Error: {e}')
return jsonify({'reply':'⚠️ De AI is momenteel overbelast. Probeer het over een minuut opnieuw.'})
if __name__ == '__main__':
print("=" * 50)
print("ZZP Dashboard running on http://127.0.0.1:5060/")
print("=" * 50)
app.run(host='0.0.0.0', port=5065, debug=False)