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 urllib.parse import urlencode
from zoneinfo import ZoneInfo

from .dashboard_segment import DASHBOARD_VIEWS, pack_segment
from .forderung_data import read_invoices
from .jinja_templates import number_format

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"):
    selection = {"view": view, "key": key}
    if segment is not ...:
        selection["department" if stack == "abteilung" else "follow_up"] = segment
    if only_invoices:
        selection["only_invoices"] = True
    return "/app/forderungen/liste?" + urlencode({"dashboard_segment": pack_segment(selection)})


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_id = _key(row.get("Verursacher_Benutzer_ID"))
        person = {"id": person_id} if person_id else {"text": _key(row.get("Verursacher_Text"))}
        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, row.get("Verursacher_Text") or person_id 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 == "Verursacher":
                    label += " (" + (group["key"].get("id") or "ohne Benutzer-ID") + ")"
                elif 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"),
    )