114 lines
5.1 KiB
Python
114 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from .assets_service import get_asset_snapshot_report
|
|
from .imports_service import list_ledger_import_batches
|
|
from .reconciliation_service import get_balance_reconciliation_report
|
|
from .review_decision_service import get_review_operating_indicators
|
|
from .shared import Any, fail, normalize_month, normalize_snapshot_at, ok
|
|
from .transaction_report_services import get_budget_vs_actual_report, get_monthly_cashflow_report
|
|
|
|
|
|
def _month_end_date(month: str) -> str:
|
|
return normalize_snapshot_at(month, default_now=False)[:10]
|
|
|
|
|
|
def _budget_overrun_count(budget_report: dict[str, Any]) -> int:
|
|
return sum(1 for item in budget_report.get("items", []) if item.get("status") == "over")
|
|
|
|
|
|
def _unresolved_blocker_summary(review_report: dict[str, Any], reconciliation_report: dict[str, Any]) -> dict[str, Any]:
|
|
indicators = review_report.get("indicators", {}) if review_report.get("success") else {}
|
|
reconciliation_items = reconciliation_report.get("items", []) if reconciliation_report.get("success") else []
|
|
reconciliation_attention_count = sum(1 for item in reconciliation_items if item.get("status") != "OK")
|
|
counts = {
|
|
"uncategorized_recommendation_backlog": int(indicators.get("uncategorized_recommendation_backlog") or 0),
|
|
"pending_review_count": int(indicators.get("pending_review_count") or 0),
|
|
"held_recheck_count": int(indicators.get("held_recheck_count") or 0),
|
|
"unreconciled_account_count": int(indicators.get("unreconciled_account_count") or 0),
|
|
"reconciliation_attention_count": reconciliation_attention_count,
|
|
}
|
|
return {"count": sum(counts.values()), "counts": counts}
|
|
|
|
|
|
def _source_coverage(
|
|
cashflow_report: dict[str, Any],
|
|
budget_report: dict[str, Any],
|
|
review_report: dict[str, Any],
|
|
reconciliation_report: dict[str, Any],
|
|
import_report: dict[str, Any],
|
|
asset_report: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
latest_import = (import_report.get("import_batches") or [None])[0]
|
|
return {
|
|
"cashflow": {"success": bool(cashflow_report.get("success")), "source": "transactions"},
|
|
"budget": {
|
|
"success": bool(budget_report.get("success")),
|
|
"budget_count": len(budget_report.get("items", [])),
|
|
},
|
|
"review": {
|
|
"success": bool(review_report.get("success")),
|
|
"indicators": review_report.get("indicators", {}) if review_report.get("success") else {},
|
|
},
|
|
"reconciliation": {
|
|
"success": bool(reconciliation_report.get("success")),
|
|
"item_count": int(reconciliation_report.get("count") or 0),
|
|
},
|
|
"imports": {
|
|
"success": bool(import_report.get("success")),
|
|
"recent_batch_count": int(import_report.get("count") or 0),
|
|
"latest_imported_at": latest_import.get("imported_at") if isinstance(latest_import, dict) else "",
|
|
},
|
|
"assets": {
|
|
"success": bool(asset_report.get("success")),
|
|
"basis_date": asset_report.get("basis_date", ""),
|
|
"source": asset_report.get("source", ""),
|
|
"is_closed_snapshot": bool(asset_report.get("is_closed_snapshot", False)),
|
|
"net_worth": asset_report.get("net_worth"),
|
|
},
|
|
}
|
|
|
|
|
|
def get_monthly_settlement_summary(month: str = "", db_path: str | None = None) -> dict[str, Any]:
|
|
"""Return the read-only monthly settlement summary foundation for dashboard/MCP consumers."""
|
|
try:
|
|
selected_month = normalize_month(month)
|
|
basis_date = _month_end_date(selected_month)
|
|
except Exception as exc:
|
|
return fail("validation_error", str(exc))
|
|
|
|
cashflow_report = get_monthly_cashflow_report(selected_month, db_path=db_path)
|
|
if not cashflow_report.get("success"):
|
|
return cashflow_report
|
|
budget_report = get_budget_vs_actual_report(selected_month, db_path=db_path)
|
|
review_report = get_review_operating_indicators(db_path=db_path)
|
|
reconciliation_report = get_balance_reconciliation_report(db_path=db_path)
|
|
import_report = list_ledger_import_batches(limit=5, db_path=db_path)
|
|
asset_report = get_asset_snapshot_report(basis_date=basis_date, db_path=db_path)
|
|
|
|
unresolved_blockers = _unresolved_blocker_summary(review_report, reconciliation_report)
|
|
return ok(
|
|
selected_month=selected_month,
|
|
month=selected_month,
|
|
basis_date=basis_date,
|
|
actual_income=int(cashflow_report.get("income") or 0),
|
|
actual_expense=int(cashflow_report.get("expenses") or 0),
|
|
net_cashflow=int(cashflow_report.get("net_cashflow") or 0),
|
|
savings_rate=cashflow_report.get("savings_rate"),
|
|
budget_overrun_count=_budget_overrun_count(budget_report),
|
|
unresolved_blocker_count=int(unresolved_blockers["count"]),
|
|
unresolved_blockers=unresolved_blockers,
|
|
source_coverage=_source_coverage(
|
|
cashflow_report,
|
|
budget_report,
|
|
review_report,
|
|
reconciliation_report,
|
|
import_report,
|
|
asset_report,
|
|
),
|
|
reports={
|
|
"cashflow": cashflow_report,
|
|
"budget": budget_report,
|
|
},
|
|
read_only=True,
|
|
)
|