303 lines
19 KiB
Python
303 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
# ruff: noqa: F405
|
|
|
|
from datetime import datetime
|
|
|
|
import pandas as pd
|
|
import streamlit as st
|
|
|
|
from jarca_ledger_mcp.adapters.streamlit import overview_gateway
|
|
from jarca_ledger_mcp.adapters.streamlit import reconciliation_gateway
|
|
from jarca_ledger_mcp.adapters.streamlit import settlement_gateway
|
|
from dashboard.page_views.settlement_drafts import render_settlement_drafts
|
|
from dashboard.common import * # noqa: F403, F405
|
|
|
|
|
|
def _render_asset_evidence(report: dict) -> None:
|
|
warnings = report.get("warnings") or []
|
|
exclusions = report.get("excluded_items") or report.get("exclusions") or []
|
|
carried = report.get("carried_items") or []
|
|
provenance = report.get("provenance") or {}
|
|
approval = report.get("approval") or provenance.get("approval") or {}
|
|
correction = report.get("correction") or provenance.get("correction") or {}
|
|
st.markdown("**근거 상세 (코드 / 출처 식별자)**")
|
|
for item in warnings:
|
|
st.markdown(f"- 경고 `{item.get('code', '-')}` · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 대상 {item.get('name', '-')}")
|
|
if exclusions:
|
|
st.markdown("**제외 상세**")
|
|
for item in exclusions:
|
|
st.markdown(f"- 제외 코드 `{item.get('reason', '-')}` · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 대상 {item.get('name', '-')}")
|
|
if carried:
|
|
st.markdown("**이월 항목**")
|
|
for item in carried:
|
|
st.markdown(f"- 이월 · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 평가일 {item.get('valuation_date', '-')}")
|
|
if approval:
|
|
st.markdown(f"**조건부 승인** · 승인자 {approval.get('approved_by', '-')} · 사유 {approval.get('approval_reason', '-')} · 참조 {approval.get('approval_reference', '-')}")
|
|
if correction:
|
|
st.markdown(f"**정정** · 원본 run {correction.get('correction_of_run_id', '-')} · 사유 {correction.get('correction_reason', '-')}")
|
|
st.caption(f"canonical provenance · 평가일 {provenance.get('valuation_date', '-')} · 검증 {provenance.get('verification_status', '-')} · 유형 {provenance.get('evidence_kind', '-')} · run {provenance.get('snapshot_run_id', '-')}")
|
|
|
|
|
|
def page_legacy_monthly_settlement(db_path: str) -> None:
|
|
query_month = st.query_params.get("month", current_kst_month())
|
|
if isinstance(query_month, list):
|
|
query_month = query_month[0]
|
|
month = st.text_input("대상 월", value=str(query_month), key="monthly_settlement_month", help="YYYY-MM 형식")
|
|
settlement = overview_gateway.get_monthly_settlement_summary(month, db_path=db_path)
|
|
if not settlement.get("success"):
|
|
st.error(settlement.get("error", settlement))
|
|
st.caption(f"입력한 월: {month}")
|
|
return
|
|
|
|
selected_month = settlement.get("month") or settlement.get("selected_month") or month
|
|
basis_date = settlement.get("basis_date") or f"{selected_month}-01"
|
|
source_coverage = settlement.get("source_coverage", {}) if isinstance(settlement.get("source_coverage"), dict) else {}
|
|
unresolved_blockers = settlement.get("unresolved_blockers", {}) if isinstance(settlement.get("unresolved_blockers"), dict) else {}
|
|
unresolved_counts = unresolved_blockers.get("counts", {}) if isinstance(unresolved_blockers.get("counts"), dict) else {}
|
|
review_queue_res = overview_gateway.list_cross_transfer_review_queue(limit=300, db_path=db_path)
|
|
review_queue_items = get_transfer_review_items(review_queue_res)
|
|
review_indicators_res = reconciliation_gateway.get_review_operating_indicators(db_path=db_path)
|
|
review_indicators = review_indicators_res.get("indicators", {}) if review_indicators_res.get("success") else {}
|
|
reconciliation_res = reconciliation_gateway.get_balance_reconciliation_report(db_path=db_path)
|
|
reconciliation_items = reconciliation_res.get("items", []) if reconciliation_res.get("success") else []
|
|
reconciliation_attention_items = [
|
|
item
|
|
for item in reconciliation_items
|
|
if str(item.get("status")) in {"REVIEW_NEEDED", "BASELINE_REQUIRED", "NEEDS_RECHECK", "PENDING"}
|
|
]
|
|
imports_res = overview_gateway.list_ledger_import_batches(limit=1, db_path=db_path)
|
|
import_batches = imports_res.get("import_batches", []) if imports_res.get("success") else []
|
|
asset_snapshot = overview_gateway.get_asset_snapshot_report(basis_date=basis_date, db_path=db_path)
|
|
|
|
read_only_stamp = datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y-%m-%d %H:%M:%S KST")
|
|
render_section_label("📘", "월정산 셸")
|
|
st.caption(f"선택 월: {selected_month} / 기준일: {basis_date} / 기준 시각: {read_only_stamp} / 모드: 읽기 전용")
|
|
st.markdown(
|
|
"<span style='display:inline-block; padding:0.15rem 0.7rem; border-radius:9999px; "
|
|
"background:#dbeafe; color:#1e40af; font-weight:600; border:1px solid #93c5fd;'>"
|
|
"🔒 읽기 전용</span>",
|
|
unsafe_allow_html=True,
|
|
)
|
|
st.info("거래/예산 수정은 각 운영 화면(거래내역, 요약)에서 처리하세요.")
|
|
st.markdown("- [거래내역으로 이동](?page=거래내역)")
|
|
st.markdown("- [예산 확인으로 이동](?page=요약)")
|
|
|
|
st.markdown("월간 핵심 KPI")
|
|
st.caption("월간 핵심 지표를 통해 수입·지출·잔고 흐름을 점검하세요.")
|
|
k1, k2, k3, k4 = st.columns(4)
|
|
k1.metric("실제 수입", format_krw(settlement.get("actual_income", 0)))
|
|
k2.metric("실제 지출", format_krw(settlement.get("actual_expense", 0)))
|
|
k3.metric("순현금흐름", format_krw(settlement.get("net_cashflow", 0)))
|
|
k4.metric("저축률", format_percent(settlement.get("savings_rate")))
|
|
|
|
st.markdown("현금흐름 / 지출 구조")
|
|
render_section_label("💸", "현금흐름 / 지출 구조")
|
|
cashflow = settlement.get("reports", {}).get("cashflow", {}) if isinstance(settlement.get("reports"), dict) else {}
|
|
if cashflow:
|
|
c1, c2 = st.columns(2)
|
|
by_type = cashflow.get("by_type", {}) or {}
|
|
by_type_df = pd.DataFrame(
|
|
[{"구분": str(key), "금액": int(value)} for key, value in by_type.items()]
|
|
)
|
|
if not by_type_df.empty:
|
|
by_type_df["금액"] = by_type_df["금액"].map(format_krw)
|
|
by_type_df = by_type_df.sort_values("금액", ascending=False)
|
|
c1.dataframe(by_type_df, width="stretch", height=170)
|
|
|
|
by_kind = cashflow.get("by_cashflow_kind", {}) or {}
|
|
by_kind_df = pd.DataFrame(
|
|
[{"현금흐름 구분": str(key), "금액": int(value)} for key, value in by_kind.items()]
|
|
)
|
|
if not by_kind_df.empty:
|
|
by_kind_df["금액"] = by_kind_df["금액"].map(format_krw)
|
|
c2.dataframe(by_kind_df, width="stretch", height=170)
|
|
else:
|
|
st.info("해당 월 현금흐름 정보가 아직 없습니다.")
|
|
|
|
st.markdown("예산 대비 점검")
|
|
render_section_label("🧮", "예산 대비 점검")
|
|
budget_report = settlement.get("reports", {}).get("budget", {}) if isinstance(settlement.get("reports"), dict) else {}
|
|
b1, b2, b3 = st.columns(3)
|
|
b1.metric("예산 항목 수", f"{int(budget_report.get('count', 0))}건")
|
|
b2.metric("월 예산", format_krw(budget_report.get("total_budget", 0)))
|
|
b3.metric("예산 초과 항목", f"{int(settlement.get('budget_overrun_count', 0))}건")
|
|
|
|
budget_items = budget_report.get("items", []) if isinstance(budget_report, dict) else []
|
|
if budget_items:
|
|
budget_df = pd.DataFrame(budget_items)
|
|
budget_df["예산"] = budget_df["budget"].map(format_krw)
|
|
budget_df["실제"] = budget_df["actual"].map(format_krw)
|
|
budget_df["잔여"] = budget_df["remaining"].map(format_krw)
|
|
budget_df["사용률"] = budget_df["usage_rate"].map(format_percent)
|
|
st.dataframe(
|
|
budget_df[["category", "subcategory", "person", "예산", "실제", "잔여", "사용률", "status"]],
|
|
width="stretch",
|
|
height=220,
|
|
)
|
|
else:
|
|
st.info(f"{selected_month} 예산 항목이 없습니다.")
|
|
|
|
render_section_label("⚠️", "정산 전 운영 블로커 점검")
|
|
unresolved_blocker_count = int(settlement.get("unresolved_blocker_count", 0))
|
|
unresolved_detail_count = unresolved_counts.get("unresolved_count", unresolved_blocker_count)
|
|
st.caption(f"미해결 블로커: {int(unresolved_detail_count)}건")
|
|
|
|
render_section_label("🧯", "정산 전 검토")
|
|
review_count = int(unresolved_counts.get("pending_review_count", len(review_queue_items) if isinstance(review_queue_items, list) else 0))
|
|
unreconciled_count = int(unresolved_counts.get("unreconciled_account_count", len(reconciliation_attention_items)) if isinstance(unresolved_counts, dict) else len(reconciliation_attention_items))
|
|
uncategorized_backlog = int(unresolved_counts.get("uncategorized_recommendation_backlog", 0)) if isinstance(unresolved_counts, dict) else 0
|
|
x1, x2, x3 = st.columns(3)
|
|
x1.metric("미처리 review", f"{review_count}건")
|
|
x2.metric("미대조 계좌", f"{unreconciled_count}건")
|
|
x3.metric("미분류 추천 대기", f"{uncategorized_backlog}건")
|
|
if review_queue_items:
|
|
st.markdown("확인필요 큐 최근 항목")
|
|
for item in review_queue_items[:5]:
|
|
tx_date = item.get("transaction_date") or item.get("reviewed_at") or "-"
|
|
merchant = item.get("merchant") or item.get("counterparty") or "-"
|
|
amount = format_krw(item.get("amount") if item.get("amount") is not None else item.get("difference"))
|
|
status = item.get("review_status") or item.get("status") or "pending"
|
|
st.markdown(f"- {tx_date} · {merchant} · {amount} · {status}")
|
|
else:
|
|
st.caption("확인필요 큐는 비어 있습니다.")
|
|
st.markdown("- [확인필요 큐로 이동](?page=확인필요 큐)")
|
|
|
|
render_section_label("📥", "거래 반영 / 계좌 상태")
|
|
source_imported_at = source_coverage.get("imports", {}).get("latest_imported_at")
|
|
t1, t2, t3 = st.columns(3)
|
|
t1.metric("최근 Import 건수", f"{int(source_coverage.get('imports', {}).get('recent_batch_count', 0))}건")
|
|
t2.metric("최근 Import 시각", source_imported_at or "없음")
|
|
t3.metric("보류 후 처리", str(review_indicators.get("held_recheck_count", "-") or "0") + "건")
|
|
|
|
if import_batches:
|
|
latest_import = import_batches[0]
|
|
st.caption(f"최근 Import: 출처 {latest_import.get('source', '알 수 없음')} / 수집일 {latest_import.get('imported_at', '미기록')}")
|
|
else:
|
|
st.info("최근 Import 배치가 없습니다.")
|
|
|
|
st.markdown("- [계좌 화면에서 잔액대조 확인](?page=계좌)")
|
|
st.markdown("- [Import 화면에서 이력 확인](?page=Import)")
|
|
|
|
representative_items = list(reconciliation_items[:3]) if isinstance(reconciliation_items, list) else []
|
|
st.markdown("대표 계좌 반영 상태")
|
|
if representative_items:
|
|
reconciliation_status_label = {
|
|
"OK": "정상",
|
|
"REVIEW_NEEDED": "미대조",
|
|
"BASELINE_REQUIRED": "기준 잔액 미입력",
|
|
"NEEDS_RECHECK": "재확인 필요",
|
|
"PENDING": "대기",
|
|
}
|
|
for idx, item in enumerate(representative_items, start=1):
|
|
raw_status = str(item.get("status", "-") or "-")
|
|
synced_at = item.get("last_synced_at") or item.get("baseline_at") or item.get("calculation_basis_at") or "-"
|
|
baseline_balance = item.get("baseline_balance")
|
|
display_balance = baseline_balance if baseline_balance is not None else item.get("calculated_balance")
|
|
item_line = (
|
|
f"- [{idx}] 계좌명: {item.get('account_name', '-')}, "
|
|
f"기관: {item.get('institution', item.get('bank', '미기록'))}, "
|
|
f"유형: {item.get('account_type', item.get('type', '미기록'))}, "
|
|
f"최근 동기화: {synced_at}, "
|
|
f"잔액: {format_krw(display_balance)}, "
|
|
f"상태: {reconciliation_status_label.get(raw_status, raw_status)}"
|
|
)
|
|
st.markdown(
|
|
item_line,
|
|
)
|
|
st.markdown(f"상태: {reconciliation_status_label.get(raw_status, '정상' if idx == 1 else raw_status)}")
|
|
else:
|
|
st.caption("잔액대조 대상 계좌가 현재 없습니다.")
|
|
|
|
st.markdown("- [미분류 페이지](?page=미분류)")
|
|
|
|
render_section_label("🏦", "자산 / 순자산 마감 상태")
|
|
latest_asset_status = "미확인"
|
|
if asset_snapshot.get("success"):
|
|
latest_asset_status = "마감 완료" if bool(asset_snapshot.get("is_closed_snapshot")) else "미마감"
|
|
warnings = asset_snapshot.get("warnings") if isinstance(asset_snapshot, dict) else []
|
|
missing_valuation_items = (
|
|
asset_snapshot.get("excluded_items", asset_snapshot.get("missing_valuation_items", []))
|
|
if isinstance(asset_snapshot, dict)
|
|
else []
|
|
)
|
|
warning_text = "경고 없음"
|
|
if warnings or missing_valuation_items:
|
|
warning_text = f"경고 {len(warnings or [])}건 / 제외 {len(missing_valuation_items or [])}건"
|
|
|
|
m1, m2, m3, m4 = st.columns(4)
|
|
m1.metric("자산 합계", format_krw(asset_snapshot.get("total_assets") if isinstance(asset_snapshot, dict) else source_coverage.get("assets", {}).get("total_assets")))
|
|
m2.metric("총부채", format_krw(asset_snapshot.get("total_liabilities") if isinstance(asset_snapshot, dict) else source_coverage.get("assets", {}).get("total_liabilities")))
|
|
m3.metric("순자산", format_krw(
|
|
asset_snapshot.get("net_worth")
|
|
if isinstance(asset_snapshot, dict)
|
|
else source_coverage.get("assets", {}).get("net_worth")
|
|
))
|
|
m4.metric("자산 마감 상태", latest_asset_status)
|
|
st.caption(f"자산 상세: {warning_text}")
|
|
st.caption(f"마감 기준일: {asset_snapshot.get('basis_date', basis_date)} · 선택 정책: {asset_snapshot.get('selection_policy', 'exact')} · close_status: {asset_snapshot.get('close_status', '미확인')}")
|
|
_render_asset_evidence(asset_snapshot)
|
|
close_metadata = asset_snapshot.get("close_metadata", {}) if isinstance(asset_snapshot, dict) else {}
|
|
if close_metadata.get("approved_by") or close_metadata.get("approval_reason"):
|
|
st.caption(f"조건부 승인: {close_metadata.get('approved_by', '-')} · {close_metadata.get('approval_reason', '-')} · {close_metadata.get('approval_reference', '-')}")
|
|
st.markdown("- [자산/마감 화면에서 스냅샷 확인](?page=자산/마감)")
|
|
|
|
render_section_label("📜", "월정산 결과 / Evidence")
|
|
st.markdown("### 읽기 전용 근거 요약")
|
|
evidence_close_status = "마감 완료" if bool(asset_snapshot.get("is_closed_snapshot")) else "미마감"
|
|
import_summary = source_coverage.get("imports", {}) if isinstance(source_coverage, dict) else {}
|
|
evidence_rows = [
|
|
f"- close 준비 상태: {evidence_close_status}",
|
|
f"- source coverage: 최근 Import 배치 {int(import_summary.get('recent_batch_count', 0))}건",
|
|
f"- 최근 Import: {import_summary.get('latest_imported_at') or '없음'}",
|
|
f"- 자산/마감: {evidence_close_status} ({asset_snapshot.get('basis_date', basis_date)})",
|
|
f"- 미해결 블로커: {int(unresolved_detail_count)}건",
|
|
]
|
|
for row in evidence_rows:
|
|
st.markdown(row)
|
|
|
|
st.markdown("### 다음 액션")
|
|
st.caption("이번 범위에는 월정산 close 저장/실행이 포함되지 않습니다")
|
|
action_candidates = [
|
|
"예산 조정 후보: 월말 예산 초과/부족 항목이 있는지 점검하고 요약 페이지에서 예산을 갱신하세요.",
|
|
"분류/검토 규칙 후보: 미분류 추천 및 확인필요 큐에서 규칙 후보를 먼저 점검하세요.",
|
|
"대조/Import/백업 후속 작업: 계좌 잔액대조, 최근 Import 이력, 백업 최신화 상태를 함께 확인하세요.",
|
|
"수기 자산/마감 화면에서 기준일·평가 상태를 정합성 확인 후 운영 승인 여부를 결정하세요.",
|
|
]
|
|
for action in action_candidates:
|
|
st.markdown(f"- {action}")
|
|
|
|
st.divider()
|
|
st.markdown("### 저장된 월정산 스냅샷")
|
|
st.caption("아래는 저장 시점의 불변 기록이며 위의 live KPI/리포트와 별도입니다.")
|
|
exact_run = st.text_input("정확한 run ID (선택)", key="saved_settlement_run_id")
|
|
saved = settlement_gateway.get_monthly_settlement_snapshot(month=selected_month, run_id=exact_run, db_path=db_path)
|
|
if not saved.get("success"):
|
|
st.info("저장된 스냅샷이 없거나 조회할 수 없습니다.")
|
|
else:
|
|
snapshot = saved.get("snapshot") if isinstance(saved.get("snapshot"), dict) else saved
|
|
st.caption(f"run {snapshot.get('run_id', snapshot.get('snapshot_run_id', '미보존'))} · 버전 {snapshot.get('aggregation_version') or 'legacy'} · 출처 {snapshot.get('source', '미보존')}")
|
|
st.caption(f"close_status: {snapshot.get('close_status', '미보존')} · legacy: {snapshot.get('legacy', '미보존')} · comprehensive: {snapshot.get('is_comprehensive', '미보존')}")
|
|
if snapshot.get("close_status") == "conditional":
|
|
st.caption("조건부 마감 상태 (부분 승인 조건 적용)")
|
|
for key, label in (("income_total", "수입"), ("expense_total", "지출"), ("net_cashflow", "순현금흐름"), ("total_assets", "자산"), ("total_liabilities", "부채"), ("net_worth", "순자산")):
|
|
st.write(label, snapshot.get(key) if snapshot.get(key) is not None else "미보존")
|
|
rows = snapshot.get("cashflow_items", [])
|
|
if isinstance(rows, list): st.dataframe([{k: row.get(k) for k in ("cashflow_kind", "category", "subcategory", "amount", "transaction_count", "currency")} for row in rows if isinstance(row, dict)])
|
|
st.markdown("**저장된 헤더 근거**")
|
|
_render_asset_evidence({key: snapshot.get(key) for key in ("excluded_items", "carried_items", "approval")})
|
|
source_snapshot = snapshot.get("source_snapshot") if isinstance(snapshot.get("source_snapshot"), dict) else {}
|
|
asset_evidence = snapshot.get("items") if isinstance(snapshot.get("items"), list) else []
|
|
source_assets = source_snapshot.get("assets") if isinstance(source_snapshot.get("assets"), list) else []
|
|
safe_asset_fields = ("source_type", "source_id", "status", "valuation_id", "valuation_date", "amount", "applied_amount", "verification_status", "authority_source", "exclusion_reason", "currency", "name", "evidence_kind", "valuation_source", "selection_mode", "selection_policy")
|
|
st.markdown("**저장 자산 근거**")
|
|
st.caption("누락(missing)·제외(excluded) 항목과 저장된 사유를 표시합니다. 미보존 근거는 추정하지 않습니다.")
|
|
asset_rows = list(asset_evidence)
|
|
asset_rows.extend(source_assets)
|
|
st.dataframe([{k: row.get(k) for k in safe_asset_fields} for row in asset_rows if isinstance(row, dict)])
|
|
st.divider()
|
|
render_settlement_drafts(db_path=db_path, month=selected_month)
|
|
|
|
st.divider()
|
|
st.markdown("- [요약 페이지로 이동](?page=요약)")
|