from __future__ import annotations import sqlite3 from datetime import datetime from html import escape from typing import Any import importlib import inspect from zoneinfo import ZoneInfo import pandas as pd import plotly.express as px import plotly.io as pio import streamlit as st from jarca_ledger_mcp.adapters.streamlit import accounts_gateway as accounts_gateway from jarca_ledger_mcp.adapters.streamlit import assets_gateway as assets_gateway from jarca_ledger_mcp.adapters.streamlit import categorization_gateway as categorization_gateway from jarca_ledger_mcp.adapters.streamlit import imports_gateway as imports_gateway from jarca_ledger_mcp.adapters.streamlit import overview_gateway as overview_gateway from jarca_ledger_mcp.adapters.streamlit import reconciliation_gateway as reconciliation_gateway from jarca_ledger_mcp.adapters.streamlit import transactions_gateway as transactions_gateway from jarca_ledger_mcp.adapters.streamlit import transfers_gateway as transfers_gateway from jarca_ledger_mcp.db import resolve_db_path JARCA_COLORS = { "navy": "#0f172a", "slate": "#475569", "muted": "#64748b", "border": "#e2e8f0", "surface": "#ffffff", "surface_soft": "#f8fafc", "blue": "#2563eb", "cyan": "#06b6d4", "emerald": "#10b981", "amber": "#f59e0b", "rose": "#f43f5e", } JARCA_PLOTLY_TEMPLATE = "jarca_clean" pio.templates[JARCA_PLOTLY_TEMPLATE] = pio.templates["plotly_white"].update( { "layout": { "font": {"family": "Inter, Pretendard, Noto Sans KR, Arial, sans-serif", "color": JARCA_COLORS["navy"]}, "paper_bgcolor": "rgba(0,0,0,0)", "plot_bgcolor": "rgba(255,255,255,0)", "colorway": [JARCA_COLORS["blue"], JARCA_COLORS["rose"], JARCA_COLORS["emerald"], JARCA_COLORS["amber"], JARCA_COLORS["cyan"]], "title": {"font": {"size": 16, "color": JARCA_COLORS["navy"]}, "x": 0.02, "xanchor": "left"}, "legend": {"orientation": "h", "yanchor": "bottom", "y": 1.02, "xanchor": "right", "x": 1}, "margin": {"l": 12, "r": 12, "t": 56, "b": 28}, "xaxis": {"gridcolor": "rgba(148, 163, 184, 0.20)", "zerolinecolor": "rgba(148, 163, 184, 0.35)"}, "yaxis": {"gridcolor": "rgba(148, 163, 184, 0.20)", "zerolinecolor": "rgba(148, 163, 184, 0.35)"}, } } ) px.defaults.template = JARCA_PLOTLY_TEMPLATE DEFAULT_CATEGORIES = [ "", "수입", "이체", "식비", "주거", "통신", "교통", "교육", "의료", "생활", "문화/여가", "금융", "가족/용돈", "미분류" ] DEFAULT_SUBCATEGORIES: dict[str, list[str]] = { "수입": ["", "급여", "이자", "환급", "배당", "기타수입"], "이체": ["", "본인계좌이동", "현금입출금", "가족송금", "투자이동", "확인필요"], "식비": ["", "마트", "외식", "카페", "배달"], "주거": ["", "관리비", "전기", "가스", "수도", "인터넷"], "통신": ["", "휴대폰", "인터넷", "통신요금"], "교통": ["", "대중교통", "택시", "주유", "주차", "차량정비"], "교육": ["", "학원", "교재", "학교", "예체능"], "의료": ["", "병원", "약국", "치과", "보험"], "생활": ["", "생필품", "의류", "미용", "세탁"], "문화/여가": ["", "여행", "숙박", "공연", "취미", "스포츠"], "금융": ["", "카드대금", "대출상환", "수수료", "보험료"], "가족/용돈": ["", "부모님", "자녀", "가족송금"], "미분류": ["", "확인필요"], } CASHFLOW_KIND_OPTIONS = ["income", "expense", "transfer", "investment", "debt_payment", "exclude", "unknown"] CASHFLOW_KIND_LABELS = { "income": "실제 수입", "expense": "실제 지출", "transfer": "이체/제외", "investment": "투자이동/제외", "debt_payment": "부채상환/별도", "exclude": "분석 제외", "unknown": "확인필요", } STATEMENT_STATUS_OPTIONS = ["unknown", "import_supported", "manual_only", "partial_import"] STATEMENT_STATUS_LABELS = { "unknown": "미정", "import_supported": "명세서 import 지원", "manual_only": "수동 증빙 전용", "partial_import": "부분 import + 수동 보완", } LIQUIDITY_BUCKET_OPTIONS = ["cash", "investment", "deposit", "real_estate", "loan", "other"] LIQUIDITY_BUCKET_LABELS = { "cash": "현금성", "investment": "투자자산", "deposit": "예적금", "real_estate": "부동산", "loan": "부채", "other": "기타", } def format_krw(value: Any) -> str: try: if value is None: return "-" return f"{int(value):,}원" except Exception: return str(value) def current_kst_month() -> str: return datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y-%m") def format_percent(value: Any) -> str: if value is None: return "-" try: return f"{float(value):.1%}" except Exception: return str(value) def get_transfer_review_items(payload: dict[str, Any] | None) -> list[dict[str, Any]]: if not payload or not payload.get("success"): return [] items = payload.get("review_items") if isinstance(items, list): return items items = payload.get("transactions") if isinstance(items, list): return items return [] @st.cache_data(ttl=5) def read_sql(db_path: str, sql: str, params: tuple[Any, ...] = ()) -> pd.DataFrame: with sqlite3.connect(db_path) as conn: return pd.read_sql_query(sql, conn, params=params) def clear_cache() -> None: read_sql.clear() def get_db_path() -> str: return str(resolve_db_path()) def get_service_callable(module, name: str, required_params: list[str] | None = None): """Return a service function, reloading once for Streamlit hot-reload edge cases. Streamlit can keep an already-imported service module in memory while the dashboard script itself reloads. When a UI change starts passing a new keyword argument, a stale service function may still exist but have the old signature. In that case reload the service module before returning it. """ required_params = required_params or [] fn = getattr(module, name, None) if callable(fn): try: params = inspect.signature(fn).parameters if all(param in params for param in required_params): return fn except (TypeError, ValueError): if not required_params: return fn importlib.reload(module) return getattr(module, name, None) def inject_app_css() -> None: st.markdown( """ """, unsafe_allow_html=True, ) def render_section_label(icon: str, label: str) -> None: st.markdown(f'
{escape(icon)}{escape(label)}
', unsafe_allow_html=True) def validate_dashboard_page(page: str | None, pages: list[str]) -> str: return page if page in pages else "요약" def _sync_dashboard_page_query() -> None: st.query_params["page"] = st.session_state["dashboard_page"] def render_sidebar_nav(pages: list[str], current_page: str) -> str: if "dashboard_page" not in st.session_state: st.session_state["dashboard_page"] = current_page return st.radio( "메뉴", pages, key="dashboard_page", label_visibility="collapsed", on_change=_sync_dashboard_page_query, ) def render_dashboard_body(page: str, renderer) -> None: body = st.empty() body.empty() try: with body.container(): with st.container(key="dashboard_body"): loading = st.empty() with loading.container(): with st.container(key="dashboard_loading"): st.info(f"{page} 불러오는 중…") with st.container(key="dashboard_content"): renderer() loading.empty() except Exception: body.empty() with body.container(): st.error("페이지를 불러오지 못했습니다. 다시 시도해 주세요.") if st.button("다시 시도", key=f"retry_{page}"): st.rerun() def parse_optional_int(value: str, field_name: str) -> int | None: value = (value or "").strip() if not value: return None try: return int(value) except ValueError as exc: raise ValueError(f"{field_name}는 정수여야 합니다.") from exc def get_category_options(db_path: str) -> tuple[list[str], dict[str, list[str]]]: result = categorization_gateway.list_ledger_categories(active_only=True, include_usage=False, db_path=db_path) if result.get("success") and result.get("count", 0) == 0: categorization_gateway.seed_default_ledger_categories(db_path=db_path) result = categorization_gateway.list_ledger_categories(active_only=True, include_usage=False, db_path=db_path) grouped = result.get("grouped", {}) if result.get("success") else {} categories = [""] + sorted(grouped.keys()) subcategories = {cat: [""] + sorted([s for s in subs if s]) for cat, subs in grouped.items()} # Keep hardcoded defaults as fallback for fresh/broken DBs and for existing transaction values. for cat in DEFAULT_CATEGORIES: if cat and cat not in categories: categories.append(cat) for cat, subs in DEFAULT_SUBCATEGORIES.items(): subcategories.setdefault(cat, [""]) for sub in subs: if sub not in subcategories[cat]: subcategories[cat].append(sub) return categories, subcategories def transaction_df(db_path: str, where: str = "", params: tuple[Any, ...] = (), limit: int = 1000) -> pd.DataFrame: sql = f""" SELECT t.id, t.transaction_datetime, t.transaction_date, t.type, t.amount, t.category, t.subcategory, COALESCE(lc.cashflow_kind, 'unknown') AS cashflow_kind, t.merchant, t.raw_summary, t.counterparty, t.person, t.is_fixed, t.balance_after, t.branch, t.source, t.import_batch_id, t.memo, t.status FROM transactions t LEFT JOIN ledger_categories lc ON lc.active=1 AND t.category=lc.category AND t.subcategory=lc.subcategory {where} ORDER BY COALESCE(NULLIF(t.transaction_datetime,''), t.transaction_date) DESC, t.created_at DESC LIMIT ? """ df = read_sql(db_path, sql, params + (limit,)) if not df.empty: df["금액"] = df["amount"].map(format_krw) df["거래후잔액"] = df["balance_after"].map(format_krw) df["고정비"] = df["is_fixed"].fillna(0).astype(bool) return df def update_transaction_form(tx: dict[str, Any], key_prefix: str, db_path: str | None = None) -> None: db_path = db_path or get_db_path() category_options, subcategory_options = get_category_options(db_path) category_key = f"{key_prefix}_category" subcategory_key = f"{key_prefix}_subcategory" person_key = f"{key_prefix}_person" fixed_key = f"{key_prefix}_fixed" memo_key = f"{key_prefix}_memo" save_rule_key = f"{key_prefix}_save_rule" tx_marker_key = f"{key_prefix}_tx_id" current_category = tx.get("category", "") or "" if current_category and current_category not in category_options: category_options = category_options + [current_category] if st.session_state.get(tx_marker_key) != tx["id"]: st.session_state[tx_marker_key] = tx["id"] st.session_state[category_key] = current_category st.session_state[subcategory_key] = tx.get("subcategory", "") or "" st.session_state[person_key] = tx.get("person", "") or "" st.session_state[fixed_key] = bool(tx.get("is_fixed") or False) st.session_state[memo_key] = tx.get("memo", "") or "" st.session_state[save_rule_key] = False st.caption(f"거래 ID: `{tx['id']}`") st.write(f"**{tx.get('transaction_datetime') or tx.get('transaction_date')}** · {tx.get('merchant')} · {format_krw(tx.get('amount'))}") category = st.selectbox( "대분류", category_options, index=category_options.index(st.session_state.get(category_key, current_category)) if st.session_state.get(category_key, current_category) in category_options else 0, key=category_key, ) sub_opts = subcategory_options.get(category, [""]) current_sub = st.session_state.get(subcategory_key, tx.get("subcategory", "") or "") or "" if current_sub and current_sub not in sub_opts: current_sub = "" st.session_state[subcategory_key] = "" subcategory = st.selectbox( "소분류", sub_opts, index=sub_opts.index(current_sub) if current_sub in sub_opts else 0, key=subcategory_key, ) person = st.text_input("가족 구성원/person", key=person_key) is_fixed = st.checkbox("고정비", key=fixed_key) memo = st.text_input("메모", key=memo_key) save_rule = st.checkbox("이 사용처 패턴을 규칙으로 저장", key=save_rule_key) submitted = st.button("저장", type="primary", key=f"{key_prefix}_submit") if submitted: result = transactions_gateway.update_ledger_transaction( tx["id"], category=category, subcategory=subcategory, person=person, is_fixed=is_fixed, memo=memo, db_path=db_path, ) if not result.get("success"): st.error(result.get("error", result)) return if save_rule and tx.get("merchant") and category: rule = categorization_gateway.create_category_rule( match_field="merchant", pattern=tx["merchant"], category=category, subcategory=subcategory, transaction_type=tx.get("type", ""), person=person, is_fixed=is_fixed, priority=70, memo="dashboard: user-confirmed rule", db_path=db_path, ) if not rule.get("success"): st.warning(f"거래는 저장됐지만 규칙 저장 실패: {rule.get('error')}") clear_cache() st.success("저장했습니다.") st.rerun() @st.dialog("거래 분류", width="large") def transaction_edit_dialog(tx: dict[str, Any], key_prefix: str, db_path: str) -> None: update_transaction_form(tx, key_prefix, db_path) @st.dialog("연결 거래 후보", width="large") def linked_transfer_candidates_dialog(tx: dict[str, Any], db_path: str) -> None: st.caption(f"기준 거래: `{tx['id']}`") st.write(f"**{tx.get('transaction_datetime') or tx.get('transaction_date')}** · {tx.get('merchant')} · {format_krw(tx.get('amount'))}") result = transfers_gateway.find_self_transfer_candidates(tx["id"], limit=20, db_path=db_path) if not result.get("success"): st.error(result.get("error", result)) return candidates = result.get("candidates", []) if not candidates: st.info("같은 금액/가까운 날짜 조건에서 연결 후보를 찾지 못했습니다.") return st.success(f"연결 후보 {result.get('count', len(candidates))}건") for idx, candidate in enumerate(candidates): cols = st.columns([1.45, 0.95, 0.7, 1.2, 1.15, 0.95]) cols[0].write(candidate.get("transaction_datetime") or candidate.get("transaction_date") or "") cols[1].write(candidate.get("source") or "") cols[2].write(candidate.get("type") or "") cols[3].write(format_krw(candidate.get("amount"))) cols[4].write(candidate.get("merchant") or candidate.get("counterparty") or "") if cols[5].button("이 거래와 연결", key=f"link_pair_{tx['id']}_{candidate['id']}_{idx}", use_container_width=True, type="primary"): link_res = transfers_gateway.link_self_transfer_pair(tx["id"], candidate["id"], db_path=db_path) if not link_res.get("success"): st.error(link_res.get("error", link_res)) else: clear_cache() st.success("양쪽 거래를 본인계좌이동으로 연결했습니다.") st.rerun() meta = [] if candidate.get("account_name"): meta.append(f"계좌: {candidate['account_name']}") if candidate.get("category") or candidate.get("subcategory"): meta.append(f"현재 분류: {candidate.get('category', '')} / {candidate.get('subcategory', '')}") if candidate.get("score") is not None: meta.append(f"score={candidate['score']}") if candidate.get("reason"): meta.append(candidate["reason"]) st.caption(" · ".join(meta)) if idx < len(candidates) - 1: st.divider() @st.dialog("원거래 후보", width="large") def cancel_original_candidates_dialog(tx: dict[str, Any], db_path: str) -> None: st.caption(f"취소/환급 거래: `{tx['id']}`") st.write(f"**{tx.get('transaction_datetime') or tx.get('transaction_date')}** · {tx.get('merchant')} · {format_krw(tx.get('amount'))}") result = transfers_gateway.find_cancel_original_candidates(tx["id"], limit=20, db_path=db_path) if not result.get("success"): st.error(result.get("error", result)) return candidates = result.get("candidates", []) if not candidates: st.info("같은 금액 기준으로 연결할 원거래 후보를 찾지 못했습니다.") return st.success(f"원거래 후보 {result.get('count', len(candidates))}건") for idx, candidate in enumerate(candidates): cols = st.columns([1.4, 0.95, 0.7, 1.25, 1.15, 0.95]) cols[0].write(candidate.get("transaction_datetime") or candidate.get("transaction_date") or "") cols[1].write(candidate.get("source") or "") cols[2].write(candidate.get("type") or "") cols[3].write(format_krw(candidate.get("amount"))) cols[4].write(candidate.get("merchant") or candidate.get("counterparty") or "") if cols[5].button("이 원거래와 연결", key=f"cancel_pair_{tx['id']}_{candidate['id']}_{idx}", use_container_width=True, type="primary"): link_res = transfers_gateway.link_cancel_to_original(tx["id"], candidate["id"], db_path=db_path) if not link_res.get("success"): st.error(link_res.get("error", link_res)) else: clear_cache() st.success("취소/환급 거래를 원거래와 연결했습니다.") st.rerun() meta = [] if candidate.get("account_name"): meta.append(f"계좌: {candidate['account_name']}") if candidate.get("category") or candidate.get("subcategory"): meta.append(f"현재 분류: {candidate.get('category', '')} / {candidate.get('subcategory', '')}") if candidate.get("score") is not None: meta.append(f"score={candidate['score']}") if candidate.get("reason"): meta.append(candidate["reason"]) st.caption(" · ".join(meta)) if idx < len(candidates) - 1: st.divider() def render_review_decision_actions(gateway: Any, target_type: str, target_id: str, key_prefix: str, db_path: str, target_payload: dict[str, Any] | None = None) -> None: decision_type = st.selectbox( "판정", ["unknown", "account_transfer", "real_expense", "real_income", "investment_transfer", "excluded_from_analysis", "duplicate_or_cancel"], key=f"{key_prefix}_decision_type", ) reason_code = st.text_input("사유 코드", value="operator_review", key=f"{key_prefix}_reason") memo = st.text_input("결정 메모", value="", key=f"{key_prefix}_memo") cols = st.columns(4) if cols[0].button("결정 저장", key=f"{key_prefix}_decided", use_container_width=True, type="primary"): res = gateway.upsert_review_decision(target_type, target_id, status="decided", decision_type=decision_type, reason_code=reason_code, memo=memo, target_payload=target_payload or {}, db_path=db_path) clear_cache(); st.success("저장했습니다." if res.get("success") else res.get("error", res)); st.rerun() if cols[1].button("보류", key=f"{key_prefix}_held", use_container_width=True): res = gateway.hold_review_decision(target_type, target_id, reason_code=reason_code or "held", memo=memo, db_path=db_path) clear_cache(); st.success("보류했습니다." if res.get("success") else res.get("error", res)); st.rerun() if cols[2].button("재검토", key=f"{key_prefix}_recheck", use_container_width=True): res = gateway.mark_review_recheck(target_type, target_id, reason_code=reason_code or "needs_recheck", memo=memo, db_path=db_path) clear_cache(); st.success("재검토로 표시했습니다." if res.get("success") else res.get("error", res)); st.rerun() if cols[3].button("해제", key=f"{key_prefix}_release", use_container_width=True): res = gateway.release_review_decision(target_type, target_id, memo=memo, db_path=db_path) clear_cache(); st.success("pending으로 해제했습니다." if res.get("success") else res.get("error", res)); st.rerun()