Jan Novak
b562ce3201
All checks were successful
Deploy to K8s / deploy (push) Successful in 8s
Pull 350+ lines of inline per-row computation out of adults_view, juniors_view, and payments into three pure builder functions with no Flask globals or IO dependencies. Route handlers now contain only cache/IO calls and a single render_template. No behaviour change — all 27 tests pass. Also moves get_month_labels, group_payments_by_person, and adapt_junior_members out of app.py. Prep for /api/* shadow endpoints (M5 Go parity). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
446 lines
16 KiB
Python
446 lines
16 KiB
Python
import json
|
|
import re
|
|
from datetime import datetime
|
|
|
|
from attendance import ADULT_MERGED_MONTHS, JUNIOR_MERGED_MONTHS
|
|
from match_payments import canonical_member_key
|
|
|
|
|
|
def get_month_labels(sorted_months, merged_months):
|
|
labels = {}
|
|
for m in sorted_months:
|
|
dt = datetime.strptime(m, "%Y-%m")
|
|
merged_in = sorted([k for k, v in merged_months.items() if v == m])
|
|
if merged_in:
|
|
all_dts = [datetime.strptime(x, "%Y-%m") for x in sorted(merged_in + [m])]
|
|
years = {d.year for d in all_dts}
|
|
if len(years) > 1:
|
|
parts = [d.strftime("%b %Y") for d in all_dts]
|
|
labels[m] = "+".join(parts)
|
|
else:
|
|
parts = [d.strftime("%b") for d in all_dts]
|
|
labels[m] = f"{'+'.join(parts)} {dt.strftime('%Y')}"
|
|
else:
|
|
labels[m] = dt.strftime("%b %Y")
|
|
return labels
|
|
|
|
|
|
def group_payments_by_person(transactions, member_names=None):
|
|
canonical_by_key = (
|
|
{canonical_member_key(n): n for n in member_names} if member_names else {}
|
|
)
|
|
grouped = {}
|
|
for tx in transactions:
|
|
person = str(tx.get("person", "")).strip()
|
|
if not person:
|
|
continue
|
|
for p in person.split(","):
|
|
p = re.sub(r"\[\?\]\s*", "", p).strip()
|
|
if not p:
|
|
continue
|
|
key = canonical_by_key.get(canonical_member_key(p), p)
|
|
grouped.setdefault(key, []).append(tx)
|
|
for rows in grouped.values():
|
|
rows.sort(key=lambda t: str(t.get("date", "")), reverse=True)
|
|
return grouped
|
|
|
|
|
|
def adapt_junior_members(junior_members):
|
|
"""Convert 4-tuple junior fee data to (fee, total_count) for reconcile."""
|
|
adapted = []
|
|
for name, tier, fees_dict in junior_members:
|
|
adapted_fees = {}
|
|
for m, fee_data in fees_dict.items():
|
|
if len(fee_data) == 4:
|
|
fee, total_count, _, _ = fee_data
|
|
adapted_fees[m] = (fee, total_count)
|
|
else:
|
|
fee, count = fee_data
|
|
adapted_fees[m] = (fee, count)
|
|
adapted.append((name, tier, adapted_fees))
|
|
return adapted
|
|
|
|
|
|
def build_adults_view_model(
|
|
members,
|
|
sorted_months,
|
|
result,
|
|
transactions,
|
|
current_month,
|
|
*,
|
|
attendance_url,
|
|
payments_url,
|
|
bank_account,
|
|
):
|
|
month_labels = get_month_labels(sorted_months, ADULT_MERGED_MONTHS)
|
|
adult_names = sorted([name for name, tier, _ in members if tier == "A"])
|
|
|
|
monthly_totals = {m: {"expected": 0, "paid": 0} for m in sorted_months}
|
|
formatted_results = []
|
|
for name in adult_names:
|
|
data = result["members"][name]
|
|
row = {
|
|
"name": name,
|
|
"months": [],
|
|
"balance": data["total_balance"],
|
|
"unpaid_periods": "",
|
|
"raw_unpaid_periods": "",
|
|
}
|
|
unpaid_months = []
|
|
raw_unpaid_months = []
|
|
for m in sorted_months:
|
|
mdata = data["months"].get(m, {
|
|
"expected": 0, "original_expected": 0,
|
|
"attendance_count": 0, "paid": 0, "exception": None,
|
|
})
|
|
expected = mdata.get("expected", 0)
|
|
original_expected = mdata.get("original_expected", 0)
|
|
count = mdata.get("attendance_count", 0)
|
|
paid = int(mdata.get("paid", 0))
|
|
exception_info = mdata.get("exception", None)
|
|
|
|
monthly_totals[m]["expected"] += expected
|
|
monthly_totals[m]["paid"] += paid
|
|
|
|
override_amount = exception_info["amount"] if exception_info else None
|
|
|
|
if override_amount is not None and override_amount != original_expected:
|
|
is_overridden = True
|
|
fee_display = (
|
|
f"{override_amount} ({original_expected}) CZK ({count})"
|
|
if count > 0
|
|
else f"{override_amount} ({original_expected}) CZK"
|
|
)
|
|
else:
|
|
is_overridden = False
|
|
fee_display = (
|
|
f"{expected} CZK ({count})" if count > 0 else f"{expected} CZK"
|
|
)
|
|
|
|
status = "empty"
|
|
cell_text = "-"
|
|
amount_to_pay = 0
|
|
|
|
if expected > 0:
|
|
amount_to_pay = max(0, expected - paid)
|
|
if paid >= expected:
|
|
status = "ok"
|
|
cell_text = f"{paid}/{fee_display}"
|
|
elif paid > 0:
|
|
status = "partial"
|
|
cell_text = f"{paid}/{fee_display}"
|
|
if m < current_month:
|
|
unpaid_months.append(month_labels[m])
|
|
raw_unpaid_months.append(
|
|
datetime.strptime(m, "%Y-%m").strftime("%m/%Y")
|
|
)
|
|
else:
|
|
status = "unpaid"
|
|
cell_text = f"0/{fee_display}"
|
|
if m < current_month:
|
|
unpaid_months.append(month_labels[m])
|
|
raw_unpaid_months.append(
|
|
datetime.strptime(m, "%Y-%m").strftime("%m/%Y")
|
|
)
|
|
elif paid > 0:
|
|
status = "surplus"
|
|
cell_text = f"PAID {paid}"
|
|
else:
|
|
cell_text = "-"
|
|
amount_to_pay = 0
|
|
|
|
if expected > 0 or paid > 0:
|
|
tooltip = f"Received: {paid}, Expected: {expected}"
|
|
else:
|
|
tooltip = ""
|
|
|
|
row["months"].append({
|
|
"text": cell_text,
|
|
"overridden": is_overridden,
|
|
"status": status,
|
|
"amount": amount_to_pay,
|
|
"month": month_labels[m],
|
|
"raw_month": m,
|
|
"tooltip": tooltip,
|
|
})
|
|
|
|
settled_balance = 0
|
|
for m, mdata in data["months"].items():
|
|
if m >= current_month:
|
|
continue
|
|
exp = mdata.get("expected", 0)
|
|
if isinstance(exp, int):
|
|
settled_balance += int(mdata.get("paid", 0)) - exp
|
|
|
|
payable_amount = max(0, -settled_balance)
|
|
row["unpaid_periods"] = ", ".join(unpaid_months)
|
|
row["raw_unpaid_periods"] = "+".join(raw_unpaid_months)
|
|
row["balance"] = settled_balance
|
|
row["payable_amount"] = payable_amount
|
|
formatted_results.append(row)
|
|
|
|
formatted_totals = []
|
|
for m in sorted_months:
|
|
t = monthly_totals[m]
|
|
status = "empty"
|
|
if t["expected"] > 0 or t["paid"] > 0:
|
|
if t["paid"] == t["expected"]:
|
|
status = "ok"
|
|
elif t["paid"] < t["expected"]:
|
|
status = "unpaid"
|
|
else:
|
|
status = "surplus"
|
|
formatted_totals.append({
|
|
"text": f"{t['paid']} / {t['expected']} CZK",
|
|
"status": status,
|
|
})
|
|
|
|
def _settled_balance(name):
|
|
data = result["members"][name]
|
|
total = 0
|
|
for m, mdata in data["months"].items():
|
|
if m >= current_month:
|
|
continue
|
|
exp = mdata.get("expected", 0)
|
|
if isinstance(exp, int):
|
|
total += int(mdata.get("paid", 0)) - exp
|
|
return total
|
|
|
|
credits = sorted(
|
|
[{"name": n, "amount": _settled_balance(n)} for n in adult_names if _settled_balance(n) > 0],
|
|
key=lambda x: x["name"],
|
|
)
|
|
debts = sorted(
|
|
[{"name": n, "amount": abs(_settled_balance(n))} for n in adult_names if _settled_balance(n) < 0],
|
|
key=lambda x: x["name"],
|
|
)
|
|
|
|
raw_payments_by_person = group_payments_by_person(
|
|
transactions, [name for name, _, _ in members]
|
|
)
|
|
|
|
return dict(
|
|
months=[month_labels[m] for m in sorted_months],
|
|
raw_months=sorted_months,
|
|
results=formatted_results,
|
|
totals=formatted_totals,
|
|
member_data=json.dumps(result["members"]),
|
|
month_labels_json=json.dumps(month_labels),
|
|
raw_payments_json=json.dumps(raw_payments_by_person),
|
|
credits=credits,
|
|
debts=debts,
|
|
unmatched=result["unmatched"],
|
|
attendance_url=attendance_url,
|
|
payments_url=payments_url,
|
|
bank_account=bank_account,
|
|
current_month=current_month,
|
|
)
|
|
|
|
|
|
def build_juniors_view_model(
|
|
junior_members,
|
|
adapted_members,
|
|
sorted_months,
|
|
result,
|
|
transactions,
|
|
current_month,
|
|
*,
|
|
attendance_url,
|
|
payments_url,
|
|
bank_account,
|
|
):
|
|
month_labels = get_month_labels(sorted_months, JUNIOR_MERGED_MONTHS)
|
|
junior_names = sorted([name for name, tier, _ in adapted_members])
|
|
junior_members_dict = {name: fees_dict for name, _, fees_dict in junior_members}
|
|
|
|
monthly_totals = {m: {"expected": 0, "paid": 0} for m in sorted_months}
|
|
formatted_results = []
|
|
for name in junior_names:
|
|
data = result["members"][name]
|
|
row = {
|
|
"name": name,
|
|
"months": [],
|
|
"balance": data["total_balance"],
|
|
"unpaid_periods": "",
|
|
"raw_unpaid_periods": "",
|
|
}
|
|
unpaid_months = []
|
|
raw_unpaid_months = []
|
|
for m in sorted_months:
|
|
mdata = data["months"].get(m, {
|
|
"expected": 0, "original_expected": 0,
|
|
"attendance_count": 0, "paid": 0, "exception": None,
|
|
})
|
|
expected = mdata.get("expected", 0)
|
|
original_expected = mdata.get("original_expected", 0)
|
|
count = mdata.get("attendance_count", 0)
|
|
paid = int(mdata.get("paid", 0))
|
|
exception_info = mdata.get("exception", None)
|
|
|
|
if expected != "?" and isinstance(expected, int):
|
|
monthly_totals[m]["expected"] += expected
|
|
monthly_totals[m]["paid"] += paid
|
|
|
|
orig_fee_data = junior_members_dict.get(name, {}).get(m)
|
|
adult_count = 0
|
|
junior_count = 0
|
|
if orig_fee_data and len(orig_fee_data) == 4:
|
|
_, _, adult_count, junior_count = orig_fee_data
|
|
|
|
breakdown = ""
|
|
if adult_count > 0 and junior_count > 0:
|
|
breakdown = f":{junior_count}J,{adult_count}A"
|
|
elif junior_count > 0:
|
|
breakdown = f":{junior_count}J"
|
|
elif adult_count > 0:
|
|
breakdown = f":{adult_count}A"
|
|
|
|
count_str = f" ({count}{breakdown})" if count > 0 else ""
|
|
|
|
override_amount = exception_info["amount"] if exception_info else None
|
|
|
|
if override_amount is not None and override_amount != original_expected:
|
|
is_overridden = True
|
|
fee_display = f"{override_amount} ({original_expected}) CZK{count_str}"
|
|
else:
|
|
is_overridden = False
|
|
fee_display = f"{expected} CZK{count_str}"
|
|
|
|
status = "empty"
|
|
cell_text = "-"
|
|
amount_to_pay = 0
|
|
|
|
if expected == "?" or (isinstance(expected, int) and expected > 0):
|
|
if expected == "?":
|
|
status = "empty"
|
|
cell_text = f"?{count_str}"
|
|
elif paid >= expected:
|
|
status = "ok"
|
|
cell_text = f"{paid}/{fee_display}"
|
|
elif paid > 0:
|
|
status = "partial"
|
|
cell_text = f"{paid}/{fee_display}"
|
|
amount_to_pay = expected - paid
|
|
if m < current_month:
|
|
unpaid_months.append(month_labels[m])
|
|
raw_unpaid_months.append(
|
|
datetime.strptime(m, "%Y-%m").strftime("%m/%Y")
|
|
)
|
|
else:
|
|
status = "unpaid"
|
|
cell_text = f"0/{fee_display}"
|
|
amount_to_pay = expected
|
|
if m < current_month:
|
|
unpaid_months.append(month_labels[m])
|
|
raw_unpaid_months.append(
|
|
datetime.strptime(m, "%Y-%m").strftime("%m/%Y")
|
|
)
|
|
elif paid > 0:
|
|
status = "surplus"
|
|
cell_text = f"PAID {paid}"
|
|
|
|
if (isinstance(expected, int) and expected > 0) or paid > 0:
|
|
tooltip = f"Received: {paid}, Expected: {expected}"
|
|
else:
|
|
tooltip = ""
|
|
|
|
row["months"].append({
|
|
"text": cell_text,
|
|
"overridden": is_overridden,
|
|
"status": status,
|
|
"amount": amount_to_pay,
|
|
"month": month_labels[m],
|
|
"raw_month": m,
|
|
"tooltip": tooltip,
|
|
})
|
|
|
|
settled_balance = 0
|
|
for m, mdata in data["months"].items():
|
|
if m >= current_month:
|
|
continue
|
|
exp = mdata.get("expected", 0)
|
|
if isinstance(exp, int):
|
|
settled_balance += int(mdata.get("paid", 0)) - exp
|
|
|
|
payable_amount = max(0, -settled_balance)
|
|
row["unpaid_periods"] = ", ".join(unpaid_months)
|
|
row["raw_unpaid_periods"] = "+".join(raw_unpaid_months)
|
|
row["balance"] = settled_balance
|
|
row["payable_amount"] = payable_amount
|
|
formatted_results.append(row)
|
|
|
|
formatted_totals = []
|
|
for m in sorted_months:
|
|
t = monthly_totals[m]
|
|
status = "empty"
|
|
if t["expected"] > 0 or t["paid"] > 0:
|
|
if t["paid"] == t["expected"]:
|
|
status = "ok"
|
|
elif t["paid"] < t["expected"]:
|
|
status = "unpaid"
|
|
else:
|
|
status = "surplus"
|
|
formatted_totals.append({
|
|
"text": f"{t['paid']} / {t['expected']} CZK",
|
|
"status": status,
|
|
})
|
|
|
|
junior_all_names = [name for name, _, _ in adapted_members]
|
|
|
|
def _junior_settled_balance(name):
|
|
data = result["members"][name]
|
|
total = 0
|
|
for m, mdata in data["months"].items():
|
|
if m >= current_month:
|
|
continue
|
|
exp = mdata.get("expected", 0)
|
|
if isinstance(exp, int):
|
|
total += int(mdata.get("paid", 0)) - exp
|
|
return total
|
|
|
|
credits = sorted(
|
|
[{"name": n, "amount": _junior_settled_balance(n)} for n in junior_all_names if _junior_settled_balance(n) > 0],
|
|
key=lambda x: x["name"],
|
|
)
|
|
debts = sorted(
|
|
[{"name": n, "amount": abs(_junior_settled_balance(n))} for n in junior_all_names if _junior_settled_balance(n) < 0],
|
|
key=lambda x: x["name"],
|
|
)
|
|
|
|
raw_payments_by_person = group_payments_by_person(
|
|
transactions, [name for name, _, _ in adapted_members]
|
|
)
|
|
|
|
return dict(
|
|
months=[month_labels[m] for m in sorted_months],
|
|
raw_months=sorted_months,
|
|
results=formatted_results,
|
|
totals=formatted_totals,
|
|
member_data=json.dumps(result["members"]),
|
|
month_labels_json=json.dumps(month_labels),
|
|
raw_payments_json=json.dumps(raw_payments_by_person),
|
|
credits=credits,
|
|
debts=debts,
|
|
unmatched=result["unmatched"],
|
|
attendance_url=attendance_url,
|
|
payments_url=payments_url,
|
|
bank_account=bank_account,
|
|
current_month=current_month,
|
|
)
|
|
|
|
|
|
def build_payments_view_model(transactions, member_names, *, attendance_url, payments_url):
|
|
grouped = group_payments_by_person(transactions, member_names)
|
|
for tx in transactions:
|
|
if not str(tx.get("person", "")).strip():
|
|
grouped.setdefault("Unmatched / Unknown", []).append(tx)
|
|
for rows in grouped.values():
|
|
rows.sort(key=lambda t: str(t.get("date", "")), reverse=True)
|
|
sorted_people = sorted(grouped.keys())
|
|
return dict(
|
|
grouped_payments=grouped,
|
|
sorted_people=sorted_people,
|
|
attendance_url=attendance_url,
|
|
payments_url=payments_url,
|
|
)
|