428 lines
14 KiB
Python
428 lines
14 KiB
Python
"""
|
||
銀行明細 CSV 取込・照会モジュール
|
||
"""
|
||
import csv
|
||
import hashlib
|
||
import io
|
||
import uuid
|
||
from datetime import datetime, date
|
||
from typing import Optional, List
|
||
|
||
from fastapi import APIRouter, HTTPException, UploadFile, File, Form
|
||
from pydantic import BaseModel
|
||
|
||
from app.core.database import get_connection
|
||
|
||
router = APIRouter(prefix="/bank-statements", tags=["銀行明細"])
|
||
|
||
# ── CSV 列名検出パターン ──────────────────────────────────────
|
||
|
||
_DATE_KEYS = ['日付', '取引日', '年月日', '取引年月日', 'date', '取引日付']
|
||
_DESC_KEYS = ['摘要', '内容', '取引内容', '支払先', '摘要/メモ', '入出金先内容',
|
||
'description', '取引先', 'お取引内容', '取引詳細']
|
||
_DEBIT_KEYS = ['お支払い金額', 'お支払金額', '支払額', '引出し', '出金', '引出額',
|
||
'借方', 'debit', '出金金額', '支払い金額', '引き出し金額']
|
||
_CREDIT_KEYS = ['お預かり金額', '預入れ', '預入', '入金', '預入金額',
|
||
'貸方', 'credit', '入金金額', 'お預り金額', '預け入れ金額']
|
||
_BALANCE_KEYS = ['残高', '差引残高', 'balance', '帳簿残高', '取引後残高']
|
||
_MEMO_KEYS = ['メモ', '備考', '通帳メモ', 'memo', '注記', '備考欄']
|
||
|
||
|
||
def _find_col(headers: List[str], patterns: List[str]) -> Optional[int]:
|
||
lowers = [h.strip().lower() for h in headers]
|
||
for pat in patterns:
|
||
p = pat.lower()
|
||
for i, h in enumerate(lowers):
|
||
if p in h:
|
||
return i
|
||
return None
|
||
|
||
|
||
def _parse_amount(s: str) -> int:
|
||
if not s:
|
||
return 0
|
||
cleaned = (s.strip()
|
||
.replace(',', '').replace('，', '')
|
||
.replace('¥', '').replace('￥', '')
|
||
.replace(' ', '').replace('\u3000', ''))
|
||
if cleaned in ('', '-', '―', '－', '▲', 'ー'):
|
||
return 0
|
||
try:
|
||
return int(float(cleaned))
|
||
except ValueError:
|
||
return 0
|
||
|
||
|
||
def _parse_date(s: str) -> Optional[date]:
|
||
s = (s.strip()
|
||
.replace('/', '-').replace('.', '-')
|
||
.replace('年', '-').replace('月', '-').replace('日', ''))
|
||
for fmt in ['%Y-%m-%d', '%y-%m-%d']:
|
||
try:
|
||
return datetime.strptime(s, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
s2 = s.replace('-', '')
|
||
if len(s2) == 8:
|
||
try:
|
||
return datetime.strptime(s2, '%Y%m%d').date()
|
||
except ValueError:
|
||
pass
|
||
if len(s2) == 6:
|
||
try:
|
||
return datetime.strptime('20' + s2, '%Y%m%d').date()
|
||
except ValueError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _detect_encoding(raw: bytes) -> str:
|
||
for enc in ['utf-8-sig', 'utf-8', 'shift_jis', 'cp932', 'euc-jp']:
|
||
try:
|
||
raw.decode(enc)
|
||
return enc
|
||
except (UnicodeDecodeError, LookupError):
|
||
continue
|
||
return 'utf-8'
|
||
|
||
|
||
def _make_hash(tx_date: str, description: str, debit: int,
|
||
credit: int, balance: Optional[int]) -> str:
|
||
s = f"{tx_date}|{description}|{debit}|{credit}|{balance}"
|
||
return hashlib.md5(s.encode()).hexdigest()
|
||
|
||
|
||
# ── 三菱UFJ MEISAI フォーマット対応 ──────────────────────────
|
||
# 行タイプ: "1"=口座ヘッダー, "2"=明細, "8"=フッター, "9"=合計
|
||
# 列: [タイプ, 日付, 摘要区分, 摘要詳細, 出金額, 入金額, 残高]
|
||
|
||
def _is_meisai_format(rows: List[List[str]]) -> bool:
|
||
"""三菱UFJ銀行 MEISAI フォーマット判定"""
|
||
type1 = any(r and r[0] == '1' for r in rows)
|
||
type2 = any(r and r[0] == '2' for r in rows)
|
||
return type1 and type2
|
||
|
||
|
||
def _parse_meisai_rows(rows: List[List[str]]) -> tuple:
|
||
"""MEISAI フォーマットをパース。(明細リスト, 銀行名ヒント) を返す"""
|
||
bank_hint = ''
|
||
parsed = []
|
||
|
||
for row in rows:
|
||
if not row:
|
||
continue
|
||
row_type = row[0]
|
||
|
||
if row_type == '1' and len(row) >= 3:
|
||
# ヘッダー行: [1, 支店コード, 支店名, ..., 科目, 口座番号, 口座名, ...]
|
||
branch = row[2].strip() if len(row) > 2 else ''
|
||
bank_hint = f"三菱UFJ銀行 {branch}" if branch else '三菱UFJ銀行'
|
||
|
||
elif row_type == '2' and len(row) >= 5:
|
||
# 明細行: [2, 日付, 摘要区分, 摘要詳細, 出金額, 入金額, 残高]
|
||
tx_date = _parse_date(row[1])
|
||
if tx_date is None:
|
||
continue
|
||
|
||
summary1 = row[2].strip() if len(row) > 2 else ''
|
||
summary2 = row[3].strip() if len(row) > 3 else ''
|
||
desc = f"{summary1} {summary2}".strip(' ') if summary2 else summary1
|
||
|
||
debit = _parse_amount(row[4]) if len(row) > 4 else 0
|
||
credit = _parse_amount(row[5]) if len(row) > 5 else 0
|
||
bal_str = row[6].strip() if len(row) > 6 else ''
|
||
balance = _parse_amount(bal_str) if bal_str else None
|
||
|
||
parsed.append({
|
||
'tx_date': tx_date.isoformat(),
|
||
'description': desc,
|
||
'debit': debit,
|
||
'credit': credit,
|
||
'balance': balance,
|
||
'memo': '',
|
||
'row_hash': _make_hash(tx_date.isoformat(), desc, debit, credit, balance),
|
||
})
|
||
|
||
return parsed, bank_hint
|
||
|
||
|
||
def _parse_csv_bytes(content: bytes) -> tuple:
|
||
"""CSVをパース。(明細リスト, 銀行名ヒント) を返す"""
|
||
enc = _detect_encoding(content)
|
||
text = content.decode(enc).replace('\x00', '').strip()
|
||
|
||
reader = csv.reader(io.StringIO(text))
|
||
rows = list(reader)
|
||
|
||
if not rows:
|
||
raise HTTPException(status_code=400, detail="CSVファイルが空です")
|
||
|
||
# MEISAI（三菱UFJ）フォーマット優先検出
|
||
if _is_meisai_format(rows):
|
||
return _parse_meisai_rows(rows)
|
||
|
||
# ── 汎用ヘッダー行検出 ──
|
||
header_row = 0
|
||
for i, row in enumerate(rows[:15]):
|
||
row_str = ' '.join(row).lower()
|
||
if any(k.lower() in row_str for k in ['日付', '取引日', 'date', '年月日']):
|
||
header_row = i
|
||
break
|
||
|
||
headers = rows[header_row]
|
||
date_col = _find_col(headers, _DATE_KEYS)
|
||
desc_col = _find_col(headers, _DESC_KEYS)
|
||
debit_col = _find_col(headers, _DEBIT_KEYS)
|
||
credit_col = _find_col(headers, _CREDIT_KEYS)
|
||
balance_col = _find_col(headers, _BALANCE_KEYS)
|
||
memo_col = _find_col(headers, _MEMO_KEYS)
|
||
|
||
if date_col is None:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="日付列が見つかりません。CSVのヘッダー行に「日付」「取引日」等の列名が必要です。"
|
||
)
|
||
|
||
parsed = []
|
||
for row in rows[header_row + 1:]:
|
||
if not row or all(c.strip() == '' for c in row):
|
||
continue
|
||
if len(row) <= date_col:
|
||
continue
|
||
raw_date = row[date_col].strip()
|
||
if not raw_date:
|
||
continue
|
||
tx_date = _parse_date(raw_date)
|
||
if tx_date is None:
|
||
continue
|
||
|
||
def _safe(col: Optional[int]) -> str:
|
||
return row[col].strip() if col is not None and len(row) > col else ''
|
||
|
||
desc = _safe(desc_col)
|
||
debit = _parse_amount(_safe(debit_col))
|
||
credit = _parse_amount(_safe(credit_col))
|
||
bal_str = _safe(balance_col)
|
||
balance = _parse_amount(bal_str) if bal_str else None
|
||
memo = _safe(memo_col)
|
||
|
||
parsed.append({
|
||
'tx_date': tx_date.isoformat(),
|
||
'description': desc,
|
||
'debit': debit,
|
||
'credit': credit,
|
||
'balance': balance,
|
||
'memo': memo,
|
||
'row_hash': _make_hash(tx_date.isoformat(), desc, debit, credit, balance),
|
||
})
|
||
|
||
return parsed, ''
|
||
|
||
|
||
# ── エンドポイント ─────────────────────────────────────────────
|
||
|
||
@router.post("/preview", summary="CSVプレビュー（重複チェック）")
|
||
async def preview_csv(
|
||
file: UploadFile = File(...),
|
||
bank_name: str = Form(""),
|
||
):
|
||
"""CSVを解析し、重複チェック結果付きでプレビューを返す（DBには保存しない）"""
|
||
content = await file.read()
|
||
if len(content) > 10_000_000:
|
||
raise HTTPException(status_code=400, detail="ファイルサイズが大きすぎます（上限10MB）")
|
||
|
||
rows, bank_hint = _parse_csv_bytes(content)
|
||
if not rows:
|
||
raise HTTPException(status_code=400, detail="有効なデータ行がありません")
|
||
|
||
# 銀行名未入力の場合はCSVから自動検出した値を使用
|
||
if not bank_name.strip() and bank_hint:
|
||
bank_name = bank_hint
|
||
|
||
# DB上の既存ハッシュを取得して重複チェック
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
cur.execute("SELECT row_hash FROM bank_statement_rows")
|
||
existing_hashes = {r['row_hash'] for r in cur.fetchall()}
|
||
|
||
result = []
|
||
for row in rows:
|
||
row['status'] = 'duplicate' if row['row_hash'] in existing_hashes else 'new'
|
||
result.append(row)
|
||
|
||
new_count = sum(1 for r in result if r['status'] == 'new')
|
||
dup_count = sum(1 for r in result if r['status'] == 'duplicate')
|
||
|
||
return {
|
||
'total': len(result),
|
||
'new': new_count,
|
||
'duplicates': dup_count,
|
||
'bank_name': bank_name,
|
||
'rows': result,
|
||
}
|
||
|
||
|
||
class ImportRow(BaseModel):
|
||
tx_date: str
|
||
description: str
|
||
debit: int
|
||
credit: int
|
||
balance: Optional[int] = None
|
||
memo: str = ''
|
||
row_hash: str = ''
|
||
|
||
|
||
class ImportRequest(BaseModel):
|
||
bank_name: str = ''
|
||
rows: List[ImportRow]
|
||
|
||
|
||
@router.post("/import", summary="CSV取込確定")
|
||
def import_rows(req: ImportRequest):
|
||
"""プレビュー確認後、新規行のみDBに取込む"""
|
||
if not req.rows:
|
||
raise HTTPException(status_code=400, detail="取込データがありません")
|
||
|
||
batch_id = str(uuid.uuid4())
|
||
inserted = 0
|
||
skipped = 0
|
||
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
for row in req.rows:
|
||
# row_hash が空の場合は再計算
|
||
rh = row.row_hash or _make_hash(
|
||
row.tx_date, row.description, row.debit, row.credit, row.balance
|
||
)
|
||
cur.execute("""
|
||
INSERT INTO bank_statement_rows
|
||
(tx_date, description, debit, credit, balance,
|
||
memo, bank_name, import_batch_id, row_hash)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
ON CONFLICT (row_hash) DO NOTHING
|
||
""", (
|
||
row.tx_date, row.description,
|
||
row.debit, row.credit, row.balance,
|
||
row.memo, req.bank_name, batch_id, rh,
|
||
))
|
||
if cur.rowcount > 0:
|
||
inserted += 1
|
||
else:
|
||
skipped += 1
|
||
conn.commit()
|
||
|
||
return {
|
||
'batch_id': batch_id,
|
||
'inserted': inserted,
|
||
'skipped': skipped,
|
||
}
|
||
|
||
|
||
@router.get("", summary="明細一覧取得")
|
||
def list_statements(
|
||
year: Optional[int] = None,
|
||
month: Optional[int] = None,
|
||
bank_name: Optional[str] = None,
|
||
limit: int = 10000,
|
||
offset: int = 0,
|
||
):
|
||
conditions: list = []
|
||
params: list = []
|
||
|
||
if year:
|
||
conditions.append("EXTRACT(YEAR FROM tx_date)::int = %s")
|
||
params.append(year)
|
||
if month:
|
||
conditions.append("EXTRACT(MONTH FROM tx_date)::int = %s")
|
||
params.append(month)
|
||
if bank_name:
|
||
conditions.append("bank_name = %s")
|
||
params.append(bank_name)
|
||
|
||
where_clause = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
# 明細行
|
||
cur.execute(f"""
|
||
SELECT id,
|
||
TO_CHAR(tx_date, 'YYYY-MM-DD') AS tx_date,
|
||
description,
|
||
debit::bigint,
|
||
credit::bigint,
|
||
balance::bigint,
|
||
memo,
|
||
bank_name,
|
||
import_batch_id
|
||
FROM bank_statement_rows
|
||
{where_clause}
|
||
ORDER BY tx_date, id
|
||
LIMIT %s OFFSET %s
|
||
""", params + [limit, offset])
|
||
rows = cur.fetchall()
|
||
|
||
# 月別集計
|
||
cur.execute(f"""
|
||
SELECT
|
||
TO_CHAR(tx_date, 'YYYY-MM') AS ym,
|
||
SUM(debit)::bigint AS total_debit,
|
||
SUM(credit)::bigint AS total_credit,
|
||
COUNT(*)::int AS count
|
||
FROM bank_statement_rows
|
||
{where_clause}
|
||
GROUP BY ym
|
||
ORDER BY ym
|
||
""", params)
|
||
monthly = cur.fetchall()
|
||
|
||
# 総件数
|
||
cur.execute(
|
||
f"SELECT COUNT(*)::int AS cnt FROM bank_statement_rows {where_clause}",
|
||
params
|
||
)
|
||
total = cur.fetchone()['cnt']
|
||
|
||
# 銀行名リスト
|
||
cur.execute("""
|
||
SELECT DISTINCT bank_name
|
||
FROM bank_statement_rows
|
||
WHERE bank_name != ''
|
||
ORDER BY bank_name
|
||
""")
|
||
banks = [r['bank_name'] for r in cur.fetchall()]
|
||
|
||
return {
|
||
'total': total,
|
||
'rows': rows,
|
||
'monthly': monthly,
|
||
'banks': banks,
|
||
}
|
||
|
||
|
||
@router.get("/batches", summary="取込バッチ一覧")
|
||
def list_batches():
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
cur.execute("""
|
||
SELECT
|
||
import_batch_id,
|
||
bank_name,
|
||
MIN(tx_date)::text AS date_from,
|
||
MAX(tx_date)::text AS date_to,
|
||
COUNT(*)::int AS count,
|
||
MIN(imported_at)::text AS imported_at
|
||
FROM bank_statement_rows
|
||
GROUP BY import_batch_id, bank_name
|
||
ORDER BY MIN(imported_at) DESC
|
||
LIMIT 100
|
||
""")
|
||
return cur.fetchall()
|
||
|
||
|
||
@router.delete("/batch/{batch_id}", summary="バッチ削除")
|
||
def delete_batch(batch_id: str):
|
||
with get_connection() as conn, conn.cursor() as cur:
|
||
cur.execute(
|
||
"DELETE FROM bank_statement_rows WHERE import_batch_id = %s",
|
||
(batch_id,)
|
||
)
|
||
deleted = cur.rowcount
|
||
conn.commit()
|
||
return {'deleted': deleted}
|