96 lines
5.1 KiB
Python
96 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
# ruff: noqa: F405
|
|
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import streamlit as st
|
|
|
|
from jarca_ledger_mcp.adapters.streamlit import imports_gateway
|
|
from jarca_ledger_mcp.adapters.streamlit import overview_gateway
|
|
from dashboard.common import * # noqa: F403, F405
|
|
|
|
def page_imports(db_path: str) -> None:
|
|
render_section_label("📈", "토스증권 거래내역 PDF 가져오기")
|
|
st.caption("토스증권 거래내역서 PDF를 업로드하면 투자계좌 거래로 가져옵니다. 매수/매도/환전/입출금은 `이체/투자이동`, 배당/이자는 실제 수입으로 분류합니다.")
|
|
uploaded = st.file_uploader("토스증권 PDF", type=["pdf"], key="toss_pdf")
|
|
if uploaded is not None:
|
|
upload_dir = Path.home() / ".local" / "share" / "jarca" / "uploads"
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
upload_path = upload_dir / uploaded.name
|
|
upload_path.write_bytes(uploaded.getbuffer())
|
|
c1, c2 = st.columns(2)
|
|
if c1.button("토스증권 Preview"):
|
|
res = imports_gateway.preview_toss_securities_pdf_import(str(upload_path), db_path=db_path)
|
|
st.session_state["toss_preview"] = res
|
|
if c2.button("토스증권 Import", type="primary"):
|
|
res = imports_gateway.import_toss_securities_pdf(str(upload_path), db_path=db_path)
|
|
clear_cache()
|
|
st.session_state["toss_import_result"] = res
|
|
st.rerun()
|
|
if "toss_preview" in st.session_state:
|
|
res = st.session_state["toss_preview"]
|
|
if res.get("success"):
|
|
st.success(f"Preview: {res.get('row_count', 0)}건, 중복 {res.get('duplicate_count', 0)}건")
|
|
st.json({"metadata": res.get("metadata"), "by_action": res.get("by_action"), "validation": res.get("validation")})
|
|
st.dataframe(pd.DataFrame(res.get("preview", [])), width="stretch", height=300)
|
|
else:
|
|
st.error(res)
|
|
if "toss_import_result" in st.session_state:
|
|
res = st.session_state["toss_import_result"]
|
|
if res.get("success"):
|
|
st.success(f"Import 완료: {res.get('inserted_count', 0)}건 추가, 중복 skip {res.get('skipped_duplicate_count', 0)}건")
|
|
st.json(res)
|
|
else:
|
|
st.error(res)
|
|
|
|
st.divider()
|
|
render_section_label("🏦", "토스뱅크 거래내역 PDF 가져오기")
|
|
st.caption("토스뱅크 계좌 거래내역 PDF를 업로드하면 입출금/이체/카드결제를 가져옵니다. 타인 이체/출금은 `이체/확인필요`로 분류됩니다.")
|
|
toss_bank_uploaded = st.file_uploader("토스뱅크 PDF", type=["pdf"], key="tossbank_pdf")
|
|
if toss_bank_uploaded is not None:
|
|
upload_dir = Path.home() / ".local" / "share" / "jarca" / "uploads"
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
upload_path = upload_dir / toss_bank_uploaded.name
|
|
upload_path.write_bytes(toss_bank_uploaded.getbuffer())
|
|
c1, c2 = st.columns(2)
|
|
if c1.button("토스뱅크 Preview"):
|
|
res = imports_gateway.preview_toss_bank_pdf_import(str(upload_path), db_path=db_path)
|
|
st.session_state["toss_bank_preview"] = res
|
|
if c2.button("토스뱅크 Import", type="primary"):
|
|
res = imports_gateway.import_toss_bank_pdf(str(upload_path), db_path=db_path)
|
|
clear_cache()
|
|
st.session_state["toss_bank_import_result"] = res
|
|
st.rerun()
|
|
if "toss_bank_preview" in st.session_state:
|
|
res = st.session_state["toss_bank_preview"]
|
|
if res.get("success"):
|
|
st.success(f"Preview: {res.get('row_count', 0)}건, 중복 {res.get('duplicate_count', 0)}건")
|
|
st.json({"metadata": res.get("metadata"), "by_action": res.get("by_action"), "validation": res.get("validation")})
|
|
st.dataframe(pd.DataFrame(res.get("preview", [])), width="stretch", height=300)
|
|
else:
|
|
st.error(res)
|
|
if "toss_bank_import_result" in st.session_state:
|
|
res = st.session_state["toss_bank_import_result"]
|
|
if res.get("success"):
|
|
st.success(f"Import 완료: {res.get('inserted_count', 0)}건 추가, 중복 skip {res.get('skipped_duplicate_count', 0)}건")
|
|
st.json(res)
|
|
else:
|
|
st.error(res)
|
|
|
|
st.divider()
|
|
render_section_label("🧾", "Import 이력")
|
|
batches = overview_gateway.list_ledger_import_batches(limit=100, db_path=db_path).get("import_batches", [])
|
|
st.dataframe(pd.DataFrame(batches), width="stretch", height=350)
|
|
if batches:
|
|
with st.form("rollback"):
|
|
batch_id = st.selectbox("Rollback batch", [b["id"] for b in batches])
|
|
hard = st.checkbox("Hard delete", value=False)
|
|
submitted = st.form_submit_button("Rollback 실행", type="secondary")
|
|
if submitted:
|
|
res = imports_gateway.rollback_ledger_import_batch(batch_id, hard_delete=hard, db_path=db_path)
|
|
clear_cache()
|
|
st.warning(res)
|
|
st.rerun()
|