BlackMarkandClaude Opus 5.5
45424e6fd6
A local web app for a small room-rental business in Austria: guest register, room ledger, occupancy plan, invoices, the monthly tourist-tax report and the quarterly VAT list. Application code only; it ships no data. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
738 lines
36 KiB
Python
738 lines
36 KiB
Python
"""Flask app: the web UI. Refuses to start while a migration is pending. All data access goes through
|
|
repo.py, so scripts get the same operations as the pages.
|
|
"""
|
|
import base64
|
|
import io
|
|
import mimetypes
|
|
import secrets
|
|
import struct
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from flask import (Flask, render_template, request, redirect, url_for, g, flash, abort, send_file)
|
|
|
|
from app import compute, db, i18n, money, pdf, plan, repo, quality, tvb, vat
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
INVOICE_ASSETS = APP_DIR / "static" / "invoice"
|
|
|
|
|
|
def get_db():
|
|
if "conn" not in g:
|
|
g.conn = db.connect(actor="web")
|
|
return g.conn
|
|
|
|
|
|
_page_cache = {}
|
|
|
|
|
|
def assets_fingerprint():
|
|
"""Newest mtime of the grid's own code — static files AND templates. It belongs in the fingerprint that
|
|
keys the FROZEN column layouts in the browser's localStorage: without it a change is invisible after a
|
|
reload, because the browser re-applies the widths it measured against the old code. The templates count
|
|
too: a template change can widen a column without touching a single static file."""
|
|
files = [APP_DIR / "static" / f for f in ("grids.js", "style.css")]
|
|
files += sorted((APP_DIR / "templates").glob("*.html"))
|
|
return str(max(int(f.stat().st_mtime) for f in files))
|
|
|
|
|
|
def _ledger_year(row):
|
|
"""Which year does a Zimmerbelegung row belong to? That of the first date it actually has —
|
|
`repo.ledger_rows` says which."""
|
|
return (row["first_date"] or "")[:4] or None
|
|
|
|
|
|
def page_fingerprint():
|
|
"""What a rendered list page depends on: the data AND the grid code. Keys both the server-side page
|
|
cache and the frozen column layouts in the browser — a cached page carrying the old fingerprint would
|
|
keep the stale layout alive just as effectively as the layout itself."""
|
|
return db_fingerprint() + "-" + assets_fingerprint()
|
|
|
|
|
|
def db_fingerprint():
|
|
"""The schema and its write counter (`data_version`, counted up by every statement that writes a data
|
|
table — this app or any script; migration 001). Keys the server-side page cache and the
|
|
browser's stored grid layouts. The schema belongs in it: the tests run each in a schema of their own."""
|
|
return f"{db.SCHEMA}.{db.row(get_db(), 'SELECT version FROM data_version')[0]}"
|
|
|
|
|
|
def cached_page(view):
|
|
"""Serve the big read-only list pages from memory while the database is unchanged.
|
|
Pages carrying flash messages are rendered fresh and never stored."""
|
|
from functools import wraps
|
|
|
|
@wraps(view)
|
|
def wrapped(**kwargs):
|
|
from flask import session
|
|
if "_flashes" in session:
|
|
return view(**kwargs)
|
|
key = (request.full_path, page_fingerprint())
|
|
if _page_cache.get(view.__name__, (None,))[0] != key:
|
|
_page_cache[view.__name__] = (key, view(**kwargs))
|
|
return _page_cache[view.__name__][1]
|
|
return wrapped
|
|
|
|
|
|
def asset_uri(filename):
|
|
"""Inline an invoice image as a data URI so it renders in both browser-print and the headless PDF."""
|
|
if not filename:
|
|
return ""
|
|
path = INVOICE_ASSETS / filename
|
|
if not path.exists():
|
|
return ""
|
|
mime = mimetypes.guess_type(str(path))[0] or "image/png"
|
|
return f"data:{mime};base64," + base64.b64encode(path.read_bytes()).decode()
|
|
|
|
|
|
ART_BOX_W, ART_BOX_H, ART_GAP = 64, 80, 4 # masthead art box (mm) — keep in sync with .masthead .art CSS
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def _image_size(path_str):
|
|
"""Pixel (w, h) of a PNG or JPEG read from the file header — avoids an imaging dependency."""
|
|
data = Path(path_str).read_bytes()
|
|
if data[:8] == b"\x89PNG\r\n\x1a\n":
|
|
w, h = struct.unpack(">II", data[16:24])
|
|
return w, h
|
|
if data[:2] == b"\xff\xd8":
|
|
i = 2
|
|
while i + 9 < len(data):
|
|
if data[i] != 0xFF:
|
|
i += 1
|
|
elif data[i + 1] == 0x01 or 0xD0 <= data[i + 1] <= 0xD9: # marker without a length field
|
|
i += 2
|
|
elif 0xC0 <= data[i + 1] <= 0xCF and data[i + 1] not in (0xC4, 0xC8, 0xCC): # SOFn
|
|
h, w = struct.unpack(">HH", data[i + 5:i + 9])
|
|
return w, h
|
|
else:
|
|
i += 2 + struct.unpack(">H", data[i + 2:i + 4])[0]
|
|
raise ValueError(f"cannot read image size: {path_str}")
|
|
|
|
|
|
def masthead_art(unit):
|
|
"""Explicit print sizes for the branding images. All images of a unit share ONE width, solved so the
|
|
stack exactly fills the art box height (capped at the box width) — a wide logo can never render bigger
|
|
than the house photo, and a lone image grows to fill the whole box."""
|
|
imgs = [{"src": name, "cls": cls} for name, cls in ((unit.get("logo"), "logo"), (unit.get("photo"), "photo"))
|
|
if name and (INVOICE_ASSETS / name).exists()]
|
|
for a in imgs:
|
|
w_px, h_px = _image_size(str(INVOICE_ASSETS / a["src"]))
|
|
a["ratio"] = h_px / w_px
|
|
if not imgs:
|
|
return []
|
|
usable_h = ART_BOX_H - ART_GAP * (len(imgs) - 1)
|
|
width = min(ART_BOX_W, usable_h / sum(a["ratio"] for a in imgs))
|
|
for a in imgs:
|
|
a["w"] = round(width, 2)
|
|
a["h"] = round(width * a.pop("ratio"), 2)
|
|
return imgs
|
|
|
|
|
|
def _tourism_rate(on_date=None):
|
|
"""The Aufenthaltsabgabe per person and night in force on that date — handed to the ledger form so the
|
|
browser can propose the amount without a second copy of the rate table."""
|
|
import datetime
|
|
return compute.tourism_rate(on_date or datetime.date.today().isoformat())
|
|
|
|
|
|
def create_app():
|
|
app = Flask(__name__)
|
|
app.secret_key = secrets.token_hex(32) # the session only carries flash messages across a redirect
|
|
|
|
# Static assets: version the URL by file mtime and cache hard. Without this the browser's heuristic
|
|
# cache kept serving OLD grids.js/style.css after updates (no revalidation for hours/days).
|
|
app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 60 * 60 * 24 * 365
|
|
|
|
@app.url_defaults
|
|
def bust_static_cache(endpoint, values):
|
|
if endpoint == "static" and "filename" in values:
|
|
f = APP_DIR / "static" / values["filename"]
|
|
if f.exists():
|
|
values["v"] = int(f.stat().st_mtime)
|
|
|
|
conn = db.connect()
|
|
pending = db.pending_migrations(conn)
|
|
conn.close()
|
|
if pending: # the app's role may not change the schema — the owner does, before the app starts
|
|
raise RuntimeError(f"database schema is behind the code (migrations {pending}) — "
|
|
f"run first: python -m app.db migrate")
|
|
|
|
# the grid pages render thousands of rows — don't ship template indentation as page weight
|
|
app.jinja_env.trim_blocks = True
|
|
app.jinja_env.lstrip_blocks = True
|
|
app.context_processor(lambda: {"db_fp": page_fingerprint()})
|
|
app.jinja_env.globals["t"] = i18n.t
|
|
app.jinja_env.globals["asset_uri"] = asset_uri
|
|
app.jinja_env.globals["masthead_art"] = masthead_art
|
|
app.jinja_env.filters["euro"] = money.euro
|
|
app.jinja_env.filters["num"] = money.de_number
|
|
app.jinja_env.filters["date"] = money.de_date
|
|
app.jinja_env.filters["datewd"] = money.de_date_wd # with the weekday, for the two lists
|
|
|
|
@app.after_request
|
|
def never_cache_pages(resp):
|
|
# dynamic pages must always re-render — without this, the browser's back button (bfcache /
|
|
# heuristic caching) shows a STALE copy after an edit. Versioned static assets keep their year.
|
|
if request.endpoint != "static":
|
|
resp.headers["Cache-Control"] = "no-store"
|
|
return resp
|
|
|
|
@app.teardown_appcontext
|
|
def close_db(exc):
|
|
conn = g.pop("conn", None)
|
|
if conn is not None:
|
|
conn.close()
|
|
|
|
@app.route("/")
|
|
def home():
|
|
return render_template("home.html", stats=repo.stats(get_db()))
|
|
|
|
# --- guest register ---
|
|
@app.route("/register")
|
|
@cached_page
|
|
def register():
|
|
search = request.args.get("q", "").strip()
|
|
return render_template("register.html", rows=repo.register_rows(get_db(), search), q=search)
|
|
|
|
@app.route("/register/new", methods=["GET", "POST"])
|
|
@app.route("/register/<int:stay_id>/edit", methods=["GET", "POST"])
|
|
def stay_form(stay_id=None):
|
|
conn = get_db()
|
|
stay = repo.get_stay(conn, stay_id) if stay_id else None
|
|
# a room change: the new sheet continues the old one — same person, same number
|
|
from_id = None if stay_id else request.args.get("from", type=int)
|
|
moved_from = repo.get_stay(conn, from_id) if from_id else None
|
|
if request.method == "POST":
|
|
f = request.form
|
|
if not f.get("last_name", "").strip():
|
|
flash(i18n.t("err_required", field=i18n.t("col_name")))
|
|
elif f.get("arrival") and f.get("departure") and f["departure"] < f["arrival"]:
|
|
flash(i18n.t("err_departure_before_arrival")) # reject at the door, not on Datenprüfung
|
|
else:
|
|
last = f["last_name"].strip()
|
|
first = f.get("first_name", "").strip() or None
|
|
gfields = dict(birth_date=f.get("birth_date") or None, nationality=f.get("nationality") or None,
|
|
doc_type=f.get("doc_type") or None, doc_number=f.get("doc_number") or None,
|
|
profession=f.get("profession") or None, home_street=f.get("home_street") or None,
|
|
home_postcode=f.get("home_postcode") or None, home_city=f.get("home_city") or None,
|
|
home_country=f.get("home_country") or None, phone=f.get("phone") or None)
|
|
person = stay or moved_from # edit, or the same person in another room
|
|
if person and person.get("guest_id"): # update this exact person
|
|
repo.update_guest(conn, person["guest_id"], last_name=last, first_name=first, **gfields)
|
|
guest_id = person["guest_id"]
|
|
else: # new: reuse only on exact person match
|
|
guest_id = repo.find_or_create_guest(conn, last, first, **gfields)
|
|
room_id = int(f["room_id"]) if f.get("room_id") else None
|
|
saved = repo.save_stay(conn, stay_id, guest_id, room_id,
|
|
int(f["unit_id"]) if f.get("unit_id") else None,
|
|
repo.find_or_create_client(conn, f.get("employer")),
|
|
f.get("arrival") or None, f.get("departure") or None,
|
|
f.get("sheet_number") or None,
|
|
int(f["hauptwohnsitz"]) if f.get("hauptwohnsitz") in ("0", "1") else None,
|
|
int(f["tourism_liable"]) if f.get("tourism_liable") in ("0", "1") else None,
|
|
f.get("notes") or None)
|
|
if f.get("action") == "room_change":
|
|
if not f.get("departure"): # the departure IS the day of the change
|
|
flash(i18n.t("err_room_change_needs_date"))
|
|
return redirect(url_for("stay_form", stay_id=saved))
|
|
return redirect(url_for("stay_form", **{"from": saved}))
|
|
if moved_from and repo.carry_ledger_to_room(conn, f.get("sheet_number") or None,
|
|
moved_from["room_id"], room_id, f.get("arrival"),
|
|
{"last_name": last, "first_name": first}):
|
|
flash(i18n.t("msg_room_changed", date=money.de_date(f["arrival"]),
|
|
room=_room_number(conn, room_id)))
|
|
return redirect(url_for("register"))
|
|
if moved_from: # the old sheet, as the new one starts
|
|
stay = dict(moved_from, id=None, room_id=None, arrival=moved_from["departure"], departure=None,
|
|
notes=i18n.t("note_room_change", room=_room_number(conn, moved_from["room_id"])))
|
|
return render_template("stay_form.html", stay=stay, room_change=bool(moved_from), rooms=repo.rooms(conn),
|
|
units=repo.units(conn), suggest=repo.field_suggestions(conn))
|
|
|
|
@app.route("/register/<int:stay_id>/delete", methods=["POST"])
|
|
def stay_delete(stay_id):
|
|
try:
|
|
repo.delete_stay(get_db(), stay_id)
|
|
except ValueError:
|
|
flash(i18n.t("err_delete_blocked"))
|
|
return redirect(url_for("register"))
|
|
|
|
# --- room ledger ---
|
|
@app.route("/ledger")
|
|
@cached_page
|
|
def ledger():
|
|
conn = get_db()
|
|
room = request.args.get("room", "").strip()
|
|
search = request.args.get("q", "").strip()
|
|
rows = repo.ledger_rows(conn, room or None, search)
|
|
years = sorted({y for y in (_ledger_year(r) for r in rows) if y}, reverse=True)
|
|
return render_template("ledger.html", rows=rows, years=years,
|
|
rooms=repo.rooms(conn), room=room, q=search)
|
|
|
|
@app.route("/ledger/new", methods=["GET", "POST"])
|
|
def ledger_new():
|
|
conn = get_db()
|
|
if request.method == "POST":
|
|
f = request.form
|
|
room = f.get("room_id", "").strip()
|
|
repo.create_ledger_row(conn, int(room) if room.isdigit() else None,
|
|
f.get("occupant", "").strip() or None,
|
|
f.get("period_from") or None, f.get("period_to") or None,
|
|
parse_number(f.get("persons")), parse_number(f.get("quantity")),
|
|
f.get("unit_type") or None, parse_number(f.get("gross_per_unit")),
|
|
parse_number(f.get("betrag")), f.get("paid_status") or None,
|
|
f.get("paid_date") or None, f.get("art") or None,
|
|
f.get("payer", "").strip() or None, f.get("rnr", "").strip() or None,
|
|
f.get("gb_ref", "").strip() or None, f.get("notes", "").strip() or None,
|
|
tourism_tax=parse_number(f.get("tourism_tax")),
|
|
payments=_payments(f) or [])
|
|
return redirect(url_for("ledger"))
|
|
return render_template("ledger_form.html", ch={}, rooms=repo.rooms(conn), is_new=True,
|
|
tourism_rate=_tourism_rate(None))
|
|
|
|
@app.route("/ledger/<int:charge_id>/edit", methods=["GET", "POST"])
|
|
def ledger_edit(charge_id):
|
|
conn = get_db()
|
|
ch = repo.get_charge(conn, charge_id)
|
|
if not ch:
|
|
abort(404)
|
|
if request.method == "POST":
|
|
f = request.form
|
|
repo.update_charge(conn, charge_id, f.get("occupant", "").strip() or None,
|
|
f.get("period_from") or None, f.get("period_to") or None,
|
|
parse_number(f.get("persons")), parse_number(f.get("quantity")),
|
|
f.get("unit_type") or None, parse_number(f.get("gross_per_unit")),
|
|
parse_number(f.get("betrag")), f.get("paid_status") or None,
|
|
f.get("paid_date") or None, f.get("art") or None,
|
|
f.get("payer", "").strip() or None, f.get("rnr", "").strip() or None,
|
|
f.get("gb_ref", "").strip() or None, f.get("notes", "").strip() or None,
|
|
tourism_tax=parse_number(f.get("tourism_tax")),
|
|
payments=_payments(f))
|
|
if f.get("action") == "continue":
|
|
# month-end billing: bill up to this date, then carry the occupancy on from it
|
|
if not f.get("period_to"):
|
|
flash(i18n.t("err_continue_needs_date"))
|
|
return redirect(url_for("ledger_edit", charge_id=charge_id))
|
|
repo.continue_ledger_row(conn, charge_id)
|
|
flash(i18n.t("msg_continued", date=money.de_date(f["period_to"])))
|
|
return redirect(url_for("ledger"))
|
|
return render_template("ledger_form.html", ch=ch, tourism_rate=_tourism_rate(ch.get("period_from")))
|
|
|
|
@app.route("/ledger/<int:charge_id>/delete", methods=["POST"])
|
|
def ledger_delete(charge_id):
|
|
repo.delete_charge(get_db(), charge_id)
|
|
return redirect(url_for("ledger"))
|
|
|
|
# --- Belegungsplan (occupancy timeline) ---
|
|
@app.route("/plan")
|
|
def plan_page():
|
|
# not @cached_page: the view depends on today's date (today marker, occupancy pills),
|
|
# and the cache key only sees URL + db fingerprint — it would survive midnight.
|
|
import datetime
|
|
weeks = 8 if request.args.get("weeks") == "8" else 4
|
|
try:
|
|
start = datetime.date.fromisoformat(request.args.get("start", ""))
|
|
except ValueError:
|
|
start = plan.default_start()
|
|
return render_template("plan.html", p=plan.build(get_db(), start, weeks * 7), weeks=weeks)
|
|
|
|
# --- invoices ---
|
|
@app.route("/invoices")
|
|
@cached_page
|
|
def invoices():
|
|
q = request.args.get("q", "").strip()
|
|
return render_template("invoices.html", rows=repo.invoices_list(get_db(), q), q=q)
|
|
|
|
def _rows_follow(conn, invoice_id):
|
|
"""The Zimmerbelegung takes over what the saved invoice bills — and the user is told it did."""
|
|
fixed, new = repo.sync_rows_to_invoice(conn, invoice_id)
|
|
if fixed or new:
|
|
flash(i18n.t("msg_rows_follow_invoice", fixed=len(fixed), new=len(new)))
|
|
|
|
@app.route("/invoices/new", methods=["GET", "POST"])
|
|
def invoice_new():
|
|
conn = get_db()
|
|
units = repo.units(conn)
|
|
f = request.form
|
|
lines = []
|
|
order_warning = False
|
|
client_choices = None
|
|
if request.method == "POST":
|
|
lines, incomplete = _parse_lines(f)
|
|
unit_id = int(f["unit_id"]) if f.get("unit_id") else None
|
|
number = f.get("number", "").strip()
|
|
if incomplete:
|
|
flash(i18n.t("err_number", field=", ".join(incomplete)))
|
|
elif not unit_id or not f.get("client", "").strip() or not lines:
|
|
flash(i18n.t("err_required", field="Einheit, Kunde, mind. 1 Zeile"))
|
|
elif number and repo.invoice_number_taken(conn, unit_id, number):
|
|
flash(i18n.t("err_number_taken", number=number))
|
|
elif _order_warning(conn, f, unit_id, number):
|
|
order_warning = True
|
|
elif (client_choices := _recipient_warning(conn, f)) is None:
|
|
links = [int(x) for x in f.get("link_charges", "").split(",") if x.strip().isdecimal()]
|
|
inv_id = repo.create_invoice(
|
|
conn, unit_id, repo.find_or_create_client(conn, f["client"]),
|
|
number, f.get("date") or None, lines, note=f.get("note"),
|
|
paid_date=f.get("paid_date") or None, payment_method=f.get("payment_method") or None,
|
|
link_charge_ids=links)
|
|
_rows_follow(conn, inv_id)
|
|
return redirect(url_for("invoice_view", invoice_id=inv_id))
|
|
lines = _raw_lines(f) # re-render exactly what was typed, incl. the rejected rows
|
|
elif request.args.get("stays") or request.args.get("charges"):
|
|
# generate a pre-filled invoice from selected Gästeblätter rows or Zimmerbelegung rows
|
|
if request.args.get("charges"):
|
|
ids = [int(x) for x in request.args["charges"].split(",") if x.strip().isdecimal()]
|
|
pre = repo.lines_from_charges(conn, ids)
|
|
else:
|
|
ids = [int(x) for x in request.args["stays"].split(",") if x.strip().isdecimal()]
|
|
pre = repo.lines_from_stays(conn, ids)
|
|
lines = pre["lines"]
|
|
f = {"unit_id": pre["unit_id"], "client": pre["client"] or "",
|
|
# remember the ledger rows so saving writes the invoice number back into them
|
|
"link_charges": ",".join(map(str, ids)) if request.args.get("charges") else ""}
|
|
if pre["client"]: # a payer without an address asks at once, not only on saving
|
|
client_choices = repo.recipient_choices(conn, pre["client"])
|
|
import datetime
|
|
suggestions = {u["id"]: repo.next_invoice_number(conn, u["id"]) for u in units}
|
|
return render_template("invoice_form.html", units=units, suggestions=suggestions, form=f, lines=lines,
|
|
suggest=repo.field_suggestions(conn), order_warning=order_warning,
|
|
client_choices=client_choices, today=datetime.date.today().isoformat())
|
|
|
|
@app.route("/invoices/<int:invoice_id>/edit", methods=["GET", "POST"])
|
|
def invoice_edit(invoice_id):
|
|
conn = get_db()
|
|
inv = repo.get_invoice(conn, invoice_id)
|
|
if not inv:
|
|
abort(404)
|
|
units = repo.units(conn)
|
|
order_warning = False
|
|
client_choices = None
|
|
if request.method == "POST":
|
|
f = request.form
|
|
lines, incomplete = _parse_lines(f)
|
|
unit_id = int(f["unit_id"]) if f.get("unit_id") else None
|
|
number = f.get("number", "").strip()
|
|
if incomplete:
|
|
flash(i18n.t("err_number", field=", ".join(incomplete)))
|
|
elif not unit_id or not f.get("client", "").strip() or not lines:
|
|
flash(i18n.t("err_required", field="Einheit, Kunde, mind. 1 Zeile"))
|
|
elif number and repo.invoice_number_taken(conn, unit_id, number, exclude_id=invoice_id):
|
|
flash(i18n.t("err_number_taken", number=number))
|
|
elif _order_warning(conn, f, unit_id, number, invoice_id):
|
|
order_warning = True
|
|
elif (client_choices := _recipient_warning(conn, f)) is None:
|
|
repo.update_invoice(conn, invoice_id, unit_id,
|
|
repo.find_or_create_client(conn, f["client"]), number,
|
|
f.get("date") or None, lines, note=f.get("note"),
|
|
paid_date=f.get("paid_date") or None, payment_method=f.get("payment_method") or None)
|
|
_rows_follow(conn, invoice_id)
|
|
return redirect(url_for("invoice_view", invoice_id=invoice_id))
|
|
return render_template("invoice_form.html", units=units, suggestions={}, form=f,
|
|
lines=_raw_lines(f), editing=inv, suggest=repo.field_suggestions(conn),
|
|
order_warning=order_warning, client_choices=client_choices)
|
|
form = {"unit_id": inv["unit_id"], "client": (inv["client"] or {}).get("name", ""),
|
|
"number": inv["number"], "date": inv["date"], "note": inv.get("notes"),
|
|
"paid_date": inv.get("paid_date"), "payment_method": inv.get("payment_method")}
|
|
return render_template("invoice_form.html", units=units, suggestions={}, form=form, lines=inv["lines"],
|
|
editing=inv, suggest=repo.field_suggestions(conn),
|
|
client_choices=repo.recipient_choices(conn, form["client"]))
|
|
|
|
@app.route("/invoices/<int:invoice_id>/delete", methods=["POST"])
|
|
def invoice_delete(invoice_id):
|
|
repo.delete_invoice(get_db(), invoice_id)
|
|
return redirect(url_for("invoices"))
|
|
|
|
@app.route("/invoices/<int:invoice_id>")
|
|
def invoice_view(invoice_id):
|
|
conn = get_db()
|
|
inv = repo.get_invoice(conn, invoice_id)
|
|
if not inv:
|
|
abort(404)
|
|
# offer "Neu generieren" only when a linked ledger row changed since the invoice was (re)generated
|
|
can_regen = bool(inv.get("lines_stale")) and bool(repo.linked_ledger_charges(conn, invoice_id))
|
|
return render_template("invoice_print.html", inv=inv, can_regen=can_regen)
|
|
|
|
@app.route("/invoices/<int:invoice_id>/regenerate", methods=["POST"])
|
|
def invoice_regenerate(invoice_id):
|
|
if not repo.regenerate_invoice(get_db(), invoice_id):
|
|
abort(404)
|
|
return redirect(url_for("invoice_view", invoice_id=invoice_id))
|
|
|
|
@app.route("/invoices/<int:invoice_id>/payment", methods=["GET", "POST"])
|
|
def invoice_payment(invoice_id):
|
|
"""Record the settlement of an already issued invoice — the ONLY writing path onto one that is not
|
|
a full edit. Deliberate and manual by design: an issued invoice never changes as a side effect."""
|
|
conn = get_db()
|
|
inv = repo.get_invoice(conn, invoice_id)
|
|
if not inv:
|
|
abort(404)
|
|
if request.method == "POST":
|
|
repo.set_invoice_payment(conn, invoice_id, request.form.get("paid_date"),
|
|
request.form.get("payment_method"))
|
|
return redirect(url_for("invoice_view", invoice_id=invoice_id))
|
|
return render_template("invoice_payment.html", inv=inv,
|
|
hint=repo.invoice_payment_hint(conn, invoice_id))
|
|
|
|
@app.route("/invoices/<int:invoice_id>/pdf")
|
|
def invoice_pdf(invoice_id):
|
|
inv = repo.get_invoice(get_db(), invoice_id)
|
|
if not inv:
|
|
abort(404)
|
|
html = render_template("invoice_print.html", inv=inv, standalone=True)
|
|
return send_file(io.BytesIO(pdf.render(html)), mimetype="application/pdf", as_attachment=True,
|
|
download_name=repo.invoice_filename(inv))
|
|
|
|
# --- companies (canonical clients + their address versions) ---
|
|
@app.route("/clients")
|
|
@cached_page
|
|
def clients():
|
|
q = request.args.get("q", "").strip()
|
|
return render_template("clients.html", rows=repo.companies_list(get_db(), q or None), q=q)
|
|
|
|
@app.route("/clients/<int:company_id>", methods=["GET", "POST"])
|
|
def client_edit(company_id):
|
|
conn = get_db()
|
|
comp = repo.get_company(conn, company_id)
|
|
if not comp:
|
|
abort(404)
|
|
if request.method == "POST":
|
|
f = request.form
|
|
if f.get("action") == "add_address":
|
|
repo.add_company_address(conn, company_id, f.get("address", "").strip(),
|
|
f.get("postcode", "").strip(), f.get("city", "").strip(),
|
|
make_current=bool(f.get("make_current")))
|
|
elif f.get("action") == "edit_address":
|
|
repo.update_client_address(conn, int(f["client_id"]), f.get("name", ""),
|
|
f.get("address", ""), f.get("postcode", ""), f.get("city", ""))
|
|
elif f.get("action") == "make_current":
|
|
repo.set_current_address(conn, company_id, int(f["client_id"]))
|
|
elif f.get("action") == "delete_address":
|
|
try:
|
|
repo.delete_client_address(conn, int(f["client_id"]))
|
|
except ValueError as e:
|
|
flash(i18n.t("err_address_last" if str(e) == "last_version" else "err_address_has_refs"))
|
|
elif f.get("action") == "merge":
|
|
target = f.get("merge_into", "").strip()
|
|
if target.isdigit() and int(target) != company_id:
|
|
repo.merge_companies(conn, int(target), company_id)
|
|
return redirect(url_for("client_edit", company_id=int(target)))
|
|
elif not f.get("name", "").strip():
|
|
flash(i18n.t("err_required", field=i18n.t("col_name")))
|
|
else:
|
|
repo.save_company(conn, company_id, f["name"].strip(), f.get("kind") or "company",
|
|
f.get("uid", "").strip() or None, f.get("email", "").strip() or None,
|
|
f.get("phone", "").strip() or None, f.get("notes", "").strip() or None,
|
|
int(f["current_client_id"]) if f.get("current_client_id", "").isdigit() else None)
|
|
return redirect(url_for("clients"))
|
|
return redirect(url_for("client_edit", company_id=company_id))
|
|
others = [c for c in repo.companies_list(conn) if c["id"] != company_id]
|
|
return render_template("client_form.html", comp=comp, others=others)
|
|
|
|
@app.route("/clients/<int:company_id>/delete", methods=["POST"])
|
|
def client_delete(company_id):
|
|
try:
|
|
repo.delete_company(get_db(), company_id)
|
|
except ValueError:
|
|
flash(i18n.t("err_company_has_refs"))
|
|
return redirect(url_for("client_edit", company_id=company_id))
|
|
return redirect(url_for("clients"))
|
|
|
|
# --- data-quality audit ---
|
|
@app.route("/quality")
|
|
@cached_page
|
|
def quality_page():
|
|
return render_template("quality.html", q=quality.overview(get_db()))
|
|
|
|
# --- monthly TVB Sammelmeldung (Aufenthaltsabgabe) ---
|
|
@app.route("/tvb")
|
|
def tvb_page():
|
|
import datetime
|
|
m = request.args.get("month", "")
|
|
try:
|
|
year, month = int(m[:4]), int(m[5:7])
|
|
datetime.date(year, month, 1)
|
|
except ValueError: # default: the month that is due now
|
|
prev = datetime.date.today().replace(day=1) - datetime.timedelta(days=1)
|
|
year, month = prev.year, prev.month
|
|
return render_template("tvb.html", r=tvb.monthly_report(get_db(), year, month),
|
|
month_value=f"{year:04d}-{month:02d}")
|
|
|
|
# --- quarterly Umsatzsteuer-Voranmeldung ---
|
|
@app.route("/vat")
|
|
def vat_page():
|
|
import datetime
|
|
conn = get_db()
|
|
today = datetime.date.today()
|
|
year, quarter = vat.previous_quarter(today) # default: the quarter that is due now
|
|
try:
|
|
year = int(request.args.get("year", year))
|
|
quarter = min(max(int(request.args.get("q", quarter)), 1), 4)
|
|
except ValueError:
|
|
pass
|
|
units = vat.units(conn)
|
|
names = {u["code"]: u["name"] for u in units}
|
|
unit = request.args.get("unit") if request.args.get("unit") in names else next(iter(names), None)
|
|
return render_template("vat.html", r=vat.quarterly_report(conn, unit, year, quarter), units=units,
|
|
unit=unit, unit_name=names.get(unit, unit), year=year, quarter=quarter,
|
|
years=range(today.year, today.year - 8, -1))
|
|
|
|
@app.route("/quality/companies/merge", methods=["POST"])
|
|
def quality_merge_companies():
|
|
f = request.form
|
|
if f.get("keep", "").isdigit() and f.get("drop", "").isdigit():
|
|
repo.merge_companies(get_db(), int(f["keep"]), int(f["drop"]))
|
|
return redirect(url_for("quality_page"))
|
|
|
|
@app.route("/quality/guests/merge", methods=["POST"])
|
|
def quality_merge_guests():
|
|
f = request.form
|
|
if f.get("keep", "").isdigit() and f.get("drop", "").isdigit():
|
|
repo.merge_guests(get_db(), int(f["keep"]), int(f["drop"]))
|
|
return redirect(url_for("quality_page"))
|
|
|
|
@app.route("/quality/dismiss", methods=["POST"])
|
|
def quality_dismiss():
|
|
f = request.form
|
|
if f.get("kind") and f.get("key"):
|
|
quality.dismiss(get_db(), f["kind"], f["key"], f.get("note", "").strip() or None)
|
|
return redirect(url_for("quality_page"))
|
|
|
|
@app.errorhandler(404)
|
|
def not_found(e):
|
|
return render_template("error.html", message=i18n.t("err_not_found")), 404
|
|
|
|
@app.errorhandler(500)
|
|
def server_error(e):
|
|
return render_template("error.html", message=i18n.t("err_generic")), 500
|
|
|
|
return app
|
|
|
|
|
|
def _payments(f):
|
|
"""The ledger form's payment lines as [(date, amount)] — None when the form carries no payment lines at
|
|
all, which leaves the row's payments untouched. A line without an amount is the spare empty one; the
|
|
form makes the date required as soon as an amount is typed."""
|
|
if "pay_amount" not in f:
|
|
return None
|
|
lines = zip(f.getlist("pay_date"), (parse_number(a) for a in f.getlist("pay_amount")))
|
|
return [(d, a) for d, a in lines if d and a]
|
|
|
|
|
|
def _room_number(conn, room_id):
|
|
return next((r["number"] for r in repo.rooms(conn) if r["id"] == room_id), "")
|
|
|
|
|
|
def parse_number(value):
|
|
"""Accept German decimals ('161,50', '1.234,56') and plain numbers; None if not parseable."""
|
|
if value is None:
|
|
return None
|
|
s = str(value).strip().replace(" ", "")
|
|
if not s:
|
|
return None
|
|
if "," in s: # German: comma = decimal, dot = thousands separator
|
|
s = s.replace(".", "").replace(",", ".")
|
|
try:
|
|
return float(s)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _raw_lines(form):
|
|
"""The submitted line rows verbatim, for re-rendering a form that failed validation — a rejected row
|
|
must come back filled in, not vanish."""
|
|
keys = ("worker", "period_from", "period_to", "quantity", "unit_type", "gross_per_unit", "tourism", "kind")
|
|
lists = {k: form.getlist(k) for k in keys}
|
|
n = max((len(v) for v in lists.values()), default=0)
|
|
rows = []
|
|
for i in range(n):
|
|
row = {k: (lists[k][i] if i < len(lists[k]) else "") for k in keys}
|
|
row["tourism_tax"] = row.pop("tourism")
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def _order_warning(conn, f, unit_id, number, exclude_id=None):
|
|
"""Stop the save and say which invoice is in the way when the date breaks the number order
|
|
(`repo.invoice_out_of_order`) — unless the user has ticked that the date is right on purpose."""
|
|
clash = repo.invoice_out_of_order(conn, unit_id, number, f.get("date") or None, exclude_id)
|
|
if not clash or f.get("confirm_order"):
|
|
return False
|
|
flash(i18n.t("warn_order_smaller" if clash["smaller"] else "warn_order_larger", number=number,
|
|
date=money.de_date(f.get("date")), other=clash["number"], other_date=money.de_date(clash["date"])))
|
|
return True
|
|
|
|
|
|
def _recipient_warning(conn, f):
|
|
"""None when the recipient prints an address; otherwise stop the save and return the companies the
|
|
name may mean (`repo.recipient_choices`) — unless the user has ticked that it is right without one."""
|
|
if f.get("confirm_client"):
|
|
return None
|
|
name = f.get("client", "").strip()
|
|
choices = repo.recipient_choices(conn, name)
|
|
if choices is not None:
|
|
flash(i18n.t("warn_client_no_address", name=name))
|
|
return choices
|
|
|
|
|
|
def _parse_lines(form):
|
|
"""Return (valid_lines, incomplete). A row with a name but no valid rate/quantity (or vice versa) is
|
|
reported as incomplete rather than silently dropped, so an invoice can't be undercharged unnoticed."""
|
|
workers = form.getlist("worker")
|
|
gross, qtys = form.getlist("gross_per_unit"), form.getlist("quantity")
|
|
pfrom, pto, utype = form.getlist("period_from"), form.getlist("period_to"), form.getlist("unit_type")
|
|
tourism, kinds = form.getlist("tourism"), form.getlist("kind")
|
|
|
|
def at(lst, i):
|
|
return lst[i] if i < len(lst) else None
|
|
|
|
lines, incomplete = [], []
|
|
for i, worker in enumerate(workers):
|
|
worker = worker.strip()
|
|
rate = parse_number(at(gross, i))
|
|
qty = parse_number(at(qtys, i))
|
|
if not worker and rate is None:
|
|
continue # empty row
|
|
if not worker or rate is None or not qty or qty <= 0: # a blank/zero Anz is a mistake, not a free line
|
|
incomplete.append(worker or "Zeile {}".format(i + 1))
|
|
continue
|
|
lines.append({
|
|
"worker": worker,
|
|
"period_from": at(pfrom, i) or None,
|
|
"period_to": at(pto, i) or None,
|
|
"quantity": qty,
|
|
"unit_type": at(utype, i) or "week",
|
|
"gross_per_unit": rate,
|
|
"tourism_tax": parse_number(at(tourism, i)) or 0,
|
|
"kind": at(kinds, i) or "stay",
|
|
})
|
|
return lines, incomplete
|
|
|
|
|
|
HOST, PORT = "127.0.0.1", 5000
|
|
|
|
|
|
def server_answers(host=HOST, port=PORT, timeout=0.4):
|
|
"""Is a server already listening? Windows lets a SECOND server bind the same port instead of refusing
|
|
it (SO_REUSEADDR), so every extra start left another instance running and Windows decided which one
|
|
answered — after an update often the OLD one, still serving the previous code."""
|
|
import socket
|
|
with socket.socket() as s:
|
|
s.settimeout(timeout)
|
|
return s.connect_ex((host, port)) == 0
|
|
|
|
|
|
def main():
|
|
import threading
|
|
import webbrowser
|
|
url = f"http://{HOST}:{PORT}"
|
|
if server_answers(): # already running: just open the window, never start twice
|
|
webbrowser.open(url)
|
|
return
|
|
app = create_app()
|
|
threading.Timer(1.0, lambda: webbrowser.open(url)).start()
|
|
app.run(host=HOST, port=PORT, debug=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|