696 lines
45 KiB
Python
696 lines
45 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from uuid import uuid4
|
|
|
|
from .shared import read_db, db, DatabaseNotInitializedError, now_iso
|
|
|
|
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 _cashflow_fingerprint(report: dict[str, Any]) -> str:
|
|
payload = {
|
|
"categories": [{k: item.get(k) for k in ("cashflow_kind", "category", "subcategory", "amount", "count", "currency")} for item in report.get("categories", [])],
|
|
"source_transactions": report.get("source_transactions", []),
|
|
}
|
|
return hashlib.sha256(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def _review_cashflow(month: str, db_path: str | None = None, conn=None) -> dict[str, Any]:
|
|
if conn is None:
|
|
return get_monthly_cashflow_report(month, db_path=db_path)
|
|
from .transaction_report_services import _month_range, CASHFLOW_KIND_SQL
|
|
start, end = _month_range(month)
|
|
rows = conn.execute(f"SELECT {CASHFLOW_KIND_SQL} AS cashflow_kind, t.category, t.subcategory, t.currency, SUM(t.amount) AS amount, COUNT(*) AS count FROM transactions t LEFT JOIN ledger_categories lc ON lc.active=1 AND t.category=lc.category AND t.subcategory=lc.subcategory WHERE t.status='active' AND t.transaction_date>=? AND t.transaction_date<? GROUP BY {CASHFLOW_KIND_SQL}, t.category, t.subcategory, t.currency ORDER BY cashflow_kind, amount DESC", (start, end)).fetchall()
|
|
categories = [dict(r) for r in rows]
|
|
source_rows = conn.execute("SELECT t.id, t.transaction_date, t.type, t.amount, t.currency, t.category, t.subcategory, t.account_id FROM transactions t LEFT JOIN ledger_categories lc ON lc.active=1 AND t.category=lc.category AND t.subcategory=lc.subcategory WHERE t.status='active' AND t.transaction_date>=? AND t.transaction_date<? ORDER BY t.id", (start, end)).fetchall()
|
|
totals = {}
|
|
for item in categories: totals[item['cashflow_kind']] = totals.get(item['cashflow_kind'], 0) + int(item['amount'] or 0)
|
|
income, expenses = totals.get('income', 0), totals.get('expense', 0)
|
|
return ok(month=month, income=income, expenses=expenses, net_cashflow=income-expenses, by_cashflow_kind=totals, categories=categories, source_transactions=[dict(r) for r in source_rows])
|
|
|
|
|
|
def review_monthly_settlement(settlement_id: str, db_path: str | None = None) -> dict[str, Any]:
|
|
try:
|
|
with read_db(db_path) as conn:
|
|
row = conn.execute("SELECT id,month,status,version FROM monthly_settlement WHERE id=?", (settlement_id,)).fetchone()
|
|
if not row:
|
|
return fail("not_found", "monthly settlement not found")
|
|
if row["status"] not in {"draft", "in_review"}:
|
|
return fail("invalid_settlement_status", "only draft or in_review settlements can be reviewed")
|
|
report = _review_cashflow(row["month"], conn=conn)
|
|
fp = _cashflow_fingerprint(report)
|
|
return ok(settlement_id=settlement_id, month=row["month"], version=row["version"], status=row["status"], cashflow=report, review_fingerprint=fp)
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement review unavailable")
|
|
|
|
|
|
def approve_monthly_settlement(settlement_id: str, expected_version: int, review_fingerprint: str, approved_by: str = "dashboard-admin", db_path: str | None = None) -> dict[str, Any]:
|
|
if not review_fingerprint:
|
|
return fail("validation_error", "review_fingerprint is required")
|
|
try:
|
|
with db(db_path) as conn:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
row = conn.execute("SELECT * FROM monthly_settlement WHERE id=?", (settlement_id,)).fetchone()
|
|
if not row: return fail("not_found", "monthly settlement not found")
|
|
if row["status"] not in {"draft", "in_review"} or row["version"] != expected_version:
|
|
return fail("stale_settlement", "settlement status or version is stale")
|
|
report = _review_cashflow(row["month"], conn=conn)
|
|
actual = _cashflow_fingerprint(report)
|
|
if actual != review_fingerprint:
|
|
return fail("stale_review", "review fingerprint is stale", current_fingerprint=actual)
|
|
ts = now_iso()
|
|
conn.execute("UPDATE monthly_settlement SET status='approved', approved_by=?, approved_at=?, approved_version=version, approved_fingerprint=COALESCE(review_fingerprint,?), approved_cashflow_fingerprint=?, updated_by=?, updated_at=? WHERE id=? AND version=?", (approved_by, ts, review_fingerprint, actual, approved_by, ts, settlement_id, expected_version))
|
|
conn.execute("UPDATE monthly_settlement SET review_fingerprint=? WHERE id=?", (review_fingerprint, settlement_id))
|
|
return ok(settlement_id=settlement_id, status="approved", version=expected_version, approved_cashflow_fingerprint=actual)
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement approval failed")
|
|
|
|
|
|
def cancel_monthly_settlement_approval(settlement_id: str, expected_version: int, approved_by: str = "dashboard-admin", db_path: str | None = None) -> dict[str, Any]:
|
|
"""Return an approved settlement to review without changing its source data."""
|
|
if not isinstance(expected_version, int) or isinstance(expected_version, bool) or expected_version < 1:
|
|
return fail("validation_error", "expected_version is required")
|
|
try:
|
|
with db(db_path) as conn:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
row = conn.execute("SELECT id, status, version FROM monthly_settlement WHERE id=?", (settlement_id,)).fetchone()
|
|
if not row:
|
|
return fail("not_found", "monthly settlement not found")
|
|
if row["status"] != "approved" or row["version"] != expected_version:
|
|
return fail("stale_settlement", "only the current approved settlement can have approval cancelled")
|
|
ts = now_iso()
|
|
cur = conn.execute("""UPDATE monthly_settlement
|
|
SET status='in_review', version=version+1,
|
|
approved_by=NULL, approved_at=NULL, approved_version=NULL,
|
|
approved_fingerprint=NULL, approved_cashflow_fingerprint=NULL,
|
|
approval_reason=NULL, approval_flags_json='{}',
|
|
review_fingerprint=NULL, updated_by=?, updated_at=?
|
|
WHERE id=? AND status='approved' AND version=?""", (approved_by, ts, settlement_id, expected_version))
|
|
if cur.rowcount != 1:
|
|
return fail("stale_settlement", "settlement status or version is stale")
|
|
return ok(settlement_id=settlement_id, status="in_review", version=expected_version + 1)
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement approval cancellation failed")
|
|
|
|
|
|
def close_monthly_settlement(settlement_id: str, expected_version: int, closed_by: str = "dashboard-admin", db_path: str | None = None) -> dict[str, Any]:
|
|
try:
|
|
with db(db_path) as conn:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
row = conn.execute("SELECT * FROM monthly_settlement WHERE id=?", (settlement_id,)).fetchone()
|
|
if not row: return fail("not_found", "monthly settlement not found")
|
|
if row["status"] == "closed": return ok(settlement_id=settlement_id, status="closed", version=row["version"], idempotent=True)
|
|
if row["status"] != "approved" or row["version"] != expected_version:
|
|
return fail("stale_settlement", "settlement status or version is stale")
|
|
if not row["approved_cashflow_fingerprint"]: return fail("missing_fingerprint", "approved cashflow fingerprint is missing")
|
|
report = _review_cashflow(row["month"], conn=conn)
|
|
actual = _cashflow_fingerprint(report)
|
|
if actual != row["approved_cashflow_fingerprint"]: return fail("stale_cashflow", "cashflow changed after approval", current_fingerprint=actual)
|
|
conn.execute("DELETE FROM settlement_cashflow_snapshots WHERE monthly_settlement_id=?", (settlement_id,))
|
|
items = report.get("categories", [])
|
|
for item in items:
|
|
conn.execute("INSERT INTO settlement_cashflow_snapshots(id,monthly_settlement_id,cashflow_kind,category,subcategory,amount,transaction_count,currency) VALUES(?,?,?,?,?,?,?,?)", (uuid4().hex, settlement_id, item.get("cashflow_kind") or "excluded", item.get("category") or "", item.get("subcategory") or "", int(item.get("amount") or 0), max(1, int(item.get("count") or 0)), item.get("currency") or ""))
|
|
ts = now_iso()
|
|
conn.execute("UPDATE monthly_settlement SET status='closed', closed_by=?, closed_at=?, updated_by=?, updated_at=? WHERE id=? AND status='approved' AND version=?", (closed_by, ts, closed_by, ts, settlement_id, expected_version))
|
|
return ok(settlement_id=settlement_id, status="closed", version=expected_version, snapshot_count=len(items), approved_cashflow_fingerprint=row["approved_cashflow_fingerprint"])
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement close failed")
|
|
|
|
|
|
def create_monthly_settlement(month: str, created_by: str = "dashboard-admin", db_path: str | None = None) -> dict[str, Any]:
|
|
"""Create only the canonical monthly-settlement parent draft."""
|
|
if not isinstance(month, str) or not __import__("re").fullmatch(r"\d{4}-\d{2}", month):
|
|
return fail("validation_error", "month must be YYYY-MM")
|
|
try:
|
|
selected = normalize_month(month)
|
|
basis = _month_end_date(selected)
|
|
ts = now_iso()
|
|
with db(db_path) as conn:
|
|
existing = conn.execute("SELECT id FROM monthly_settlement WHERE month=?", (selected,)).fetchone()
|
|
if existing:
|
|
return fail("duplicate_month", "monthly settlement already exists", settlement_id=existing["id"], month=selected)
|
|
settlement_id = uuid4().hex
|
|
conn.execute(
|
|
"INSERT INTO monthly_settlement(id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at) VALUES(?,?,?,?,?,?,?,?,?)",
|
|
(settlement_id, selected, basis, "draft", 1, created_by, ts, created_by, ts),
|
|
)
|
|
return ok(settlement_id=settlement_id, month=selected, status="draft", created=True)
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception as exc:
|
|
if "UNIQUE" in str(exc).upper():
|
|
return fail("duplicate_month", "monthly settlement already exists", month=month)
|
|
return fail("operation_failed", "monthly settlement creation failed")
|
|
|
|
|
|
def list_monthly_settlement_overview(year: int | None = None, status: str = "", db_path: str | None = None) -> dict[str, Any]:
|
|
"""List monthly-settlement parent rows without consulting removed run history."""
|
|
try:
|
|
with read_db(db_path) as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
WITH live AS (
|
|
SELECT substr(transaction_date, 1, 7) AS month,
|
|
SUM(CASE WHEN type IN ('income','credit') THEN amount ELSE 0 END) AS income,
|
|
SUM(CASE WHEN type IN ('expense','debit') THEN amount ELSE 0 END) AS expense
|
|
FROM transactions
|
|
GROUP BY substr(transaction_date, 1, 7)
|
|
)
|
|
SELECT m.month, m.id AS settlement_id, m.basis_date, m.status AS settlement_status,
|
|
m.version AS settlement_version, m.created_at, m.updated_at,
|
|
l.income AS live_income, l.expense AS live_expense
|
|
FROM monthly_settlement m
|
|
LEFT JOIN live l ON l.month = m.month
|
|
WHERE (? IS NULL OR ? = '' OR substr(m.month, 1, 4) = ?)
|
|
AND (? = '' OR m.status = ?)
|
|
ORDER BY m.month DESC
|
|
""",
|
|
(str(year) if year is not None else None, str(year) if year is not None else '', str(year) if year is not None else '', status or '', status or ''),
|
|
).fetchall()
|
|
result = []
|
|
for row in rows:
|
|
d = dict(row)
|
|
parent_exists = bool(d.get("settlement_id"))
|
|
lifecycle = d.get("settlement_status") or "unregistered"
|
|
label = {
|
|
"closed": "마감완료",
|
|
"approved": "승인완료",
|
|
"in_review": "검토중",
|
|
"draft": "초안",
|
|
"cancelled": "취소",
|
|
"unregistered": "정산 미생성·상세 선택 불가·현재집계(미확정)",
|
|
}.get(lifecycle, lifecycle)
|
|
d.update(
|
|
status_label=label,
|
|
amounts_are_snapshot=False,
|
|
income=d.get("live_income") or 0,
|
|
expense=d.get("live_expense") or 0,
|
|
income_display=d.get("live_income") or 0,
|
|
expense_display=d.get("live_expense") or 0,
|
|
amount_confirmation="미확정",
|
|
net_cashflow_display=None,
|
|
net_worth_display=None,
|
|
run_id=None,
|
|
run_version=None,
|
|
close_status=None,
|
|
closed_at=None,
|
|
income_total=None,
|
|
expense_total=None,
|
|
net_cashflow=None,
|
|
net_worth=None,
|
|
correction_active=False,
|
|
draft_active=parent_exists,
|
|
)
|
|
result.append(d)
|
|
return ok(months=result, read_only=True)
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement overview unavailable")
|
|
|
|
|
|
def get_monthly_settlement_record(settlement_id: str, db_path: str | None = None) -> dict[str, Any]:
|
|
"""Read one monthly-settlement parent by canonical ID."""
|
|
if not isinstance(settlement_id, str) or not settlement_id.strip():
|
|
return fail("validation_error", "settlement_id is required")
|
|
try:
|
|
with read_db(db_path) as conn:
|
|
row = conn.execute(
|
|
"SELECT id, month, basis_date, status, version, created_at, updated_at "
|
|
"FROM monthly_settlement WHERE id = ?", (settlement_id,)
|
|
).fetchone()
|
|
if row is None:
|
|
return fail("not_found", "monthly settlement not found")
|
|
return ok(settlement=dict(row), read_only=True)
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement unavailable")
|
|
|
|
|
|
def _ledger_candidates(conn, account_id: str, currency: str, basis_date: str) -> dict[str, Any]:
|
|
base = """SELECT t.id,t.transaction_date,t.transaction_datetime,t.created_at,t.balance_after,t.currency,t.account_id
|
|
FROM transactions t
|
|
WHERE t.account_id=? AND t.currency=? AND t.status='active'
|
|
AND t.balance_after IS NOT NULL AND t.transaction_date {op} ?
|
|
ORDER BY t.transaction_date {direction}, t.transaction_datetime {direction}, t.created_at {direction}, t.id {direction} LIMIT 1"""
|
|
prev = conn.execute(base.format(op="<=", direction="DESC"), (account_id, currency, basis_date)).fetchone()
|
|
nxt = conn.execute(base.format(op=">", direction="ASC"), (account_id, currency, basis_date)).fetchone()
|
|
return {"previous_ledger": dict(prev) if prev else None, "next_ledger": dict(nxt) if nxt else None}
|
|
|
|
|
|
def list_account_snapshots_for_settlement(settlement_id: str, db_path: str | None = None) -> dict[str, Any]:
|
|
"""Return the account evaluation rows for a settlement without inventing values.
|
|
|
|
Draft/review/approved/cancelled settlements use the complete registered-account
|
|
set and exact monthly snapshot matches. Closed settlements are historical:
|
|
they read only rows stored against that settlement, so later accounts cannot
|
|
leak into the report. Duplicate snapshot rows are deliberately preserved.
|
|
"""
|
|
if not isinstance(settlement_id, str) or not settlement_id.strip():
|
|
return fail("validation_error", "settlement_id is required")
|
|
try:
|
|
with read_db(db_path) as conn:
|
|
settlement = conn.execute(
|
|
"SELECT status, basis_date FROM monthly_settlement WHERE id=?", (settlement_id,)
|
|
).fetchone()
|
|
if settlement is None:
|
|
return fail("not_found", "monthly settlement not found")
|
|
closed = settlement["status"] == "closed"
|
|
if closed:
|
|
rows = conn.execute(
|
|
"""SELECT s.name, s.source_type, s.source_id, s.amount, s.currency,
|
|
a.type AS account_type, a.liquidity_bucket AS account_liquidity_bucket,
|
|
s.liquidity_bucket AS snapshot_account_category,
|
|
s.basis_date, s.verification_status, s.memo, s.source_note,
|
|
s.id AS snapshot_id, 'snapshot' AS match_status
|
|
FROM account_snapshot s
|
|
LEFT JOIN accounts a ON a.id=s.source_id
|
|
WHERE s.monthly_settlement_id=? AND s.source_type='account'
|
|
ORDER BY s.name, s.source_id, s.created_at, s.id""", (settlement_id,)
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"""SELECT a.name, 'account' AS source_type, a.id AS source_id,
|
|
a.type AS account_type, a.liquidity_bucket AS account_liquidity_bucket,
|
|
s.liquidity_bucket AS snapshot_account_category,
|
|
s.amount, COALESCE(s.currency, a.currency) AS currency,
|
|
COALESCE(s.basis_date, ?) AS basis_date,
|
|
s.verification_status, s.memo, s.source_note,
|
|
s.id AS snapshot_id,
|
|
CASE WHEN s.id IS NULL THEN 'unregistered' ELSE 'snapshot' END AS match_status
|
|
FROM accounts a
|
|
LEFT JOIN account_snapshot s
|
|
ON s.source_id=a.id AND s.source_type='account'
|
|
AND s.monthly_settlement_id=?
|
|
ORDER BY a.name, a.id, s.created_at, s.id""",
|
|
(settlement["basis_date"], settlement_id),
|
|
).fetchall()
|
|
result = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item["account_name"] = item.get("name") or ""
|
|
item["account_type"] = item.get("account_type") or ""
|
|
item["account_liquidity_bucket"] = item.get("account_liquidity_bucket") or ""
|
|
item["snapshot_account_category"] = item.get("snapshot_account_category") or ""
|
|
item["display_status"] = "미등록" if item["match_status"] == "unregistered" else ("미확인" if item.get("amount") is None else "확인")
|
|
if item["match_status"] == "unregistered":
|
|
with read_db(db_path) as candidate_conn:
|
|
item.update(_ledger_candidates(candidate_conn, item["source_id"], item["currency"], item["basis_date"]))
|
|
if item["match_status"] == "unregistered":
|
|
with read_db(db_path) as candidate_conn:
|
|
candidates = candidate_conn.execute(
|
|
"""SELECT s.id AS snapshot_id, s.basis_date, s.amount, s.currency,
|
|
s.verification_status, s.created_at
|
|
FROM account_snapshot s
|
|
WHERE s.source_type='account' AND s.source_id=?
|
|
AND s.monthly_settlement_id IS NULL AND s.currency=?
|
|
ORDER BY CASE WHEN s.basis_date=? THEN 0 ELSE 1 END,
|
|
s.basis_date DESC, s.created_at DESC, s.id DESC""",
|
|
(item["source_id"], item["currency"], item["basis_date"]),
|
|
).fetchall()
|
|
item["unlinked_candidates"] = [dict(candidate) for candidate in candidates]
|
|
result.append(item)
|
|
return ok(snapshots=result, count=len(result), registered_count=sum(r["match_status"] == "snapshot" for r in result), unregistered_count=sum(r["match_status"] == "unregistered" for r in result), closed=closed, status=settlement["status"], read_only=True)
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception:
|
|
return fail("operation_failed", "account snapshot details unavailable")
|
|
|
|
|
|
def link_account_snapshot(snapshot_id: str, monthly_settlement_id: str, *, allow_basis_date_mismatch: bool = False, db_path: str | None = None) -> dict[str, Any]:
|
|
"""Atomically attach one existing, unlinked account snapshot."""
|
|
if not isinstance(snapshot_id, str) or not snapshot_id.strip() or not isinstance(monthly_settlement_id, str) or not monthly_settlement_id.strip():
|
|
return fail("validation_error", "snapshot_id and monthly_settlement_id are required")
|
|
try:
|
|
with db(db_path) as conn:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
settlement = conn.execute("SELECT id,basis_date,status FROM monthly_settlement WHERE id=?", (monthly_settlement_id,)).fetchone()
|
|
if settlement is None:
|
|
conn.rollback(); return fail("not_found", "monthly settlement not found")
|
|
if settlement["status"] not in {"draft", "in_review"}:
|
|
conn.rollback(); return fail("invalid_settlement_status", "only draft or in_review settlements can be changed")
|
|
snapshot = conn.execute("SELECT id,source_type,source_id,currency,basis_date,monthly_settlement_id FROM account_snapshot WHERE id=?", (snapshot_id,)).fetchone()
|
|
if snapshot is None:
|
|
conn.rollback(); return fail("not_found", "account snapshot not found")
|
|
account = conn.execute("SELECT id,currency FROM accounts WHERE id=?", (snapshot["source_id"],)).fetchone()
|
|
if snapshot["source_type"] != "account" or account is None or snapshot["currency"] != account["currency"]:
|
|
conn.rollback(); return fail("validation_error", "snapshot account or currency is invalid")
|
|
if snapshot["monthly_settlement_id"] is not None:
|
|
conn.rollback(); return fail("already_linked", "account snapshot is already linked")
|
|
if snapshot["basis_date"] != settlement["basis_date"] and not allow_basis_date_mismatch:
|
|
conn.rollback(); return fail("basis_date_mismatch", "basis date mismatch requires explicit approval")
|
|
duplicate = conn.execute("SELECT id FROM account_snapshot WHERE monthly_settlement_id=? AND source_type='account' AND source_id=?", (monthly_settlement_id, snapshot["source_id"])).fetchone()
|
|
if duplicate:
|
|
conn.rollback(); return fail("duplicate", "account evaluation already registered", snapshot_id=duplicate["id"])
|
|
updated = conn.execute("UPDATE account_snapshot SET monthly_settlement_id=? WHERE id=? AND monthly_settlement_id IS NULL", (monthly_settlement_id, snapshot_id))
|
|
if updated.rowcount != 1:
|
|
conn.rollback(); return fail("stale_link", "account snapshot link is stale")
|
|
return ok(snapshot_id=snapshot_id, monthly_settlement_id=monthly_settlement_id, linked=True)
|
|
except Exception:
|
|
return fail("db_error", "account snapshot link failed")
|
|
|
|
|
|
def unlink_account_snapshot(snapshot_id: str, monthly_settlement_id: str, db_path: str | None = None) -> dict[str, Any]:
|
|
"""Detach exactly one account snapshot from one editable monthly settlement."""
|
|
if not isinstance(snapshot_id, str) or not snapshot_id.strip() or not isinstance(monthly_settlement_id, str) or not monthly_settlement_id.strip():
|
|
return fail("validation_error", "snapshot_id and monthly_settlement_id are required")
|
|
try:
|
|
with db(db_path) as conn:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
settlement = conn.execute("SELECT id,status FROM monthly_settlement WHERE id=?", (monthly_settlement_id,)).fetchone()
|
|
if settlement is None:
|
|
conn.rollback(); return fail("not_found", "monthly settlement not found")
|
|
if settlement["status"] not in {"draft", "in_review"}:
|
|
conn.rollback(); return fail("invalid_settlement_status", "only draft or in_review settlements can be changed")
|
|
snapshot = conn.execute(
|
|
"SELECT id,source_type,source_id,monthly_settlement_id FROM account_snapshot WHERE id=?", (snapshot_id,)
|
|
).fetchone()
|
|
if snapshot is None:
|
|
conn.rollback(); return fail("not_found", "account snapshot not found")
|
|
if snapshot["source_type"] != "account":
|
|
conn.rollback(); return fail("validation_error", "snapshot source_type must be account")
|
|
if not snapshot["source_id"] or conn.execute("SELECT 1 FROM accounts WHERE id=?", (snapshot["source_id"],)).fetchone() is None:
|
|
conn.rollback(); return fail("validation_error", "snapshot account not found")
|
|
if snapshot["monthly_settlement_id"] != monthly_settlement_id:
|
|
conn.rollback(); return fail("validation_error", "snapshot is not linked to this monthly settlement")
|
|
result = conn.execute(
|
|
"UPDATE account_snapshot SET monthly_settlement_id=NULL WHERE id=? AND monthly_settlement_id=?",
|
|
(snapshot_id, monthly_settlement_id),
|
|
)
|
|
if result.rowcount != 1:
|
|
conn.rollback(); return fail("validation_error", "snapshot link is stale")
|
|
return ok(snapshot_id=snapshot_id, monthly_settlement_id=monthly_settlement_id, unlinked=True)
|
|
except Exception:
|
|
return fail("db_error", "account snapshot unlink failed")
|
|
|
|
def list_monthly_settlement_history(month: str = "", db_path: str | None = None) -> dict[str, Any]:
|
|
"""Return immutable close/correction records; lifecycle history is authoritative."""
|
|
try:
|
|
selected = normalize_month(month) if month else ""
|
|
with read_db(db_path) as conn:
|
|
rows = conn.execute(
|
|
"""SELECT id AS run_id, settlement_month, draft_version AS version,
|
|
COALESCE(approved_at, created_at) AS closed_at,
|
|
correction_of_run_id AS correction_target,
|
|
correction_reason AS reason, close_status, created_at
|
|
FROM asset_snapshot_runs
|
|
WHERE (? = '' OR settlement_month = ? OR (settlement_month IS NULL AND substr(basis_date,1,7)=?))
|
|
ORDER BY COALESCE(approved_at, created_at) DESC, rowid DESC""",
|
|
(selected, selected, selected),
|
|
).fetchall()
|
|
return ok(history=[dict(row) for row in rows], month=selected, read_only=True)
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement history unavailable")
|
|
|
|
|
|
def get_monthly_settlement_history_record(run_id: str, db_path: str | None = None) -> dict[str, Any]:
|
|
"""Read one immutable close/correction record by run ID."""
|
|
if not isinstance(run_id, str) or not run_id:
|
|
return fail("validation_error", "run_id is required")
|
|
try:
|
|
with read_db(db_path) as conn:
|
|
row = conn.execute(
|
|
"""SELECT id AS run_id, settlement_month, draft_version AS version,
|
|
COALESCE(approved_at, created_at) AS closed_at,
|
|
correction_of_run_id AS correction_target,
|
|
correction_reason AS reason, close_status, created_at,
|
|
basis_date, income_total, expense_total, net_cashflow, net_worth
|
|
FROM asset_snapshot_runs WHERE id=?""", (run_id,)
|
|
).fetchone()
|
|
if not row:
|
|
return fail("not_found", "settlement history record not found")
|
|
return ok(history=dict(row), read_only=True)
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception:
|
|
return fail("operation_failed", "monthly settlement history unavailable")
|
|
|
|
|
|
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"),
|
|
"close_status": asset_report.get("close_status", ""),
|
|
"approval": asset_report.get("approval", {}),
|
|
"correction": asset_report.get("correction", {}),
|
|
"warnings": asset_report.get("warnings", []),
|
|
"provenance": asset_report.get(
|
|
"provenance", {"valuation_date": "", "warnings": [], "exclusions": []}
|
|
),
|
|
"missing_items": [item for item in asset_report.get("excluded_items", []) if "missing" in str(item.get("reason", ""))],
|
|
"excluded_items": asset_report.get("excluded_items", []),
|
|
"carried_items": asset_report.get("carried_items", asset_report.get("included_items", [])),
|
|
"monthly_source_snapshot_id": asset_report.get("snapshot_run_id", ""),
|
|
"draft_id": asset_report.get("close_metadata", {}).get("draft_id", ""),
|
|
"draft_version": asset_report.get("close_metadata", {}).get("draft_version", 0),
|
|
"review_fingerprint": asset_report.get("close_metadata", {}).get("review_fingerprint", ""),
|
|
"contains_substitutions": asset_report.get("close_metadata", {}).get("contains_substitutions", False),
|
|
"contains_carried": asset_report.get("close_metadata", {}).get("contains_carried", False),
|
|
"has_approved_missing": asset_report.get("close_metadata", {}).get("has_approved_missing", False),
|
|
"applied_evidence": asset_report.get("close_metadata", {}).get("applied_evidence", []),
|
|
},
|
|
}
|
|
|
|
|
|
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,
|
|
"assets": asset_report,
|
|
},
|
|
read_only=True,
|
|
)
|
|
|
|
|
|
def get_monthly_settlement_snapshot(month: str = "", run_id: str = "", db_path: str | None = None) -> dict[str, Any]:
|
|
"""Read a persisted monthly settlement close without consulting live ledger state."""
|
|
if not isinstance(month, str) or not isinstance(run_id, str):
|
|
return fail("validation_error", "month and run_id must be strings")
|
|
try:
|
|
selected_month = normalize_month(month) if month else ""
|
|
if selected_month and (len(selected_month) != 7 or selected_month[4] != "-"):
|
|
raise ValueError("invalid month")
|
|
except (TypeError, ValueError):
|
|
return fail("validation_error", "invalid settlement month")
|
|
|
|
def parsed(value: Any, default: Any):
|
|
if value in (None, ""):
|
|
return default
|
|
return json.loads(value)
|
|
|
|
try:
|
|
with read_db(db_path) as conn:
|
|
tables = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
if "asset_snapshot_runs" not in tables or "asset_snapshots" not in tables:
|
|
return fail("not_found", "settlement snapshot not found")
|
|
run = None
|
|
rows = []
|
|
legacy = False
|
|
metadata: dict[str, Any] = {}
|
|
if run_id:
|
|
run = conn.execute("SELECT rowid AS _run_rowid,* FROM asset_snapshot_runs WHERE id=?", (run_id,)).fetchone()
|
|
if run:
|
|
from .assets_service import _snapshot_rows_for_run
|
|
rows, metadata = _snapshot_rows_for_run(conn, run_id)
|
|
else:
|
|
cols = {r["name"] for r in conn.execute("PRAGMA table_info(asset_snapshots)")}
|
|
if {"snapshot_run_id", "closed_at"} <= cols:
|
|
rows = conn.execute("SELECT rowid AS _rowid,* FROM asset_snapshots WHERE snapshot_run_id=? AND closed_at<>'' ORDER BY rowid", (run_id,)).fetchall()
|
|
if rows:
|
|
legacy = True
|
|
metadata = {"snapshot_run_id": run_id, "legacy_closed": True, "close_status": "legacy_read_only"}
|
|
else:
|
|
if selected_month:
|
|
run_columns = {r["name"] for r in conn.execute("PRAGMA table_info(asset_snapshot_runs)")}
|
|
comprehensive_columns = {"snapshot_schema_version", "aggregation_version"}
|
|
if comprehensive_columns <= run_columns:
|
|
run = conn.execute(
|
|
"""SELECT rowid AS _run_rowid,* FROM asset_snapshot_runs
|
|
WHERE basis_date=? AND snapshot_schema_version=1
|
|
AND aggregation_version='comprehensive-v1'
|
|
ORDER BY rowid DESC LIMIT 1""",
|
|
(_month_end_date(selected_month),),
|
|
).fetchone()
|
|
if run is None:
|
|
run = conn.execute("SELECT rowid AS _run_rowid,* FROM asset_snapshot_runs WHERE basis_date=? ORDER BY rowid DESC LIMIT 1", (_month_end_date(selected_month),)).fetchone()
|
|
if run is None:
|
|
basis = _month_end_date(selected_month) if selected_month else ""
|
|
if basis:
|
|
from .assets_service import _snapshot_rows_for_latest_run
|
|
rows, metadata = _snapshot_rows_for_latest_run(conn, basis)
|
|
legacy = bool(metadata.get("legacy_closed"))
|
|
if legacy:
|
|
rows = [r for r in rows if r["snapshot_run_id"] == metadata.get("snapshot_run_id") and r["closed_at"]]
|
|
elif not metadata.get("snapshot_run_id"):
|
|
rows = []
|
|
if metadata.get("snapshot_run_id"):
|
|
run = conn.execute("SELECT rowid AS _run_rowid,* FROM asset_snapshot_runs WHERE id=?", (metadata["snapshot_run_id"],)).fetchone()
|
|
if not run and not rows:
|
|
return fail("not_found", "settlement snapshot not found")
|
|
if run:
|
|
rid = run["id"]
|
|
basis_date = run["basis_date"]
|
|
run_month = run["settlement_month"] if "settlement_month" in run.keys() else None
|
|
if selected_month and ((run_month and run_month != selected_month) or (not run_month and basis_date[:7] != selected_month)):
|
|
return fail("validation_error", "run does not match settlement month")
|
|
if not metadata:
|
|
from .assets_service import _snapshot_rows_for_run
|
|
rows, metadata = _snapshot_rows_for_run(conn, rid)
|
|
else:
|
|
rid = metadata.get("snapshot_run_id", run_id)
|
|
basis_date = rows[0]["basis_date"]
|
|
if selected_month and basis_date[:7] != selected_month:
|
|
return fail("validation_error", "run does not match settlement month")
|
|
legacy = True
|
|
if metadata.get("evidence_integrity_error"):
|
|
return fail("data_integrity_error", "snapshot evidence is malformed", snapshot_run_id=rid)
|
|
keys = set(run.keys()) if run else set()
|
|
def field(name): return run[name] if run and name in keys else None
|
|
from .settlement_cashflow_service import _AGGREGATION_VERSION
|
|
if (field("snapshot_schema_version") is not None or field("aggregation_version") is not None) and (
|
|
field("snapshot_schema_version") != 1 or field("aggregation_version") != _AGGREGATION_VERSION
|
|
):
|
|
raise ValueError("unsupported comprehensive snapshot version")
|
|
comprehensive = bool(field("snapshot_schema_version") == 1 and not legacy)
|
|
legacy = not comprehensive
|
|
cashflows = []
|
|
if comprehensive and "settlement_cashflow_snapshots" in tables:
|
|
cashflows = [dict(r) for r in conn.execute("SELECT cashflow_kind,category,subcategory,amount,transaction_count,currency FROM settlement_cashflow_snapshots WHERE settlement_run_id=? ORDER BY cashflow_kind,category,subcategory,currency", (rid,))]
|
|
totals = {"income_total": field("income_total") if comprehensive else None, "expense_total": field("expense_total") if comprehensive else None, "net_cashflow": field("net_cashflow") if comprehensive else None, "total_assets": field("total_assets") if comprehensive else None, "total_liabilities": field("total_liabilities") if comprehensive else None, "net_worth": field("net_worth") if comprehensive else None}
|
|
safe_items = []
|
|
for row in rows:
|
|
d = dict(row)
|
|
safe_items.append({k: d.get(k) for k in ("id","source_type","source_id","name","amount","currency","liquidity_bucket","valuation_source","snapshot_run_id","correction_of_run_id","correction_reason","is_correction","closed_at","valuation_id","selection_policy","verification_status","evidence_kind","exclusion_reason","created_at") if k in d})
|
|
source = parsed(field("source_snapshot_json"), {}) if comprehensive else {}
|
|
if not isinstance(source, dict): raise ValueError()
|
|
if comprehensive:
|
|
tx_fields = ("id", "transaction_date", "type", "amount", "currency", "category", "subcategory", "original_cashflow_kind", "originalcashflow_kind", "cashflow_kind", "applied_amount", "sourceprovenance", "category_cashflow_kind", "category_active")
|
|
asset_fields = ("id", "source_id", "source_type", "valuation_id", "valuation_date", "basis_date", "amount", "currency", "liquidity_bucket", "selection_mode", "applied_amount", "authority_source", "is_liability", "status", "verified", "verification_status", "evidence_kind", "exclusion_reason", "state_hash")
|
|
def project(records, fields):
|
|
if not isinstance(records, list) or any(not isinstance(x, dict) for x in records):
|
|
raise ValueError()
|
|
result = []
|
|
for record in records:
|
|
item = {k: record[k] for k in fields if k in record}
|
|
if any(value is not None and not isinstance(value, (str, int, bool)) for value in item.values()):
|
|
raise ValueError()
|
|
for key, allowed in (("decision", ("category", "subcategory", "cashflow_kind", "reason")), ("source_classification", ("id", "category", "subcategory", "cashflow_kind", "active"))):
|
|
value = record.get(key)
|
|
if isinstance(value, dict):
|
|
item[key] = {k: value[k] for k in allowed if k in value and (value[k] is None or isinstance(value[k], (str, int, bool)))}
|
|
result.append(item)
|
|
return result
|
|
source = {"transactions": project(source.get("transactions", []), tx_fields), "assets": project(source.get("assets", []), asset_fields)}
|
|
policy = parsed(field("cashflow_policy_json"), {}) if comprehensive else None
|
|
if comprehensive:
|
|
from .settlement_cashflow_service import _normalize_cashflow_policy
|
|
if not isinstance(policy, dict):
|
|
raise ValueError()
|
|
policy = _normalize_cashflow_policy(policy)
|
|
from .assets_service import _closed_run_metadata
|
|
public_metadata = _closed_run_metadata(metadata)
|
|
header_evidence = {key: public_metadata[key] for key in ("excluded_items", "carried_items", "applied_evidence", "approval")}
|
|
return ok(**header_evidence, source="settlement_snapshot", snapshot_run_id=rid, run_id=rid, settlement_month=field("settlement_month") or basis_date[:7], basis_date=basis_date, currency=field("currency"), is_closed_snapshot=True, is_comprehensive=comprehensive, legacy=legacy, aggregation_version=field("aggregation_version") if comprehensive else None, snapshot_schema_version=field("snapshot_schema_version") if comprehensive else None, cashflow_policy=policy, cashflow_items=cashflows, items=safe_items, source_snapshot=source, totals=totals, **totals, close_status=(field("close_status") or metadata.get("close_status", "legacy_read_only")), correction={"is_correction": bool(field("correction_of_run_id") or metadata.get("is_correction")), "correction_of_run_id": field("correction_of_run_id") or metadata.get("correction_of_run_id", ""), "correction_reason": field("correction_reason") or metadata.get("correction_reason", "")})
|
|
except DatabaseNotInitializedError:
|
|
return fail("db_not_initialized", "ledger database is not initialized")
|
|
except Exception:
|
|
return fail("db_error", "settlement snapshot read failed")
|