"""Service layer: all reads/writes for the app, callable by the web routes and by scripts alike.
Keeps SQL in one place (with db.py). Returns plain dicts with a few computed display fields added.
"""
import re
from collections import Counter

from app import db, compute, normalize


def _dicts(rows):
    return [dict(r) for r in rows]


# --- home ---
def stats(conn):
    n = lambda sql: db.row(conn, sql)[0]  # noqa: E731
    return {
        "guests": n("SELECT COUNT(*) FROM guest"),
        "stays": n("SELECT COUNT(*) FROM stay"),
        "invoices": n("SELECT COUNT(*) FROM invoice"),
        "open": n("SELECT COUNT(*) FROM charge WHERE paid_status='open'"),
    }


# --- lookups for forms ---
def units(conn):
    return _dicts(db.rows(conn, "SELECT id, code, name FROM unit WHERE is_current=1 ORDER BY code"))


def rooms(conn):
    return _dicts(db.rows(conn, "SELECT id, number, category, unit_id FROM room ORDER BY number"))


def field_suggestions(conn):
    """Distinct past values for form autosuggest (rendered as <datalist> options)."""
    def vals(sql):
        return [r[0] for r in db.rows(conn, sql) if r[0]]
    return {
        "nationality": vals("SELECT DISTINCT nationality FROM guest WHERE nationality<>'' ORDER BY nationality"),
        "doc_type": vals("SELECT DISTINCT doc_type FROM guest WHERE doc_type<>'' ORDER BY doc_type"),
        "profession": vals("SELECT DISTINCT profession FROM guest WHERE profession<>'' ORDER BY profession"),
        "city": vals("SELECT DISTINCT home_city FROM guest WHERE home_city<>'' ORDER BY home_city"),
        "country": vals("SELECT DISTINCT home_country FROM guest WHERE home_country<>'' ORDER BY home_country"),
        # canonical company names only — variants resolve to the same company anyway
        "client": vals("SELECT name FROM company ORDER BY name"),
    }


def lines_from_stays(conn, stay_ids):
    """Pre-fill invoice lines from selected guest-register rows: worker + period + unit + employer→client;
    the rate/quantity are left blank (not derivable from the guest sheet)."""
    if not stay_ids:
        return {"lines": [], "unit_id": None, "client": None}
    ph = ",".join("?" * len(stay_ids))
    rows = db.rows(conn, f"""
        SELECT s.unit_id, cl.name AS employer, s.sheet_number,
               TRIM(COALESCE(g.last_name || ' ' || COALESCE(g.first_name,''), s.label)) AS worker,
               s.arrival, s.departure
        FROM stay s LEFT JOIN guest g ON s.guest_id = g.id LEFT JOIN client cl ON s.employer_client_id = cl.id
        WHERE s.id IN ({ph}) ORDER BY s.arrival NULLS FIRST, s.id""", list(stay_ids))
    lines = [{"worker": r["worker"], "period_from": r["arrival"], "period_to": r["departure"],
              "quantity": compute.weeks_for_period(r["arrival"], r["departure"]) or "",
              "unit_type": "week", "gross_per_unit": "",
              "tourism_tax": compute.tourism_for_period(r["arrival"], r["departure"]) or "", "kind": "stay"}
             for r in rows]
    employer = next((r["employer"] for r in rows if r["employer"]), None)
    return {"lines": lines, "unit_id": rows[0]["unit_id"] if rows else None,
            "client": billing_client_name(conn, employer,
                                          " ".join(r["sheet_number"] or "" for r in rows)) if employer else None}


def _ledger_run(conn, worker, day):
    """(nights, still_running) of the guest's UNBROKEN presence covering `day`, stitched across month and
    room boundaries from the ledger's week-segments. Drives the >10-night Ortstaxe exemption: a stay that
    began in an earlier (separately billed) month is already exempt, so its short current slice carries no
    tax either. A segment WITHOUT a 'bis' continues the run — the guest is still in the house, so the stay
    can never be judged as short (billing the first week must not charge Ortstaxe while the guest stays
    on). (None, False) when no run covers the day."""
    if not worker or not day:
        return None, False
    rows = db.rows(conn,
        "SELECT c.period_from AS pf, c.period_to AS pt FROM charge c JOIN stay s ON s.id = c.stay_id "
        "WHERE s.source_sheet = 'ledger' AND c.period_from IS NOT NULL "
        "AND s.label ILIKE '%' || ? || '%' ORDER BY c.period_from NULLS FIRST, c.id", (worker,))
    runs = []
    for r in rows:
        end = r["pt"] or r["pf"]                     # an open segment carries no end yet
        if runs and r["pf"] <= runs[-1][1]:          # segment touches the open run → extend it
            runs[-1][1] = max(runs[-1][1], end)
            runs[-1][2] = runs[-1][2] or r["pt"] is None
        else:
            runs.append([r["pf"], end, r["pt"] is None])
    for a, b, running in runs:
        if a <= day <= b:
            return compute.days_between(a, b), running
    return None, False


def billing_client_name(conn, payer, gb_ref=None):
    """The recipient name to pre-fill an invoice with — one that carries a FULL address.

    The ledger's 'Zlg. durch' is often a short form ('Muster Personal'), and a short form that matches no
    known spelling becomes a bare company with no address at all: the invoice then prints half a recipient
    block. So the
    payer is only used as-is when it resolves to an address version that HAS an address; otherwise the
    employer recorded on the linked Gästeblätter answers, since that is where the postal address lives.
    Never creates anything — this runs while only previewing an invoice."""
    def with_address(company_id):
        r = db.row(conn, """SELECT cl.name FROM company co JOIN client cl ON cl.id = co.current_client_id
                            WHERE co.id=? AND COALESCE(cl.address,'') <> ''""", (company_id,))
        if not r:
            r = db.row(conn, "SELECT name FROM client WHERE company_id=? AND COALESCE(address,'') <> '' "
                             "ORDER BY id DESC LIMIT 1", (company_id,))
        return r["name"] if r else None

    key = normalize.company_key(payer or "")
    comp = (_company_for_key(conn, key) or _alias_company(conn, key)) if key else None
    if comp is not None:
        name = with_address(comp)
        if name:
            return name
    for num in re.findall(r"\d+", gb_ref or ""):
        r = db.row(conn, """SELECT cl.name FROM stay s JOIN client cl ON cl.id = s.employer_client_id
                            WHERE s.source_sheet='guestsheet' AND s.sheet_number=?
                              AND COALESCE(cl.address,'') <> '' ORDER BY s.id DESC LIMIT 1""", (num,))
        if r:
            return r["name"]
    return payer


def lines_from_charges(conn, charge_ids):
    """Pre-fill invoice lines from selected Zimmerbelegung rows. The ledger bills per room × week-segment
    with the persons combined ('Muster, Beispiel, Probe' · 2 Wochen); an invoice bills per PERSON — so
    the names are split, each person's contiguous segments merge into ONE line with a continuous period
    (quantities added), and a gap in between starts a new line. A 'Storno' row comes in as a
    nicht-steuerbare storno line, a lump row (Betrag but no per-unit rate) as one 'total' line at the
    Betrag. Unit follows the rooms; the client is taken from the payer ('Zlg. Durch') when present."""
    if not charge_ids:
        return {"lines": [], "unit_id": None, "client": None}
    ph = ",".join("?" * len(charge_ids))
    rows = db.rows(conn, f"""
        SELECT c.period_from, c.period_to, c.quantity, c.unit_type, c.gross_per_unit, c.line_net_override,
               c.persons, c.art, c.gb_ref, s.label AS occupant, s.unit_id,
               COALESCE(c.payer, iv.payment_method) AS payer
        FROM charge c JOIN stay s ON c.stay_id = s.id LEFT JOIN invoice iv ON c.invoice_id = iv.id
        WHERE c.id IN ({ph}) ORDER BY c.period_from NULLS FIRST, c.id""", list(charge_ids))

    def line(worker, period_from, period_to, quantity, unit_type, gross_per_unit, kind="stay"):
        if quantity is not None and abs(quantity - round(quantity)) < 1e-9:
            quantity = int(round(quantity))             # 5.0 -> 5 in the form
        return {"worker": worker, "period_from": period_from, "period_to": period_to,
                "quantity": quantity if quantity is not None else "", "unit_type": unit_type,
                "gross_per_unit": gross_per_unit if gross_per_unit is not None else "",
                "tourism_tax": "", "kind": kind}

    specials, segments = [], []
    for r in rows:
        storno = is_storno(r)
        gpu, qty, utype = r["gross_per_unit"], r["quantity"], r["unit_type"] or "week"
        if not gpu and r["line_net_override"] is not None:   # lump row: bill the exact Betrag as one 'total' line
            specials.append(line(r["occupant"], r["period_from"], r["period_to"],
                                 qty or 1, "total", round(r["line_net_override"] * 1.1, 2)))
        elif storno:
            # invoice bills per PERSON: split the names, one storno line each at the room Satz. So a half-week
            # storno in a DZ (2 Pers.) prints as TWO lines of 0,5 Woche × Satz (= Anz × Satz per line, honest),
            # summing to 0,5 × Satz × 2 — the same total the ledger row carries.
            names = [n.strip() for n in (r["occupant"] or "").split(",")
                     if n.strip() and "storno" not in n.lower()]
            workers = names or ["Stornogebühr"] * int(r["persons"] or 1)
            for w in workers:
                specials.append(line(w, r["period_from"], r["period_to"], qty, utype, gpu, "storno"))
        else:
            names = [n.strip() for n in (r["occupant"] or "").split(",") if n.strip()] or [r["occupant"]]
            for name in names:                               # the row's Anz/Satz apply per person
                segments.append({"worker": name, "period_from": r["period_from"], "period_to": r["period_to"],
                                 "quantity": qty, "unit_type": utype, "gross_per_unit": gpu})
    merged = []
    for seg in sorted(segments, key=lambda x: ((x["worker"] or "").lower(), str(x["gross_per_unit"]),
                                               x["period_from"] or "")):
        last = merged[-1] if merged else None
        contiguous = (last and last["worker"] == seg["worker"] and last["gross_per_unit"] == seg["gross_per_unit"]
                      and last["unit_type"] == seg["unit_type"] and last["period_to"] and seg["period_from"]
                      and seg["period_from"] <= last["period_to"])
        if contiguous:                                       # same man, no break -> one continuous line
            last["period_to"] = max(last["period_to"], seg["period_to"] or last["period_to"])
            last["quantity"] = (last["quantity"] or 0) + (seg["quantity"] or 0)
        else:
            merged.append(dict(seg))
    lines = specials + [line(**m) for m in merged]
    for ln in lines:                                         # the billing rules on the final periods:
        if ln["kind"] != "stay":
            continue
        if ln["unit_type"] == "week":                        # >= 4 nights -> whole weeks, rounded up
            weeks = compute.weeks_for_period(ln["period_from"], ln["period_to"])
            if weeks:
                ln["quantity"] = weeks
        # Ortstaxe on the WHOLE continuous stay, not just this month's slice: a worker already here last
        # month is past the >10-night exemption, so a short slice of the stay this month carries no tax
        # either — and a guest who has not left yet is not a short stay at all.
        nights, still_running = _ledger_run(conn, ln["worker"], ln["period_from"])
        ln["tourism_tax"] = "" if still_running else             (compute.tourism_amount(ln["period_from"], ln["period_to"], nights) or "")
    lines.sort(key=lambda x: (x["period_from"] or "", (x["worker"] or "").lower()))
    # suggest the payer as the client — resolved to the canonical company; payment CHANNELS are no client
    payer = next((r["payer"] for r in rows if r["payer"]), None)
    if payer and payer.strip().lower().split()[0] in ("bar", "bank"):
        payer = None
    gb_refs = " ".join(r["gb_ref"] or "" for r in rows)
    return {"lines": lines, "unit_id": rows[0]["unit_id"] if rows else None,
            "client": billing_client_name(conn, payer, gb_refs) if payer else None}


