"""Aufteilung (Ausbaustufe 13): Brücke DB -> engine/breakdown.py. scenario=None ist „System-Plan" (reine Planung, alle Kredite, keine Modifikatoren) - dieselbe Datenbasis wie load_scenario_inputs(None). """ from dataclasses import dataclass from datetime import date from decimal import Decimal from sqlalchemy import select from sqlalchemy.orm import Session from app.engine.breakdown import (BreakdownRow, Filters, PlainLoan, Summary, breakdown_rows, summarize) from app.engine.loans import loan_schedule from app.models.tables import Category, Scenario from app.reserved import SYSTEM_PREFIX from app.services.double_count import Konflikt, all_conflicts from app.services.planning_chart import KREDITRATEN, UNKATEGORISIERT from app.services.projection_service import load_scenario_inputs PLAN_NAME = f"{SYSTEM_PREFIX}Plan" @dataclass(frozen=True) class Breakdown: month: date rows: list[BreakdownRow] summary: Summary conflicts: list[Konflikt] def compute_breakdown(session: Session, month: date, scenario: Scenario | None, filters: Filters) -> Breakdown: inputs = load_scenario_inputs(session, scenario) names = {c.id: c.name for c in session.execute(select(Category)).scalars()} loans = [PlainLoan(l.id, l.name, loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct), l.term_months, l.payout_date, l.repayment_type)) for l in inputs.loans] rows = breakdown_rows(month, inputs.recurring, inputs.planned, loans, inputs.modifiers, filters, category_names=names, loan_category=KREDITRATEN, uncategorized=UNKATEGORISIERT) # Doppelzählungs-Hinweis nur, wenn Fixposten UND Kredit hier als Zeile stehen. posten = {r.recurring_id for r in rows if r.recurring_id is not None} kredite = {r.loan_id for r in rows if r.loan_id is not None} conflicts = [k for k in all_conflicts(session) if k.posten_id in posten and k.kredit_id in kredite] return Breakdown(month.replace(day=1), rows, summarize(rows), conflicts)