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()