def _company_for_key(conn, key):
    r = db.row(conn, "SELECT company_id FROM company_key WHERE key=?", (key,))
    return r[0] if r else None


def _alias_company(conn, key):
    """A short alias ('Alpenbau', 'Muster Personal' — as the ledger's 'Zlg. durch' writes them) resolves
    to the ONE company whose registered spelling starts with it. Ambiguous or very short prefixes resolve
    to nothing (a new name must not silently attach to the wrong company)."""
    if not key or len(key) < 5:
        return None
    rows = db.rows(conn, "SELECT DISTINCT company_id FROM company_key WHERE key ILIKE ? ORDER BY company_id", (key + "%",))
    return rows[0][0] if len(rows) == 1 else None


def recipient_choices(conn, name):
    """Read-only, before an invoice is saved to `name`: None when the name resolves, as
    `find_or_create_client` resolves it, to an address version that carries an address. Otherwise the
    invoice would print no recipient address (§11 UStG) — an unknown name even creates an empty company,
    and a short name may fit several companies of one group. Then the companies the name may mean come
    back instead, with their current address — those whose spelling starts with the whole name, else with
    its first word ('Muster Bau' → 'Muster Bau GmbH' and 'Muster Bau Holding', not every 'Muster…'); []
    when none."""
    key = normalize.company_key(name or "")
    comp_id = (_company_for_key(conn, key) or _alias_company(conn, key)) if key else None
    if comp_id is not None and db.row(conn, """
            SELECT 1 FROM client WHERE COALESCE(address,'') <> '' AND id = COALESCE(
                (SELECT current_client_id FROM company WHERE id=?), (SELECT max(id) FROM client WHERE company_id=?))""",
            (comp_id, comp_id)):
        return None
    words = (name or "").split()
    for stem in (key, normalize.company_key(words[0]) if words else ""):
        if len(stem) < 3:
            continue
        choices = [dict(r) for r in db.rows(conn, """
            SELECT co.name, cl.address, cl.postcode, cl.city
            FROM company co JOIN client cl ON cl.id = co.current_client_id
            WHERE COALESCE(cl.address,'') <> '' AND co.id IN (SELECT company_id FROM company_key WHERE key LIKE ?)
            ORDER BY co.name""", (stem + "%",))]
        if choices:
            return choices
    return []


def _new_company(conn, name, kind, key):
    comp_id = db.execute(conn, "INSERT INTO company (name, kind) VALUES (?,?)", (name, kind))
    db.execute(conn, "INSERT INTO company_key (key, company_id) VALUES (?,?)", (key, comp_id))
    return comp_id


def _new_client_version(conn, comp_id, name, kind, address=None, postcode=None, city=None, phone=None,
                        make_current=True):
    cid = db.execute(conn, "INSERT INTO client (kind, name, address, postcode, city, company_id) "
                     "VALUES (?,?,?,?,?,?)", (kind, name, address, postcode, city, comp_id))
    if phone:   # phone is a COMPANY attribute (single source) — versions only carry name + address
        db.execute(conn, "UPDATE company SET phone=COALESCE(phone,?) WHERE id=?", (phone, comp_id))
    if make_current:
        db.execute(conn, "UPDATE company SET current_client_id=? WHERE id=?", (cid, comp_id))
    return cid


def find_or_create_client(conn, name, kind="company"):
    """Resolve a typed name to a client row via the company registry: ANY known spelling of a company maps
    to its current address version, so typing a variant never creates a duplicate. A new name creates a
    company + its first version."""
    name = (name or "").strip()
    if not name:
        return None
    key = normalize.company_key(name)
    if not key:   # punctuation-only garbage — fall back to a bare client row
        existing = db.row(conn, "SELECT id FROM client WHERE lower(name)=lower(?) ORDER BY id", (name,))
        return existing[0] if existing else db.execute(conn,
            "INSERT INTO client (kind, name) VALUES (?, ?)", (kind, name))
    comp_id = _company_for_key(conn, key)
    if comp_id is None:
        comp_id = _alias_company(conn, key)
        if comp_id is not None:   # learn the alias so it keeps resolving (and survives consolidation)
            db.execute(conn, "INSERT INTO company_key (key, company_id) VALUES (?,?)", (key, comp_id))
    if comp_id is None:
        comp_id = _new_company(conn, name, kind, key)
        return _new_client_version(conn, comp_id, name, kind)
    cur = db.row(conn, "SELECT current_client_id FROM company WHERE id=?", (comp_id,))
    if cur and cur[0]:
        return cur[0]
    cl = db.row(conn, "SELECT id FROM client WHERE company_id=? ORDER BY id DESC LIMIT 1", (comp_id,))
    if cl:
        db.execute(conn, "UPDATE company SET current_client_id=? WHERE id=?", (cl[0], comp_id))
        return cl[0]
    return _new_client_version(conn, comp_id, name, kind)


# --- guest register view ---
def register_rows(conn, search=None):
    sql = """
        SELECT s.id AS stay_id, g.id AS guest_id, g.last_name, g.first_name, g.birth_date,
               g.nationality, g.doc_type, g.doc_number, g.profession,
               g.home_street, g.home_postcode, g.home_city, g.home_country, g.phone,
               r.number AS room, u.code AS unit, cl.name AS employer,
               -- the address version this stay was registered under; if that one carries no address
               -- (a sheet may name only the firm), fall back to the company's current one — as a
               -- BLOCK, never mixing a street from one version with the town of another
               CASE WHEN cl.address IS NOT NULL THEN cl.address   ELSE cur.address   END AS employer_street,
               CASE WHEN cl.address IS NOT NULL THEN cl.postcode  ELSE cur.postcode  END AS employer_postcode,
               CASE WHEN cl.address IS NOT NULL THEN cl.city      ELSE cur.city      END AS employer_city,
               s.arrival, s.departure, s.sheet_number, s.hauptwohnsitz, s.tourism_liable, s.notes
        FROM stay s
        LEFT JOIN guest g ON s.guest_id = g.id
        LEFT JOIN room r ON s.room_id = r.id
        LEFT JOIN unit u ON s.unit_id = u.id
        LEFT JOIN client cl ON s.employer_client_id = cl.id
        LEFT JOIN company co ON cl.company_id = co.id
        LEFT JOIN client cur ON co.current_client_id = cur.id
        WHERE s.source_sheet = 'guestsheet'
    """
    params = []
    if search:
        like = f"%{search}%"
        sql += (" AND (g.last_name ILIKE ? OR g.first_name ILIKE ? OR cl.name ILIKE ? OR r.number = ?)")
        params = [like, like, like, search]
    sql += " ORDER BY s.id"   # entry order breaks ties in the sort below
    out = _dicts(db.rows(conn, sql, params))
    for row in out:
        row["days"] = compute.days_between(row["arrival"], row["departure"])
        # the employer's postal address as ONE grid column, in the order it is written on an envelope
        town = " ".join(x for x in (row["employer_postcode"], row["employer_city"]) if x)
        row["employer_address"] = ", ".join(x for x in (row["employer_street"], town) if x)
    return _grouped_by_sheet_number(out)


def _grouped_by_sheet_number(rows):
    """Order the register by ARRIVAL, chronologically — a guest entered late must land at the arrival
    date, not at the end of the list where the sheet happened to be typed.

    Every row of ONE Gästeblatt stays together, directly under the first of that group: a room change
    continues the same GB number, and the group travels with its earliest arrival. The group is
    (unit, GB number, person) — the same number is reused in later years for a different guest, and those
    must NOT be pulled together; a row without a number is its own group. Within a group: by arrival, and
    equal dates keep the entry order.

    The same grouping answers 'Abg' where the sheet leaves it empty: the Abgabe exemption counts the guest's
    WHOLE continuous stay (T-AAG §4), so the first segment of a room change must not read as a short stay."""
    def group(r):
        return (r["unit"], r["sheet_number"], r["guest_id"]) if r["sheet_number"] else ("", "", r["stay_id"])
    first, span = {}, {}
    for r in rows:
        g = group(r)
        # the group sorts by its earliest arrival; the row id only breaks ties (= the entry order)
        stamp = (r["arrival"] or "", r["stay_id"])
        first[g] = min(first[g], stamp) if g in first else stamp
        arr, dep, open_end = span.get(g, (r["arrival"], r["departure"], False))
        span[g] = (min(x for x in (arr, r["arrival"]) if x) if (arr or r["arrival"]) else None,
                   max((x for x in (dep, r["departure"]) if x), default=None),
                   open_end or r["departure"] is None)
    for r in rows:
        if r["tourism_liable"] is None:
            arr, dep, open_end = span[group(r)]
            liable = compute.tourism_liable(arr, dep)
            # still running: certain only once the stay has already passed the exempt length
            r["tourism_liable"] = None if open_end and liable != 0 else liable
    return sorted(rows, key=lambda r: (first[group(r)], r["arrival"] or "", r["stay_id"]))


# --- room ledger view ---
def _payment(row):
    """Is a Zimmerbelegung row paid ('paid' / 'cash' / 'open' / 'unknown'), and on which day?

    The status picked in the row goes before its date: a part payment carries both (175,00 paid, the
    16,00 Ortstaxe still open). A paid INVOICE pays its rows too — the payment is often recorded on the
    invoice only — but only where the row itself says nothing, or a part payment would suddenly read as
    settled. An empty cell on a billed row IS open; without an invoice number nothing is due yet."""
    status = row["paid_status"]
    payment = ("open" if status == "open"
               else "paid" if status == "paid" or (not status and row["paid_date"])
               else "cash" if status == "cash"
               else "paid" if not status and row["invoice_paid_date"]
               else "open" if row["invoice_no"] else "unknown")
    return payment, row["paid_date"] or (row["invoice_paid_date"] if payment == "paid" else None)


def is_storno(row):
    """A Storno row says so in the name ('Stornogebühr Muster') or the Art. It is nicht steuerbarer
    Schadenersatz: invoiced without USt, and outside the USt-Voranmeldung's base."""
    return "storno" in ((row["occupant"] or "") + " " + (row["art"] or "")).lower()


