AI가 먹을 수 있게 요리하라, 회계 데이터 자동수집기 (ETL)

import pandas as pd
from pathlib import Path

file = Path("transactions.xlsx")
ext = file.suffix.lower()

if ext == ".xlsx":
    df = pd.read_excel(file)
elif ext == ".csv":
    # 한글 깨짐 방지를 위해 utf-8-sig 인코딩 사용 권장
    df = pd.read_csv(file, encoding="utf-8-sig")

def normalize_transactions(df):
    # 1. 컬럼명 표준화
    df = df.rename(columns={
        "거래일": "transaction_date", "전표일자": "transaction_date",
        "거래처명": "customer", "적요": "description", "금액": "amount"
    })
    
    # 2. 날짜 및 금액 형변환 (결측치는 강제로 NaT, NaN 처리)
    df["transaction_date"] = pd.to_datetime(df["transaction_date"], errors="coerce")
    df["amount"] = df["amount"].astype(str).str.replace(",", "").str.replace("원", "")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    
    return df

df["error"] = ""
df.loc[df["customer"].isnull(), "error"] = "거래처 누락"
df.loc[df["amount"].isnull(), "error"] = "금액 오류"

# 에러가 있는 행만 골라서 엑셀로 떨궈줍니다. (회계사가 고쳐서 다시 올릴 수 있게)
errors_df = df[df["error"] != ""]
errors_df.to_excel("error_report.xlsx", index=False)

valid_df = df[df["error"] == ""]
valid_df["office_id"] = current_office_id
valid_df["customer_id"] = customer_id

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://user:pass@localhost/accounting")
valid_df.to_sql("transactions", engine, if_exists="append", index=False)