189 lines
7.4 KiB
Python
189 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
# ruff: noqa: F405
|
|
|
|
from html import escape
|
|
from typing import Any
|
|
|
|
import streamlit as st
|
|
|
|
from jarca_ledger_mcp.adapters.streamlit import accounts_gateway
|
|
from dashboard.components import JarcaList
|
|
from dashboard.common import * # noqa: F403, F405
|
|
|
|
|
|
ACCOUNT_TYPE_LABELS = {
|
|
"bank": "은행",
|
|
"card": "카드",
|
|
"cash": "현금",
|
|
"investment": "투자",
|
|
"loan": "대출",
|
|
"other": "기타",
|
|
}
|
|
|
|
ACCOUNT_TYPE_STYLES = {
|
|
"bank": "#0969da",
|
|
"card": "#bf3989",
|
|
"cash": "#1a7f37",
|
|
"investment": "#8250df",
|
|
"loan": "#bc4c00",
|
|
"other": "#57606a",
|
|
}
|
|
|
|
def render_accounts_list_header(items: list[dict[str, Any]]) -> None:
|
|
st.markdown(
|
|
f'{len(items)}개 계좌',
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
|
|
def render_account_item(account: dict[str, Any]) -> None:
|
|
active = bool(account.get("active", True))
|
|
status_label = ":blue-badge[활성]" if active else ":gray-badge[비활성]"
|
|
account_type = ACCOUNT_TYPE_LABELS.get(account.get("type"), account.get("type") or "기타")
|
|
statement_status = STATEMENT_STATUS_LABELS.get(account.get("statement_status"), account.get("statement_status") or "-")
|
|
balance = format_krw(account.get("last_balance"))
|
|
synced_at = account.get("last_synced_at") or "미입력"
|
|
memo = account.get("memo") or "메모 없음"
|
|
with st.container(gap=None):
|
|
columns = st.columns([9, 1], vertical_alignment="center", gap=None)
|
|
columns[0].markdown(
|
|
f"""
|
|
**{escape(str(account.get("name") or "이름 없음"))}** {status_label}
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
if columns[1].button(" ", key=f"account_edit_{account['id']}", help="수정", icon=":material/edit:", type="tertiary", width="content"):
|
|
st.session_state["account_edit_id"] = account["id"]
|
|
columns = st.columns([2, 80, 18], vertical_alignment="center")
|
|
columns[1].markdown(
|
|
f"""
|
|
{escape(str(memo))}<br>
|
|
<small>#{escape(str(account["id"])[-6:])} · {escape(str(synced_at))} 기준 · {escape(str(account.get("institution") or "기관 미입력"))} · {escape(str(account_type))} · {escape(str(statement_status))}</small>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
columns[2].text(escape(balance), width="stretch", text_alignment="right")
|
|
|
|
def render_account_list(accounts: list[dict[str, Any]]) -> None:
|
|
st.markdown(
|
|
"""
|
|
<style>
|
|
[class*="st-key-account_edit_"] {
|
|
width: fit-content;
|
|
margin-left: auto;
|
|
}
|
|
[class*="st-key-account_edit_"] [data-testid="stBaseButton-tertiary"] {
|
|
width: auto;
|
|
margin: 0;
|
|
padding: 0;
|
|
}
|
|
</style>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
search = st.text_input("계좌 검색", placeholder="계좌를 검색하세요", label_visibility="collapsed")
|
|
query = search.strip().lower()
|
|
visible_accounts = [
|
|
account
|
|
for account in accounts
|
|
if not query or query in " ".join(str(account.get(field) or "") for field in ("name", "institution", "type")).lower()
|
|
]
|
|
|
|
JarcaList(
|
|
render_header=render_accounts_list_header,
|
|
render_item=render_account_item,
|
|
items=visible_accounts,
|
|
empty_message="등록된 계좌가 없습니다.",
|
|
).render()
|
|
|
|
|
|
@st.dialog("계좌 수정", width="large")
|
|
def account_edit_dialog(account: dict[str, Any], db_path: str) -> None:
|
|
st.caption(f"{account.get('institution') or '기관 미입력'} · ID: `{account['id']}`")
|
|
with st.form(f"account_edit_form_{account['id']}"):
|
|
identity_columns = st.columns(3)
|
|
name = identity_columns[0].text_input("계좌명", value=account.get("name") or "")
|
|
institution = identity_columns[1].text_input("금융기관", value=account.get("institution") or "")
|
|
current_type = account.get("type", "other") or "other"
|
|
account_type = identity_columns[2].selectbox(
|
|
"유형",
|
|
list(ACCOUNT_TYPE_LABELS),
|
|
index=list(ACCOUNT_TYPE_LABELS).index(current_type) if current_type in ACCOUNT_TYPE_LABELS else 0,
|
|
format_func=lambda value: ACCOUNT_TYPE_LABELS[value],
|
|
)
|
|
|
|
balance_columns = st.columns(2)
|
|
last_balance_raw = balance_columns[0].text_input(
|
|
"기준 잔액",
|
|
value="" if account.get("last_balance") is None else str(account.get("last_balance")),
|
|
help="원 단위 정수",
|
|
)
|
|
last_synced_at = balance_columns[1].text_input(
|
|
"기준 시각",
|
|
value=account.get("last_synced_at") or "",
|
|
help="예: 2026-08-20 또는 2026-08-20T23:59:00+09:00",
|
|
)
|
|
|
|
operating_columns = st.columns(3)
|
|
current_status = account.get("statement_status", "unknown") or "unknown"
|
|
statement_status = operating_columns[0].selectbox(
|
|
"명세서 상태",
|
|
STATEMENT_STATUS_OPTIONS,
|
|
index=STATEMENT_STATUS_OPTIONS.index(current_status) if current_status in STATEMENT_STATUS_OPTIONS else 0,
|
|
format_func=lambda value: STATEMENT_STATUS_LABELS[value],
|
|
)
|
|
current_bucket = account.get("liquidity_bucket", "other") or "other"
|
|
liquidity_bucket = operating_columns[1].selectbox(
|
|
"유동성 버킷",
|
|
LIQUIDITY_BUCKET_OPTIONS,
|
|
index=LIQUIDITY_BUCKET_OPTIONS.index(current_bucket) if current_bucket in LIQUIDITY_BUCKET_OPTIONS else LIQUIDITY_BUCKET_OPTIONS.index("other"),
|
|
format_func=lambda value: LIQUIDITY_BUCKET_LABELS[value],
|
|
)
|
|
active = operating_columns[2].checkbox("활성", value=bool(account.get("active", True)))
|
|
memo = st.text_input("메모", value=account.get("memo") or "")
|
|
submitted = st.form_submit_button("변경 저장", type="primary")
|
|
|
|
if submitted:
|
|
try:
|
|
last_balance = int(last_balance_raw.replace(",", "")) if last_balance_raw.strip() else None
|
|
except ValueError:
|
|
st.error("기준 잔액은 원 단위 정수로 입력하세요.")
|
|
return
|
|
result = accounts_gateway.update_ledger_account(
|
|
account["id"],
|
|
name=name,
|
|
institution=institution,
|
|
type=account_type,
|
|
last_balance=last_balance,
|
|
last_synced_at=last_synced_at,
|
|
statement_status=statement_status,
|
|
liquidity_bucket=liquidity_bucket,
|
|
active=active,
|
|
memo=memo,
|
|
db_path=db_path,
|
|
)
|
|
if not result.get("success"):
|
|
st.error(result.get("error", result))
|
|
return
|
|
st.session_state.pop("account_edit_id", None)
|
|
clear_cache()
|
|
st.rerun()
|
|
|
|
|
|
def page_accounts(db_path: str) -> None:
|
|
accounts_res = accounts_gateway.list_ledger_accounts(active_only=False, db_path=db_path)
|
|
accounts_data = accounts_res.get("accounts", []) if accounts_res.get("success") else []
|
|
if accounts_data:
|
|
render_account_list(accounts_data)
|
|
account_edit_id = st.session_state.get("account_edit_id")
|
|
if account_edit_id:
|
|
account = next((item for item in accounts_data if item["id"] == account_edit_id), None)
|
|
if account:
|
|
account_edit_dialog(account, db_path)
|
|
else:
|
|
st.session_state.pop("account_edit_id", None)
|
|
st.rerun()
|
|
else:
|
|
st.info("등록된 계좌가 없습니다.")
|