def ledger_rows(conn, room=None, search=None):
    sql = """
        SELECT c.id AS charge_id, s.id AS stay_id, r.number AS room, u.code AS unit,
               COALESCE(g.last_name || ' ' || COALESCE(g.first_name,''), s.label) AS occupant,
               c.period_from, c.period_to, c.gb_ref, c.persons, c.quantity, c.unit_type,
               c.gross_per_unit, c.art, c.notes, c.tourism_tax, COALESCE(c.payer, i.payment_method) AS payer,
               c.line_net_override, c.paid_status, c.paid_date,
               COALESCE(c.rnr, i.number) AS invoice_no, i.id AS invoice_id, i.date AS invoice_date,
               -- a paid invoice pays its rows too: a payment recorded on the invoice must show in the
               -- Zimmerbelegung
               i.paid_date AS invoice_paid_date
        FROM charge c
        JOIN stay s ON c.stay_id = s.id
        LEFT JOIN room r ON s.room_id = r.id
        LEFT JOIN unit u ON s.unit_id = u.id
        LEFT JOIN guest g ON s.guest_id = g.id
        LEFT JOIN invoice i ON c.invoice_id = i.id
        WHERE s.source_sheet = 'ledger'
    """
    params = []
    if room:
        sql += " AND r.number = ?"
        params.append(room)
    if search:
        like = f"%{search}%"
        sql += " AND (COALESCE(g.last_name,'') ILIKE ? OR s.label ILIKE ? OR i.number ILIKE ?)"
        params += [like, like, like]
    sql += " ORDER BY c.id"   # entry order breaks ties in the sort below
    out = _dicts(db.rows(conn, sql, params))
    payments = {}
    for p in db.rows(conn, "SELECT charge_id, paid_date, amount FROM payment ORDER BY paid_date, id"):
        payments.setdefault(p["charge_id"], []).append((p["paid_date"], p["amount"]))
    when, last = {}, {}
    for row in out:
        # the first date the row shows: its Von — but a long-term tenant's monthly rent may carry only a
        # 'bis', or only the date the rent was paid
        row["first_date"] = row["period_from"] or row["period_to"] or row["paid_date"]
        # a row without any date stays behind the row entered before it in the same room
        when[row["charge_id"]] = last[row["room"]] = row["first_date"] or last.get(row["room"], "")
        # the ledger bills a row for EVERYONE in it: Betrag = Anz × Satz × Personen (2 Pers. × 1 Woche ×
        # 161 = 322). A Betrag typed into the row always wins; only a row without one is computed, and then
        # the person count belongs in it — otherwise the list shows half the money.
        row["amount"] = compute.line_gross((row["quantity"] or 0) * (row["persons"] or 1),
                                           row["gross_per_unit"] or 0, row["line_net_override"])
        row["days"] = compute.days_between(row["period_from"], row["period_to"])
        # part payment: what is still owed of Betrag + Ortstaxe (175 of 191 paid: 16 open)
        row["payments"] = payments.get(row["charge_id"], [])
        row["paid_amount"] = round(sum(a for _, a in row["payments"]), 2) if row["payments"] else None
        row["total"] = round(row["amount"] + (row["tourism_tax"] or 0), 2)
        row["open_amount"] = (round(row["total"] - row["paid_amount"], 2)
                              if row["paid_amount"] is not None and row["paid_amount"] + 0.005 < row["total"]
                              else None)
        row["payment"], row["paid_on"] = _payment(row)
    # Chronologically, earliest date first: in entry order a booking typed in advance would sit between
    # unrelated months. One day's rows go by room, and within a room the entry order stands.
    return sorted(out, key=lambda r: (when[r["charge_id"]], r["room"] or "", r["charge_id"]))


# --- stay CRUD (guest register) ---
def get_stay(conn, stay_id):
    r = db.row(conn, "SELECT * FROM stay WHERE id=?", (stay_id,))
    if not r:
        return None
    stay = dict(r)
    stay["guest"] = {}
    if stay["guest_id"]:
        stay["guest"] = dict(db.row(conn, "SELECT * FROM guest WHERE id=?", (stay["guest_id"],)))
    stay["employer_name"] = ""
    if stay["employer_client_id"]:
        c = db.row(conn, "SELECT name FROM client WHERE id=?", (stay["employer_client_id"],))
        stay["employer_name"] = c[0] if c else ""
    if stay["tourism_liable"] is None:      # pre-select the rule's answer, so saving the form records it
        # over the guest's WHOLE stay under this Gästeblatt number, room changes included
        sp = db.row(conn, "SELECT MIN(arrival) a, MAX(departure) d, SUM((departure IS NULL)::int) open_end "
                          "FROM stay WHERE source_sheet='guestsheet' AND guest_id IS NOT DISTINCT FROM ? "
                          "AND COALESCE(sheet_number,'')=COALESCE(?,'')",
                    (stay["guest_id"], stay["sheet_number"]))
        liable = compute.tourism_liable(sp["a"], sp["d"])
        stay["tourism_liable"] = None if sp["open_end"] and liable != 0 else liable
    return stay


def find_or_create_guest(conn, last_name, first_name, **fields):
    """Reuse a guest only on an exact person match (name + birth date); otherwise a distinct person with the
    same name must not be merged. Never updates an existing guest — use update_guest for edits."""
    existing = db.row(conn,
        "SELECT id FROM guest WHERE lower(last_name)=lower(?) AND lower(COALESCE(first_name,''))=lower(?) "
        "AND COALESCE(birth_date,'')=COALESCE(?,'') ORDER BY id",
        (last_name, first_name or "", fields.get("birth_date")))
    if existing:
        return existing[0]
    cols = ["last_name", "first_name"] + list(fields.keys())
    vals = [last_name, first_name] + list(fields.values())
    ph = ",".join("?" * len(cols))
    return db.execute(conn, f"INSERT INTO guest ({','.join(cols)}) VALUES ({ph})", vals)


def merge_guests(conn, keep_id, merge_id):
    """Two guest rows that are the same person (typo variants): move the stays, drop the duplicate.
    Master fields of the surviving guest win; missing ones are filled from the merged row."""
    if keep_id == merge_id:
        return
    keep = db.row(conn, "SELECT * FROM guest WHERE id=?", (keep_id,))
    merged = db.row(conn, "SELECT * FROM guest WHERE id=?", (merge_id,))
    if not keep or not merged:
        return
    for col in keep.keys():
        if col != "id" and keep[col] in (None, "") and merged[col] not in (None, ""):
            db.execute(conn, f"UPDATE guest SET {col}=? WHERE id=?", (merged[col], keep_id))
    db.execute(conn, "UPDATE stay SET guest_id=? WHERE guest_id=?", (keep_id, merge_id))
    db.execute(conn, "DELETE FROM guest WHERE id=?", (merge_id,))


def update_guest(conn, guest_id, **fields):
    cols = ", ".join(f"{k}=?" for k in fields)
    db.execute(conn, f"UPDATE guest SET {cols} WHERE id=?", list(fields.values()) + [guest_id])


def save_stay(conn, stay_id, guest_id, room_id, unit_id, employer_client_id, arrival, departure, sheet_number,
              hauptwohnsitz=None, tourism_liable=None, notes=None):
    args = (guest_id, room_id, unit_id, employer_client_id, arrival, departure, sheet_number,
            hauptwohnsitz, tourism_liable, notes)
    if stay_id:
        db.execute(conn, "UPDATE stay SET guest_id=?, room_id=?, unit_id=?, employer_client_id=?, arrival=?, "
                   "departure=?, sheet_number=?, hauptwohnsitz=?, tourism_liable=?, notes=? WHERE id=?", args + (stay_id,))
    else:
        stay_id = db.execute(conn, "INSERT INTO stay (guest_id,room_id,unit_id,employer_client_id,arrival,"
                             "departure,sheet_number,hauptwohnsitz,tourism_liable,notes,source_sheet) "
                             "VALUES (?,?,?,?,?,?,?,?,?,?, 'guestsheet')", args)
    close_ledger_on_departure(conn, sheet_number, room_id)
    return stay_id


PER_UNIT = {"week": 7, "day": 1, "night": 1}   # nights per Anz — month and lump rows derive nothing


def _ledger_row(conn, charge_id):
    return db.row(conn, """SELECT c.id, c.stay_id, s.label, s.room_id, c.gb_ref, c.period_from, c.period_to,
                                  c.unit_type, c.gross_per_unit, c.persons, c.rnr, c.invoice_id, c.quantity,
                                  c.line_net_override, c.tourism_tax, c.payer, c.art, c.notes,
                                  c.paid_status, c.paid_date
                           FROM charge c JOIN stay s ON s.id = c.stay_id WHERE c.id=?""", (charge_id,))


def _gb_refs(gb_ref):
    return [t for t in re.split(r"\D+", gb_ref or "") if t]


def _departures(conn, refs, room_id, since):
    """{Gästeblatt-Nr.: departure} for the people on a row who have left. Only stays overlapping the row
    count: the numbers are reused years later, and an ancient namesake must not pass as departed."""
    if not refs:
        return {}
    return {r["sheet_number"]: r["departure"] for r in db.rows(conn, """
        SELECT sheet_number, departure FROM stay
        WHERE source_sheet='guestsheet' AND sheet_number IN (%s) AND room_id IS NOT DISTINCT FROM ?
          AND departure IS NOT NULL AND departure >= ? ORDER BY id""" % ",".join("?" * len(refs)),
        refs + [room_id, since])}


def _closing(conn, r, refs, bis, persons):
    """Anz, Betrag (as line_net_override) and Ortstaxe of a row ending at `bis`, the way the form computes
    them: Anz = Nächte / 7 for Wochen, Betrag = Anz x Satz x Personen. The Ortstaxe is nights x Satz x
    Personen, and nothing at all once the WHOLE stay — the span of these Gästeblätter — passes 10 nights
    (T-AAG §4). A room change keeps the number, so the span runs through the Betrieb's other rooms too — there
    only for the row's own people, since an old number written on two guests' sheets must not lend one the
    other's stay: the stays that reach this row and, without a break, those before them. A number reused
    years earlier is not part of it."""
    per = PER_UNIT.get(r["unit_type"])
    nights = compute.days_between(r["period_from"], bis)
    # a started week counts in full from 4 nights on, as on the invoice; shorter: the fraction
    weeks = compute.weeks_for_period(r["period_from"], bis) if r["unit_type"] == "week" else None
    qty = weeks or (round(nights / per, 6) if per and nights else r["quantity"])
    override = (round(qty * (r["gross_per_unit"] or 0) * persons / 1.1, 6) if per and nights
                else r["line_net_override"])
    stays = db.rows(conn, """SELECT s.arrival, s.departure, s.room_id, g.last_name FROM stay s
                             LEFT JOIN room rm ON rm.id = s.room_id LEFT JOIN guest g ON g.id = s.guest_id
                             WHERE s.source_sheet='guestsheet' AND s.sheet_number IN (%s)
                               AND s.arrival IS NOT NULL AND s.arrival <= ? AND s.departure IS NOT NULL
                               AND rm.unit_id IS NOT DISTINCT FROM (SELECT unit_id FROM room WHERE id = ?)
                             ORDER BY s.departure DESC""" % ",".join("?" * len(refs)), refs + [bis, r["room_id"]])
    label = (r["label"] or "").lower()
    stays = [s for s in stays if s["room_id"] == r["room_id"]
             or any(w in label for w in re.findall(r"\w{3,}", (s["last_name"] or "").lower()))]
    start = end = None
    for s in stays:                                   # newest first; the run ends at the first gap
        if end is None and s["departure"] < r["period_from"]:
            break                                     # nothing reaches this row
        if end is not None and s["departure"] < start:
            break
        start = min(start or s["arrival"], s["arrival"])
        end = end or s["departure"]
    taxe = round(compute.tourism_amount(r["period_from"], bis,
                                        compute.days_between(start, end)) * persons, 2) or None
    return qty, override, taxe


