Initial commit: intercompany ledger app

This commit is contained in:
leefer
2026-08-06 23:46:04 +08:00
commit b3043f9430
60 changed files with 4280 additions and 0 deletions
+9
View File
@@ -0,0 +1,9 @@
from .models import NormalizedTransaction, StatementBatch
from .parser import parse_directory, parse_statement
__all__ = [
"NormalizedTransaction",
"StatementBatch",
"parse_directory",
"parse_statement",
]
+43
View File
@@ -0,0 +1,43 @@
from __future__ import annotations
import argparse
import json
from pathlib import Path
from .parser import StatementParseError, parse_directory, parse_statement
def main() -> int:
parser = argparse.ArgumentParser(description="Parse bank statements by header signature.")
parser.add_argument("path", type=Path, help="Statement file or directory")
args = parser.parse_args()
try:
batches = (
parse_directory(args.path)
if args.path.is_dir()
else parse_statement(args.path)
)
except (OSError, StatementParseError) as exc:
parser.error(str(exc))
summaries = [
{
"file": batch.source_file.name,
"sheet": batch.sheet_name,
"bank": batch.bank_name,
"template": batch.template_id,
"header_row": batch.header_row,
"period_start": batch.period_start.isoformat() if batch.period_start else None,
"period_end": batch.period_end.isoformat() if batch.period_end else None,
"transactions": len(batch.transactions),
"warnings": list(batch.warnings),
}
for batch in batches
]
print(json.dumps(summaries, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
+41
View File
@@ -0,0 +1,41 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
@dataclass(frozen=True)
class NormalizedTransaction:
source_file: Path
sheet_name: str
source_row: int
transaction_at: datetime
income: Decimal
expense: Decimal
balance: Decimal | None
own_account: str | None
own_name: str | None
counterparty_account: str | None
counterparty_name: str | None
counterparty_bank: str | None
summary: str | None
purpose: str | None
reference: str | None
currency: str | None
@dataclass(frozen=True)
class StatementBatch:
source_file: Path
sheet_name: str
bank_name: str
template_id: str
header_row: int
own_account: str | None
own_name: str | None
period_start: date | None
period_end: date | None
transactions: tuple[NormalizedTransaction, ...]
warnings: tuple[str, ...]
+399
View File
@@ -0,0 +1,399 @@
from __future__ import annotations
from datetime import date, datetime, time
from decimal import Decimal, InvalidOperation
from itertools import groupby
from pathlib import Path
import re
from typing import Any, Iterable
from .models import NormalizedTransaction, StatementBatch
from .reader import RawSheet, read_workbook
from .templates import BankTemplate, TEMPLATES, normalize_header
class StatementParseError(RuntimeError):
pass
class UnknownTemplateError(StatementParseError):
pass
class AmbiguousTemplateError(StatementParseError):
pass
def detect_header(
rows: tuple[tuple[Any, ...], ...], scan_limit: int = 50
) -> tuple[BankTemplate, int, dict[str, int]]:
candidates: list[tuple[int, BankTemplate, int, dict[str, int]]] = []
for row_index, row in enumerate(rows[:scan_limit]):
normalized = [normalize_header(value) for value in row]
for template in TEMPLATES:
column_map: dict[str, int] = {}
for field, aliases in template.columns.items():
for column_index, header in enumerate(normalized):
if header and header in aliases:
column_map[field] = column_index
break
if all(field in column_map for field in template.required):
candidates.append((len(column_map), template, row_index, column_map))
if not candidates:
inspected = min(len(rows), scan_limit)
candidate_rows = _header_candidate_summary(rows[:scan_limit])
detail = f";候选表头:{candidate_rows}" if candidate_rows else ""
raise UnknownTemplateError(
f"未识别到受支持的银行表头(已扫描前 {inspected} 行){detail}"
)
candidates.sort(key=lambda item: item[0], reverse=True)
best_score = candidates[0][0]
best = [candidate for candidate in candidates if candidate[0] == best_score]
identities = {(item[1].template_id, item[2]) for item in best}
if len(identities) > 1:
matches = ", ".join(f"{item[1].bank_name}@{item[2] + 1}" for item in best)
raise AmbiguousTemplateError(f"Ambiguous bank header signatures: {matches}")
_, template, row_index, column_map = best[0]
return template, row_index, column_map
def parse_statement(path: str | Path) -> tuple[StatementBatch, ...]:
source = Path(path)
batches: list[StatementBatch] = []
errors: list[str] = []
for sheet in read_workbook(source):
if not any(any(_text(value) for value in row) for row in sheet.rows):
continue
try:
batches.append(_parse_sheet(source, sheet))
except UnknownTemplateError as exc:
errors.append(f"{sheet.name}: {exc}")
if not batches:
detail = "; ".join(errors) or "Workbook contains no readable worksheets."
raise UnknownTemplateError(f"{source.name}: {detail}")
return tuple(batches)
def parse_directory(path: str | Path) -> tuple[StatementBatch, ...]:
root = Path(path)
files = sorted(
file
for file in root.rglob("*")
if file.is_file()
and file.suffix.lower() in {".xls", ".xlsx"}
and not file.name.startswith("~$")
and file.stat().st_size > 0
)
return tuple(batch for file in files for batch in parse_statement(file))
def _header_candidate_summary(
rows: tuple[tuple[Any, ...], ...], limit: int = 3
) -> str:
known_headers = {
alias
for template in TEMPLATES
for aliases in template.columns.values()
for alias in aliases
}
candidates: list[tuple[int, int, int, tuple[str, ...]]] = []
for row_index, row in enumerate(rows):
values = tuple(_text(value) for value in row if _text(value))
if len(values) < 2:
continue
matched = sum(normalize_header(value) in known_headers for value in values)
candidates.append((matched, len(values), row_index, values))
candidates.sort(key=lambda item: (item[0], item[1]), reverse=True)
summaries: list[str] = []
for _, _, row_index, values in candidates[:limit]:
preview = "".join(value[:24] for value in values[:8])
if len(values) > 8:
preview += "……"
summaries.append(f"{row_index + 1} 行「{preview}")
return "".join(summaries)
def _parse_sheet(source: Path, sheet: RawSheet) -> StatementBatch:
template, header_index, columns = detect_header(sheet.rows)
metadata = _extract_metadata(sheet.rows[:header_index])
transactions: list[NormalizedTransaction] = []
warnings: list[str] = []
for row_index, row in enumerate(sheet.rows[header_index + 1 :], header_index + 2):
transaction_at = _transaction_datetime(row, columns)
if transaction_at is None:
continue
income = _decimal(_value(row, columns.get("income")))
expense = _decimal(_value(row, columns.get("expense")))
if income == 0 and expense == 0:
continue
if income > 0 and expense > 0:
warnings.append(f"Row {row_index} has both income and expense.")
transactions.append(
NormalizedTransaction(
source_file=source,
sheet_name=sheet.name,
source_row=row_index,
transaction_at=transaction_at,
income=income,
expense=expense,
balance=_optional_decimal(_value(row, columns.get("balance"))),
own_account=_optional_text(_value(row, columns.get("own_account"))),
own_name=_optional_text(_value(row, columns.get("own_name"))),
counterparty_account=_optional_text(
_value(row, columns.get("counterparty_account"))
),
counterparty_name=_optional_text(
_value(row, columns.get("counterparty_name"))
),
counterparty_bank=_optional_text(
_value(row, columns.get("counterparty_bank"))
),
summary=_optional_text(_value(row, columns.get("summary"))),
purpose=_optional_text(_value(row, columns.get("purpose"))),
reference=_optional_text(_value(row, columns.get("reference"))),
currency=_optional_text(_value(row, columns.get("currency"))),
)
)
if not transactions:
raise StatementParseError(
f"{source.name}/{sheet.name}: header found but no transaction rows parsed."
)
own_account = metadata.get("own_account") or _first_value(
transaction.own_account for transaction in transactions
)
own_name = metadata.get("own_name") or _first_value(
transaction.own_name for transaction in transactions
)
warnings.extend(_balance_warnings(transactions))
return StatementBatch(
source_file=source,
sheet_name=sheet.name,
bank_name=template.bank_name,
template_id=template.template_id,
header_row=header_index + 1,
own_account=own_account if isinstance(own_account, str) else None,
own_name=own_name if isinstance(own_name, str) else None,
period_start=metadata.get("period_start")
if isinstance(metadata.get("period_start"), date)
else None,
period_end=metadata.get("period_end")
if isinstance(metadata.get("period_end"), date)
else None,
transactions=tuple(transactions),
warnings=tuple(warnings),
)
def _extract_metadata(rows: tuple[tuple[Any, ...], ...]) -> dict[str, str | date]:
result: dict[str, str | date] = {}
labels = {
"账号": "own_account",
"账/卡号": "own_account",
"户名": "own_name",
"账户名称": "own_name",
"起始日期": "period_start",
"开始日期": "period_start",
"截止日期": "period_end",
"结束日期": "period_end",
}
for row in rows:
for index, value in enumerate(row):
text = _text(value)
if not text:
continue
key_text, inline_value = _split_label(text)
key = labels.get(key_text)
if key:
candidate: Any = inline_value or _next_nonempty(row, index + 1)
if key.startswith("period_"):
parsed = _date_value(candidate)
if parsed:
result[key] = parsed
elif candidate is not None:
result[key] = _text(candidate)
if "起止日期" in text:
dates = _dates_in_text(text)
if len(dates) >= 2:
result["period_start"], result["period_end"] = dates[:2]
return result
def _transaction_datetime(row: tuple[Any, ...], columns: dict[str, int]) -> datetime | None:
if "transaction_at" in columns:
return _datetime_value(_value(row, columns["transaction_at"]))
transaction_date = _date_value(_value(row, columns.get("transaction_date")))
if transaction_date is None:
return None
transaction_time = _time_value(_value(row, columns.get("transaction_time")))
return datetime.combine(transaction_date, transaction_time or time.min)
def _datetime_value(value: Any) -> datetime | None:
if isinstance(value, datetime):
return value
if isinstance(value, date):
return datetime.combine(value, time.min)
text = _text(value)
if not text:
return None
normalized = text.replace("", "-").replace("", "-").replace("", " ")
normalized = " ".join(normalized.split())
for pattern in (
"%Y-%m-%d %H:%M:%S",
"%Y/%m/%d %H:%M:%S",
"%Y%m%d %H:%M:%S",
"%Y-%m-%d",
"%Y/%m/%d",
"%Y%m%d",
):
try:
return datetime.strptime(normalized, pattern)
except ValueError:
continue
return None
def _date_value(value: Any) -> date | None:
parsed = _datetime_value(value)
return parsed.date() if parsed else None
def _time_value(value: Any) -> time | None:
if isinstance(value, datetime):
return value.time()
if isinstance(value, time):
return value
text = _text(value)
if not text:
return None
for pattern in ("%H:%M:%S", "%H:%M"):
try:
return datetime.strptime(text, pattern).time()
except ValueError:
continue
return None
def _dates_in_text(value: str) -> list[date]:
date_strings = re.findall(
r"20\d{2}(?:年|-|/)?\d{1,2}(?:月|-|/)?\d{1,2}日?", value
)
return [parsed for item in date_strings if (parsed := _date_value(item))]
def _decimal(value: Any) -> Decimal:
if value is None:
return Decimal("0")
if isinstance(value, Decimal):
return value
if isinstance(value, (int, float)):
return Decimal(str(value))
text = _text(value)
if not text:
return Decimal("0")
text = text.replace(",", "").replace("¥", "").replace("", "")
text = text.removesuffix("")
negative = text.startswith("(") and text.endswith(")")
if negative:
text = text[1:-1]
try:
amount = Decimal(text)
except InvalidOperation as exc:
raise StatementParseError(f"Invalid amount: {value!r}") from exc
return -amount if negative else amount
def _optional_decimal(value: Any) -> Decimal | None:
if value is None or _text(value) == "":
return None
return _decimal(value)
def _balance_warnings(
transactions: list[NormalizedTransaction],
) -> list[str]:
warnings: list[str] = []
if len(transactions) < 2:
return warnings
descending_source = transactions[0].transaction_at > transactions[-1].transaction_at
ordered = sorted(transactions, key=lambda item: item.transaction_at)
groups = [
list(items)
for _, items in groupby(ordered, key=lambda item: item.transaction_at)
]
def source_endpoint(group: list[NormalizedTransaction]) -> NormalizedTransaction:
key = lambda item: item.source_row
return min(group, key=key) if descending_source else max(group, key=key)
previous_balance = source_endpoint(groups[0]).balance
for group in groups[1:]:
if previous_balance is None:
previous_balance = source_endpoint(group).balance
continue
expected = previous_balance + sum(
(item.income - item.expense for item in group), Decimal("0")
)
matching_balance = next(
(
item.balance
for item in group
if item.balance is not None
and abs(expected - item.balance) <= Decimal("0.01")
),
None,
)
if matching_balance is not None:
previous_balance = matching_balance
continue
rows = ",".join(str(item.source_row) for item in group)
warnings.append(f"Rows {rows} do not reconcile by balance.")
previous_balance = source_endpoint(group).balance
return warnings
def _value(row: tuple[Any, ...], column: int | None) -> Any:
if column is None or column >= len(row):
return None
return row[column]
def _text(value: Any) -> str:
return "" if value is None else str(value).strip()
def _optional_text(value: Any) -> str | None:
text = _text(value)
return text or None
def _split_label(value: str) -> tuple[str, str]:
parts = re.split(r"[:]", value, maxsplit=1)
key = parts[0].strip()
return key, parts[1].strip() if len(parts) == 2 else ""
def _next_nonempty(row: tuple[Any, ...], start: int) -> Any:
for value in row[start:]:
if _text(value):
return value
return None
def _first_value(values: Iterable[str | None]) -> str | None:
return next((value for value in values if value), None)
+75
View File
@@ -0,0 +1,75 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
class UnsupportedWorkbookError(RuntimeError):
pass
@dataclass(frozen=True)
class RawSheet:
name: str
rows: tuple[tuple[Any, ...], ...]
def read_workbook(path: Path) -> tuple[RawSheet, ...]:
suffix = path.suffix.lower()
if suffix == ".xlsx":
return _read_xlsx(path)
if suffix == ".xls":
return _read_xls(path)
raise UnsupportedWorkbookError(f"Unsupported workbook type: {suffix}")
def _read_xlsx(path: Path) -> tuple[RawSheet, ...]:
try:
from openpyxl import load_workbook
except ImportError as exc:
raise UnsupportedWorkbookError(
"Reading .xlsx files requires openpyxl. Install requirements.txt."
) from exc
workbook = load_workbook(path, read_only=True, data_only=True)
try:
return tuple(
RawSheet(
name=worksheet.title,
rows=tuple(tuple(row) for row in worksheet.iter_rows(values_only=True)),
)
for worksheet in workbook.worksheets
)
finally:
workbook.close()
def _read_xls(path: Path) -> tuple[RawSheet, ...]:
try:
import xlrd
except ImportError as exc:
raise UnsupportedWorkbookError(
"Reading .xls files requires xlrd. Install requirements.txt."
) from exc
workbook = xlrd.open_workbook(path, on_demand=True)
sheets: list[RawSheet] = []
try:
for sheet_name in workbook.sheet_names():
worksheet = workbook.sheet_by_name(sheet_name)
rows: list[tuple[Any, ...]] = []
for row_index in range(worksheet.nrows):
values: list[Any] = []
for column_index in range(worksheet.ncols):
cell = worksheet.cell(row_index, column_index)
value = cell.value
if cell.ctype == xlrd.XL_CELL_DATE:
value = xlrd.xldate_as_datetime(value, workbook.datemode)
values.append(value)
rows.append(tuple(values))
sheets.append(RawSheet(name=sheet_name, rows=tuple(rows)))
finally:
workbook.release_resources()
return tuple(sheets)
+178
View File
@@ -0,0 +1,178 @@
from __future__ import annotations
from dataclasses import dataclass
import unicodedata
@dataclass(frozen=True)
class BankTemplate:
template_id: str
bank_name: str
columns: dict[str, tuple[str, ...]]
required: tuple[str, ...]
def normalize_header(value: object) -> str:
if value is None:
return ""
text = unicodedata.normalize("NFKC", str(value)).strip().casefold()
return "".join(text.split())
def _aliases(*values: str) -> tuple[str, ...]:
return tuple(normalize_header(value) for value in values)
TEMPLATES: tuple[BankTemplate, ...] = (
BankTemplate(
template_id="citic-account-detail-v1",
bank_name="中信银行",
columns={
"transaction_date": _aliases("交易日期"),
"transaction_time": _aliases("交易时间"),
"counterparty_account": _aliases("对方账号"),
"counterparty_name": _aliases("对方账户名称"),
"counterparty_bank": _aliases("对方账号开户网点名称"),
"expense": _aliases("借方发生额"),
"income": _aliases("贷方发生额"),
"balance": _aliases("账户余额"),
"summary": _aliases("摘要"),
"purpose": _aliases("附言"),
"currency": _aliases("币种"),
"own_account": _aliases("交易账号"),
"reference": _aliases("柜员交易号", "发起方流水号"),
},
required=(
"transaction_date",
"transaction_time",
"counterparty_account",
"expense",
"income",
"balance",
),
),
BankTemplate(
template_id="abc-account-detail-v1",
bank_name="中国农业银行",
columns={
"transaction_at": _aliases("交易时间"),
"income": _aliases("收入金额"),
"expense": _aliases("支出金额"),
"balance": _aliases("账户余额"),
"counterparty_account": _aliases("对方账号"),
"counterparty_name": _aliases("对方户名"),
"counterparty_bank": _aliases("对方开户行"),
"summary": _aliases("摘要"),
},
required=(
"transaction_at",
"income",
"expense",
"balance",
"counterparty_account",
"counterparty_name",
"summary",
),
),
BankTemplate(
template_id="icbc-history-detail-v1",
bank_name="中国工商银行",
columns={
"reference": _aliases("凭证号"),
"own_account": _aliases("本方账号"),
"counterparty_account": _aliases("对方账号"),
"transaction_at": _aliases("交易时间"),
"expense": _aliases("借方发生额"),
"income": _aliases("贷方发生额"),
"summary": _aliases("摘要"),
"purpose": _aliases("用途"),
"counterparty_name": _aliases("对方单位名称"),
"balance": _aliases("余额"),
},
required=(
"reference",
"own_account",
"transaction_at",
"expense",
"income",
"counterparty_name",
"balance",
),
),
BankTemplate(
template_id="ccb-account-detail-v1",
bank_name="中国建设银行",
columns={
"own_account": _aliases("客户账号"),
"own_name": _aliases("账户名称"),
"transaction_at": _aliases("交易时间"),
"expense": _aliases("借方发生额(支取)"),
"income": _aliases("贷方发生额(收入)"),
"balance": _aliases("余额"),
"currency": _aliases("币种"),
"counterparty_name": _aliases("对方户名"),
"counterparty_account": _aliases("对方账号"),
"counterparty_bank": _aliases("对方开户机构"),
"summary": _aliases("摘要"),
"purpose": _aliases("备注"),
},
required=(
"own_account",
"transaction_at",
"expense",
"income",
"balance",
"counterparty_account",
),
),
BankTemplate(
template_id="henan-rural-history-v1",
bank_name="河南农商银行",
columns={
"transaction_at": _aliases("交易时间"),
"reference": _aliases("交易流水号"),
"income": _aliases("收入"),
"expense": _aliases("支出"),
"balance": _aliases("账户余额"),
"summary": _aliases("交易类型"),
"counterparty_account": _aliases("对方帐号"),
"counterparty_name": _aliases("对方户名"),
"counterparty_bank": _aliases("交易网点"),
"purpose": _aliases("备注"),
},
required=(
"transaction_at",
"reference",
"income",
"expense",
"balance",
"counterparty_account",
"counterparty_name",
),
),
BankTemplate(
template_id="bank-of-zhengzhou-detail-v1",
bank_name="郑州银行",
columns={
"transaction_at": _aliases("交易时间"),
"income": _aliases("收入金额(元)"),
"expense": _aliases("支出金额(元)"),
"balance": _aliases("账户余额(元)"),
"counterparty_name": _aliases("对方户名"),
"counterparty_account": _aliases("对方账号"),
"counterparty_bank": _aliases("对方账户开户行"),
"purpose": _aliases("用途"),
"summary": _aliases("摘要"),
"currency": _aliases("币种"),
"reference": _aliases("交易流水号"),
},
required=(
"transaction_at",
"income",
"expense",
"balance",
"counterparty_account",
"reference",
),
),
)