154 lines
6.4 KiB
Python
154 lines
6.4 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
import io
|
||
from pathlib import Path
|
||
import tempfile
|
||
import unittest
|
||
|
||
from openpyxl import Workbook
|
||
|
||
from bank_importer.parser import (
|
||
UnknownTemplateError,
|
||
analyze_workbook,
|
||
detect_header,
|
||
parse_directory,
|
||
)
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SAMPLES = ROOT / "流水模板"
|
||
|
||
|
||
class StatementParserTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.batches = parse_directory(SAMPLES)
|
||
cls.by_file = {batch.source_file.name: batch for batch in cls.batches}
|
||
|
||
def test_all_current_bank_samples_are_recognized(self) -> None:
|
||
expected = {
|
||
"中信银行账户流水.xlsx": ("中信银行", 16, 4),
|
||
"中国农业银行账户流水.xls": ("中国农业银行", 3, 4),
|
||
"中国工商银行账户流水.xlsx": ("中国工商银行", 2, 5),
|
||
"中国建设银行账户流水.xls": ("中国建设银行", 1, 1),
|
||
"河南农商银行账户流水.xlsx": ("河南农商银行", 11, 1),
|
||
"郑州银行账户流水.xls": ("郑州银行", 7, 5),
|
||
}
|
||
self.assertEqual(set(expected), set(self.by_file))
|
||
for filename, (bank, header_row, transaction_count) in expected.items():
|
||
with self.subTest(filename=filename):
|
||
batch = self.by_file[filename]
|
||
self.assertEqual(bank, batch.bank_name)
|
||
self.assertEqual(header_row, batch.header_row)
|
||
self.assertEqual(transaction_count, len(batch.transactions))
|
||
|
||
def test_known_statement_periods_are_extracted(self) -> None:
|
||
expected = {
|
||
"中信银行账户流水.xlsx": (date(2026, 4, 26), date(2026, 5, 23)),
|
||
"中国农业银行账户流水.xls": (date(2025, 11, 26), date(2025, 12, 27)),
|
||
"河南农商银行账户流水.xlsx": (date(2026, 2, 24), date(2026, 3, 28)),
|
||
"郑州银行账户流水.xls": (date(2026, 2, 24), date(2026, 3, 27)),
|
||
}
|
||
for filename, period in expected.items():
|
||
with self.subTest(filename=filename):
|
||
batch = self.by_file[filename]
|
||
self.assertEqual(period, (batch.period_start, batch.period_end))
|
||
|
||
def test_normalized_transactions_have_one_cash_direction(self) -> None:
|
||
for batch in self.batches:
|
||
for transaction in batch.transactions:
|
||
with self.subTest(
|
||
filename=batch.source_file.name, row=transaction.source_row
|
||
):
|
||
self.assertNotEqual(transaction.income > 0, transaction.expense > 0)
|
||
|
||
def test_current_samples_reconcile_by_running_balance(self) -> None:
|
||
for batch in self.batches:
|
||
with self.subTest(filename=batch.source_file.name):
|
||
self.assertEqual((), batch.warnings)
|
||
|
||
def test_column_order_does_not_affect_header_detection(self) -> None:
|
||
header = (
|
||
"余额",
|
||
"对方单位名称",
|
||
"贷方发生额",
|
||
"交易时间",
|
||
"本方账号",
|
||
"借方发生额",
|
||
"凭证号",
|
||
)
|
||
template, row_index, columns = detect_header((header,))
|
||
self.assertEqual("icbc-history-detail-v1", template.template_id)
|
||
self.assertEqual(0, row_index)
|
||
self.assertEqual(3, columns["transaction_at"])
|
||
|
||
def test_unknown_header_is_rejected(self) -> None:
|
||
with self.assertRaises(UnknownTemplateError) as raised:
|
||
detect_header((("日期", "金额", "备注"),))
|
||
message = str(raised.exception)
|
||
self.assertIn("未识别到受支持的银行表头", message)
|
||
self.assertIn("第 1 行", message)
|
||
self.assertIn("日期、金额、备注", message)
|
||
|
||
|
||
class SheetResultTests(unittest.TestCase):
|
||
"""Per-worksheet outcomes: parsed / exception / ignored with evidence."""
|
||
|
||
def setUp(self) -> None:
|
||
self.temp_dir = tempfile.TemporaryDirectory()
|
||
self.addCleanup(self.temp_dir.cleanup)
|
||
|
||
def _write(self, sheets) -> Path:
|
||
workbook = Workbook()
|
||
workbook.remove(workbook.active)
|
||
for name, rows in sheets.items():
|
||
worksheet = workbook.create_sheet(name)
|
||
for row in rows:
|
||
worksheet.append(row)
|
||
path = Path(self.temp_dir.name) / "workbook.xlsx"
|
||
workbook.save(path)
|
||
return path
|
||
|
||
def test_every_sheet_has_an_independent_result(self) -> None:
|
||
path = self._write(
|
||
{
|
||
"正常": [
|
||
["客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）", "余额", "对方账号"],
|
||
["6228480000000000", "测试", "2026-01-05 10:00:00", "100.00", "", "99900.00", "1002003004"],
|
||
],
|
||
"未知": [["日期", "金额", "备注"], ["2026-01-01", "100", "x"]],
|
||
"空表": [],
|
||
}
|
||
)
|
||
results = analyze_workbook(path)
|
||
by_name = {result.sheet_name: result for result in results}
|
||
self.assertEqual({"正常", "未知", "空表"}, set(by_name))
|
||
self.assertEqual("parsed", by_name["正常"].outcome)
|
||
self.assertIsNotNone(by_name["正常"].batch)
|
||
self.assertEqual("exception", by_name["未知"].outcome)
|
||
self.assertIn("未识别到受支持的银行表头", by_name["未知"].message)
|
||
self.assertGreaterEqual(by_name["未知"].scanned_rows, 2)
|
||
self.assertTrue(any("日期、金额、备注" in row for row in by_name["未知"].candidate_headers))
|
||
self.assertEqual("ignored", by_name["空表"].outcome)
|
||
self.assertEqual(0, by_name["空表"].scanned_rows)
|
||
|
||
def test_single_cell_rows_are_candidate_evidence(self) -> None:
|
||
path = self._write({"候选": [["只有一个单元格"], ["又一格"]]})
|
||
(result,) = analyze_workbook(path)
|
||
self.assertEqual("exception", result.outcome)
|
||
self.assertEqual(2, result.scanned_rows)
|
||
self.assertTrue(result.candidate_headers)
|
||
self.assertIn("只有一个单元格", "；".join(result.candidate_headers))
|
||
|
||
def test_messages_never_contain_stored_filename(self) -> None:
|
||
path = self._write({"流水": [["日期", "金额", "备注"]]})
|
||
(result,) = analyze_workbook(path)
|
||
self.assertEqual("exception", result.outcome)
|
||
self.assertNotIn(path.name, result.message)
|
||
self.assertNotIn(str(path), result.message)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|