def _name_match(guest, names):
    """The one name on a row that is this register guest — 'Last First' first, else the last name alone;
    None unless exactly one (father and son of the same name on one row: rather not guess)."""
    full = f"{guest['last_name']} {guest['first_name'] or ''}".strip().lower()
    hits = ([n for n in names if n.lower().startswith(full)]
            or [n for n in names if guest["last_name"].lower() in n.lower()])
    return hits[0] if len(hits) == 1 else None


def _split_off(conn, r, gb, departure):
    """Give the person behind Gästeblatt `gb` a row of their own that ends on their departure; the row
    keeps the others. Only where the row says unambiguously who that is — one name and one Gästeblatt per
    person — and the money derives from Anz x Satz; otherwise None and the row stays as it is."""
    refs = _gb_refs(r["gb_ref"])
    names = [n.strip() for n in (r["label"] or "").split(",") if n.strip()]
    if (len(refs) < 2 or len(names) != len(refs) or (r["persons"] or 1) != len(refs)
            or not PER_UNIT.get(r["unit_type"]) or r["gross_per_unit"] is None):
        return None
    g = db.row(conn, """SELECT g.last_name, g.first_name FROM stay s JOIN guest g ON g.id = s.guest_id
                        WHERE s.source_sheet='guestsheet' AND s.sheet_number=? AND s.room_id IS NOT DISTINCT FROM ?
                          AND s.departure=? ORDER BY s.id""", (gb, r["room_id"], departure))
    name = _name_match(g, names) if g else None
    if not name:
        return None
    qty, override, taxe = _closing(conn, r, [gb], departure, 1)
    cid = create_ledger_row(conn, r["room_id"], name, r["period_from"], departure, 1, qty, r["unit_type"],
                            r["gross_per_unit"], override * 1.1 if override is not None else None, None, None,
                            r["art"], payer=r["payer"], gb_ref=gb, notes=r["notes"], tourism_tax=taxe)
    names.remove(name)
    refs.remove(gb)
    db.execute(conn, "UPDATE stay SET label=? WHERE id=?", (", ".join(names), r["stay_id"]))
    share = len(refs) / (len(refs) + 1)          # an already closed row keeps its price per person
    db.execute(conn, "UPDATE charge SET gb_ref=?, persons=?, line_net_override=line_net_override*?, "
                     "tourism_tax=tourism_tax*? WHERE id=?", (", ".join(refs), len(refs), share, share, r["id"]))
    return cid


def _rejoin(conn, charge_id):
    """Rows that ended up alike — same room, same start and end, same terms, nothing billed — are ONE row
    again: a group that leaves together is entered sheet by sheet, and the first sheet split its person
    off. Returns the surviving row (the oldest), or None."""
    r = _ledger_row(conn, charge_id)
    if not r or not r["period_to"] or r["rnr"] or r["invoice_id"] or not r["gb_ref"]:
        return None
    twins = [t["id"] for t in db.rows(conn, """
        SELECT c.id FROM charge c JOIN stay s ON s.id = c.stay_id
        WHERE c.source_sheet='ledger' AND s.room_id IS NOT DISTINCT FROM ? AND c.id <> ? AND c.period_from = ?
          AND c.period_to = ? AND c.gross_per_unit IS NOT DISTINCT FROM ? AND c.unit_type IS NOT DISTINCT FROM ?
          AND c.payer IS NOT DISTINCT FROM ? AND c.art IS NOT DISTINCT FROM ?
          AND c.rnr IS NULL AND c.invoice_id IS NULL AND c.gb_ref IS NOT NULL
          AND (c.notes IS NOT DISTINCT FROM ? OR c.notes IS NULL OR CAST(? AS text) IS NULL)
        ORDER BY c.id""",
        (r["room_id"], r["id"], r["period_from"], r["period_to"], r["gross_per_unit"], r["unit_type"],
         r["payer"], r["art"], r["notes"], r["notes"]))]
    if not twins:
        return None
    rows = [_ledger_row(conn, i) for i in sorted([r["id"]] + twins)]
    keep = rows[0]
    names = [n.strip() for x in rows for n in (x["label"] or "").split(",") if n.strip()]
    refs = [gb for x in rows for gb in _gb_refs(x["gb_ref"])]
    persons = sum(x["persons"] or 1 for x in rows)
    notes = next((x["notes"] for x in rows if x["notes"]), None)
    qty, override, taxe = _closing(conn, keep, refs, keep["period_to"], persons)
    db.execute(conn, "UPDATE stay SET label=? WHERE id=?", (", ".join(names), keep["stay_id"]))
    db.execute(conn, "UPDATE charge SET gb_ref=?, persons=?, quantity=?, line_net_override=?, tourism_tax=?, "
                     "notes=? WHERE id=?", (", ".join(refs), persons, qty, override, taxe, notes, keep["id"]))
    for x in rows[1:]:
        delete_charge(conn, x["id"])
    return keep["id"]


def close_ledger_on_departure(conn, sheet_number, room_id):
    """A departure typed into a Gästeblatt writes itself into the room's Zimmerbelegung row — and a
    CHANGED departure moves it again, which is the same job: the row's 'bis' is simply the last departure
    of the people it bills. Rules the data forces:
      * a row ends only when EVERY Gästeblatt on it has left, and gets Anz, Betrag and Ortstaxe with the
        date, the way the form computes them — unless it already ends that day: then what was typed stands
        (a guest may pay half a week for one night);
      * whoever leaves while the others stay on — or leave later — gets a row of their own that ends on
        their own departure, or the row bills that person for nights not stayed; rows that end up alike
        are one row again (`_rejoin`), so a group leaving together stays one row;
      * the departure belongs to the NEWEST row of that Gästeblatt in the room: an older one was closed at
        a month end and continued, and its dates are that month's;
      * Gästeblatt numbers are reused years later, so only stays overlapping the row count;
      * a row that is already billed (RNr or invoice) is left untouched — its money sits on a document,
        and correcting that is a deliberate act.
    Returns the id of the row it last wrote, or None."""
    if not sheet_number:
        return None
    rows = [r for r in db.rows(conn, """SELECT c.id, c.gb_ref FROM charge c JOIN stay s ON s.id = c.stay_id
                                        WHERE c.source_sheet='ledger' AND s.room_id IS NOT DISTINCT FROM ? AND c.gb_ref IS NOT NULL
                                          AND c.period_from IS NOT NULL
                                        ORDER BY c.period_from DESC NULLS LAST, c.id DESC""", (room_id,))
            if str(sheet_number) in _gb_refs(r["gb_ref"])]
    if not rows:
        return None
    r = _ledger_row(conn, rows[0]["id"])
    if r["rnr"] or r["invoice_id"]:
        return None                           # already billed: dates and money belong to a document
    written = None
    refs = _gb_refs(r["gb_ref"])
    done = _departures(conn, refs, room_id, r["period_from"])
    last = max(done.values()) if set(done) == set(refs) else None    # None: somebody is still here
    end = max((d for d in (r["period_to"], last) if d), default=None)
    split = []
    for gb in sorted(done):
        if end and done[gb] >= end:
            continue                          # leaves with the row's end, as it stands or as it moves
        if last is None or done[gb] < last:
            new = _split_off(conn, r, gb, done[gb])
            if new:
                split.append(new)
                r = _ledger_row(conn, r["id"])   # its names, Gästeblätter and Personen just changed
    refs = _gb_refs(r["gb_ref"])
    if last is not None and set(refs) <= set(done):
        bis = max(done[gb] for gb in refs)
        if r["period_to"] != bis or r["quantity"] is None:   # it already ends that day: the typed Anz and Betrag stand
            qty, override, taxe = _closing(conn, r, refs, bis, r["persons"] or 1)
            if (r["period_to"], r["quantity"], r["line_net_override"], r["tourism_tax"]) != (bis, qty, override, taxe):
                db.execute(conn, "UPDATE charge SET period_to=?, quantity=?, line_net_override=?, tourism_tax=? "
                                 "WHERE id=?", (bis, qty, override, taxe, r["id"]))
                written = r["id"]
    # join alike rows only now: before the row moved on, a row split off it would look alike and join back
    for new in split:
        written = _rejoin(conn, new) or new
    return _rejoin(conn, r["id"]) or written


def carry_ledger_to_room(conn, sheet_number, old_room, new_room, day, guest):
    """A room change moves the Zimmerbelegung along with the Gästeblatt: the guest's row in the old room
    ends on `day` — the departure from it closed it — and a row in the new room starts there, like
    'Speichern und weiterführen': same name, Einheit, Satz and Zahlung durch, nothing billed yet. A row
    already continued in the old room from that day is moved instead of doubled. The ended row's Ortstaxe
    is cleared as at a month end: the stay goes on, so its slice cannot be judged alone. Returns the new
    room's row, or None when the ledger holds nothing to carry (it is entered as usual)."""
    if not sheet_number or not day or not new_room or new_room == old_room:
        return None
    rows = [_ledger_row(conn, r["id"]) for r in db.rows(conn, """
                SELECT c.id, c.gb_ref FROM charge c JOIN stay s ON s.id = c.stay_id
                WHERE c.source_sheet='ledger' AND s.room_id IS NOT DISTINCT FROM ? AND c.gb_ref IS NOT NULL
                  AND (c.period_from = ? OR c.period_to = ?) ORDER BY c.id""", (old_room, day, day))
            if str(sheet_number) in _gb_refs(r["gb_ref"])]
    ended = next((r for r in rows if r["period_to"] == day and r["period_from"] < day), None)
    started = next((r for r in rows if r["period_from"] == day and _gb_refs(r["gb_ref"]) == [str(sheet_number)]
                    and not (r["rnr"] or r["invoice_id"])), None)
    if ended and not (ended["rnr"] or ended["invoice_id"]):
        db.execute(conn, "UPDATE charge SET tourism_tax=NULL WHERE id=?", (ended["id"],))
    if started:
        db.execute(conn, "UPDATE stay SET room_id=?, unit_id=(SELECT unit_id FROM room WHERE id=?) WHERE id=?",
                   (new_room, new_room, started["stay_id"]))
        if started["period_to"] == day:                  # the departure closed it after no night at all
            db.execute(conn, "UPDATE charge SET period_to=NULL, quantity=NULL, line_net_override=NULL, "
                             "tourism_tax=NULL WHERE id=?", (started["id"],))
        return started["id"]
    if not ended:
        return None
    names = [n.strip() for n in (ended["label"] or "").split(",") if n.strip()]
    alone = _gb_refs(ended["gb_ref"]) == [str(sheet_number)]
    name = (ended["label"] if alone else _name_match(guest, names)
            or f"{guest['last_name']} {guest['first_name'] or ''}".strip())
    return create_ledger_row(conn, new_room, name, day, None, (ended["persons"] or 1) if alone else 1, None,
                             ended["unit_type"], ended["gross_per_unit"], None, None, None, None,
                             payer=ended["payer"], gb_ref=str(sheet_number))


def delete_stay(conn, stay_id):
    if db.row(conn, "SELECT 1 FROM charge WHERE stay_id=?", (stay_id,)):
        raise ValueError("has_charges")
    db.execute(conn, "DELETE FROM stay WHERE id=?", (stay_id,))


