from __future__ import annotations # ruff: noqa: F405 import pandas as pd import streamlit as st from jarca_ledger_mcp.adapters.streamlit import assets_gateway from jarca_ledger_mcp.adapters.streamlit import overview_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: """Render allowlisted evidence, never free-form source notes or memos.""" 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("**근거 상세 (코드 / 출처 식별자)**") if warnings: for item in warnings: st.markdown(f"- 경고 `{item.get('code', '-')}` · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 대상 {item.get('name', '-')}") else: st.caption("경고 상세 없음") 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', '-')} · 근거 {item.get('evidence_kind', '-')}") 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_assets(db_path: str) -> None: st.caption("표시 기준: preview는 live_authority 추정값, close 이후 조회는 closed snapshot 값입니다. 실운영 전 synthetic/validation DB에서 먼저 검증하세요.") render_section_label("✍️", "수기 자산/부채 등록") with st.form("manual_asset_create"): c1, c2, c3 = st.columns([1.3, 1, 1]) name = c1.text_input("자산/부채명", placeholder="예: 전세보증금, 주택담보대출") amount = c2.number_input("평가금액(원)", min_value=0, value=0, step=10000) valuation_date = c3.date_input("평가 기준일") c4, c5 = st.columns([1, 2]) bucket = c4.selectbox("유동성 버킷", LIQUIDITY_BUCKET_OPTIONS, format_func=lambda x: LIQUIDITY_BUCKET_LABELS.get(x, x)) source_note = c5.text_input("출처 메모", placeholder="예: 월말 앱 잔액, 계약서, synthetic") memo = st.text_input("운영 메모", value="") submitted = st.form_submit_button("수기 자산 추가", type="primary") if submitted: res = assets_gateway.create_manual_asset(name, int(amount), valuation_date.isoformat(), bucket, source_note=source_note, memo=memo, db_path=db_path) if res.get("success"): clear_cache() st.success("수기 자산을 추가했습니다.") st.rerun() else: st.error(res.get("error", res)) assets_res = assets_gateway.list_manual_assets(active_only=False, db_path=db_path) assets = assets_res.get("manual_assets", []) if assets_res.get("success") else [] render_section_label("📋", "수기 자산/부채 목록") if assets: df = pd.DataFrame(assets) df["bucket_label"] = df["liquidity_bucket"].map(LIQUIDITY_BUCKET_LABELS).fillna(df["liquidity_bucket"]) df["valuation_display"] = df["valuation_amount"].map(format_krw) st.dataframe(df[["name", "active", "bucket_label", "valuation_display", "valuation_date", "id"]], width="stretch", height=220) with st.form("manual_asset_update"): asset_options = {f"{a['name']} ({a['id']})": a for a in assets} selected_label = st.selectbox("수정/비활성 대상", list(asset_options.keys())) selected = asset_options[selected_label] u1, u2, u3 = st.columns([1.3, 1, 1]) new_name = u1.text_input("이름", value=selected.get("name", "")) new_amount = u2.number_input("평가금액(원)", min_value=0, value=int(selected.get("valuation_amount") or 0), step=10000, key="asset_update_amount") new_date = u3.text_input("평가 기준일", value=selected.get("valuation_date", "")) u4, u5 = st.columns([1, 2]) current_bucket = selected.get("liquidity_bucket", "other") if selected.get("liquidity_bucket", "other") in LIQUIDITY_BUCKET_OPTIONS else "other" new_bucket = u4.selectbox("버킷", LIQUIDITY_BUCKET_OPTIONS, index=LIQUIDITY_BUCKET_OPTIONS.index(current_bucket), format_func=lambda x: LIQUIDITY_BUCKET_LABELS.get(x, x), key="asset_update_bucket") new_source = u5.text_input("출처 메모", value=selected.get("source_note", "")) new_memo = st.text_input("메모", value=selected.get("memo", ""), key="asset_update_memo") make_active = st.checkbox("활성", value=bool(selected.get("active", True))) col_update, col_deactivate = st.columns(2) do_update = col_update.form_submit_button("선택 항목 수정", type="primary") do_deactivate = col_deactivate.form_submit_button("선택 항목 비활성", type="secondary") if do_update: res = assets_gateway.update_manual_asset(selected["id"], name=new_name, valuation_amount=int(new_amount), valuation_date=new_date, liquidity_bucket=new_bucket, source_note=new_source, memo=new_memo, active=make_active, db_path=db_path) if res.get("success"): clear_cache(); st.success("수정했습니다."); st.rerun() else: st.error(res.get("error", res)) if do_deactivate: res = assets_gateway.deactivate_manual_asset(selected["id"], db_path=db_path) if res.get("success"): clear_cache(); st.warning("비활성 처리했습니다."); st.rerun() else: st.error(res.get("error", res)) else: st.info("등록된 수기 자산/부채가 없습니다.") st.divider() render_section_label("🔒", "월말 스냅샷 Preview / Close") st.caption("Preview는 DB를 변경하지 않습니다. 같은 기준일을 다시 close하면 정정 사유가 필요하며 기존 행을 덮어쓰지 않고 새 correction run을 append합니다.") c1, c2, c3, c4 = st.columns([1, 1, 1.5, 1.5]) basis_date = c1.text_input("마감 기준일", value=f"{current_kst_month()}-01", help="YYYY-MM-DD") selection_policy = c2.selectbox("선택 정책", ["exact", "as_of"], format_func=lambda value: "정확히 해당일" if value == "exact" else "기준일 현재(as-of)") snapshot_report = overview_gateway.get_asset_snapshot_report(basis_date, selection_policy=selection_policy, db_path=db_path) if snapshot_report.get("success"): _render_asset_evidence(snapshot_report) close_note = c3.text_input("Close 출처 메모", value="month-end close") correction_reason = c4.text_input("정정 사유(재마감 시 필수)", value="") p1, p2 = st.columns(2) approval_required = st.checkbox("조건부 close 승인 명시", value=False) approval_cols = st.columns(3) approved_by = approval_cols[0].text_input("승인자") approval_reason = approval_cols[1].text_input("승인 사유") approval_reference = approval_cols[2].text_input("승인 근거 참조") allow_missing = st.checkbox("누락 제외 허용", value=False) allow_carry = st.checkbox("as-of 이월 허용", value=False) if p1.button("스냅샷 Preview", type="secondary"): st.session_state["asset_snapshot_preview"] = assets_gateway.preview_asset_snapshot_close(basis_date, selection_policy=selection_policy, db_path=db_path) if p2.button("스냅샷 Close 실행", type="primary"): st.session_state["asset_snapshot_close"] = assets_gateway.close_asset_snapshot( basis_date, source_note=close_note, correction_reason=correction_reason, selection_policy=selection_policy, allow_missing_exclusions=allow_missing, allow_as_of_carry_forward=allow_carry, approved_by=approved_by if approval_required else "", approval_reason=approval_reason if approval_required else "", approval_reference=approval_reference if approval_required else "", db_path=db_path, ) clear_cache() preview = st.session_state.get("asset_snapshot_preview") close_res = st.session_state.get("asset_snapshot_close") if preview: if preview.get("success"): st.success(f"Preview: source={preview.get('source')}, closed={preview.get('is_closed_snapshot')}, warnings={len(preview.get('warnings', []))}") m1, m2, m3 = st.columns(3) m1.metric("자산 합계", format_krw(preview.get("total_assets"))) m2.metric("총부채", format_krw(preview.get("total_liabilities"))) m3.metric("순자산", format_krw(preview.get("net_worth"))) st.dataframe(pd.DataFrame(preview.get("included_items", [])), width="stretch", height=220) if preview.get("warnings"): st.warning(preview.get("warnings")) _render_asset_evidence(preview) else: st.error(preview.get("error", preview)) if close_res: if close_res.get("success"): st.success(f"Close 완료: run={close_res.get('snapshot_run_id')}, correction={close_res.get('is_correction')}, inserted={close_res.get('inserted_count')}") st.json({k: close_res.get(k) for k in ["basis_date", "source", "is_closed_snapshot", "snapshot_run_id", "correction_of_run_id", "correction_reason", "closed_at", "net_worth"]}) else: st.error(close_res.get("error", close_res)) st.divider() render_settlement_drafts(db_path=db_path)