Reisacher-Forderung/app/forderung_dashboard.py

"""Aggregate authorized invoices by rubric and selectable stack dimension."""

import hashlib
from collections import Counter
from datetime import date, datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo

from .dashboard_segment import DASHBOARD_VIEWS, pack_segment
from .forderung_data import read_invoices
from .jinja_templates import number_format
from .navigation import list_url
from .query_filter import EMPTY_FILTER_VALUE, encode_filter_values

BERLIN = ZoneInfo("Europe/Berlin")
STACKS = ("abteilung", "wiedervorlage")


def _key(value):
    return None if value is None or str(value) == "" else str(value)


def _date(value):
    return (
        value.date()
        if isinstance(value, datetime)
        else value if isinstance(value, date) else date.fromisoformat(str(value)[:10]) if value else None
    )


def color(department):
    hue = int.from_bytes(hashlib.sha256((department or "").encode()).digest()[:2], "big") % 360
    return f"hsl({hue}, 23%, 70%)"


def follow_up_bands(today):
    return [
        dict(
            id="overdue",
            label="Überfällig",
            color="#d4aaaa",
            key={"from": None, "to": (today - timedelta(days=1)).isoformat()},
        ),
        dict(
            id="soon",
            label="In 3 Tagen fällig",
            color="#e0cd9c",
            key={"from": today.isoformat(), "to": (today + timedelta(days=3)).isoformat()},
        ),
        dict(
            id="not_due",
            label="Noch nicht fällig",
            color="#b4c5b5",
            key={"from": (today + timedelta(days=4)).isoformat(), "to": (today + timedelta(days=14)).isoformat()},
        ),
        dict(
            id="deferred",
            label="zurückgestellt",
            color="#b2bccd",
            key={"from": (today + timedelta(days=15)).isoformat(), "to": None},
        ),
        dict(id="undated", label="Ohne Datum", color="#c5c5c5", key=None),
    ]


def _url(view, key, segment=..., only_invoices=False, stack="abteilung"):
    filters = {}

    def select(name, value):
        filters[name] = encode_filter_values([EMPTY_FILTER_VALUE if value is None or value == "" else value])

    if view == "Standort":
        select("Mandant", key["client"])
        select("Filiale", key["site"])
    elif view == "Altersstaffel":
        select("Staffel", key)
    elif view == "Verursacher":
        select("Verursacher", key["text"])
    if segment is not ... and stack == "abteilung":
        select("Abteilung", segment)
    if only_invoices:
        select("RechnungGutschrift", "Rechnung")
    return list_url(filters)


def build_dashboard(db, username, now=None, only_invoices=False, stack="abteilung"):
    if stack not in STACKS:
        raise ValueError("Ungültige Stapelung")
    now = now or datetime.now(BERLIN)
    today = now.astimezone(BERLIN).date()
    if only_invoices:
        columns, source = read_invoices(
            db, username, ["F.[Rechnung_Gutschrift] = :dashboard_invoice_type"], {"dashboard_invoice_type": "Rechnung"}
        )
    else:
        columns, source = read_invoices(db, username)
    groups = {view: {} for view in DASHBOARD_VIEWS}
    total, departments, observed = Decimal(0), set(), set()
    bands = follow_up_bands(today)
    for values in source:
        row = dict(zip(columns, values))
        department = _key(row.get("Abteilung"))
        departments.add(department)
        if stack == "abteilung":
            segment_id = department
        else:
            when = _date(row.get("Wiedervorlage") or row.get("Beleg_Faellig_Datum"))
            bucket = (
                4
                if when is None
                else (
                    0
                    if when < today
                    else 1 if when <= today + timedelta(days=3) else 2 if when <= today + timedelta(days=14) else 3
                )
            )
            segment_id = bands[bucket]["id"]
        observed.add(segment_id)
        amount = Decimal(str(row["Beleg_Offen"] or 0))
        total += amount
        age = _key(row.get("Staffel_Text"))
        site = {"client": str(row["Client_DB"]), "site": _key(row.get("Standort_ID"))}
        person_name = _key(row.get("Verursacher_Text"))
        person = {"text": person_name}
        definitions = [
            (
                "Altersstaffel",
                age,
                age or "Ohne Altersstaffel",
                int(row["Beleg_Tage_offen"]) if age and row.get("Beleg_Tage_offen") is not None else 10**12,
            ),
            (
                "Standort",
                site,
                f"{row.get('Standort_Name') or site['site'] or 'Ohne Standort'} ({row.get('Hauptbetrieb_Name') or 'Mandant ' + site['client']})",
                0,
            ),
            ("Verursacher", person, person_name or "Ohne Verursacher", 0),
        ]
        for view, key, label, order in definitions:
            identity = pack_segment(key)
            group = groups[view].setdefault(
                identity, dict(key=key, label=label, order=order, count=0, amount=Decimal(0), segments={})
            )
            group["order"] = min(group["order"], order)
            group["count"] += 1
            group["amount"] += amount
            segment = group["segments"].setdefault(segment_id, dict(count=0, amount=Decimal(0)))
            segment["count"] += 1
            segment["amount"] += amount
    items = (
        [
            dict(id=dep, label=dep or "Ohne Abteilung", color=color(dep), key=dep)
            for dep in sorted(departments, key=lambda value: (value is None, value or ""))
        ]
        if stack == "abteilung"
        else [band for band in bands if band["id"] != "undated" or "undated" in observed]
    )

    def serialized(amount):
        return str(amount.quantize(Decimal("0.01")))

    output = {}
    for view, mapping in groups.items():
        result = []
        labels = Counter(group["label"].casefold() for group in mapping.values())
        for group in mapping.values():
            label = group["label"]
            if labels[label.casefold()] > 1:
                if view == "Standort":
                    label += " · Standort-ID " + (group["key"]["site"] or "ohne ID")
            segments = [
                dict(
                    id=item["id"],
                    department=item["id"] if stack == "abteilung" else None,
                    label=item["label"],
                    count=group["segments"][item["id"]]["count"],
                    amount=serialized(group["segments"][item["id"]]["amount"]),
                    color=item["color"],
                    url=_url(view, group["key"], item["key"], only_invoices, stack),
                )
                for item in items
                if item["id"] in group["segments"]
            ]
            result.append(
                dict(
                    label=label,
                    order=group["order"],
                    count=group["count"],
                    amount=serialized(group["amount"]),
                    segments=segments,
                    url=_url(view, group["key"], only_invoices=only_invoices),
                )
            )
        output[view] = result
    return dict(
        count=len(source),
        amount=serialized(total),
        amount_display=number_format(total),
        views=output,
        only_invoices=only_invoices,
        stack=stack,
        stack_label="Abteilung" if stack == "abteilung" else "Wiedervorlage",
        stack_items=[{name: value for name, value in item.items() if name != "key"} for item in items],
        generated=now.astimezone(BERLIN).strftime("%d.%m.%Y %H:%M"),
    )