# --- ledger row (charge) editing ---
def get_charge(conn, charge_id):
    r = db.row(conn, "SELECT c.*, s.label AS occupant, s.room_id, rm.number AS room "
               "FROM charge c JOIN stay s ON c.stay_id=s.id LEFT JOIN room rm ON s.room_id=rm.id WHERE c.id=?",
               (charge_id,))
    if not r:
        return None
    d = dict(r)
    d["betrag"] = round((d["line_net_override"] or 0) * 1.1, 2) if d.get("line_net_override") is not None else None
    d["payments"] = charge_payments(conn, charge_id)
    return d


def charge_payments(conn, charge_id):
    """The payments recorded on a Zimmerbelegung row, oldest first: [(paid_date, amount)]."""
    return [(r["paid_date"], r["amount"]) for r in db.rows(
        conn, "SELECT paid_date, amount FROM payment WHERE charge_id=? ORDER BY paid_date, id", (charge_id,))]


def _save_payments(conn, charge_id, payments):
    db.execute(conn, "DELETE FROM payment WHERE charge_id=?", (charge_id,))
    for paid_date, amount in payments:
        db.execute(conn, "INSERT INTO payment (charge_id, paid_date, amount) VALUES (?,?,?)",
                   (charge_id, paid_date, amount))


def _paid_status(paid_status, paid_date, paid=None, total=None):
    """Writing a date into 'bezahlt' IS marking the row paid — a row with a payment date must not read
    OFFEN just because the status dropdown stayed empty. The date decides — but only where nothing else
    was said: a status picked by hand always wins.
    Recorded PAYMENTS (`paid` = their sum) decide on their own: a part payment keeps the row OPEN (175,00
    paid, 16,00 Ortstaxe open), and once they cover Betrag + Ortstaxe it is PAID (241,50 in May, the
    remaining 174,50 in October)."""
    if paid is not None and total is not None:
        return "open" if paid + 0.005 < total else "paid"
    return paid_status or ("paid" if paid_date else None)


def _settle(paid_status, paid_date, payments, total):
    """A row's 'Bezahlt' and 'bezahlt am'. With payments recorded, they decide both — bezahlt am is the
    day of the last one; without, the dropdown and date stand (the usual single full payment)."""
    if not payments:
        return _paid_status(paid_status, paid_date), paid_date
    return (_paid_status(paid_status, paid_date, sum(a for _, a in payments), total),
            max(d for d, _ in payments))


def charge_total(quantity, gross_per_unit, line_net_override, persons, tourism_tax):
    """What this ledger row owes altogether: Betrag (Anz × Satz × Personen) plus the Ortstaxe — the figure
    a part payment is measured against."""
    amount = compute.line_gross((quantity or 0) * (persons or 1), gross_per_unit or 0, line_net_override)
    return round(amount + (tourism_tax or 0), 2)


def _link_typed_rnr(conn, charge_id, rnr):
    """An RNr typed into a Zimmerbelegung row that names exactly one invoice of the row's own Betrieb LINKS
    the row to it — the link is what fills 'Re vom' and makes the number clickable. Nothing on the invoice
    changes; a number matching nothing leaves the row as is. Only a row WITHOUT a link, and only an invoice
    dated within a year of the stay: bare numbers ('15') recur year after year — re-saving such a row must
    never move its link."""
    if not rnr:
        return
    hits = db.rows(conn, "SELECT i.id FROM invoice i JOIN stay s ON s.unit_id = i.unit_id "
                         "JOIN charge c ON c.stay_id = s.id WHERE c.id=? AND i.number=? AND c.invoice_id IS NULL "
                         "AND ABS(i.date::date - COALESCE(c.period_to, c.period_from)::date) <= 366 ORDER BY i.id",
                   (charge_id, rnr))
    if len(hits) == 1:
        db.execute(conn, "UPDATE charge SET invoice_id=? WHERE id=?", (hits[0]["id"], charge_id))


def update_charge(conn, charge_id, occupant, period_from, period_to, persons, quantity, unit_type,
                  gross_per_unit, betrag, paid_status, paid_date, art, payer=None, rnr=None, gb_ref=None,
                  notes=None, tourism_tax=None, payments=None):
    """`payments`: [(paid_date, amount)] replaces the row's payments; None leaves them as they are — they
    still decide 'Bezahlt' against the (possibly changed) Betrag."""
    override = round(betrag / 1.1, 6) if betrag is not None else None   # pin the exact Betrag
    if payments is None:
        payments = charge_payments(conn, charge_id)
    paid_status, paid_date = _settle(paid_status, paid_date, payments,
                                     charge_total(quantity, gross_per_unit, override, persons, tourism_tax))
    old = db.row(conn, "SELECT c.period_from, c.period_to, c.quantity, c.unit_type, c.gross_per_unit, "
                 "c.line_net_override, c.art, c.invoice_id, c.source_sheet, c.stay_id, s.label AS occupant "
                 "FROM charge c JOIN stay s ON c.stay_id=s.id WHERE c.id=?", (charge_id,))
    db.execute(conn, "UPDATE charge SET period_from=?, period_to=?, persons=?, quantity=?, unit_type=?, "
               "gross_per_unit=?, line_net_override=?, paid_status=?, paid_date=?, art=?, "
               "payer=?, rnr=?, gb_ref=?, notes=?, tourism_tax=? WHERE id=?",
               (period_from, period_to, persons, quantity, unit_type, gross_per_unit, override,
                paid_status, paid_date, art, payer, rnr, gb_ref, notes, tourism_tax, charge_id))
    _save_payments(conn, charge_id, payments)
    db.execute(conn, "UPDATE stay SET label=? WHERE id=?", (occupant, old["stay_id"]))
    _link_typed_rnr(conn, charge_id, rnr)
    # a billing-relevant change on a linked ledger row marks its invoice for regeneration (payment fields don't).
    # Of the Art only whether it makes the row a Storno counts: 'M' (mailed) flagged every sent invoice.
    billing = (occupant, period_from, period_to, quantity, unit_type, gross_per_unit, override,
               is_storno({"occupant": "", "art": art}))
    was = tuple(old[k] for k in ("occupant", "period_from", "period_to", "quantity", "unit_type",
                                 "gross_per_unit", "line_net_override")) + (is_storno({"occupant": "", "art": old["art"]}),)
    if billing != was and old["invoice_id"] and old["source_sheet"] == "ledger":
        db.execute(conn, "UPDATE invoice SET lines_stale=1 WHERE id=?", (old["invoice_id"],))


def create_ledger_row(conn, room_id, occupant, period_from, period_to, persons, quantity, unit_type,
                      gross_per_unit, betrag, paid_status, paid_date, art, payer=None, rnr=None, gb_ref=None,
                      notes=None, tourism_tax=None, payments=None):
    """Add a room-ledger occupancy: its own ledger-sourced stay + charge. The unit follows the chosen room;
    Betrag is pinned via line_net_override. The stay carries NO dates — the charge's period is the single
    source."""
    unit_id = None
    if room_id:
        r = db.row(conn, "SELECT unit_id FROM room WHERE id=?", (room_id,))
        unit_id = r[0] if r else None
    override = round(betrag / 1.1, 6) if betrag is not None else None
    payments = payments or []
    paid_status, paid_date = _settle(paid_status, paid_date, payments,
                                     charge_total(quantity, gross_per_unit, override, persons, tourism_tax))
    sid = db.execute(conn, "INSERT INTO stay (room_id,unit_id,label,source_sheet) "
                     "VALUES (?,?,?, 'ledger')", (room_id, unit_id, occupant))
    cid = db.execute(conn, "INSERT INTO charge (stay_id,period_from,period_to,persons,quantity,unit_type,"
                     "gross_per_unit,line_net_override,paid_status,paid_date,art,payer,rnr,"
                     "gb_ref,notes,tourism_tax,source_sheet) "
                     "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, 'ledger')",
                     (sid, period_from, period_to, persons, quantity, unit_type, gross_per_unit, override,
                      paid_status, paid_date, art, payer, rnr, gb_ref, notes, tourism_tax))
    _save_payments(conn, cid, payments)
    _link_typed_rnr(conn, cid, rnr)
    return cid


def continue_ledger_row(conn, charge_id):
    """Month-end: the row is closed at the invoicing date, but the guest stays on. Continue the occupancy
    in a NEW row that starts exactly where the closed one ends — same room, names, persons, Satz, Einheit,
    Zahlung durch and Gästeblatt-Nr., with Anz/Betrag/Ortstaxe left open because nothing is billed yet.
    The REGISTER stay is deliberately untouched: a month-end 'bis' is not a departure."""
    r = db.row(conn, """SELECT c.period_to, c.persons, c.unit_type, c.gross_per_unit, c.art, c.payer,
                               c.gb_ref, s.room_id, s.label
                        FROM charge c JOIN stay s ON s.id = c.stay_id
                        WHERE c.id=? AND c.source_sheet='ledger'""", (charge_id,))
    if not r or not r["period_to"]:
        return None
    # the guest stays on, so this month's slice is part of a longer stay and its Ortstaxe cannot be judged
    # from the slice alone (>10 nights over the WHOLE stay is exempt, T-AAG §4). Leave it to the invoice,
    # which counts the continuous run — an auto-proposed figure here would simply be wrong.
    db.execute(conn, "UPDATE charge SET tourism_tax=NULL WHERE id=?", (charge_id,))
    return create_ledger_row(conn, r["room_id"], r["label"], r["period_to"], None, r["persons"], None,
                             r["unit_type"], r["gross_per_unit"], None, None, None, r["art"],
                             payer=r["payer"], gb_ref=r["gb_ref"])


def delete_charge(conn, charge_id):
    row = db.row(conn, "SELECT stay_id, invoice_id, source_sheet FROM charge WHERE id=?", (charge_id,))
    if not row:
        return
    sid = row["stay_id"]
    if row["invoice_id"] and row["source_sheet"] == "ledger":   # its invoice no longer matches the ledger
        db.execute(conn, "UPDATE invoice SET lines_stale=1 WHERE id=?", (row["invoice_id"],))
    db.execute(conn, "DELETE FROM charge WHERE id=?", (charge_id,))
    if not db.row(conn, "SELECT 1 FROM charge WHERE stay_id=?", (sid,)):
        db.execute(conn, "DELETE FROM stay WHERE id=? AND source_sheet='ledger'", (sid,))


