from __future__ import annotations
# ruff: noqa: F405
from datetime import date, datetime
import streamlit as st
from jarca_ledger_mcp.adapters.streamlit import overview_gateway
from jarca_ledger_mcp.adapters.streamlit import settlement_gateway
from dashboard.common import * # noqa: F403, F405
def _query_value(name: str, default: str = "") -> str:
value = st.query_params.get(name, default)
if isinstance(value, list):
value = value[0] if value else default
return str(value or default)
def _valid_month(value: str) -> bool:
try:
datetime.strptime(value, "%Y-%m")
return len(value) == 7
except (TypeError, ValueError):
return False
def _monthly_list_url(month: str | None, year: str, status: str) -> str:
params = ["page=월정산"]
if month:
params.append(f"month={month}")
if year and year != "전체":
params.append(f"year={year}")
if status and status != "전체":
params.append(f"status={status}")
return "?" + "&".join(params)
def _set_monthly_detail(settlement_id: str, year: str, status: str) -> None:
# The sidebar radio owns this widget key and may already have been
# instantiated in the current run. Mutating it here raises
# StreamlitAPIException during a real sidebar-backed navigation event.
if "dashboard_page" not in st.session_state:
st.session_state["dashboard_page"] = "월정산"
st.query_params["page"] = "monthly_settlement"
st.query_params["id"] = settlement_id
st.query_params.pop("month", None)
if year and year != "전체":
st.query_params["year"] = year
else:
st.query_params.pop("year", None)
if status and status != "전체":
st.query_params["status"] = status
else:
st.query_params.pop("status", None)
def _clear_monthly_detail(year: str, status: str) -> None:
# Keep the sidebar-owned widget state untouched; query params drive the
# rerun and the existing radio value is already 월정산.
if "dashboard_page" not in st.session_state:
st.session_state["dashboard_page"] = "월정산"
# Clear any detail selection before rerunning; otherwise the app's
# query-to-session restore can resurrect the stale detail on this run.
st.session_state.pop("monthly_settlement_detail_id", None)
st.session_state.pop("monthly_settlement_detail_month", None)
# Do not let the previous overview selection fire again when the list
# widget is recreated after returning from detail.
st.session_state.pop("monthly_settlement_overview_table", None)
st.query_params["page"] = "monthly_settlement"
st.query_params.pop("id", None)
st.query_params.pop("month", None)
if year and year != "전체":
st.query_params["year"] = year
else:
st.query_params.pop("year", None)
if status and status != "전체":
st.query_params["status"] = status
else:
st.query_params.pop("status", None)
st.rerun()
def _open_monthly_settlement_dialog(db_path: str) -> None:
@st.dialog("월정산 추가")
def dialog() -> None:
target_month = st.text_input("대상 월", placeholder="YYYY-MM", key="monthly_settlement_dialog_month")
if st.button("추가", type="primary", key="monthly_settlement_dialog_submit"):
result = settlement_gateway.create_monthly_settlement(target_month, db_path=db_path)
settlement_id = result.get("settlement_id")
if result.get("success") and settlement_id:
st.query_params["page"] = "monthly_settlement"
st.query_params["id"] = settlement_id
st.query_params.pop("month", None)
st.rerun()
elif result.get("error_code") == "duplicate_month" and settlement_id:
st.error("이미 생성된 월정산입니다. 기존 상세로 이동합니다.")
st.query_params["page"] = "monthly_settlement"
st.query_params["id"] = settlement_id
st.rerun()
else:
st.error(result.get("message", "월정산을 추가할 수 없습니다."))
dialog()
def _render_monthly_list(db_path: str) -> None:
overview = overview_gateway.list_monthly_settlement_overview(db_path=db_path)
title_col, action_col = st.columns([7, 1])
with title_col:
st.header("월정산")
with action_col:
if st.button("+", type="primary", key="monthly_settlement_add", help="월정산 추가"):
_open_monthly_settlement_dialog(db_path)
st.caption("월을 선택하면 해당 월의 독립 상세 화면으로 이동합니다. 조회만으로 초안을 만들지 않습니다.")
years = sorted({str(r.get("month", ""))[:4] for r in overview.get("months", []) if r.get("month")}, reverse=True)
year = st.selectbox("연도", ["전체"] + years, index=(["전체"] + years).index(_query_value("year", "전체")) if _query_value("year", "전체") in (["전체"] + years) else 0, key="settlement_overview_year")
statuses = ["전체", "마감완료", "미마감", "정정진행중"]
status = st.selectbox("상태", statuses, index=statuses.index(_query_value("status", "전체")) if _query_value("status", "전체") in statuses else 0, key="settlement_overview_status")
rows = overview.get("months", []) if overview.get("success") else []
if year != "전체":
rows = [r for r in rows if str(r.get("month", "")).startswith(year)]
if status != "전체":
rows = [r for r in rows if status in str(r.get("status_label", ""))]
if not rows:
st.info("표시할 월정산이 없습니다.")
return
# Keep the seven-column table shape while using an in-app button only for
# the month cell. A markdown href reloads the parent document and flashes
# the authenticated iframe; the callback changes query params in-place.
st.caption("월 텍스트를 클릭하면 해당 월정산 상세 화면으로 이동합니다.")
st.markdown(
"""""",
unsafe_allow_html=True,
)
widths = [1.25, 1.55, 1.25, 1.25, 1.5, 1.45, 1.65]
headers = ["월", "정산상태", "수입", "지출", "순현금흐름", "순자산", "최종마감일"]
with st.container(key="monthly_settlement_table"):
header_cells = st.columns(widths)
for cell, label in zip(header_cells, headers):
with cell:
st.markdown(label)
for row_index, row in enumerate(rows):
month = str(row.get("month", ""))
settlement_id = row.get("settlement_id")
cells = st.columns(widths)
with cells[0]:
if settlement_id:
st.button(
month,
type="tertiary",
key=f"monthly_settlement_month_{row_index}_{settlement_id}",
on_click=_set_monthly_detail,
args=(str(settlement_id), year, status),
)
else:
st.markdown(month)
values = [
row.get("status_label", ""),
format_krw(row.get("income")) if row.get("income") is not None else "미확정",
format_krw(row.get("expense")) if row.get("expense") is not None else "미확정",
format_krw(row.get("net_cashflow_display")) if row.get("net_cashflow_display") is not None else "미확정",
format_krw(row.get("net_worth_display")) if row.get("net_worth_display") is not None else "미확정",
row.get("closed_at") or "미마감",
]
for cell, value in zip(cells[1:], values):
with cell:
st.markdown(str(value))
def page_monthly_settlement(db_path: str) -> None:
if _query_value("legacy") == "1":
from .legacy_monthly_settlement import page_legacy_monthly_settlement
page_legacy_monthly_settlement(db_path)
return
settlement_id = _query_value("id")
legacy_month = _query_value("month")
if not settlement_id and legacy_month:
overview = overview_gateway.list_monthly_settlement_overview(db_path=db_path)
settlement_id = next((str(row.get("settlement_id")) for row in overview.get("months", []) if row.get("month") == legacy_month and row.get("settlement_id")), "")
if settlement_id:
st.query_params["page"] = "monthly_settlement"
st.query_params["id"] = settlement_id
st.query_params.pop("month", None)
else:
from .legacy_monthly_settlement import page_legacy_monthly_settlement
page_legacy_monthly_settlement(db_path)
return
if settlement_id:
year = _query_value("year", "전체")
status = _query_value("status", "전체")
parent = settlement_gateway.get_monthly_settlement_record(settlement_id, db_path=db_path)
if not parent.get("success"):
st.error("월정산을 찾을 수 없습니다.")
return
month = str(parent["settlement"]["month"])
settlement_row = parent.get("settlement") or {}
settlement_status = settlement_row.get("status")
settlement_version = settlement_row.get("version")
# Fetch the review payload once and feed the same read-only result to
# both the header action and the cashflow renderer.
review_payload = None
if settlement_status in {"draft", "in_review"}:
review_payload = settlement_gateway.review_monthly_settlement(settlement_id, db_path=db_path)
with st.container(key="monthly-settlement-detail-header"):
st.markdown(
"""""",
unsafe_allow_html=True,
)
try:
back_col, title_col, menu_col = st.columns([1.5, 6, 1], vertical_alignment="center")
except TypeError:
# Keep lightweight navigation fakes compatible; real Streamlit
# supports vertical_alignment (verified in the active env).
back_col, title_col, menu_col = st.columns([1.5, 6, 1])
with back_col:
st.button(
"← 월정산 목록",
key="monthly_settlement_back",
on_click=_clear_monthly_detail,
args=(year, status),
)
with title_col:
st.title(f"월정산 상세 · {month}")
with menu_col:
with st.popover("⋯", type="tertiary", help="월정산 메뉴"):
if settlement_status in {"draft", "in_review"}:
approve_disabled = not bool((review_payload or {}).get("success"))
if st.button("검토승인", key=f"monthly_approve_{settlement_id}", disabled=approve_disabled):
fingerprint = (review_payload or {}).get("review_fingerprint")
if settlement_version is None or not fingerprint:
st.error("검토 승인에 필요한 정보가 없습니다.")
result = {"success": False}
else:
result = settlement_gateway.approve_monthly_settlement(
settlement_id, int(settlement_version), str(fingerprint), db_path=db_path,
)
if result.get("success"):
st.success("승인되었습니다.")
st.rerun()
else:
st.error(result.get("error", "승인에 실패했습니다."))
elif settlement_status == "approved":
if st.button("승인 취소", key=f"monthly_cancel_approval_{settlement_id}"):
_open_monthly_approval_cancel_dialog(db_path, settlement_id, month, settlement_version)
if st.button("마감", key=f"monthly_close_{settlement_id}"):
_open_monthly_close_dialog(db_path, settlement_id, month, settlement_version)
try:
_render_monthly_detail(db_path, month, settlement_id, review_payload=review_payload)
except TypeError as exc:
# Keep lightweight navigation fakes that predate the payload arg usable.
if "review_payload" not in str(exc):
raise
_render_monthly_detail(db_path, month)
return
_render_monthly_list(db_path)
st.link_button("기존 자산 마감 · 초안 및 이력", "?page=monthly_settlement&legacy=1")
def _account_evaluation_totals(rows: list[dict]) -> dict:
"""Aggregate linked snapshots into signed assets, debt, and net assets.
Unknown categories stay out of the net calculation; NULL amounts stay unknown.
Debt follows the existing signed debt-positive contract (no unconditional abs()).
"""
registered = [row for row in rows if row.get("snapshot_id")]
registered_accounts = {row.get("account_id") or row.get("source_id") for row in registered}
all_accounts = {row.get("account_id") or row.get("source_id") for row in rows}
unregistered_accounts = all_accounts - registered_accounts
metrics: dict[str, dict[str, int]] = {}
legacy_totals: dict[str, int] = {}
unknown_count = 0
unknown_by_currency: dict[str, int] = {}
for row in registered:
currency = str(row.get("currency") or "미지정")
amount = row.get("amount")
if amount is None:
unknown_count += 1
continue
legacy_totals[currency] = legacy_totals.get(currency, 0) + int(amount)
account_type = str(row.get("account_type") or "").lower()
category = str(row.get("snapshot_account_category") or row.get("account_liquidity_bucket") or "").lower()
is_debt = account_type in {"loan", "card", "liability", "debt", "credit_card"} or category in {"loan", "card", "liability", "debt"}
is_asset = account_type in {"asset", "bank", "cash", "checking", "savings", "investment", "securities"} or category in {"asset", "cash", "bank", "investment"}
bucket = metrics.setdefault(currency, {"assets": 0, "liabilities": 0, "net_assets": 0})
if is_debt:
bucket["liabilities"] += int(amount)
elif is_asset:
bucket["assets"] += int(amount)
else:
if account_type or category:
unknown_by_currency[currency] = unknown_by_currency.get(currency, 0) + 1
continue
bucket["net_assets"] = bucket["assets"] - bucket["liabilities"]
return {
"metrics": metrics,
"totals": legacy_totals,
"unknown_count": unknown_count + sum(unknown_by_currency.values()),
"unknown_by_currency": unknown_by_currency,
"registered_count": len(registered_accounts),
"unregistered_count": len(unregistered_accounts),
"total_count": len(all_accounts),
"has_amount": bool(metrics),
}
def _format_evaluation_total(currency: str, amount: int) -> str:
return f"{amount:,}원" if currency == "KRW" else f"{amount:,} {currency}"
_ACCOUNT_EVALUATION_BUCKETS = (
("cash", "현금(cash)"),
("deposit", "예금(deposit)"),
("investment", "투자(investment)"),
("real_estate", "부동산(real_estate)"),
("loan", "대출·부채(loan)"),
("other", "기타(other)"),
)
_ACCOUNT_EVALUATION_BUCKET_CODES = {code for code, _ in _ACCOUNT_EVALUATION_BUCKETS}
def _snapshot_display_bucket(item: dict) -> str:
"""Use frozen snapshot classification; unknown/missing values are other."""
raw = item.get("snapshot_account_category")
if raw is None:
raw = item.get("account_liquidity_bucket")
value = str(raw or "").strip().lower()
return value if value in _ACCOUNT_EVALUATION_BUCKET_CODES else "other"
def _open_monthly_approval_cancel_dialog(db_path: str, settlement_id: str, month: str, expected_version: int | None) -> None:
@st.dialog("승인 취소")
def dialog() -> None:
st.markdown(f"**대상 월:** {month}")
st.warning("승인을 취소하면 검토중으로 돌아가며 재승인이 필요합니다.")
cancel_col, confirm_col = st.columns(2)
with cancel_col:
st.button("취소", key=f"monthly_approval_cancel_abort_{settlement_id}")
with confirm_col:
if st.button("승인 취소", key=f"monthly_approval_cancel_confirm_{settlement_id}", type="primary"):
if expected_version is None:
st.error("승인 취소에 필요한 버전 정보가 없습니다.")
return
result = settlement_gateway.cancel_monthly_settlement_approval(settlement_id, int(expected_version), db_path=db_path)
if result.get("success"):
st.success("승인이 취소되었습니다. 검토 후 다시 승인해야 합니다.")
st.rerun()
else:
st.error(result.get("error", "승인 취소에 실패했습니다."))
dialog()
def _open_monthly_close_dialog(db_path: str, settlement_id: str, month: str, expected_version: int | None) -> None:
@st.dialog("월정산 마감")
def dialog() -> None:
st.markdown(f"**대상 월:** {month}")
st.warning("마감 후에는 이 월정산의 평가 등록·분류 수정·연결 해제 등을 변경할 수 없습니다.")
cancel_col, confirm_col = st.columns(2)
with cancel_col:
st.button("취소", key=f"monthly_close_cancel_{settlement_id}")
with confirm_col:
if st.button("마감", key=f"monthly_close_confirm_{settlement_id}", type="primary"):
if expected_version is None:
st.error("마감에 필요한 버전 정보가 없습니다.")
return
result = settlement_gateway.close_monthly_settlement(settlement_id, int(expected_version), db_path=db_path)
if result.get("success"):
st.success("마감되었습니다.")
st.rerun()
else:
st.error(result.get("error", "마감에 실패했습니다."))
dialog()
def _render_monthly_detail(db_path: str, month: str, settlement_id: str | None = None, *, review_payload: dict | None = None) -> None:
settlement_id = settlement_id or _query_value("id")
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
st.subheader("계정별 평가 내역")
# The summary is a month-level report and does not carry lifecycle fields.
# Read the canonical parent row by the selected settlement id instead of
# guessing status/version from the nested report payload.
settlement_record = settlement_gateway.get_monthly_settlement_record(settlement_id, db_path=db_path) if settlement_id else {}
parent = settlement_record.get("settlement") or {}
settlement_status = parent.get("status")
settlement_version = parent.get("version")
snapshots = settlement_gateway.list_account_snapshots_for_settlement(settlement_id, db_path=db_path)
snapshot_rows = snapshots.get("snapshots", []) if snapshots.get("success") else []
rows = []
for item in snapshot_rows:
amount = item.get("amount")
status = item.get("display_status") or ("미등록" if not item.get("snapshot_id") else ("미확인" if amount is None else "확인"))
rows.append({
"snapshot_id": item.get("snapshot_id"),
"account_id": item.get("account_id") or item.get("source_id"),
"account_name": item.get("account_name") or item.get("name") or "",
"currency": item.get("currency") or "",
"amount": amount,
"basis_date": item.get("basis_date") or "",
"previous_ledger": item.get("previous_ledger"),
"next_ledger": item.get("next_ledger"),
"unlinked_candidates": item.get("unlinked_candidates") or [],
"account_type": item.get("account_type"),
"account_liquidity_bucket": item.get("account_liquidity_bucket"),
"snapshot_account_category": item.get("snapshot_account_category"),
"display_bucket": _snapshot_display_bucket(item),
"projection": item.get("projection"),
"계정명": item.get("account_name") or item.get("name") or "",
"평가금액": "—" if status == "미등록" else ("미확인" if amount is None else f"{amount:,}"),
"통화": item.get("currency") or "",
"기준일": item.get("basis_date") or "",
"확인상태": status,
"등록가능": not snapshots.get("closed", False) and status == "미등록",
"연결해제가능": snapshots.get("status") in {"draft", "in_review"} and status != "미등록" and bool(item.get("snapshot_id")),
})
registered = [r for r in rows if r.get("snapshot_id")]
unregistered = [r for r in rows if not r.get("snapshot_id")]
totals = _account_evaluation_totals(rows)
if snapshots.get("success"):
st.caption(f"평가등록 {totals['registered_count']} / 전체 {totals['total_count']} · 금액 미확인 {totals['unknown_count']} · 미등록 {totals['unregistered_count']}")
st.caption("미검증 평가금액이 포함될 수 있습니다.")
for currency, values in sorted(totals["metrics"].items()):
suffix = "원" if currency == "KRW" else currency
metric_columns = st.columns(3)
with metric_columns[0]:
st.metric(f"자산 합계 ({currency})", f"{values['assets']:,}{suffix}")
with metric_columns[1]:
st.metric(f"부채 합계 ({currency})", f"{values['liabilities']:,}{suffix}")
with metric_columns[2]:
st.metric(f"순자산 ({currency})", f"{values['net_assets']:,}{suffix}")
st.markdown("### 평가 등록 계좌")
if not registered:
st.caption("등록된 평가 내역이 없습니다.")
else:
# Keep the group header and every row on the same four-column grid.
widths = [2.6, 1.5, 0.45, 0.45]
st.markdown("""""", unsafe_allow_html=True)
for bucket, label in _ACCOUNT_EVALUATION_BUCKETS:
group_rows = [item for item in registered if item["display_bucket"] == bucket]
if not group_rows:
continue
group_totals = _account_evaluation_totals(group_rows)["totals"]
subtotal = " · ".join(_format_evaluation_total(currency, amount) for currency, amount in sorted(group_totals.items())) or "—"
with st.container(key=f"monthly-account-evaluation-group-{bucket}"):
header_cells = st.columns(widths)
with header_cells[0]: st.markdown(label)
with header_cells[1]: st.markdown(f"**{subtotal}**")
for cell in header_cells[2:]:
with cell: st.markdown("")
for item in group_rows:
cells = st.columns(widths)
with cells[0]: st.markdown(f'{item["계정명"]}
', unsafe_allow_html=True)
amount = item["amount"]
amount_display = "미확인" if amount is None else f'{amount:,}{"원" if item["currency"] == "KRW" else " " + str(item["currency"] or "")}'
with cells[1]: st.markdown(amount_display)
with cells[2]:
if item["연결해제가능"] and st.button("×", key=f"monthly_unlink_{settlement_id}_{item['snapshot_id']}", help="연결 해제", type="tertiary", use_container_width=True):
_render_snapshot_unlink_dialog(db_path, settlement_id, item)
with cells[3]:
if item["연결해제가능"] and st.button("✎", key=f"monthly_category_{settlement_id}_{item['snapshot_id']}", help="분류 수정", type="tertiary", use_container_width=True):
_render_snapshot_category_dialog(db_path, settlement_id, item)
st.markdown("### 평가 미등록 계좌")
if not unregistered:
st.caption("모든 계좌의 평가가 등록되었습니다.")
else:
for item in unregistered:
if item["등록가능"]:
if st.button(item["계정명"], type="tertiary", key=f"monthly_evaluation_account_{settlement_id}_{item['account_id']}"):
_render_account_registration_dialog(db_path, settlement_id, settlement, item)
else:
st.markdown(item["계정명"])
_render_monthly_cashflow(db_path, month, settlement_id, bool(snapshots.get("closed", False)), settlement_status, settlement_version, review_payload=review_payload)
def _render_monthly_cashflow(db_path: str, month: str, settlement_id: str, closed: bool, settlement_status: str | None = None, settlement_version: int | None = None, *, review_payload: dict | None = None) -> None:
report = settlement_gateway.get_monthly_cashflow(month, settlement_id=settlement_id, closed=closed, db_path=db_path)
if not report.get("success"):
st.warning("현금흐름을 불러오지 못했습니다.")
return
st.subheader("수입·지출")
st.caption("정산 스냅샷 기준" if closed else "현재 원장 기준")
if not report.get("saved", True):
st.info(report.get("message", "저장된 현금흐름 스냅샷이 없습니다."))
return
cols = st.columns(3)
for col, label, key in zip(cols, ("수입", "지출", "수지"), ("income", "expenses", "net_cashflow")):
with col:
value = report.get(key)
st.metric(label, "통화별" if value is None and report.get("by_currency") else ("미확정" if value is None else f"{int(value):,}"))
if report.get("by_currency"):
for currency, totals in report["by_currency"].items():
st.caption(f"{currency}: 수입 {totals['income']:,} · 지출 {totals['expenses']:,} · 순 {totals['net_cashflow']:,}")
unknown_rows = report.get("unknown_category_items") or [r for r in (report.get("categories") or []) if r.get("cashflow_kind") == "unknown"]
if unknown_rows:
unknown_count = sum(int(r.get("count") or 0) for r in unknown_rows)
st.warning(f"분류가 필요한 거래가 {unknown_count}건 있습니다. 미분류 금액은 수입·지출 합계에 포함되지 않습니다.")
st.markdown("### 미분류 내역")
st.dataframe([{"카테고리": r.get("category") or "미지정", "하위카테고리": r.get("subcategory") or "미지정", "금액": r.get("amount"), "거래수": int(r.get("count") or 0), "통화": r.get("currency") or "미지정"} for r in unknown_rows], hide_index=True, use_container_width=True)
st.markdown("### 카테고리별 내역")
categories = report.get("categories") or []
for kind, title in (("income", "수입"), ("expense", "지출")):
rows = [r for r in categories if r.get("cashflow_kind") == kind]
st.markdown(f"**{title}**")
if not rows:
st.caption("내역이 없습니다.")
continue
st.dataframe([{"카테고리": r.get("category") or "미분류", "하위카테고리": r.get("subcategory") or "미분류", "금액": int(r.get("amount") or 0), "거래수": int(r.get("count") or 0), "통화": r.get("currency") or "미지정"} for r in rows], hide_index=True, use_container_width=True)
def _render_snapshot_unlink_dialog(db_path: str, settlement_id: str, item: dict) -> None:
@st.dialog("평가 연결 해제")
def dialog() -> None:
account_name = str(item.get("account_name") or item.get("name") or "")
amount = item.get("amount")
amount_display = "미확인" if amount is None else f"{amount:,}"
st.markdown(f"**계좌명:** {account_name}")
st.markdown(f"**평가금액:** {amount_display}")
st.info("스냅샷은 보존되고 이 월정산과의 연결만 해제됩니다.")
confirm, cancel = st.columns(2)
if cancel.button("닫기", key=f"monthly_unlink_cancel_{settlement_id}_{item['snapshot_id']}"):
st.rerun()
if confirm.button("연결해제", key=f"monthly_unlink_confirm_{settlement_id}_{item['snapshot_id']}"):
result = settlement_gateway.unlink_account_snapshot(
snapshot_id=str(item["snapshot_id"]), monthly_settlement_id=str(settlement_id), db_path=db_path,
)
if result.get("success"):
st.rerun()
else:
st.error(result.get("error", "연결 해제에 실패했습니다."))
dialog()
def _render_snapshot_category_dialog(db_path: str, settlement_id: str, item: dict) -> None:
@st.dialog("평가 분류 수정")
def dialog() -> None:
buckets = ["cash", "deposit", "investment", "real_estate", "loan", "other"]
current = item.get("snapshot_account_category") or item.get("account_liquidity_bucket") or "other"
value = st.selectbox("분류", buckets, index=buckets.index(current) if current in buckets else 5, key=f"monthly_category_value_{settlement_id}_{item['snapshot_id']}")
if st.button("저장", key=f"monthly_category_save_{settlement_id}_{item['snapshot_id']}"):
result = settlement_gateway.update_account_snapshot_category(str(item["snapshot_id"]), str(settlement_id), value, db_path=db_path)
if result.get("success"):
st.rerun()
else:
st.error(result.get("error", "분류 수정에 실패했습니다."))
dialog()
def _render_account_registration_dialog(db_path: str, settlement_id: str, settlement: dict, item: dict) -> None:
@st.dialog("미등록 계정 평가 등록")
def dialog() -> None:
account_id = str(item.get("account_id") or item.get("source_id"))
account_name = str(item.get("account_name") or item.get("name") or account_id)
prefix = f"monthly_evaluation_{settlement_id}_{account_id}"
st.text_input("계정", value=account_name, disabled=True, key=f"{prefix}_account")
candidates = item.get("unlinked_candidates") or []
modes = ["기존 스냅샷 연결", "새 평가 등록"] if candidates else ["새 평가 등록"]
mode = st.radio("등록 방식", modes, index=0, horizontal=True, key=f"{prefix}_mode")
if mode == "기존 스냅샷 연결":
labels = []
by_label = {}
for candidate in candidates:
amount = "미확인" if candidate.get("amount") is None else f"{candidate['amount']:,}"
label = f"기준일 {candidate.get('basis_date','')} · {amount} {candidate.get('currency','')} · {candidate.get('verification_status') or '미확인'}"
labels.append(label); by_label[label] = candidate
selected_label = st.selectbox("기존 스냅샷", labels, key=f"{prefix}_candidate")
selected = by_label[selected_label]
st.text_input("선택 금액", value="미확인" if selected.get("amount") is None else f"{selected['amount']:,}", disabled=True, key=f"{prefix}_candidate_amount")
mismatch = str(selected.get("basis_date")) != str(settlement.get("basis_date"))
allow_mismatch = False
if mismatch:
st.warning("기준일이 다른 스냅샷을 연결합니다")
allow_mismatch = st.checkbox("기준일 불일치를 확인하고 연결", key=f"{prefix}_mismatch")
connect, cancel = st.columns(2)
if cancel.button("취소", key=f"{prefix}_candidate_cancel"):
st.rerun()
if connect.button("연결", key=f"{prefix}_connect"):
if mismatch and not allow_mismatch:
st.error("기준일 불일치 동의가 필요합니다.")
return
result = settlement_gateway.link_account_snapshot(str(selected["snapshot_id"]), str(settlement_id), allow_basis_date_mismatch=allow_mismatch, db_path=db_path)
if result.get("success"):
st.rerun()
else:
st.error(result.get("error", "스냅샷 연결에 실패했습니다."))
return
settlement_basis_date = settlement.get("basis_date") or settlement.get("selected_basis_date")
if not settlement_basis_date:
settlement_basis_date = f"{settlement.get('month') or settlement.get('selected_month')}-01"
basis_date = st.date_input("기준일", value=date.fromisoformat(str(settlement_basis_date)[:10]), key=f"{prefix}_basis_date")
currency = st.text_input("통화", value=str(item.get("currency") or ""), disabled=True, key=f"{prefix}_currency")
previous = item.get("previous_ledger")
following = item.get("next_ledger")
choices = []
choice_map = {}
if previous:
label = f"이전 기록 · {previous['transaction_date']} · 잔액 {previous['balance_after']} {previous['currency']}"
choices.append(label); choice_map[label] = ("previous", previous)
if following:
label = f"이후 기록 · {following['transaction_date']} · 잔액 {following['balance_after']} {following['currency']}"
choices.append(label); choice_map[label] = ("next", following)
choices.append("수동 기입")
selected = st.selectbox("평가 출처", choices, index=0 if previous else len(choices)-1, key=f"{prefix}_source")
policy, ledger = choice_map.get(selected, ("manual", None))
if ledger:
amount_text = str(ledger["balance_after"])
st.text_input("평가금액 (정수)", value=amount_text, disabled=True, key=f"{prefix}_amount_ledger")
else:
amount_text = st.text_input("평가금액 (정수)", key=f"{prefix}_amount_manual")
if policy == "next":
st.caption("기준일 이후 기록입니다.")
buckets = ["cash", "deposit", "investment", "real_estate", "loan", "other"]
account_bucket = item.get("account_liquidity_bucket") or "other"
liquidity_bucket = st.selectbox("분류", buckets, index=buckets.index(account_bucket) if account_bucket in buckets else 5, key=f"{prefix}_liquidity_bucket")
memo = st.text_area("메모 (선택)", key=f"{prefix}_memo")
st.caption("초기 등록은 draft이며 검증 완료가 아닙니다. 금액이 있어도 확인 상태로 바뀌지 않습니다.")
save, cancel = st.columns(2)
if cancel.button("취소", key=f"{prefix}_cancel"):
st.rerun()
if save.button("저장", key=f"{prefix}_save"):
if not amount_text.strip():
st.error("평가금액을 입력하세요.")
return
try:
amount = int(amount_text.strip())
except ValueError:
st.error("평가금액은 정수여야 합니다.")
return
result = settlement_gateway.register_account_valuation(
account_id=account_id, monthly_settlement_id=settlement_id,
basis_date=basis_date.isoformat(), amount=amount, currency=currency,
memo=memo, selection_policy=policy, valuation_source="ledger" if ledger else "manual",
source_note=ledger["id"] if ledger else "", db_path=db_path,
liquidity_bucket=liquidity_bucket,
)
if result.get("success"):
st.rerun()
else:
st.error(result.get("error", "평가 등록에 실패했습니다."))
dialog()