from __future__ import annotations # ruff: noqa: F405 import streamlit as st from jarca_ledger_mcp.adapters.streamlit import categorization_gateway from dashboard.components import JarcaList from dashboard.common import * # noqa: F403, F405 def render_uncategorized_action_table(rows: list[dict[str, Any]], db_path: str) -> None: header_cols = st.columns([1.45, 0.6, 0.8, 1.35, 1.3, 0.7, 1.0, 1.0]) headers = ["일시", "구분", "금액", "사용처", "요약", "", "", ""] for col, label in zip(header_cols, headers): col.markdown(f"**{label}**") for idx, row in enumerate(rows): cols = st.columns([1.45, 0.6, 0.8, 1.35, 1.3, 0.7, 1.0, 1.0]) cols[0].write(row.get("transaction_datetime") or row.get("transaction_date") or "") cols[1].write(row.get("type") or "") cols[2].write(format_krw(row.get("amount"))) cols[3].write(row.get("merchant") or "") cols[4].write(row.get("raw_summary") or "") if cols[5].button("분류", key=f"uncat_dialog_{row['id']}_{idx}", use_container_width=True, type="tertiary"): transaction_edit_dialog(row, f"uncat_dialog_{row['id']}", db_path) if cols[6].button("연결거래찾기", key=f"uncat_link_{row['id']}_{idx}", use_container_width=True, type="tertiary"): linked_transfer_candidates_dialog(row, db_path) if cols[7].button("원거래찾기", key=f"uncat_cancel_{row['id']}_{idx}", use_container_width=True, type="tertiary"): cancel_original_candidates_dialog(row, db_path) def recommendation_text(rec: dict[str, Any] | None) -> str: rec = rec or {} if not rec.get("category"): return "추천 없음" pct = int(round(float(rec.get("confidence") or 0) * 100)) return f"{rec.get('category')} / {rec.get('subcategory') or '-'} · {pct}%" def save_merchant_rule_from_recommendation(merchant: str, tx_type: str, rec: dict[str, Any], db_path: str) -> dict[str, Any]: if not merchant: return {"success": False, "error": "merchant가 비어 있어 규칙을 만들 수 없습니다."} if not rec.get("category"): return {"success": False, "error": "추천 분류가 없어 규칙을 만들 수 없습니다."} return categorization_gateway.create_category_rule( match_field="merchant", match_type="contains", pattern=merchant, transaction_type=tx_type or "", category=rec.get("category", ""), subcategory=rec.get("subcategory", ""), priority=70, confidence=float(rec.get("confidence") or 0.8), memo="dashboard: uncategorized workbench rule", db_path=db_path, ) def render_uncategorized_groups_header(groups: list[dict[str, Any]]) -> None: st.markdown(f"{len(groups)}개 사용처 그룹") def render_uncategorized_group(group: dict[str, Any], db_path: str) -> None: rec = group.get("recommendation") or {} cols = st.columns([2.0, 0.7, 1.0, 1.55, 1.0], vertical_alignment="center") merchant = group.get("merchant") or group.get("raw_summary") or "-" group_key = group.get("sample_transaction_id") or merchant cols[0].markdown(f"**{merchant}** \n{group.get('raw_summary') or '-'} · {group.get('type') or '-'}", unsafe_allow_html=True) cols[1].write(f"{int(group.get('count') or 0):,}건") cols[2].write(format_krw(group.get("total_amount"))) cols[3].write(recommendation_text(rec)) apply_disabled = not bool(rec.get("category")) action = cols[4].menu_button( "작업", options=["그룹 적용", "규칙 만들기"], key=f"uncat_group_action_{group_key}", disabled=apply_disabled, width="stretch", ) if action == "그룹 적용": res = categorization_gateway.apply_uncategorized_group_recommendation( group.get("merchant", ""), group.get("raw_summary", ""), group.get("type", ""), db_path=db_path ) clear_cache() if res.get("success"): st.success(f"그룹 {res.get('applied_count', 0):,}건을 {recommendation_text(res.get('recommendation'))}로 분류했습니다.") st.rerun() else: st.error(res.get("error", res)) elif action == "규칙 만들기": res = save_merchant_rule_from_recommendation(group.get("merchant", ""), group.get("type", ""), rec, db_path) clear_cache() if res.get("success"): st.success("사용처 규칙을 저장했습니다.") st.rerun() else: st.warning(res.get("error", res)) st.caption(f"최근: {group.get('latest_date') or '-'} · 최대금액: {format_krw(group.get('max_amount'))} · 근거: {rec.get('reason') or '추천 없음'}") def render_uncategorized_groups(groups: list[dict[str, Any]], db_path: str, repeated_only: bool = False) -> None: visible = [g for g in groups if not repeated_only or int(g.get("count") or 0) >= 2][:10] JarcaList( render_header=render_uncategorized_groups_header, render_item=lambda group: render_uncategorized_group(group, db_path), items=visible, empty_message="반복 후보가 없습니다.", ).render() def render_uncategorized_work_items(rows: list[dict[str, Any]], db_path: str, recommended_only: bool = False) -> None: visible = [r for r in rows if not recommended_only or r.get("recommended_category")] if not visible: st.info("조건에 맞는 미분류 거래가 없습니다.") return for idx, row in enumerate(visible): rec = row.get("recommendation") or {} date_text = row.get("transaction_datetime") or row.get("transaction_date") or "" title = f"{date_text} · {format_krw(row.get('amount'))} · {row.get('merchant') or row.get('counterparty') or '-'} · 추천: {recommendation_text(rec)}" with st.expander(title, expanded=False): c1, c2, c3, c4 = st.columns(4) c1.metric("거래유형", row.get("type") or "-") c2.metric("금액", format_krw(row.get("amount"))) c3.metric("추천", recommendation_text(rec)) c4.metric("추천 근거", rec.get("source") or "none") st.caption(f"사용처: {row.get('merchant') or '-'} · 요약: {row.get('raw_summary') or '-'} · 상대: {row.get('counterparty') or '-'} · source: {row.get('source') or '-'} · ID: `{row.get('id')}`") if rec.get("reason"): st.info(rec.get("reason")) action_cols = st.columns([1.0, 1.0, 1.0, 1.1, 1.1, 2.5], vertical_alignment="center") apply_disabled = not bool(rec.get("category")) if action_cols[0].button("추천 적용", key=f"uncat_apply_rec_{row['id']}_{idx}", disabled=apply_disabled, use_container_width=True, type="primary"): res = categorization_gateway.apply_uncategorized_recommendation(row["id"], db_path=db_path) clear_cache() if res.get("success"): st.success(f"{recommendation_text(res.get('recommendation'))}로 분류했습니다.") st.rerun() else: st.error(res.get("error", res)) if action_cols[1].button("직접 분류", key=f"uncat_direct_{row['id']}_{idx}", use_container_width=True): transaction_edit_dialog(row, f"uncat_direct_{row['id']}", db_path) if action_cols[2].button("규칙 만들기", key=f"uncat_rule_{row['id']}_{idx}", disabled=apply_disabled, use_container_width=True): res = save_merchant_rule_from_recommendation(row.get("merchant", ""), row.get("type", ""), rec, db_path) clear_cache() if res.get("success"): st.success("사용처 규칙을 저장했습니다.") st.rerun() else: st.warning(res.get("error", res)) if action_cols[3].button("연결거래찾기", key=f"uncat_link_{row['id']}_{idx}", use_container_width=True): linked_transfer_candidates_dialog(row, db_path) if action_cols[4].button("원거래찾기", key=f"uncat_cancel_{row['id']}_{idx}", use_container_width=True): cancel_original_candidates_dialog(row, db_path) with st.expander(f"Review decision · 현재 {row.get('review_status', 'pending')}", expanded=False): render_review_decision_actions(categorization_gateway, "uncategorized_transaction", row["id"], f"uncat_review_decision_{row['id']}_{idx}", db_path, target_payload=row) def page_uncategorized(db_path: str) -> None: filter_cols = st.columns([1.0, 1.2, 1.15, 1.15, 1.25]) min_amount = filter_cols[0].number_input("최소 금액", min_value=0, value=0, step=10000) limit = filter_cols[1].slider("표시 건수", 10, 500, 100) recommended_only = filter_cols[2].checkbox("추천 있는 항목만", value=False) repeated_only = filter_cols[3].checkbox("반복 후보만", value=False) sort_label = filter_cols[4].selectbox("작업목록 정렬", ["금액순", "날짜순(최신순)"], index=0) sort_by = "date_desc" if sort_label == "날짜순(최신순)" else "amount_desc" workbench_fn = get_service_callable(categorization_gateway, "get_uncategorized_workbench", required_params=["sort_by"]) if not callable(workbench_fn): st.error("현재 실행 중인 서비스에서 미분류 작업대 함수를 찾지 못했습니다. Streamlit 서버를 완전히 재시작해 주세요.") return result = workbench_fn(limit=limit, min_amount=int(min_amount), sort_by=sort_by, db_path=db_path) if not result.get("success"): st.error(result.get("error", result)) return summary = result.get("summary", {}) metric_cols = st.columns(5) metric_cols[0].metric("미분류", f"{summary.get('uncategorized_count', 0):,}건") metric_cols[1].metric("미분류 총액", format_krw(summary.get("uncategorized_total_amount"))) metric_cols[2].metric("사용처 그룹", f"{summary.get('merchant_group_count', 0):,}개") metric_cols[3].metric("추천 가능", f"{summary.get('recommendable_count', 0):,}건") metric_cols[4].metric("반복 후보", f"{summary.get('repeated_group_count', 0):,}개") rows = result.get("transactions", []) groups = result.get("groups", []) if not rows: st.success("미분류 거래가 없습니다.") return render_section_label("🔁", "반복 사용처 후보") st.caption("같은 사용처/요약/거래유형으로 묶인 미분류 거래입니다. 추천이 있는 그룹은 한 번에 적용하거나 규칙으로 저장할 수 있습니다.") render_uncategorized_groups(groups, db_path, repeated_only=repeated_only) render_section_label("🛠️", "미분류 작업 목록") st.caption("거래를 펼치면 추천 근거와 작업 버튼이 보입니다. 추천 적용은 카테고리가 비어 있는 현재 거래에만 적용됩니다.") render_uncategorized_work_items(rows, db_path, recommended_only=recommended_only)