# --- invoices ---
def invoices_list(conn, search=None):
    sql = """
        SELECT i.id, i.unit_id, i.number, i.date, i.paid_date, u.code AS unit, cl.name AS client,
               led.n_led, led.n_paid, led.n_open, led.led_paid_date
        FROM invoice i
        LEFT JOIN unit u ON i.unit_id = u.id
        LEFT JOIN client cl ON i.client_id = cl.id
        LEFT JOIN (SELECT invoice_id AS iid, COUNT(*) AS n_led,
                          -- a status picked by hand beats the date: a part payment carries BOTH
                          -- (175,00 paid on 04.09., 16,00 Ortstaxe open) and stays open; cash is paid
                          SUM((paid_status IN ('paid', 'cash')
                               OR (paid_status IS NULL AND paid_date IS NOT NULL))::int) AS n_paid,
                          SUM((paid_status = 'open')::int) AS n_open,
                          MAX(paid_date) AS led_paid_date
                   FROM charge WHERE source_sheet = 'ledger' AND invoice_id IS NOT NULL
                   GROUP BY invoice_id) led ON led.iid = i.id
    """
    params = []
    if search:
        like = f"%{search}%"
        sql += " WHERE i.number ILIKE ? OR cl.name ILIKE ?"
        params = [like, like]
    sql += " ORDER BY i.date DESC NULLS LAST, i.id DESC"   # no LIMIT: the grid is client-side sorted/filtered, show every year
    out = _dicts(db.rows(conn, sql, params))
    totals = _invoice_totals(conn)
    for row in out:
        row["gross"] = totals.get(row["id"], 0.0)
        row.update(_payment_state(row))
        row["number_key"] = invoice_number_key(row["number"], row["date"])
    # Newest DATE first; within a date unit by unit, in the order they were set up, each with its own
    # numbers in order: the units number independently, so sorting by the number alone would interleave
    # 2026-35 with 26-35, and by id alone a month-end run would read 2026-49/51/50/48.
    out.sort(key=lambda r: (r["date"] or "", -(r["unit_id"] or 0), r["number_key"][1], r["id"]), reverse=True)
    return out


def invoice_number_key(number, date):
    """Sortable (year, sequence) behind an invoice number, whatever era spelled it: '2026-52', '26-31',
    '49-2015' (SEQ-YEAR), '202201' (glued), '2018 -58'. The NUMBER's own year wins — a January invoice
    still numbered with the old year belongs to that year — with the invoice date as the fallback. A
    combined number ('41/46') has no single sequence, so its first group orders it."""
    import datetime
    num_year, date_year = normalize.invoice_year(number) or "", (date or "")[:4]
    year = ""
    if num_year.isdigit():
        # the 2-digit-prefix rule reads a COMBINED number ('27 30', '69, 73') as a year — believe the
        # number's year only next to the invoice's own date (a January invoice still numbered with last
        # year is one apart; a decade apart means it was never a year)
        if date_year.isdigit():
            year = num_year if abs(int(num_year) - int(date_year)) <= 1 else ""
        elif 1990 <= int(num_year) <= datetime.date.today().year + 1:
            year = num_year
    year = year or date_year
    if not year.isdigit():
        return (0, 0)
    seq = normalize.invoice_seq(number, year)
    if seq is None:
        m = re.search(r"\d+", str(number or ""))
        seq = int(m.group()) % 10000 if m else 0
    return (int(year), seq)


def _payment_state(row):
    """Is this invoice settled? The invoice's own paid_date is the answer when it has one — most payments
    are recorded in the ZIMMERBELEGUNG though ('bezahlt' per row), so the linked ledger rows answer for the
    rest. An invoice nobody has said anything about (no paid_date, no linked row) is UNKNOWN, not open:
    the receivables must not fill up with invoices nobody tracks. Read-only — writing the date onto an
    issued invoice stays a deliberate act."""
    if row["paid_date"]:
        return {"payment_state": "paid", "paid_on": row["paid_date"]}
    if row["n_open"]:
        return {"payment_state": "open", "paid_on": None}
    if row["n_paid"]:
        return {"payment_state": "paid", "paid_on": row["led_paid_date"]}
    if row["n_led"]:
        return {"payment_state": "open", "paid_on": None}   # row exists, 'bezahlt' still empty = not yet
    return {"payment_state": "unknown", "paid_on": None}


def _invoice_totals(conn):
    """Rechnungsbetrag per invoice for the list — the SAME number the invoice itself prints, so it comes
    from the same computation over the same lines (see `get_invoice`): its invoice-sourced charges.
    One query for all of them; hundreds of invoices must not become hundreds of round trips."""
    lines = {}
    for r in db.rows(conn, """SELECT c.invoice_id AS iid, c.quantity, c.gross_per_unit, c.line_net_override,
                                     c.tourism_tax, c.vat_rate, c.kind
                              FROM charge c JOIN stay s ON s.id = c.stay_id
                              WHERE c.invoice_id IS NOT NULL AND s.source_sheet = 'invoice' ORDER BY c.id"""):
        line = dict(r)
        line["tourism_tax"] = line["tourism_tax"] or 0
        line["kind"] = line["kind"] or "stay"
        lines.setdefault(r["iid"], []).append(line)
    return {iid: compute.invoice_gross(ls, sum(ln["tourism_tax"] for ln in ls)) for iid, ls in lines.items()}


def next_invoice_number(conn, unit_id, today=None):
    """Suggest the next per-unit number for the CURRENT year. Numbering restarts yearly, so scope by the
    invoice date's year; ignore concatenated historical tokens (e.g. '5556', '303132') by taking only a
    clean 1-3 digit trailing sequence. The seeded prefix carries a year ('2026-' / '26-') — swap it for
    the current one so the suggestion survives New Year."""
    import re
    import datetime
    row = db.row(conn, "SELECT invoice_prefix FROM unit WHERE id=?", (unit_id,))
    prefix = (row["invoice_prefix"] if row else "") or ""
    year = str((today or datetime.date.today()).year)
    if re.match(r"\d{4}\D", prefix):
        prefix = year + prefix[4:]
    elif re.match(r"\d{2}\D", prefix):
        prefix = year[2:] + prefix[2:]
    biggest = 0
    for r in db.rows(conn, "SELECT number FROM invoice WHERE unit_id=? AND date LIKE ? ORDER BY id", (unit_id, year + "%")):
        m = re.search(r"(?<!\d)(\d{1,3})\s*$", (r["number"] or "").strip())
        if m:
            biggest = max(biggest, int(m.group(1)))
    return f"{prefix}{biggest + 1}"


def invoice_number_taken(conn, unit_id, number, exclude_id=None):
    """True if this unit already has an invoice with this number (§11 fortlaufende Nummer). The forms
    enforce it; the Datenprüfung lists duplicates that got in by any other way."""
    number = (number or "").strip()
    if not number:
        return False
    r = db.row(conn, "SELECT id FROM invoice WHERE unit_id=? AND number=? AND id<>COALESCE(?, -1)",
               (unit_id, number, exclude_id))
    return r is not None


def invoice_out_of_order(conn, unit_id, number, date, exclude_id=None):
    """Invoice dates follow the numbers: within a Betrieb and year, no smaller number may carry a later
    date and no larger one an earlier date — said at once, before the invoice is saved. Returns the nearest
    invoice that breaks the order as {number, date, smaller} — `smaller` when its number is below this
    one — or None."""
    key = invoice_number_key(number, date) if (unit_id and number and date) else (0, 0)
    if key == (0, 0):
        return None
    clashes = []
    for r in db.rows(conn, "SELECT number, date FROM invoice WHERE unit_id=? AND id<>COALESCE(?, -1) "
                           "AND number IS NOT NULL AND date IS NOT NULL ORDER BY id", (unit_id, exclude_id)):
        k = invoice_number_key(r["number"], r["date"])
        if k[0] == key[0] and ((k < key and r["date"] > date) or (k > key and r["date"] < date)):
            clashes.append((abs(k[1] - key[1]), {"number": r["number"], "date": r["date"], "smaller": k < key}))
    return min(clashes, key=lambda c: c[0])[1] if clashes else None


def create_invoice(conn, unit_id, client_id, number, date, lines, note=None,
                   paid_date=None, payment_method=None, link_charge_ids=None):
    """lines: {worker, period_from, period_to, quantity, unit_type, gross_per_unit, tourism_tax}.
    Each line becomes a charge on a fresh invoice-sourced stay (kept out of the ledger/register views).
    Tourism tax lives per line ONLY; invoice totals derive from the lines at read time.
    `note` overrides the printed payment note (e.g. "already paid"); None keeps the default text.
    `link_charge_ids`: the Zimmerbelegung rows this invoice was generated from — they get the invoice
    number written back (the RNr column) and are linked, so the ledger shows the invoice."""
    note = (note or "").strip() or None
    inv_id = db.execute(conn,
        "INSERT INTO invoice (unit_id,client_id,number,date,notes,paid_date,payment_method) "
        "VALUES (?,?,?,?,?,?,?)",
        (unit_id, client_id, number, date, note, paid_date, payment_method))
    _insert_charge_lines(conn, inv_id, unit_id, lines)
    if link_charge_ids:
        ph = ",".join("?" * len(link_charge_ids))
        db.execute(conn, f"UPDATE charge SET invoice_id=?, rnr=? WHERE id IN ({ph}) AND source_sheet='ledger'",
                   [inv_id, number] + list(link_charge_ids))
    return inv_id


def _insert_charge_lines(conn, invoice_id, unit_id, lines):
    """Each line → a fresh invoice-sourced stay + charge. A 'storno' line defaults to vat_rate 0 (nicht steuerbar)."""
    for ln in lines:
        kind = ln.get("kind") or "stay"
        vat_rate = ln.get("vat_rate")
        if vat_rate is None:
            vat_rate = 0.0 if kind == "storno" else 0.10
        label = ln.get("worker") or ln.get("description") or ""
        sid = db.execute(conn, "INSERT INTO stay (unit_id,label,source_sheet) VALUES (?,?, 'invoice')",
                         (unit_id, label))
        db.execute(conn,
            "INSERT INTO charge (stay_id,invoice_id,period_from,period_to,quantity,unit_type,gross_per_unit,"
            "tourism_tax,vat_rate,kind,description) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            (sid, invoice_id, ln["period_from"], ln["period_to"], ln["quantity"], ln["unit_type"],
             ln["gross_per_unit"], ln.get("tourism_tax") or 0, vat_rate, kind, ln.get("description")))


def _clear_invoice_lines(conn, invoice_id):
    """Remove the invoice's OWN (app-created) charges + their stays. Linked ledger (Zimmerbelegung) rows are
    LEFT in place and still linked — they belong to the ledger and keep this invoice's number so the ledger's
    RNr stays a working link across an edit/regenerate. Only deleting the whole invoice unlinks them."""
    inv_stays = [r[0] for r in db.rows(conn,
        "SELECT s.id FROM stay s JOIN charge c ON c.stay_id=s.id "
        "WHERE c.invoice_id=? AND s.source_sheet='invoice' ORDER BY s.id", (invoice_id,))]
    db.execute(conn, "DELETE FROM charge WHERE invoice_id=? AND stay_id IN "
               "(SELECT id FROM stay WHERE source_sheet='invoice')", (invoice_id,))
    for sid in set(inv_stays):
        db.execute(conn, "DELETE FROM stay WHERE id=?", (sid,))


def update_invoice(conn, invoice_id, unit_id, client_id, number, date, lines, note=None,
                   paid_date=None, payment_method=None):
    """Edit an invoice header + lines. Its own lines are replaced; linked ledger rows stay in the ledger and stay
    linked (`_clear_invoice_lines`)."""
    note = (note or "").strip() or None
    db.execute(conn, "UPDATE invoice SET unit_id=?, client_id=?, number=?, date=?, notes=?, "
               "paid_date=?, payment_method=? WHERE id=?",
               (unit_id, client_id, number, date, note, paid_date, payment_method, invoice_id))
    _clear_invoice_lines(conn, invoice_id)
    _insert_charge_lines(conn, invoice_id, unit_id, lines)
    return invoice_id


def linked_ledger_charges(conn, invoice_id):
    """The Zimmerbelegung rows this invoice was generated from (linked via the RNr write-back)."""
    return [r[0] for r in db.rows(conn, "SELECT id FROM charge WHERE invoice_id=? AND source_sheet='ledger' ORDER BY id",
                                  (invoice_id,))]


