from __future__ import annotations # ruff: noqa: F405 import pandas as pd import streamlit as st from jarca_ledger_mcp.adapters.streamlit import categorization_gateway from dashboard.common import * # noqa: F403, F405 RULE_MATCH_FIELDS = ["merchant", "raw_summary", "counterparty", "payment_method", "branch", "description", "memo"] RULE_MATCH_TYPES = ["contains", "equals", "starts_with", "ends_with", "regex"] RULE_TX_TYPES = ["", "income", "expense", "transfer", "refund", "adjustment"] RULE_FIXED_LABELS = ["변경안함", "예", "아니오"] @st.dialog("분류 규칙 추가", width="large") def category_rule_create_dialog(db_path: str) -> None: category_options, _ = get_category_options(db_path) editable_categories = [c for c in category_options if c] with st.form("rule_create_dialog"): cols = st.columns(4) match_field = cols[0].selectbox("대상 필드", RULE_MATCH_FIELDS) match_type = cols[1].selectbox("매칭 방식", RULE_MATCH_TYPES) pattern = cols[2].text_input("패턴") tx_type = cols[3].selectbox("거래 유형", RULE_TX_TYPES) cols2 = st.columns(5) category = cols2[0].selectbox("대분류", editable_categories) subcategory = cols2[1].text_input("소분류") person = cols2[2].text_input("person") priority = cols2[3].number_input("우선순위", value=100, step=10) is_fixed = cols2[4].selectbox("고정비", RULE_FIXED_LABELS) cols3 = st.columns(3) amount_min_text = cols3[0].text_input("최소 금액(선택)", placeholder="예: 1200000") amount_max_text = cols3[1].text_input("최대 금액(선택)", placeholder="예: 1200000") day_of_month_text = cols3[2].text_input("매월 일자(선택)", placeholder="1~31") memo = st.text_input("메모", value="dashboard: user rule") submitted = st.form_submit_button("규칙 저장", type="primary") if submitted: try: amount_min = parse_optional_int(amount_min_text, "최소 금액") amount_max = parse_optional_int(amount_max_text, "최대 금액") day_of_month = parse_optional_int(day_of_month_text, "매월 일자") except ValueError as exc: st.error(str(exc)) else: fixed_value = None if is_fixed == "변경안함" else is_fixed == "예" res = categorization_gateway.create_category_rule(match_field, pattern, category, subcategory, tx_type, match_type, int(priority), person, fixed_value, 1.0, memo, amount_min, amount_max, day_of_month, db_path=db_path) clear_cache() if res.get("success"): st.success("규칙을 저장했습니다.") st.rerun() else: st.error(res.get("error", res)) @st.dialog("분류 규칙 수정", width="large") def category_rule_edit_dialog(rule: dict[str, Any], db_path: str) -> None: category_options, _ = get_category_options(db_path) current_category = rule.get("category", "") or "" editable_categories = [c for c in category_options if c] if current_category and current_category not in editable_categories: editable_categories.append(current_category) fixed_value = "변경안함" if rule.get("is_fixed") is None else ("예" if rule.get("is_fixed") else "아니오") st.caption(f"규칙 ID: `{rule['id']}`") st.write(f"**{rule.get('pattern') or '-'}**") with st.form(f"rule_edit_dialog_{rule['id']}"): cols = st.columns(4) edit_match_field = cols[0].selectbox("대상 필드", RULE_MATCH_FIELDS, index=RULE_MATCH_FIELDS.index(rule.get("match_field", "merchant")) if rule.get("match_field", "merchant") in RULE_MATCH_FIELDS else 0) edit_match_type = cols[1].selectbox("매칭 방식", RULE_MATCH_TYPES, index=RULE_MATCH_TYPES.index(rule.get("match_type", "contains")) if rule.get("match_type", "contains") in RULE_MATCH_TYPES else 0) edit_pattern = cols[2].text_input("패턴", value=rule.get("pattern", "") or "") edit_tx_type = cols[3].selectbox("거래 유형", RULE_TX_TYPES, index=RULE_TX_TYPES.index(rule.get("transaction_type", "") or "") if (rule.get("transaction_type", "") or "") in RULE_TX_TYPES else 0) cols2 = st.columns(5) edit_category = cols2[0].selectbox("대분류", editable_categories, index=editable_categories.index(current_category) if current_category in editable_categories else 0) edit_subcategory = cols2[1].text_input("소분류", value=rule.get("subcategory", "") or "") edit_person = cols2[2].text_input("person", value=rule.get("person", "") or "") edit_priority = cols2[3].number_input("우선순위", value=int(rule.get("priority", 100) or 100), step=10) edit_is_fixed = cols2[4].selectbox("고정비", RULE_FIXED_LABELS, index=RULE_FIXED_LABELS.index(fixed_value)) cols3 = st.columns(4) edit_amount_min_text = cols3[0].text_input("최소 금액(선택)", value="" if rule.get("amount_min") is None else str(rule.get("amount_min"))) edit_amount_max_text = cols3[1].text_input("최대 금액(선택)", value="" if rule.get("amount_max") is None else str(rule.get("amount_max"))) edit_day_of_month_text = cols3[2].text_input("매월 일자(선택)", value="" if rule.get("day_of_month") is None else str(rule.get("day_of_month"))) edit_active = cols3[3].checkbox("활성", value=bool(rule.get("active", True))) edit_memo = st.text_input("메모", value=rule.get("memo", "") or "") edit_submitted = st.form_submit_button("규칙 수정", type="primary") if edit_submitted: try: edit_amount_min = parse_optional_int(edit_amount_min_text, "최소 금액") edit_amount_max = parse_optional_int(edit_amount_max_text, "최대 금액") edit_day_of_month = parse_optional_int(edit_day_of_month_text, "매월 일자") except ValueError as exc: st.error(str(exc)) else: edit_fixed_value = None if edit_is_fixed == "변경안함" else edit_is_fixed == "예" update_rule = getattr(categorization_gateway, "update_category_rule", None) if update_rule is None: st.error("현재 실행 중인 서비스에 update_category_rule 함수가 없습니다. 앱을 재시작해 최신 코드를 다시 불러오세요.") else: res = update_rule( rule["id"], edit_match_field, edit_pattern, edit_category, edit_subcategory, edit_tx_type, edit_match_type, int(edit_priority), edit_person, edit_fixed_value, 1.0, edit_memo, edit_amount_min, edit_amount_max, edit_day_of_month, edit_active, db_path=db_path, ) clear_cache() if res.get("success"): st.success("규칙을 수정했습니다.") st.rerun() else: st.error(res.get("error", res)) @st.dialog("규칙 적용 미리보기", width="large") def category_rule_preview_dialog(rule: dict[str, Any], db_path: str) -> None: rule_id = rule.get("id", "") pattern = rule.get("pattern") or "-" result_text = f"{rule.get('category') or '-'} / {rule.get('subcategory') or '-'}" st.caption(f"규칙 ID: `{rule_id}`") st.write(f"**{pattern}** → **{result_text}**") preview_fn = get_service_callable(categorization_gateway, "preview_category_rule_matches") if not callable(preview_fn): st.error("현재 실행 중인 서비스에서 단일 규칙 미리보기 함수를 찾지 못했습니다. Streamlit 서버를 완전히 종료한 뒤 다시 시작해 주세요.") return preview = preview_fn(rule_id, limit=10000, sample_limit=3, db_path=db_path) if not preview.get("success"): st.error(preview.get("error", preview)) return m1, m2, m3 = st.columns(3) m1.metric("스캔한 활성 거래", f"{preview.get('scanned', 0):,}건") m2.metric("이 규칙과 매칭", f"{preview.get('matched_count', 0):,}건") m3.metric("미분류 중 적용 대상", f"{preview.get('uncategorized_matched_count', 0):,}건") sample = preview.get("sample", []) if sample: df = pd.DataFrame(sample) if "amount" in df: df["금액"] = df["amount"].map(format_krw) display_cols = [ c for c in [ "transaction_date", "transaction_datetime", "type", "금액", "merchant", "raw_summary", "counterparty", "current_category", "current_subcategory", "new_category", "new_subcategory", ] if c in df.columns ] st.dataframe(df[display_cols], width="stretch", height=160) else: st.info("이 규칙에 부합하는 거래가 없습니다.") def render_rule_list_items(rules: list[dict[str, Any]], db_path: str) -> None: if not rules: st.info("등록된 규칙이 없습니다.") return st.markdown( """ """, unsafe_allow_html=True, ) for idx, rule in enumerate(rules): rule_id = rule.get("id", "") active = bool(rule.get("active", True)) status_label = "활성" if active else "비활성" tx_type = rule.get("transaction_type") or "전체" person = rule.get("person") or "-" fixed = rule.get("is_fixed") fixed_label = "예" if fixed is True else "아니오" if fixed is False else "변경안함" amount_bits = [] if rule.get("amount_min") is not None: amount_bits.append(f"최소 {format_krw(rule.get('amount_min'))}") if rule.get("amount_max") is not None: amount_bits.append(f"최대 {format_krw(rule.get('amount_max'))}") if rule.get("day_of_month") is not None: amount_bits.append(f"매월 {rule.get('day_of_month')}일") condition_text = " · ".join(amount_bits) if amount_bits else "추가 조건 없음" match_field = rule.get("match_field") or "-" match_type = rule.get("match_type") or "-" pattern_text = rule.get("pattern") or "-" result_text = f"{rule.get('category') or '-'} / {rule.get('subcategory') or '-'}" memo = rule.get("memo") or "" priority = rule.get("priority", "-") summary = f"{'🟢' if active else '⚪'} {pattern_text} · {match_field}/{match_type} → {result_text} · p{priority} · {tx_type}" with st.expander(summary, expanded=False): st.markdown( f"""
IF
{escape(match_field)} {escape(match_type)} “{escape(pattern_text)}”
→
THEN
{escape(result_text)}
상태: {escape(status_label)} 우선순위: p{escape(str(priority))} 거래유형: {escape(tx_type)} 조건: {escape(condition_text)} person: {escape(person)} 고정비: {escape(fixed_label)} ID: {escape(rule_id)}
{f'
메모: {escape(memo)}
' if memo else ''} """, unsafe_allow_html=True, ) action_cols = st.columns([1.35, 0.9, 0.8, 1.4, 3.2], vertical_alignment="center") if action_cols[0].button("🔎 적용(미리보기)", key=f"rule_expander_preview_{rule_id}_{idx}", use_container_width=True): category_rule_preview_dialog(rule, db_path) if action_cols[1].button("✅ 적용", key=f"rule_expander_apply_{rule_id}_{idx}", use_container_width=True): apply_fn = get_service_callable(categorization_gateway, "apply_category_rule") if not callable(apply_fn): st.error("현재 실행 중인 서비스에서 단일 규칙 적용 함수를 찾지 못했습니다. Streamlit 서버를 완전히 종료한 뒤 다시 시작해 주세요.") else: res = apply_fn(rule_id, limit=10000, db_path=db_path) clear_cache() if res.get("success"): st.success(f"미분류 매칭 {res.get('applied_count', 0):,}건에 규칙을 적용했습니다.") st.rerun() else: st.error(res.get("error", res)) if action_cols[2].button("✏️ 수정", key=f"rule_expander_edit_{rule_id}_{idx}", use_container_width=True): category_rule_edit_dialog(rule, db_path) with action_cols[3]: st.markdown('
삭제보다 비활성화를 권장합니다.
', unsafe_allow_html=True) if action_cols[4].button("🗑️ 삭제", key=f"rule_expander_delete_{rule_id}_{idx}"): delete_rule = getattr(categorization_gateway, "delete_category_rule", None) if delete_rule is None: st.error("현재 실행 중인 서비스에 delete_category_rule 함수가 없습니다. 앱을 재시작해 최신 코드를 다시 불러오세요.") else: res = delete_rule(rule_id, db_path=db_path) clear_cache() if res.get("success"): if st.session_state.get("rules_edit_rule_id") == rule_id: st.session_state.pop("rules_edit_rule_id", None) st.success("규칙을 삭제했습니다.") st.rerun() else: st.error(res.get("error", res)) def page_rules(db_path: str) -> None: action_cols = st.columns([5, 1], vertical_alignment="center") with action_cols[1]: if st.button("추가", key="rule_add_dialog_open", use_container_width=True, type="primary"): category_rule_create_dialog(db_path) rules = categorization_gateway.list_category_rules(active_only=False, db_path=db_path).get("rules", []) render_section_label("📚", "현재 분류 규칙") render_rule_list_items(rules, db_path) st.caption("규칙 추가는 제목 오른쪽 `추가` 버튼, 기존 규칙 수정은 각 항목의 `⋮ → 변경` 팝업을 사용하세요.") c1, c2, c3 = st.columns(3) if c1.button("기본 규칙 seed/update"): res = categorization_gateway.seed_default_category_rules(db_path=db_path) clear_cache(); st.success(res) if c2.button("규칙 preview"): st.session_state["rule_preview"] = categorization_gateway.preview_categorization(limit=5000, db_path=db_path) if c3.button("규칙 적용"): res = categorization_gateway.apply_categorization_rules(limit=5000, db_path=db_path) clear_cache(); st.success(f"{res.get('applied_count', 0)}건 적용") if "rule_preview" in st.session_state: prev = st.session_state["rule_preview"] st.info(f"스캔 {prev.get('scanned')}건 · 매칭 {prev.get('matched_count')}건 · 미분류 {prev.get('unmatched_count')}건") st.write("카테고리별 예상") st.dataframe(pd.DataFrame([{"category": k, **v} for k, v in prev.get("by_category", {}).items()]), width="stretch") with st.expander("매칭 샘플"): st.dataframe(pd.DataFrame(prev.get("matches", [])), width="stretch") with st.expander("미매칭 샘플"): st.dataframe(pd.DataFrame(prev.get("unmatched", [])), width="stretch")