119 lines
6.0 KiB
Python
119 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
from .shared import (
|
|
Any,
|
|
BALANCE_RECONCILIATION_REFERENCE_RATIO,
|
|
BALANCE_RECONCILIATION_TOLERANCE_KRW,
|
|
RECONCILIATION_TARGET_STATUSES,
|
|
db,
|
|
fail,
|
|
normalize_date,
|
|
normalize_snapshot_at,
|
|
now_iso,
|
|
ok,
|
|
row_to_dict,
|
|
sqlite3,
|
|
)
|
|
from .review_decision_service import _fetch_latest_decisions
|
|
|
|
|
|
def _transaction_signed_amount(tx_type: str, amount: int) -> int:
|
|
if tx_type in {"income", "refund", "adjustment"}:
|
|
return int(amount)
|
|
if tx_type in {"expense", "transfer"}:
|
|
return -int(amount)
|
|
return 0
|
|
|
|
|
|
def _account_balance_row(conn: sqlite3.Connection, account_id: str, baseline_at: str) -> dict[str, Any]:
|
|
date_cutoff = normalize_date(baseline_at[:10])
|
|
rows = conn.execute(
|
|
"""SELECT type, amount, balance_after, transaction_date, transaction_datetime, created_at
|
|
FROM transactions
|
|
WHERE status='active' AND account_id=? AND transaction_date<=?
|
|
ORDER BY transaction_date ASC, transaction_datetime ASC, created_at ASC""",
|
|
(account_id, date_cutoff),
|
|
).fetchall()
|
|
signed_total = sum(_transaction_signed_amount(row["type"], int(row["amount"] or 0)) for row in rows)
|
|
latest_balance_row = next((row for row in reversed(rows) if row["balance_after"] is not None), None)
|
|
if latest_balance_row is not None:
|
|
calculated_balance = int(latest_balance_row["balance_after"])
|
|
basis = "latest_transaction_balance_after"
|
|
basis_at = latest_balance_row["transaction_datetime"] or latest_balance_row["transaction_date"]
|
|
else:
|
|
calculated_balance = int(signed_total)
|
|
basis = "net_transaction_delta_without_opening_balance"
|
|
basis_at = baseline_at
|
|
return {
|
|
"calculated_balance": calculated_balance,
|
|
"calculation_basis": basis,
|
|
"calculation_basis_at": basis_at,
|
|
"transaction_count": len(rows),
|
|
"signed_transaction_total": int(signed_total),
|
|
}
|
|
|
|
|
|
def get_balance_reconciliation_report(account_id: str = "", baseline_balance: int | None = None, baseline_at: str = "", source_note: str = "", memo: str = "", tolerance: int = BALANCE_RECONCILIATION_TOLERANCE_KRW, db_path: str | None = None) -> dict[str, Any]:
|
|
"""Preview manual/partial-import account balance reconciliation without mutating transactions."""
|
|
try:
|
|
baseline_at_norm = normalize_snapshot_at(baseline_at.strip(), default_now=False) if baseline_at and baseline_at.strip() else None
|
|
tolerance_int = int(tolerance)
|
|
if tolerance_int < 0:
|
|
raise ValueError("tolerance must be non-negative")
|
|
baseline_balance_int = int(baseline_balance) if baseline_balance is not None else None
|
|
except Exception as e:
|
|
return fail("validation_error", str(e))
|
|
with db(db_path) as conn:
|
|
clauses = ["active=1"]
|
|
params: list[Any] = []
|
|
if account_id:
|
|
clauses.append("id=?")
|
|
params.append(account_id)
|
|
else:
|
|
placeholders = ",".join("?" for _ in RECONCILIATION_TARGET_STATUSES)
|
|
clauses.append(f"statement_status IN ({placeholders})")
|
|
params.extend(sorted(RECONCILIATION_TARGET_STATUSES))
|
|
accounts = conn.execute(f"SELECT * FROM accounts WHERE {' AND '.join(clauses)} ORDER BY name", params).fetchall()
|
|
account_targets = [("reconciliation_account", account["id"]) for account in accounts]
|
|
latest_decisions = _fetch_latest_decisions(conn, account_targets)
|
|
session_rows = conn.execute("SELECT * FROM reconciliation_sessions ORDER BY basis_date DESC, updated_at DESC").fetchall()
|
|
latest_sessions: dict[str, dict[str, Any]] = {}
|
|
for row in session_rows:
|
|
session = row_to_dict(row)
|
|
latest_sessions.setdefault(session["account_id"], session)
|
|
items: list[dict[str, Any]] = []
|
|
for account in accounts:
|
|
effective_baseline = baseline_balance_int if baseline_balance_int is not None else account["last_balance"]
|
|
effective_at = baseline_at_norm or account["last_synced_at"] or now_iso()
|
|
calc = _account_balance_row(conn, account["id"], effective_at)
|
|
if effective_baseline is None:
|
|
difference = None
|
|
ratio = None
|
|
status = "BASELINE_REQUIRED"
|
|
else:
|
|
difference = int(effective_baseline) - int(calc["calculated_balance"])
|
|
ratio = abs(difference) / abs(int(effective_baseline)) if int(effective_baseline) else None
|
|
status = "OK" if abs(difference) <= tolerance_int else "REVIEW_NEEDED"
|
|
items.append({
|
|
"account_id": account["id"],
|
|
"account_name": account["name"],
|
|
"statement_status": account["statement_status"],
|
|
"is_reconciliation_target": account["statement_status"] in RECONCILIATION_TARGET_STATUSES,
|
|
"baseline_balance": effective_baseline,
|
|
"baseline_at": effective_at,
|
|
"calculated_balance": calc["calculated_balance"],
|
|
"difference": difference,
|
|
"tolerance": tolerance_int,
|
|
"reference_ratio_threshold": BALANCE_RECONCILIATION_REFERENCE_RATIO,
|
|
"difference_ratio": ratio,
|
|
"status": status,
|
|
"recommended_decisions": ["계좌이동", "실지출", "실수입", "투자이동", "분석제외", "확인필요"] if status == "REVIEW_NEEDED" else [],
|
|
"reconciliation_session": latest_sessions.get(account["id"]),
|
|
"review_status": latest_decisions.get(("reconciliation_account", account["id"]), {}).get("status", "pending"),
|
|
"review_decision": latest_decisions.get(("reconciliation_account", account["id"])),
|
|
"source_note": source_note,
|
|
"memo": memo,
|
|
**calc,
|
|
})
|
|
return ok(items=items, count=len(items), tolerance=tolerance_int, baseline_at=baseline_at_norm, routing="preview_only_manual_confirmation_required")
|