def invoice_payment_hint(conn, invoice_id):
    """What the linked Zimmerbelegung rows say about the settlement — a PROPOSAL for the manual payment
    action, nothing more. Nothing copies it onto an issued invoice by itself; taking it over is a deliberate
    click. Values that disagree across the rows are offered as a list, not picked."""
    rows = db.rows(conn, "SELECT paid_date, payer FROM charge "
                         "WHERE invoice_id=? AND source_sheet='ledger' ORDER BY id", (invoice_id,))
    def uniq(field):
        return sorted({(r[field] or "").strip() for r in rows if (r[field] or "").strip()})
    dates, payers = uniq("paid_date"), uniq("payer")
    if not (dates or payers):
        return None
    return {"paid_date": dates[0] if len(dates) == 1 else None, "dates": dates,
            "payment_method": payers[0] if len(payers) == 1 else None, "payers": payers}


def set_invoice_payment(conn, invoice_id, paid_date=None, payment_method=None):
    """Record how an invoice was settled. Touches ONLY the two settlement fields — never a line, a number,
    a date or the recipient — so it is safe on an already issued document. Manual callers only."""
    db.execute(conn, "UPDATE invoice SET paid_date=?, payment_method=? WHERE id=?",
               (paid_date or None, (payment_method or "").strip() or None, invoice_id))


def regenerate_invoice(conn, invoice_id):
    """Rebuild the invoice's lines from its linked Zimmerbelegung rows at their CURRENT values — the one-click
    path after a price/period fix in the ledger. The header (client, number, date, notes, payment) stays; the
    ledger links (and their RNr as typed) survive; the stale flag clears."""
    inv = db.row(conn, "SELECT unit_id FROM invoice WHERE id=?", (invoice_id,))
    ids = linked_ledger_charges(conn, invoice_id)
    if not inv or not ids:
        return False
    lines = lines_from_charges(conn, ids)["lines"]
    for ln in lines:                                    # pre-fill lines use '' for empty form fields
        ln["quantity"] = ln["quantity"] or None
        ln["gross_per_unit"] = ln["gross_per_unit"] or None
        ln["tourism_tax"] = ln["tourism_tax"] or 0
    _clear_invoice_lines(conn, invoice_id)
    _insert_charge_lines(conn, invoice_id, inv["unit_id"], lines)
    ph = ",".join("?" * len(ids))
    db.execute(conn, f"UPDATE charge SET invoice_id=? WHERE id IN ({ph})", [invoice_id] + ids)
    db.execute(conn, "UPDATE invoice SET lines_stale=0 WHERE id=?", (invoice_id,))
    return True


def _gb_by_name(conn, r, names):
    """{name: Gästeblatt-Nr.} for the people on a row, through the register — the stays of its Gästeblätter
    in its room that overlap it (the numbers recur years later). {} for a row without Gästeblätter; None
    unless every name and every Gästeblatt pair up exactly once."""
    refs = _gb_refs(r["gb_ref"])
    if not refs:
        return {}
    if len(refs) != len(names):
        return None
    pairs = {}
    for gb in refs:
        g = db.row(conn, """SELECT g.last_name, g.first_name FROM stay s JOIN guest g ON g.id = s.guest_id
                            WHERE s.source_sheet='guestsheet' AND s.sheet_number=? AND s.room_id IS NOT DISTINCT FROM ?
                              AND s.arrival <= ? AND COALESCE(s.departure, '9999') >= ? ORDER BY s.id""",
                     (gb, r["room_id"], r["period_to"] or "9999", r["period_from"]))
        name = _name_match(g, names) if g else None
        if not name or name in pairs:
            return None
        pairs[name] = gb
    return pairs


def _row_says(r, t):
    """Does ledger row `r` already bill what target `t` (one invoice line's worth) bills — same people,
    same dates, same money? How the money is written does not count: the invoice may bill 11 Nächte x 23
    where the row says 1,57 Wochen x 161, and the weekly Satz stays on the row."""
    names = ", ".join(n.strip() for n in (r["label"] or "").split(",") if n.strip())
    betrag = charge_total(r["quantity"], r["gross_per_unit"], r["line_net_override"], r["persons"], None)
    return (names == t["label"] and (r["persons"] or 1) == t["persons"]
            and (r["period_from"], r["period_to"]) == (t["period_from"], t["period_to"])
            and abs(betrag - t["betrag"]) < 0.005 and abs((r["tourism_tax"] or 0) - (t["tourism_tax"] or 0)) < 0.005)


def sync_rows_to_invoice(conn, invoice_id):
    """Make the Zimmerbelegung rows an invoice was made from say what the invoice bills, so the ledger
    corrects itself instead of drifting from the invoices. The price is settled on the
    invoice — alone in a double room one company pays 175, the next stays at 161 — and the rows follow: a
    changed period, Anz, Satz or Ortstaxe is corrected on the row, a second line for the same person
    (another price or period) becomes a row of its own, and people billed apart get a row each. Only where
    the lines are unambiguously that row's: every person on it billed by name and on no other linked row,
    the money Anz x Satz, no part payment recorded yet. Not an invoice whose rows were changed AFTER it
    (`lines_stale`): there the row is the newer word, and "Neu generieren" the way to bring them together.
    Returns (corrected row ids, new row ids)."""
    inv = db.row(conn, "SELECT lines_stale FROM invoice WHERE id=?", (invoice_id,))
    if not inv or inv["lines_stale"]:
        return [], []
    lines = {}
    for ln in db.rows(conn, """SELECT s.label AS worker, c.period_from, c.period_to, c.quantity, c.unit_type,
                                      c.gross_per_unit, c.tourism_tax
                               FROM charge c JOIN stay s ON s.id = c.stay_id
                               WHERE c.invoice_id=? AND s.source_sheet='invoice' AND c.kind='stay'
                               ORDER BY c.period_from NULLS FIRST, c.id""", (invoice_id,)):
        lines.setdefault((ln["worker"] or "").strip(), []).append(ln)
    rows = [_ledger_row(conn, i) for i in linked_ledger_charges(conn, invoice_id)]
    names_of = {r["id"]: [n.strip() for n in (r["label"] or "").split(",") if n.strip()] for r in rows}
    billed_on = Counter(n for names in names_of.values() for n in names)
    fixed, new = [], []
    for r in rows:
        names = names_of[r["id"]]
        if (not names or len(names) != (r["persons"] or 1) or r["gross_per_unit"] is None
                or is_storno({"occupant": r["label"], "art": r["art"]}) or charge_payments(conn, r["id"])
                or any(billed_on[n] > 1 or n not in lines for n in names)):
            continue
        groups = {}                                   # people billed alike stay together on one row
        for n in names:
            key = tuple((l["period_from"], l["period_to"], l["quantity"], l["unit_type"], l["gross_per_unit"])
                        for l in lines[n])
            groups.setdefault(key, []).append(n)
        gbs = _gb_by_name(conn, r, names) if len(groups) > 1 else {}
        if gbs is None:
            continue
        targets = [{"label": ", ".join(people), "persons": len(people), "period_from": frm, "period_to": to,
                    "quantity": qty, "unit_type": unit_type, "gross_per_unit": rate,
                    "betrag": round((qty or 0) * (rate or 0) * len(people), 2),
                    "tourism_tax": round(sum(lines[n][i]["tourism_tax"] or 0 for n in people), 2) or None,
                    "gb_ref": (", ".join(gbs[n] for n in people) if gbs
                               else None if len(groups) > 1 else r["gb_ref"])}
                   for key, people in groups.items() for i, (frm, to, qty, unit_type, rate) in enumerate(key)]
        if len(targets) == 1 and _row_says(r, targets[0]):
            continue
        t = targets[0]
        db.execute(conn, "UPDATE stay SET label=? WHERE id=?", (t["label"], r["stay_id"]))
        db.execute(conn, "UPDATE charge SET period_from=?, period_to=?, persons=?, quantity=?, unit_type=?, "
                         "gross_per_unit=?, line_net_override=?, tourism_tax=?, gb_ref=? WHERE id=?",
                   (t["period_from"], t["period_to"], t["persons"], t["quantity"], t["unit_type"],
                    t["gross_per_unit"], round(t["betrag"] / 1.1, 6), t["tourism_tax"], t["gb_ref"], r["id"]))
        fixed.append(r["id"])
        for t in targets[1:]:
            cid = create_ledger_row(conn, r["room_id"], t["label"], t["period_from"], t["period_to"],
                                    t["persons"], t["quantity"], t["unit_type"], t["gross_per_unit"], t["betrag"],
                                    r["paid_status"], r["paid_date"], r["art"], payer=r["payer"], rnr=r["rnr"],
                                    gb_ref=t["gb_ref"], notes=r["notes"], tourism_tax=t["tourism_tax"])
            db.execute(conn, "UPDATE charge SET invoice_id=? WHERE id=?", (invoice_id, cid))
            new.append(cid)
    return fixed, new


def get_invoice(conn, invoice_id):
    r = db.row(conn, "SELECT * FROM invoice WHERE id=?", (invoice_id,))
    if not r:
        return None
    inv = dict(r)
    inv["unit"] = dict(db.row(conn, "SELECT * FROM unit WHERE id=?", (inv["unit_id"],))) if inv["unit_id"] else {}
    inv["client"] = dict(db.row(conn, "SELECT * FROM client WHERE id=?", (inv["client_id"],))) \
        if inv["client_id"] else {}
    if inv["client"].get("company_id"):   # identity fields come from the company (kind decides Herrn/Firma)
        comp = db.row(conn, "SELECT kind, uid FROM company WHERE id=?", (inv["client"]["company_id"],))
        if comp:
            inv["client"]["kind"] = comp["kind"]
            inv["client"]["uid"] = comp["uid"]   # single source: uid lives on the company only
    # the invoice's OWN lines are the invoice-sourced charges; linked ledger rows (RNr write-back) are
    # bookkeeping, not lines
    lines = db.rows(conn, """
        SELECT c.quantity, c.unit_type, c.gross_per_unit, c.line_net_override, c.tourism_tax,
               c.vat_rate, c.kind, c.period_from, c.period_to, COALESCE(c.description, s.label) AS worker
        FROM charge c JOIN stay s ON c.stay_id = s.id
        WHERE c.invoice_id = ? AND s.source_sheet = 'invoice' ORDER BY c.id""", (invoice_id,))
    inv["lines"] = []
    for ln in lines:
        d = dict(ln)
        d["tourism_tax"] = d.get("tourism_tax") or 0
        d["kind"] = d.get("kind") or "stay"
        r = 0 if d["kind"] == "storno" else d.get("vat_rate")   # storno is nicht steuerbar → display at rate 0
        d["net_per_unit"] = compute.net_per_unit(d["gross_per_unit"] or 0, r)
        d["line_net"] = compute.line_net(d["quantity"] or 0, d["gross_per_unit"] or 0, d["line_net_override"], r)
        d["line_vat"] = compute.line_vat(d["quantity"] or 0, d["gross_per_unit"] or 0, d["line_net_override"], r)
        d["line_gross"] = compute.line_gross(d["quantity"] or 0, d["gross_per_unit"] or 0, d["line_net_override"], r)
        inv["lines"].append(d)
    inv["tourism_tax"] = round(sum(d["tourism_tax"] for d in inv["lines"]), 2)  # sum of the per-line column
    inv["net"] = compute.invoice_net(inv["lines"])          # steuerbarer Nettobetrag (storno excluded)
    inv["vat"] = compute.invoice_vat(inv["lines"])
    inv["vat_breakdown"] = compute.invoice_vat_breakdown(inv["lines"])   # one 'USt X %' row per rate present
    inv["storno"] = compute.invoice_storno(inv["lines"])    # nicht steuerbar subtotal
    inv["gross"] = compute.invoice_gross(inv["lines"], inv["tourism_tax"])
    # short company form for the PDF filename: the payer ('Zlg. durch', e.g. 'Alpenbau') off the linked
    # ledger rows — the invoice's own lines carry no payer. Most frequent non-null wins; None -> full name.
    pr = db.row(conn, "SELECT payer FROM charge WHERE invoice_id=? AND payer IS NOT NULL "
                      "GROUP BY payer ORDER BY COUNT(*) DESC, payer NULLS FIRST LIMIT 1", (invoice_id,))
    inv["payer"] = pr[0] if pr else None
    return inv


def _period_de(a, b):
    """'2026-05-25','2026-06-28' -> '25.05.-28.06.2026' (year once when equal, else on both)."""
    def parts(iso):
        y, m, d = iso.split("-")
        return d, m, y
    if a and b:
        da, ma, ya = parts(a); db_, mb, yb = parts(b)
        return f"{da}.{ma}.-{db_}.{mb}.{yb}" if ya == yb else f"{da}.{ma}.{ya}-{db_}.{mb}.{yb}"
    one = a or b
    if one:
        d, m, y = parts(one); return f"{d}.{m}.{y}"
    return ""


def invoice_filename(inv):
    """PDF name: '<Nr> <Firma-Kurz>, <Name1, Name2, …> <von.-bis.jahr>.pdf'. Number kept verbatim (units
    may number '26-23' or '2026-…'); company is the payer short form with the full client name as fallback;
    occupants are the invoice's stay lines, de-duplicated in order."""
    number = (inv.get("number") or f"Rechnung_{inv.get('id')}").strip()
    company = (inv.get("payer") or (inv.get("client") or {}).get("name") or "").strip()
    seen, workers = set(), []
    for ln in inv.get("lines", []):
        w = (ln.get("worker") or "").strip()
        if ln.get("kind") == "stay" and w and w not in seen:
            seen.add(w); workers.append(w)
    froms = [ln["period_from"] for ln in inv.get("lines", []) if ln.get("period_from")]
    tos = [ln["period_to"] for ln in inv.get("lines", []) if ln.get("period_to")]
    period = _period_de(min(froms) if froms else None, max(tos) if tos else None)
    head = ", ".join(p for p in [company, *workers] if p)
    name = " ".join(p for p in [number, head, period] if p)
    name = re.sub(r"\s+", " ", re.sub(r'[\\/:*?"<>|\r\n\t]+', " ", name)).strip()
    return (name or number) + ".pdf"


def delete_invoice(conn, invoice_id):
    """Delete an invoice; its own lines are removed, linked ledger rows only unlinked. An RNr this app wrote
    into a ledger row (== the invoice number) is cleared; an RNr typed by hand that differs from the number
    survives."""
    inv = db.row(conn, "SELECT number FROM invoice WHERE id=?", (invoice_id,))
    if inv and inv["number"]:
        db.execute(conn, "UPDATE charge SET rnr=NULL WHERE invoice_id=? AND rnr=?", (invoice_id, inv["number"]))
    _clear_invoice_lines(conn, invoice_id)
    db.execute(conn, "UPDATE charge SET invoice_id=NULL WHERE invoice_id=?", (invoice_id,))  # unlink ledger rows
    db.execute(conn, "DELETE FROM invoice WHERE id=?", (invoice_id,))


# --- companies (canonical client identities; client rows are their address/spelling versions) ---
def companies_list(conn, search=None):
    sql = """
        SELECT co.id, co.name, co.kind, co.uid,
               cur.address AS cur_address, cur.postcode AS cur_postcode, cur.city AS cur_city,
               (SELECT COUNT(*) FROM client c WHERE c.company_id = co.id) AS versions,
               (SELECT COUNT(*) FROM stay s JOIN client c ON s.employer_client_id=c.id
                WHERE c.company_id = co.id) AS stays,
               (SELECT COUNT(*) FROM invoice i JOIN client c ON i.client_id=c.id
                WHERE c.company_id = co.id) AS invoices
        FROM company co LEFT JOIN client cur ON co.current_client_id = cur.id
    """
    params = []
    if search:
        sql += " WHERE co.name ILIKE ?"
        params = [f"%{search}%"]
    sql += " ORDER BY lower(co.name), co.id"
    return _dicts(db.rows(conn, sql, params))


def get_company(conn, company_id):
    r = db.row(conn, "SELECT * FROM company WHERE id=?", (company_id,))
    if not r:
        return None
    comp = dict(r)
    comp["versions"] = _dicts(db.rows(conn, """
        SELECT c.*,
               (SELECT COUNT(*) FROM stay s WHERE s.employer_client_id=c.id) AS stays,
               (SELECT COUNT(*) FROM invoice i WHERE i.client_id=c.id) AS invoices,
               (SELECT MIN(i.date) FROM invoice i WHERE i.client_id=c.id) AS first_invoice,
               (SELECT MAX(i.date) FROM invoice i WHERE i.client_id=c.id) AS last_invoice
        FROM client c WHERE c.company_id=? ORDER BY c.id""", (company_id,)))
    comp["aliases"] = [r[0] for r in db.rows(conn,
        "SELECT DISTINCT c.name FROM client c WHERE c.company_id=? AND c.name<>? ORDER BY c.name",
        (company_id, comp["name"]))]
    return comp


def save_company(conn, company_id, name, kind, uid, email, phone, notes, current_client_id=None):
    db.execute(conn, "UPDATE company SET name=?, kind=?, uid=?, email=?, phone=?, notes=? WHERE id=?",
               (name, kind, uid, email, phone, notes, company_id))
    if current_client_id and db.row(conn, "SELECT 1 FROM client WHERE id=? AND company_id=?",
                                    (current_client_id, company_id)):
        db.execute(conn, "UPDATE company SET current_client_id=? WHERE id=?", (current_client_id, company_id))
    key = normalize.company_key(name)   # a rename must keep resolving: register the new spelling too
    if key and not db.row(conn, "SELECT 1 FROM company_key WHERE key=?", (key,)):
        db.execute(conn, "INSERT INTO company_key (key, company_id) VALUES (?,?)", (key, company_id))


def add_company_address(conn, company_id, address, postcode, city, make_current=True):
    """A new address version for a company (it moved / has another location). New invoices use it when
    make_current is set; old invoices keep their old version."""
    comp = db.row(conn, "SELECT name, kind FROM company WHERE id=?", (company_id,))
    if not comp:
        return None
    return _new_client_version(conn, company_id, comp["name"], comp["kind"],
                               address or None, postcode or None, city or None, make_current=make_current)


def update_client_address(conn, client_id, name, address, postcode, city):
    """Edit one address version in place (fix a typo / a name-part that leaked into the street). Old invoices
    on this version pick the correction up too — it's the same recipient, corrected."""
    db.execute(conn, "UPDATE client SET name=?, address=?, postcode=?, city=? WHERE id=?",
               (name.strip() or None, address.strip() or None, postcode.strip() or None,
                city.strip() or None, client_id))


def set_current_address(conn, company_id, client_id):
    """Make this address version the one new invoices use."""
    if db.row(conn, "SELECT 1 FROM client WHERE id=? AND company_id=?", (client_id, company_id)):
        db.execute(conn, "UPDATE company SET current_client_id=? WHERE id=?", (client_id, company_id))


def delete_client_address(conn, client_id):
    """Delete one address version. Refuses if any stay/invoice still uses it ('has_refs') or if it's the
    company's last version ('last_version'); a deleted current version hands off to another version."""
    row = db.row(conn, "SELECT company_id FROM client WHERE id=?", (client_id,))
    if not row:
        return
    company_id = row[0]
    if (db.row(conn, "SELECT COUNT(*) FROM stay WHERE employer_client_id=?", (client_id,))[0]
            or db.row(conn, "SELECT COUNT(*) FROM invoice WHERE client_id=?", (client_id,))[0]):
        raise ValueError("has_refs")
    other = db.row(conn, "SELECT id FROM client WHERE company_id=? AND id<>? ORDER BY id DESC LIMIT 1",
                   (company_id, client_id))
    if not other:
        raise ValueError("last_version")
    if db.row(conn, "SELECT current_client_id FROM company WHERE id=?", (company_id,))[0] == client_id:
        db.execute(conn, "UPDATE company SET current_client_id=? WHERE id=?", (other[0], company_id))
    db.execute(conn, "DELETE FROM client WHERE id=?", (client_id,))


def merge_companies(conn, keep_id, merge_id):
    """Fold company `merge_id` into `keep_id`: all spellings keep resolving (keys move over), every client
    version survives under the surviving company, references never break."""
    if keep_id == merge_id or not db.row(conn, "SELECT 1 FROM company WHERE id=?", (merge_id,)):
        return
    db.execute(conn, "UPDATE company_key SET company_id=? WHERE company_id=?", (keep_id, merge_id))
    db.execute(conn, "UPDATE client SET company_id=? WHERE company_id=?", (keep_id, merge_id))
    keep = db.row(conn, "SELECT * FROM company WHERE id=?", (keep_id,))
    merged = db.row(conn, "SELECT * FROM company WHERE id=?", (merge_id,))
    for col in ("uid", "email", "phone", "notes"):
        if not keep[col] and merged[col]:
            db.execute(conn, f"UPDATE company SET {col}=? WHERE id=?", (merged[col], keep_id))
    if not keep["current_client_id"] and merged["current_client_id"]:
        db.execute(conn, "UPDATE company SET current_client_id=? WHERE id=?",
                   (merged["current_client_id"], keep_id))
    db.execute(conn, "DELETE FROM company WHERE id=?", (merge_id,))


def company_ref_counts(conn, company_id):
    """How many stays / invoices still point at any address version of this company."""
    sub = "SELECT id FROM client WHERE company_id=?"
    stays = db.row(conn, f"SELECT COUNT(*) FROM stay WHERE employer_client_id IN ({sub})", (company_id,))[0]
    invoices = db.row(conn, f"SELECT COUNT(*) FROM invoice WHERE client_id IN ({sub})", (company_id,))[0]
    return stays, invoices


def delete_company(conn, company_id):
    """Delete a company and its address versions. Refuses when any stay or invoice still references it
    (raise 'has_refs') so no register/billing row is orphaned — merge such a company instead."""
    stays, invoices = company_ref_counts(conn, company_id)
    if stays or invoices:
        raise ValueError("has_refs")
    db.execute(conn, "UPDATE company SET current_client_id=NULL WHERE id=?", (company_id,))  # break the self-ref
    db.execute(conn, "DELETE FROM company_key WHERE company_id=?", (company_id,))
    db.execute(conn, "DELETE FROM client WHERE company_id=?", (company_id,))
    db.execute(conn, "DELETE FROM company WHERE id=?", (company_id,))