From 486842963e1612fa8541cb4bfb78db7b06a9e421 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=85=BE=E8=AE=AFWorkBuddy?= Date: Mon, 17 Aug 2026 01:00:19 +0800 Subject: [PATCH] =?UTF-8?q?B-42:=20=E5=AF=BC=E5=85=A5=E5=8A=9F=E8=83=BD?= =?UTF-8?q?=E5=8A=A0=E5=9B=BA=E2=80=94=E2=80=94=E9=80=90=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E8=A1=A8=E7=A1=AE=E8=AE=A4=E3=80=81=E6=B5=81=E5=BC=8F=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0=E4=B8=8E=E8=AF=8A=E6=96=AD=E8=AF=81=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 6 + server.py | 316 ++++++++++++------ src/bank_importer/db.py | 51 ++- src/bank_importer/importing.py | 370 ++++++++++++++++++++- src/bank_importer/models.py | 20 ++ src/bank_importer/multipart.py | 275 +++++++++++++++ src/bank_importer/parser.py | 100 ++++-- src/bank_importer/reader.py | 72 +++- tests/test_import_api.py | 587 +++++++++++++++++++++++++++++++++ tests/test_parser.py | 68 +++- tests/test_persistence.py | 9 +- tests/test_server_auth.py | 27 +- web/app.js | 284 ++++++++++++++-- web/company.html | 4 +- web/styles.css | 13 + 15 files changed, 2019 insertions(+), 183 deletions(-) create mode 100644 src/bank_importer/multipart.py create mode 100644 tests/test_import_api.py diff --git a/.gitignore b/.gitignore index 4d147ed..6ab6476 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,9 @@ server.out.log server.err.log nul +# 本地协作工具运行目录 +.multica/ +.opencode/ +.agent_context/ +.kimi/ + diff --git a/server.py b/server.py index fa3b48e..5aaae25 100644 --- a/server.py +++ b/server.py @@ -10,9 +10,8 @@ from http.cookies import SimpleCookie from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse -from bank_importer import auth, master_data +from bank_importer import auth, importing, master_data, multipart from bank_importer.db import connect, migrate, utc_now -from bank_importer.importing import import_statement ROOT = Path(__file__).resolve().parent @@ -49,6 +48,10 @@ class AppHandler(SimpleHTTPRequestHandler): if rows_match: self._handle_batch_rows(int(rows_match.group(1))) return + sheets_match = re.fullmatch(r"/api/batches/(\d+)/sheets", path) + if sheets_match: + self._handle_batch_sheets(int(sheets_match.group(1))) + return if path == "/api/export.csv": self._handle_export_csv(query) return @@ -119,6 +122,14 @@ class AppHandler(SimpleHTTPRequestHandler): if alias_create: self._handle_admin_add_alias(int(alias_create.group(1))) return + confirm_match = re.fullmatch(r"/api/batches/(\d+)/confirm", path) + if confirm_match: + self._handle_batch_review(int(confirm_match.group(1)), "confirm") + return + ignore_match = re.fullmatch(r"/api/batches/(\d+)/ignore", path) + if ignore_match: + self._handle_batch_review(int(ignore_match.group(1)), "ignore") + return self._send_json(404, {"status": "error", "message": "接口不存在。"}) # ------------------------------------------------------------------ @@ -308,15 +319,21 @@ class AppHandler(SimpleHTTPRequestHandler): def _handle_parse(self) -> None: connection = connect(DB_PATH) + upload = None try: user = self._require_user(connection) if user is None: return try: - filename, content, fields = self._read_upload() + fields, upload = self._receive_upload() + filename = os.path.basename(upload.filename) suffix = Path(filename).suffix.lower() if suffix not in {".xls", ".xlsx"}: raise ValueError("仅支持 .xls 或 .xlsx 银行流水文件。") + if not multipart.valid_file_signature(upload.path, suffix): + raise ValueError( + "文件内容与扩展名不符，可能已损坏或不是有效的 Excel 文件。" + ) if user["role"] == "company": # Tenant binding comes from the session only; any @@ -325,12 +342,15 @@ class AppHandler(SimpleHTTPRequestHandler): if not self._validate_upload_account(connection, user, fields): return else: - company_id = self._parse_company_field(connection, fields.get("company_id")) + company_id = self._parse_company_field( + connection, fields.get("company_id") + ) if company_id is None: return - result = import_statement( - connection, STORAGE_DIR, filename, content, company_id=company_id + result = importing.import_statement_path( + connection, STORAGE_DIR, filename, upload.path, + company_id=company_id, ) auth.audit( connection, @@ -356,8 +376,40 @@ class AppHandler(SimpleHTTPRequestHandler): else: self._send_json(200, payload) finally: + if upload is not None: + upload.path.unlink(missing_ok=True) connection.close() + def _receive_upload(self) -> tuple[dict[str, str], multipart.UploadedFile]: + """Stream the multipart body into a validated temp file. + + The whole request is never buffered in memory; file bytes are written + to a temp file under the storage directory as they arrive and are + never trimmed. Only the first file part is accepted. + """ + raw_length = self.headers.get("Content-Length") + if raw_length is None: + raise ValueError("上传请求缺少 Content-Length。") + try: + content_length = int(raw_length) + except ValueError: + raise ValueError("上传请求的 Content-Length 无效。") + if content_length <= 0: + raise ValueError("文件为空或超过 20 MB 限制。") + if content_length > MAX_UPLOAD_BYTES + multipart.MAX_FIELD_BYTES: + raise ValueError("文件超过 20 MB 限制。") + try: + fields, upload = multipart.parse_upload( + self.rfile, + content_length, + self.headers.get("Content-Type", ""), + STORAGE_DIR / ".uploads", + max_file_bytes=MAX_UPLOAD_BYTES, + ) + except multipart.MultipartError as exc: + raise ValueError(str(exc)) from exc + return fields, upload + def _validate_upload_account(self, connection, user, fields) -> bool: """Validate the optional bank_account_id on a company upload. @@ -435,15 +487,40 @@ class AppHandler(SimpleHTTPRequestHandler): f""" SELECT b.id, b.status, b.created_at, b.company_id, f.original_filename, c.name AS company_name, - COALESCE(SUM(s.transaction_count), 0) AS transactions, - MIN(s.period_start) AS period_start, - MAX(s.period_end) AS period_end + COALESCE(( + SELECT SUM(s.transaction_count) FROM sheet_batches s + WHERE s.import_batch_id = b.id + ), 0) AS transactions, + (SELECT s.bank_name FROM sheet_batches s + WHERE s.import_batch_id = b.id + ORDER BY s.id LIMIT 1) AS bank_name, + (SELECT MIN(s.period_start) FROM sheet_batches s + WHERE s.import_batch_id = b.id) AS period_start, + (SELECT MAX(s.period_end) FROM sheet_batches s + WHERE s.import_batch_id = b.id) AS period_end, + (SELECT COUNT(*) FROM source_rows r + JOIN sheet_batches s ON s.id = r.sheet_batch_id + JOIN sheet_reviews rv ON rv.sheet_batch_id = s.id + WHERE s.import_batch_id = b.id AND rv.review_status = 'confirmed') + AS confirmed_transactions, + (SELECT COUNT(*) FROM sheet_reviews r + WHERE r.import_batch_id = b.id AND r.review_status = 'confirmed') + AS confirmed_sheets, + (SELECT COUNT(*) FROM sheet_reviews r + WHERE r.import_batch_id = b.id + AND r.outcome = 'parsed' AND r.review_status = 'pending') + AS pending_sheets, + (SELECT COUNT(*) FROM sheet_reviews r + WHERE r.import_batch_id = b.id AND r.outcome = 'exception') + AS exception_sheets, + (SELECT COUNT(*) FROM sheet_reviews r + WHERE r.import_batch_id = b.id + AND (r.outcome = 'ignored' OR r.review_status = 'ignored')) + AS ignored_sheets FROM import_batches b JOIN source_files f ON f.id = b.source_file_id LEFT JOIN companies c ON c.id = b.company_id - LEFT JOIN sheet_batches s ON s.import_batch_id = b.id {where} - GROUP BY b.id ORDER BY b.id DESC LIMIT 500 """, @@ -453,6 +530,89 @@ class AppHandler(SimpleHTTPRequestHandler): finally: connection.close() + def _handle_batch_sheets(self, batch_id: int) -> None: + connection = connect(DB_PATH) + try: + user = self._require_user(connection) + if user is None: + return + company_id = user["company_id"] if user["role"] == "company" else None + batch = importing.scoped_batch(connection, batch_id, company_id) + if batch is None: + self._send_json(404, {"status": "error", "message": "批次不存在。"}) + return + file_row = connection.execute( + """ + SELECT f.original_filename FROM source_files f + JOIN import_batches b ON b.source_file_id = f.id + WHERE b.id = ? + """, + (batch_id,), + ).fetchone() + rows = importing.sheet_review_rows(connection, batch_id) + self._send_json( + 200, + { + "status": "ok", + "batch_id": batch_id, + "company_id": batch["company_id"], + "original_filename": file_row["original_filename"] + if file_row is not None + else None, + "sheets": [self._sheet_payload(row) for row in rows], + }, + ) + finally: + connection.close() + + def _handle_batch_review(self, batch_id: int, decision: str) -> None: + connection = connect(DB_PATH) + try: + user = self._require_user(connection) + if user is None: + return + data = self._read_json_body() + if data is None: + return + sheets = data.get("sheets") + if not isinstance(sheets, list): + self._send_json( + 400, {"status": "error", "message": "sheets 必须是工作表名称数组。"} + ) + return + company_id = user["company_id"] if user["role"] == "company" else None + try: + outcome = importing.review_sheets( + connection, + batch_id, + sheets, + decision, + user, + company_id, + reason=str(data.get("reason") or "") if decision == "ignore" else None, + ) + except LookupError as exc: + self._send_json(404, {"status": "error", "message": str(exc)}) + return + except importing.SheetReviewError as exc: + self._send_json(409, {"status": "error", "message": str(exc)}) + return + except ValueError as exc: + self._send_json(400, {"status": "error", "message": str(exc)}) + return + rows = importing.sheet_review_rows(connection, batch_id) + self._send_json( + 200, + { + "status": "ok", + "updated": outcome["updated"], + "already": outcome["already"], + "sheets": [self._sheet_payload(row) for row in rows], + }, + ) + finally: + connection.close() + def _handle_batch_rows(self, batch_id: int) -> None: connection = connect(DB_PATH) try: @@ -474,9 +634,11 @@ class AppHandler(SimpleHTTPRequestHandler): SELECT r.id, r.sheet_batch_id, r.source_row, r.transaction_at, r.income, r.expense, r.balance, r.own_account, r.own_name, r.counterparty_account, r.counterparty_name, r.counterparty_bank, - r.summary, r.purpose, r.reference, r.currency + r.summary, r.purpose, r.reference, r.currency, + s.sheet_name, rv.review_status FROM source_rows r JOIN sheet_batches s ON s.id = r.sheet_batch_id + JOIN sheet_reviews rv ON rv.sheet_batch_id = s.id WHERE s.import_batch_id = ? ORDER BY s.id, r.source_row """, @@ -529,6 +691,10 @@ class AppHandler(SimpleHTTPRequestHandler): FROM source_rows r JOIN sheet_batches s ON s.id = r.sheet_batch_id JOIN import_batches b ON b.id = s.import_batch_id + -- Only worksheets confirmed by the cashier may leave the + -- ledger: parse success is not business confirmation. + JOIN sheet_reviews rv + ON rv.sheet_batch_id = s.id AND rv.review_status = 'confirmed' {where} ORDER BY b.id, s.id, r.source_row """, @@ -1117,8 +1283,37 @@ class AppHandler(SimpleHTTPRequestHandler): return None return data + @staticmethod + def _sheet_payload(row) -> dict[str, object]: + """Serialize one persisted per-sheet review record.""" + return { + "sheet_name": row["sheet_name"], + "outcome": row["outcome"], + "review_status": row["review_status"], + "message": row["message"], + "scanned_rows": row["scanned_rows"], + "candidate_headers": json.loads(row["candidate_headers"]) + if row["candidate_headers"] + else [], + "review_reason": row["review_reason"], + "bank": row["bank_name"], + "template": row["template_id"], + "header_row": row["header_row"], + "period_start": row["period_start"], + "period_end": row["period_end"], + "transactions": row["transaction_count"] or 0, + "warnings": json.loads(row["warnings"]) if row["warnings"] else [], + } + @staticmethod def _import_payload(connection, result) -> dict[str, object]: + """Build the parse response from persisted state, not from memory. + + All worksheet results are returned (no more ``batches[0]``). A + cross-company duplicate is opaque: only the generic status, the + uploader's own batch id and the sha256 are returned, never the other + company's batch id, summary or diagnostics. + """ payload: dict[str, object] = { "status": result.status, "batch_id": result.batch_id, @@ -1126,96 +1321,19 @@ class AppHandler(SimpleHTTPRequestHandler): } if result.message: payload["message"] = result.message + if result.status == "duplicate" and not result.duplicate_same_company: + return payload - if result.status == "parsed": - batch = result.batches[0] - payload.update( - { - "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), - } - ) - elif result.status == "duplicate": - # Same-company duplicates keep their idempotent summary so the - # cashier sees the reused batch's bank/template/period/count. - # Cross-company duplicates are opaque: only the generic duplicate - # status and the uploader's own new batch id are returned, never - # the other company's batch id, summary or diagnostics. - if result.duplicate_same_company: - sheet = connection.execute( - """ - SELECT bank_name, template_id, header_row, period_start, - period_end, transaction_count, warnings - FROM sheet_batches - WHERE import_batch_id = ? - ORDER BY id LIMIT 1 - """, - (result.batch_id,), - ).fetchone() - if sheet is not None: - payload.update( - { - "bank": sheet["bank_name"], - "template": sheet["template_id"], - "header_row": sheet["header_row"], - "period_start": sheet["period_start"], - "period_end": sheet["period_end"], - "transactions": sheet["transaction_count"], - "warnings": json.loads(sheet["warnings"]), - } - ) + file_row = connection.execute( + "SELECT original_filename FROM source_files WHERE id = ?", + (result.source_file_id,), + ).fetchone() + if file_row is not None: + payload["original_filename"] = file_row["original_filename"] + sheets = importing.sheet_review_rows(connection, result.batch_id) + payload["sheets"] = [AppHandler._sheet_payload(row) for row in sheets] return payload - def _read_upload(self) -> tuple[str, bytes, dict[str, str]]: - """Parse the multipart body into (filename, content, text fields).""" - content_length = int(self.headers.get("Content-Length", "0")) - if content_length <= 0 or content_length > MAX_UPLOAD_BYTES: - raise ValueError("文件为空或超过 20 MB 限制。") - - content_type = self.headers.get("Content-Type", "") - boundary_match = re.search(r"boundary=(?:\"([^\"]+)\"|([^;]+))", content_type) - if not boundary_match: - raise ValueError("上传请求缺少文件边界。") - boundary = (boundary_match.group(1) or boundary_match.group(2)).encode() - body = self.rfile.read(content_length) - - fields: dict[str, str] = {} - for part in body.split(b"--" + boundary): - header, separator, content = part.partition(b"\r\n\r\n") - if not separator: - continue - name_match = re.search(br'name="([^\"]+)"', header) - if not name_match: - continue - name = name_match.group(1).decode("utf-8", errors="replace") - if name == "file": - filename_match = re.search(br'filename="([^\"]+)"', header) - filename = ( - filename_match.group(1).decode("utf-8", errors="replace") - if filename_match - else "statement.xlsx" - ) - file_content = content.rstrip(b"\r\n") - elif b"filename=" not in header: - fields[name] = content.rstrip(b"\r\n").decode("utf-8", errors="replace") - try: - return os.path.basename(filename), file_content, fields - except UnboundLocalError: - raise ValueError("上传请求中没有找到文件。") from None - - def _read_uploaded_file(self) -> tuple[str, bytes]: - filename, content, _fields = self._read_upload() - return filename, content - def _send_json( self, status: int, diff --git a/src/bank_importer/db.py b/src/bank_importer/db.py index eda6221..bc6e60f 100644 --- a/src/bank_importer/db.py +++ b/src/bank_importer/db.py @@ -307,6 +307,53 @@ MIGRATIONS: tuple[Migration, ...] = ( ALTER TABLE bank_accounts_new RENAME TO bank_accounts; """, ), + Migration( + version=4, + name="0004_per_sheet_reviews", + # Per-worksheet lifecycle: the parse outcome (parsed/exception/ignored) + # is immutable evidence captured at import time; the human decision + # (pending/confirmed/ignored) is the audit-gated gate that lets a + # worksheet participate in later matching and calculation. + up=""" + CREATE TABLE sheet_reviews ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + import_batch_id INTEGER NOT NULL REFERENCES import_batches (id), + sheet_name TEXT NOT NULL, + outcome TEXT NOT NULL + CHECK (outcome IN ('parsed', 'exception', 'ignored')), + message TEXT, + scanned_rows INTEGER, + candidate_headers TEXT, + sheet_batch_id INTEGER REFERENCES sheet_batches (id), + review_status TEXT NOT NULL DEFAULT 'pending' + CHECK (review_status IN ('pending', 'confirmed', 'ignored')), + review_reason TEXT, + reviewed_by INTEGER REFERENCES users (id), + reviewed_at TEXT, + created_at TEXT NOT NULL, + UNIQUE (import_batch_id, sheet_name) + ); + + CREATE INDEX idx_sheet_reviews_batch ON sheet_reviews (import_batch_id); + + CREATE TRIGGER sheet_reviews_evidence_immutable BEFORE UPDATE ON sheet_reviews + BEGIN + SELECT RAISE (ABORT, 'sheet_reviews parse evidence is immutable') + WHERE OLD.outcome != NEW.outcome + OR OLD.message IS NOT NEW.message + OR OLD.scanned_rows IS NOT NEW.scanned_rows + OR OLD.candidate_headers IS NOT NEW.candidate_headers + OR OLD.sheet_batch_id IS NOT NEW.sheet_batch_id + OR OLD.import_batch_id IS NOT NEW.import_batch_id + OR OLD.sheet_name IS NOT NEW.sheet_name; + END; + """, + down=""" + DROP TRIGGER IF EXISTS sheet_reviews_evidence_immutable; + DROP INDEX IF EXISTS idx_sheet_reviews_batch; + DROP TABLE IF EXISTS sheet_reviews; + """, + ), ) @@ -314,7 +361,9 @@ def connect(path: str | Path) -> sqlite3.Connection: db_path = Path(path) if str(db_path) != ":memory:": db_path.parent.mkdir(parents=True, exist_ok=True) - connection = sqlite3.connect(str(db_path)) + # Long busy timeout so concurrent uploads/confirmations wait for the + # single SQLite writer instead of surfacing "database is locked" 500s. + connection = sqlite3.connect(str(db_path), timeout=30) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") return connection diff --git a/src/bank_importer/importing.py b/src/bank_importer/importing.py index 88a8589..8985a9d 100644 --- a/src/bank_importer/importing.py +++ b/src/bank_importer/importing.py @@ -1,10 +1,16 @@ -"""Immutable statement import pipeline. +"""Immutable statement import pipeline with per-worksheet review lifecycle. Every uploaded file is hashed (SHA-256) and written once to content-addressed storage before parsing. A repeated upload of identical bytes never creates a second set of facts: it records a ``duplicate`` batch that points at the original batch. Parse failures keep the batch and its diagnostics as an ``exception`` batch without producing any confirmed source rows. + +Each worksheet is parsed independently into a :class:`SheetResult` +(``parsed`` / ``exception`` / ``ignored``) that is persisted in +``sheet_reviews`` as immutable evidence. Whether a parsed sheet may take part +in later matching and calculation is a separate, auditable human decision +(``review_status``): only ``confirmed`` sheets participate. """ from __future__ import annotations @@ -14,11 +20,18 @@ import hashlib import json import os from pathlib import Path +import shutil import sqlite3 +from . import auth from .db import utc_now -from .models import StatementBatch -from .parser import StatementParseError, parse_statement +from .models import SheetResult, StatementBatch +from .parser import StatementParseError, analyze_workbook +from .reader import CorruptWorkbookError + + +class SheetReviewError(ValueError): + """A review decision cannot be applied to the worksheet (mapped to 409).""" @dataclass(frozen=True) @@ -28,6 +41,7 @@ class ImportResult: sha256: str source_file_id: int batches: tuple[StatementBatch, ...] = () + sheets: tuple[SheetResult, ...] = () message: str | None = None duplicate_same_company: bool = False # only meaningful when status == 'duplicate' @@ -48,6 +62,78 @@ def import_statement( return _record_duplicate(connection, existing_file["id"], sha256, company_id) stored_path = _store_immutable(Path(storage_dir), original_filename, content, sha256) + try: + source_file_id, batch_id = _create_batch_records( + connection, sha256, original_filename, len(content), stored_path, company_id + ) + except sqlite3.IntegrityError: + # Lost a concurrent-insert race on the sha256 UNIQUE constraint: the + # identical content was already persisted by another request, so this + # upload is a duplicate. The content-addressed file already exists. + existing = connection.execute( + "SELECT id FROM source_files WHERE sha256 = ?", (sha256,) + ).fetchone() + if existing is not None: + return _record_duplicate(connection, existing["id"], sha256, company_id) + raise + return _parse_and_persist( + connection, stored_path, sha256, original_filename, source_file_id, batch_id + ) + + +def import_statement_path( + connection: sqlite3.Connection, + storage_dir: str | Path, + original_filename: str, + upload_path: str | Path, + company_id: int | None = None, +) -> ImportResult: + """Import from an already-downloaded upload file (streaming-friendly). + + ``upload_path`` names the temp file the multipart handler streamed to + disk; it is hashed incrementally and either discarded (duplicate) or + published into the content-addressed store. The upload file itself is + never modified. + """ + source = Path(upload_path) + sha256 = _file_sha256(source) + existing_file = connection.execute( + "SELECT id FROM source_files WHERE sha256 = ?", (sha256,) + ).fetchone() + + if existing_file is not None: + source.unlink(missing_ok=True) + return _record_duplicate(connection, existing_file["id"], sha256, company_id) + + stored_path = _publish_immutable(Path(storage_dir), original_filename, source, sha256) + size = stored_path.stat().st_size + try: + source_file_id, batch_id = _create_batch_records( + connection, sha256, original_filename, size, stored_path, company_id + ) + except sqlite3.IntegrityError: + # Lost a concurrent-insert race on the sha256 UNIQUE constraint: the + # identical content was already persisted by another request, so this + # upload is a duplicate. The content-addressed file already exists. + existing = connection.execute( + "SELECT id FROM source_files WHERE sha256 = ?", (sha256,) + ).fetchone() + if existing is not None: + return _record_duplicate(connection, existing["id"], sha256, company_id) + raise + return _parse_and_persist( + connection, stored_path, sha256, original_filename, source_file_id, batch_id + ) + + +def _create_batch_records( + connection: sqlite3.Connection, + sha256: str, + original_filename: str, + size_bytes: int, + stored_path: Path, + company_id: int | None, +) -> tuple[int, int]: now = utc_now() with connection: cursor = connection.execute( @@ -55,7 +141,7 @@ def import_statement( INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at) VALUES (?, ?, ?, ?, ?) """, - (sha256, original_filename, len(content), str(stored_path), now), + (sha256, original_filename, size_bytes, str(stored_path), now), ) source_file_id = cursor.lastrowid cursor = connection.execute( @@ -66,27 +152,64 @@ def import_statement( (source_file_id, company_id, now, now), ) batch_id = cursor.lastrowid + return source_file_id, batch_id + +def _parse_and_persist( + connection: sqlite3.Connection, + stored_path: Path, + sha256: str, + original_filename: str, + source_file_id: int, + batch_id: int, +) -> ImportResult: try: - batches = parse_statement(stored_path) - except StatementParseError as exc: + sheets = analyze_workbook(stored_path) + except CorruptWorkbookError as exc: message = _clean_message(str(exc), stored_path, original_filename) with connection: - _insert_exception(connection, batch_id, "parse", message, original_filename) + _insert_exception( + connection, batch_id, "parse", message, original_filename, + diagnostics={"original_filename": original_filename}, + ) _set_batch_status(connection, batch_id, "exception") return ImportResult(batch_id, "exception", sha256, source_file_id, message=message) except Exception as exc: message = f"文件解析失败，请检查文件是否完整。（{type(exc).__name__}）" with connection: - _insert_exception(connection, batch_id, "internal", message, original_filename) + _insert_exception( + connection, batch_id, "internal", message, original_filename, + diagnostics={"original_filename": original_filename}, + ) _set_batch_status(connection, batch_id, "failed") raise with connection: - for batch in batches: - _insert_sheet_batch(connection, batch_id, batch) + parsed_batches: list[StatementBatch] = [] + for sheet in sheets: + if sheet.outcome == "parsed" and sheet.batch is not None: + sheet_batch_id = _insert_sheet_batch(connection, batch_id, sheet.batch) + _insert_sheet_review(connection, batch_id, sheet, sheet_batch_id) + parsed_batches.append(sheet.batch) + else: + _insert_sheet_review(connection, batch_id, sheet, None) + if not parsed_batches: + message = _whole_file_message(original_filename, sheets) + with connection: + _insert_exception( + connection, batch_id, "parse", message, original_filename, + diagnostics=_exception_diagnostics(original_filename, sheets), + ) + _set_batch_status(connection, batch_id, "exception") + return ImportResult( + batch_id, "exception", sha256, source_file_id, + message=message, sheets=sheets, + ) _set_batch_status(connection, batch_id, "parsed") - return ImportResult(batch_id, "parsed", sha256, source_file_id, batches=batches) + return ImportResult( + batch_id, "parsed", sha256, source_file_id, + batches=tuple(parsed_batches), sheets=sheets, + ) def _record_duplicate( @@ -169,9 +292,39 @@ def _store_immutable( return target +def _publish_immutable( + storage_dir: Path, original_filename: str, upload_path: Path, sha256: str +) -> Path: + suffix = Path(original_filename).suffix.lower() + target_dir = storage_dir / sha256[:2] + target_dir.mkdir(parents=True, exist_ok=True) + target = target_dir / f"{sha256}{suffix}" + temp = target_dir / f".{sha256}.tmp" + try: + # Copy keeps the same-filesystem guarantee even if the upload temp + # lives elsewhere; the hard link then publishes atomically. + shutil.copyfile(upload_path, temp) + try: + os.link(temp, target) + except FileExistsError: + pass + finally: + temp.unlink(missing_ok=True) + upload_path.unlink(missing_ok=True) + return target + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + def _insert_sheet_batch( connection: sqlite3.Connection, batch_id: int, batch: StatementBatch -) -> None: +) -> int: now = utc_now() cursor = connection.execute( """ @@ -226,6 +379,45 @@ def _insert_sheet_batch( now, ), ) + return sheet_batch_id + + +def _insert_sheet_review( + connection: sqlite3.Connection, + batch_id: int, + sheet: SheetResult, + sheet_batch_id: int | None, +) -> None: + # Parse-side ignored sheets are terminal: there is nothing to confirm, so + # their review lifecycle is closed with a machine reason. + if sheet.outcome == "ignored": + review_status = "ignored" + review_reason = "空表或无有效内容，自动忽略。" + else: + review_status = "pending" + review_reason = None + connection.execute( + """ + INSERT INTO sheet_reviews ( + import_batch_id, sheet_name, outcome, message, scanned_rows, + candidate_headers, sheet_batch_id, review_status, review_reason, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + batch_id, + sheet.sheet_name, + sheet.outcome, + sheet.message, + sheet.scanned_rows, + json.dumps(list(sheet.candidate_headers), ensure_ascii=False) + if sheet.candidate_headers + else None, + sheet_batch_id, + review_status, + review_reason, + utc_now(), + ), + ) def _insert_exception( @@ -234,7 +426,12 @@ def _insert_exception( stage: str, message: str, original_filename: str, + *, + diagnostics: dict[str, object] | None = None, ) -> None: + payload = {"original_filename": original_filename} + if diagnostics: + payload.update(diagnostics) connection.execute( """ INSERT INTO import_exceptions (import_batch_id, stage, message, diagnostics, created_at) @@ -244,7 +441,7 @@ def _insert_exception( batch_id, stage, message, - json.dumps({"original_filename": original_filename}, ensure_ascii=False), + json.dumps(payload, ensure_ascii=False), utc_now(), ), ) @@ -257,7 +454,154 @@ def _set_batch_status(connection: sqlite3.Connection, batch_id: int, status: str ) +def _whole_file_message(original_filename: str, sheets: tuple[SheetResult, ...]) -> str: + detail = "；".join( + f"{sheet.sheet_name}：{sheet.message}" for sheet in sheets if sheet.message + ) + return f"文件「{original_filename}」没有可确认的工作表。{detail}" + + +def _exception_diagnostics( + original_filename: str, sheets: tuple[SheetResult, ...] +) -> dict[str, object]: + return { + "original_filename": original_filename, + "sheets": [ + { + "sheet_name": sheet.sheet_name, + "outcome": sheet.outcome, + "message": sheet.message, + "scanned_rows": sheet.scanned_rows, + "candidate_headers": list(sheet.candidate_headers), + } + for sheet in sheets + ], + } + + def _clean_message(message: str, stored_path: Path, original_filename: str) -> str: return message.replace(str(stored_path), original_filename).replace( stored_path.name, original_filename ) + + +# --------------------------------------------------------------------------- +# Per-sheet review lifecycle +# --------------------------------------------------------------------------- + + +def sheet_review_rows(connection: sqlite3.Connection, batch_id: int) -> list[sqlite3.Row]: + return connection.execute( + """ + SELECT r.id, r.sheet_name, r.outcome, r.message, r.scanned_rows, + r.candidate_headers, r.review_status, r.review_reason, + r.reviewed_by, r.reviewed_at, + s.bank_name, s.template_id, s.header_row, s.own_account, s.own_name, + s.period_start, s.period_end, s.transaction_count, s.warnings + FROM sheet_reviews r + LEFT JOIN sheet_batches s ON s.id = r.sheet_batch_id + WHERE r.import_batch_id = ? + ORDER BY r.id + """, + (batch_id,), + ).fetchall() + + +def scoped_batch( + connection: sqlite3.Connection, batch_id: int, company_id: int | None +) -> sqlite3.Row | None: + """Load a batch, scoping company users to their own tenant. + + ``company_id is None`` (admin) sees any batch; a company user only sees + batches owned by that company. Returns None when out of scope so callers + can answer 404 without leaking the batch's existence. + """ + if company_id is None: + return connection.execute( + "SELECT id, company_id FROM import_batches WHERE id = ?", (batch_id,) + ).fetchone() + return connection.execute( + "SELECT id, company_id FROM import_batches WHERE id = ? AND company_id = ?", + (batch_id, company_id), + ).fetchone() + + +def review_sheets( + connection: sqlite3.Connection, + batch_id: int, + sheet_names: list[str], + decision: str, + actor: sqlite3.Row | None, + company_id: int | None, + reason: str | None = None, +) -> dict[str, object]: + """Apply ``confirm`` or ``ignore`` to whole worksheets atomically. + + Multi-sheet writes run in one transaction; a failure rolls back every + sheet in the request. Idempotent repeats of the same decision succeed + without touching the audit log; changing a settled decision is a + conflict. Returns ``{"updated": [...], "already": [...]}``. + """ + if decision not in ("confirm", "ignore"): + raise ValueError("decision 必须是 confirm 或 ignore。") + names = [str(name).strip() for name in sheet_names] + if not names: + raise ValueError("必须至少指定一个工作表。") + reason = (reason or "").strip() or None + if decision == "ignore" and not reason: + raise ValueError("忽略工作表必须填写原因。") + + batch = scoped_batch(connection, batch_id, company_id) + if batch is None: + raise LookupError("批次不存在。") + + placeholders = ",".join("?" for _ in names) + rows = connection.execute( + f""" + SELECT * FROM sheet_reviews + WHERE import_batch_id = ? AND sheet_name IN ({placeholders}) + """, + (batch_id, *names), + ).fetchall() + found = {row["sheet_name"] for row in rows} + missing = [name for name in names if name not in found] + if missing: + raise ValueError(f"工作表不存在或不属于该批次：{'、'.join(missing)}") + + target = "confirmed" if decision == "confirm" else "ignored" + now = utc_now() + updated: list[str] = [] + already: list[str] = [] + with connection: + for row in rows: + if row["review_status"] == target: + already.append(row["sheet_name"]) + continue + if decision == "confirm" and row["outcome"] != "parsed": + raise SheetReviewError( + f"工作表「{row['sheet_name']}」没有可确认的交易，只能忽略或保留待处理。" + ) + if row["review_status"] != "pending": + raise SheetReviewError( + f"工作表「{row['sheet_name']}」已处理，不能更改决定。" + ) + cursor = connection.execute( + """ + UPDATE sheet_reviews + SET review_status = ?, review_reason = ?, reviewed_by = ?, reviewed_at = ? + WHERE id = ? AND review_status = 'pending' + """, + (target, reason, actor["id"] if actor is not None else None, now, row["id"]), + ) + if cursor.rowcount: + updated.append(row["sheet_name"]) + + if updated: + auth.audit( + connection, + f"sheet_{decision}", + actor=actor, + target=f"batch:{batch_id}", + detail=f"sheets:{','.join(updated)}" + (f";reason:{reason}" if reason else ""), + ) + return {"updated": updated, "already": already} diff --git a/src/bank_importer/models.py b/src/bank_importer/models.py index 3f0ac50..4c75ba5 100644 --- a/src/bank_importer/models.py +++ b/src/bank_importer/models.py @@ -40,3 +40,23 @@ class StatementBatch: transactions: tuple[NormalizedTransaction, ...] warnings: tuple[str, ...] template_version: int = 1 + + +@dataclass(frozen=True) +class SheetResult: + """Per-worksheet parse outcome with the four diagnostic evidence items. + + ``outcome`` is the deterministic parse verdict: ``parsed`` (transactions + normalized), ``exception`` (template/header/balance problems) or + ``ignored`` (empty or single-cell sheets with no bank content). The human + decision about whether a parsed sheet may participate in calculations is + a separate lifecycle stage (``review_status``), persisted in + ``sheet_reviews``; it is never decided here. + """ + + sheet_name: str + outcome: str # parsed | exception | ignored + message: str | None = None + scanned_rows: int | None = None + candidate_headers: tuple[str, ...] = () + batch: StatementBatch | None = None diff --git a/src/bank_importer/multipart.py b/src/bank_importer/multipart.py new file mode 100644 index 0000000..ff1a0b4 --- /dev/null +++ b/src/bank_importer/multipart.py @@ -0,0 +1,275 @@ +"""Streaming multipart/form-data parser for the stdlib HTTP server. + +The upload body is consumed in chunks from the request stream and never +assembled into memory as a whole. File parts are streamed straight to a temp +file on disk with an incremental size cap; text fields are capped at a small +field size. File bytes are preserved exactly (no trailing-byte trimming), the +first file part is the only one accepted, and both the extension and the file +signature are validated by the caller. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +import re +from typing import BinaryIO + +MAX_FIELD_BYTES = 16 * 1024 +MAX_HEADER_BYTES = 32 * 1024 +CHUNK_SIZE = 64 * 1024 +# .xlsx files are ZIP containers (PK\x03\x04 local header or an empty-archive +# PK\x05\x06); .xls files are OLE2 compound documents. +XLSX_MAGIC = (b"PK\x03\x04", b"PK\x05\x06") +XLS_MAGIC = (b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1",) + + +class MultipartError(ValueError): + pass + + +@dataclass(frozen=True) +class UploadedFile: + filename: str + path: Path + size: int + + +def valid_file_signature(path: Path, suffix: str) -> bool: + """True when the file's leading bytes match its declared extension.""" + with path.open("rb") as handle: + head = handle.read(8) + if suffix == ".xlsx": + return any(head.startswith(magic) for magic in XLSX_MAGIC) + if suffix == ".xls": + return any(head.startswith(magic) for magic in XLS_MAGIC) + return False + + +class _MultipartStream: + def __init__( + self, + rfile: BinaryIO, + content_length: int, + content_type: str, + *, + max_body_bytes: int, + ) -> None: + lowered = content_type.lower() + if not lowered.startswith("multipart/form-data"): + raise MultipartError("上传请求必须是 multipart/form-data。") + match = re.search(r"boundary=(?:\"([^\"]+)\"|([^;]+))", content_type) + if not match: + raise MultipartError("上传请求缺少文件边界。") + boundary = (match.group(1) or match.group(2)).strip().encode("utf-8") + if not boundary or len(boundary) > 200: + raise MultipartError("上传请求的文件边界无效。") + if content_length <= 0: + raise MultipartError("上传请求为空。") + if content_length > max_body_bytes: + raise MultipartError(f"上传请求超过 {max_body_bytes // (1024 * 1024)} MB 限制。") + self.rfile = rfile + self.remaining = content_length + self.total_read = 0 + self.buffer = bytearray() + self.boundary = boundary + self._fill() + + # ------------------------------------------------------------------ + # Low-level stream helpers + # ------------------------------------------------------------------ + + def _fill(self) -> None: + if self.total_read >= self.remaining: + return + want = min(CHUNK_SIZE, self.remaining - self.total_read) + data = self.rfile.read(want) + if not data: + # Client closed early; clamp remaining so every helper sees EOF. + self.remaining = self.total_read + return + self.total_read += len(data) + self.buffer.extend(data) + + def _take(self, count: int) -> bytes: + while len(self.buffer) < count: + if self.total_read >= self.remaining: + raise MultipartError("multipart 请求体不完整。") + self._fill() + out = bytes(self.buffer[:count]) + del self.buffer[:count] + return out + + def _read_line(self) -> bytes: + while True: + index = self.buffer.find(b"\n") + if index >= 0: + line = bytes(self.buffer[: index + 1]) + del self.buffer[: index + 1] + return line + if self.total_read >= self.remaining: + if not self.buffer: + raise MultipartError("multipart 请求体不完整。") + line = bytes(self.buffer) + self.buffer.clear() + return line + self._fill() + + def _skip_preamble(self) -> None: + marker = b"--" + self.boundary + while True: + index = self.buffer.find(marker) + if index >= 0: + del self.buffer[: index + len(marker)] + return + keep = len(marker) - 1 + if len(self.buffer) > keep: + del self.buffer[: len(self.buffer) - keep] + if self.total_read >= self.remaining: + raise MultipartError("上传请求中没有找到文件。") + self._fill() + + def _iter_content(self, boundary: bytes): + """Yield content bytes up to (excluding) the ``\\r\\n--boundary`` marker.""" + marker = b"\r\n--" + boundary + keep = len(marker) - 1 + while True: + index = self.buffer.find(marker) + if index >= 0: + content = bytes(self.buffer[:index]) + del self.buffer[:index] + if content: + yield content + return + if len(self.buffer) > keep: + safe = len(self.buffer) - keep + content = bytes(self.buffer[:safe]) + del self.buffer[:safe] + if content: + yield content + if self.total_read >= self.remaining: + raise MultipartError("multipart 请求体缺少结束边界。") + self._fill() + + def _after_content_boundary(self) -> str: + """Consume the content-ending marker; return 'part' or 'end'.""" + marker = b"\r\n--" + self.boundary + del self.buffer[: len(marker)] + indicator = self._take(2) + if indicator == b"--": + if self.buffer.startswith(b"\r\n"): + del self.buffer[:2] + elif self.buffer.startswith(b"\n"): + del self.buffer[:1] + return "end" + if indicator != b"\r\n": + raise MultipartError("上传请求格式无效。") + return "part" + + # ------------------------------------------------------------------ + # Part parsing + # ------------------------------------------------------------------ + + def _read_headers(self) -> bytes: + total = 0 + lines: list[bytes] = [] + while True: + line = self._read_line() + total += len(line) + if total > MAX_HEADER_BYTES: + raise MultipartError("multipart 头部过长。") + if line in (b"\r\n", b"\n"): + return b"".join(lines) + lines.append(line) + + @staticmethod + def _parse_disposition(header: bytes) -> tuple[str | None, str | None]: + name = None + filename = None + name_match = re.search(br'name="([^"]*)"', header) + if name_match: + name = name_match.group(1).decode("utf-8", errors="replace") + filename_match = re.search(br'filename="([^"]*)"', header) + if filename_match: + filename = filename_match.group(1).decode("utf-8", errors="replace") + return name, filename + + def parse(self, work_dir: Path) -> tuple[dict[str, str], UploadedFile | None]: + import tempfile + + self._skip_preamble() + next_bytes = self._take(2) + if next_bytes == b"--": + # Empty multipart body: opening boundary immediately closes. + return {}, None + if next_bytes != b"\r\n": + raise MultipartError("上传请求格式无效。") + + fields: dict[str, str] = {} + uploaded: UploadedFile | None = None + while True: + header = self._read_headers() + name, filename = self._parse_disposition(header) + if name is None: + raise MultipartError("上传请求缺少字段名。") + + if filename is not None: + if uploaded is not None: + raise MultipartError("一次只能上传一个文件。") + with tempfile.NamedTemporaryFile( + dir=str(work_dir), prefix="upload-", delete=False + ) as sink: + temp_path = Path(sink.name) + size = 0 + try: + for chunk in self._iter_content(self.boundary): + size += len(chunk) + sink.write(chunk) + except Exception: + temp_path.unlink(missing_ok=True) + raise + uploaded = UploadedFile( + filename=filename, path=temp_path, size=size + ) + if self._after_content_boundary() == "end": + break + else: + chunks: list[bytes] = [] + total = 0 + for chunk in self._iter_content(self.boundary): + total += len(chunk) + if total > MAX_FIELD_BYTES: + raise MultipartError("表单字段超过大小限制。") + chunks.append(chunk) + fields[name] = b"".join(chunks).decode( + "utf-8", errors="replace" + ).strip() + if self._after_content_boundary() == "end": + break + return fields, uploaded + + +def parse_upload( + rfile: BinaryIO, + content_length: int, + content_type: str, + work_dir: str | Path, + *, + max_file_bytes: int, +) -> tuple[dict[str, str], UploadedFile | None]: + """Stream a multipart upload; returns ``(fields, uploaded_file)``. + + Raises :class:`MultipartError` for malformed or oversized bodies. The + caller owns the uploaded temp file and must remove it when done. + """ + work = Path(work_dir) + work.mkdir(parents=True, exist_ok=True) + stream = _MultipartStream( + rfile, content_length, content_type, max_body_bytes=max_file_bytes + MAX_FIELD_BYTES + ) + fields, uploaded = stream.parse(work) + if uploaded is None: + raise MultipartError("上传请求中没有找到文件。") + if uploaded.size > max_file_bytes: + raise MultipartError(f"文件超过 {max_file_bytes // (1024 * 1024)} MB 限制。") + return fields, uploaded diff --git a/src/bank_importer/parser.py b/src/bank_importer/parser.py index 0e83b15..8bd80f5 100644 --- a/src/bank_importer/parser.py +++ b/src/bank_importer/parser.py @@ -7,7 +7,7 @@ from pathlib import Path import re from typing import Any, Iterable -from .models import NormalizedTransaction, StatementBatch +from .models import NormalizedTransaction, SheetResult, StatementBatch from .reader import RawSheet, read_workbook from .templates import BankTemplate, TEMPLATES, normalize_header @@ -24,8 +24,11 @@ class AmbiguousTemplateError(StatementParseError): pass +SCAN_LIMIT = 50 + + def detect_header( - rows: tuple[tuple[Any, ...], ...], scan_limit: int = 50 + rows: tuple[tuple[Any, ...], ...], scan_limit: int = SCAN_LIMIT ) -> tuple[BankTemplate, int, dict[str, int]]: candidates: list[tuple[int, BankTemplate, int, dict[str, int]]] = [] for row_index, row in enumerate(rows[:scan_limit]): @@ -42,8 +45,8 @@ def detect_header( 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 "" + candidate_rows = _header_candidate_rows(rows[:scan_limit]) + detail = f"；候选表头：{'；'.join(candidate_rows)}" if candidate_rows else "" raise UnknownTemplateError( f"未识别到受支持的银行表头（已扫描前 {inspected} 行）{detail}。" ) @@ -60,21 +63,69 @@ def detect_header( 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}") - + batches = tuple(result.batch for result in analyze_workbook(path) if result.batch) if not batches: - detail = "; ".join(errors) or "Workbook contains no readable worksheets." - raise UnknownTemplateError(f"{source.name}: {detail}") - return tuple(batches) + raise UnknownTemplateError(f"{Path(path).name}: 工作簿中没有可解析的工作表。") + return batches + + +def analyze_workbook(path: str | Path) -> tuple[SheetResult, ...]: + """Parse every worksheet into an independent result. + + Each worksheet yields exactly one :class:`SheetResult` whose ``outcome`` + is ``parsed``, ``exception`` or ``ignored``. Unreadable workbooks raise + ``CorruptWorkbookError``; a workbook that reads but contains no parsable + sheet still returns one result per sheet so the UI can surface the + filename / sheet / scanned range / candidate headers evidence. + """ + source = Path(path) + return tuple(_analyze_sheet(source, sheet) for sheet in read_workbook(source)) + + +def _analyze_sheet(source: Path, sheet: RawSheet) -> SheetResult: + rows = sheet.rows + scanned = min(len(rows), SCAN_LIMIT) + if not rows: + return SheetResult( + sheet_name=sheet.name, + outcome="ignored", + message=f"工作表「{sheet.name}」为空，已跳过。", + scanned_rows=0, + ) + if not any(any(_text(value) for value in row) for row in rows): + return SheetResult( + sheet_name=sheet.name, + outcome="ignored", + message=f"工作表「{sheet.name}」无有效内容，已跳过。", + scanned_rows=scanned, + ) + try: + batch = _parse_sheet(source, sheet) + except UnknownTemplateError as exc: + return SheetResult( + sheet_name=sheet.name, + outcome="exception", + message=_clean_sheet_message(str(exc), source), + scanned_rows=scanned, + candidate_headers=_header_candidate_rows(rows), + ) + except (AmbiguousTemplateError, StatementParseError) as exc: + return SheetResult( + sheet_name=sheet.name, + outcome="exception", + message=_clean_sheet_message(str(exc), source), + scanned_rows=scanned, + ) + return SheetResult( + sheet_name=sheet.name, + outcome="parsed", + scanned_rows=scanned, + batch=batch, + ) + + +def _clean_sheet_message(message: str, source: Path) -> str: + return message.replace(str(source), source.name).replace(source.name, "本文件") def parse_directory(path: str | Path) -> tuple[StatementBatch, ...]: @@ -90,9 +141,14 @@ def parse_directory(path: str | Path) -> tuple[StatementBatch, ...]: return tuple(batch for file in files for batch in parse_statement(file)) -def _header_candidate_summary( +def _header_candidate_rows( rows: tuple[tuple[Any, ...], ...], limit: int = 3 -) -> str: +) -> tuple[str, ...]: + """Preview up to ``limit`` plausible header rows. + + Every non-empty row counts as a candidate, including single-cell rows, so + an ambiguous workbook always has explicit scan evidence for the UI. + """ known_headers = { alias for template in TEMPLATES @@ -102,7 +158,7 @@ def _header_candidate_summary( 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: + if not values: continue matched = sum(normalize_header(value) in known_headers for value in values) candidates.append((matched, len(values), row_index, values)) @@ -114,7 +170,7 @@ def _header_candidate_summary( if len(values) > 8: preview += "……" summaries.append(f"第 {row_index + 1} 行「{preview}」") - return "；".join(summaries) + return tuple(summaries) def _parse_sheet(source: Path, sheet: RawSheet) -> StatementBatch: diff --git a/src/bank_importer/reader.py b/src/bank_importer/reader.py index 5c15a7e..ba52bbb 100644 --- a/src/bank_importer/reader.py +++ b/src/bank_importer/reader.py @@ -10,6 +10,23 @@ class UnsupportedWorkbookError(RuntimeError): pass +class CorruptWorkbookError(RuntimeError): + """The file cannot be opened as a valid Excel workbook. + + Raised for corrupt/truncated files, wrong signatures and files that + exceed the resource limits. The message is user-safe: it never contains + server-side paths or the content-addressed storage filename. + """ + + +# Workbook resource guards (defense against decompression bombs and +# accidentally giant exports). Bank statements are small; these bounds are +# generous enough for real exports while keeping memory bounded. +MAX_SHEETS = 50 +MAX_ROWS_PER_SHEET = 200_000 +MAX_COLS_PER_SHEET = 64 + + @dataclass(frozen=True) class RawSheet: name: str @@ -18,11 +35,18 @@ class RawSheet: 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}") + try: + if suffix == ".xlsx": + return _read_xlsx(path) + if suffix == ".xls": + return _read_xls(path) + raise UnsupportedWorkbookError(f"Unsupported workbook type: {suffix}") + except (UnsupportedWorkbookError, CorruptWorkbookError): + raise + except Exception as exc: + raise CorruptWorkbookError( + "文件无法读取，可能已损坏或不是有效的 Excel 文件。" + ) from exc def _read_xlsx(path: Path) -> tuple[RawSheet, ...]: @@ -35,13 +59,24 @@ def _read_xlsx(path: Path) -> tuple[RawSheet, ...]: 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 - ) + sheets: list[RawSheet] = [] + for worksheet in workbook.worksheets: + rows: list[tuple[Any, ...]] = [] + for row_index, row in enumerate(worksheet.iter_rows(values_only=True)): + if row_index >= MAX_ROWS_PER_SHEET: + raise CorruptWorkbookError( + f"工作表「{worksheet.title}」超过 {MAX_ROWS_PER_SHEET} 行，已拒绝读取。" + ) + values = tuple(row) + if len(values) > MAX_COLS_PER_SHEET: + raise CorruptWorkbookError( + f"工作表「{worksheet.title}」列数超过 {MAX_COLS_PER_SHEET}，已拒绝读取。" + ) + rows.append(values) + sheets.append(RawSheet(name=worksheet.title, rows=tuple(rows))) + if len(sheets) > MAX_SHEETS: + raise CorruptWorkbookError(f"工作簿工作表数量超过 {MAX_SHEETS} 个，已拒绝读取。") + return tuple(sheets) finally: workbook.close() @@ -57,8 +92,19 @@ def _read_xls(path: Path) -> tuple[RawSheet, ...]: workbook = xlrd.open_workbook(path, on_demand=True) sheets: list[RawSheet] = [] try: - for sheet_name in workbook.sheet_names(): + names = workbook.sheet_names() + if len(names) > MAX_SHEETS: + raise CorruptWorkbookError(f"工作簿工作表数量超过 {MAX_SHEETS} 个，已拒绝读取。") + for sheet_name in names: worksheet = workbook.sheet_by_name(sheet_name) + if worksheet.nrows > MAX_ROWS_PER_SHEET: + raise CorruptWorkbookError( + f"工作表「{sheet_name}」超过 {MAX_ROWS_PER_SHEET} 行，已拒绝读取。" + ) + if worksheet.ncols > MAX_COLS_PER_SHEET: + raise CorruptWorkbookError( + f"工作表「{sheet_name}」列数超过 {MAX_COLS_PER_SHEET}，已拒绝读取。" + ) rows: list[tuple[Any, ...]] = [] for row_index in range(worksheet.nrows): values: list[Any] = [] diff --git a/tests/test_import_api.py b/tests/test_import_api.py new file mode 100644 index 0000000..9855df0 --- /dev/null +++ b/tests/test_import_api.py @@ -0,0 +1,587 @@ +"""HTTP integration tests for the hardened import API (B-42). + +Covers streaming multipart limits, per-worksheet parse results with the four +diagnostic evidence items, the confirm/ignore review lifecycle with tenant +scoping and idempotency, unconfirmed-sheet export gating, and concurrent +duplicate uploads. Uses a real ``ThreadingHTTPServer`` like +``test_server_auth``. +""" + +from __future__ import annotations + +import io +import itertools +import json +import os +from pathlib import Path +import tempfile +import threading +import unittest + +from openpyxl import Workbook + +from bank_importer import auth +from bank_importer.db import connect, migrate + +import server +from test_server_auth import Client, as_json + +ROOT = Path(__file__).resolve().parents[1] +SAMPLES = ROOT / "流水模板" +CCB_SAMPLE = SAMPLES / "中国建设银行账户流水.xls" +CITIC_SAMPLE = SAMPLES / "中信银行账户流水.xlsx" + +BOOTSTRAP_PASSWORD = "BootAdmin123" +ADMIN_PASSWORD = "AdminPass123" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +CCB_HEADER = [ + "客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）", + "余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注", +] +CCB_DATA = [ + ["6228480000000000", "测试公司", "2026-01-05 10:00:00", "100.00", "", "99900.00", "RMB", "供应商", "1002003004", "某银行", "货款", ""], + ["6228480000000000", "测试公司", "2026-01-06 11:00:00", "", "200.00", "100100.00", "RMB", "客户", "2003004005", "某银行", "收款", ""], +] + + +def workbook_bytes(sheets): + """Build an xlsx in memory. + + ``sheets`` is a list of ``(name, rows)``; an empty ``rows`` list means an + empty worksheet. + """ + workbook = Workbook() + workbook.remove(workbook.active) + for name, rows in sheets: + worksheet = workbook.create_sheet(name) + for row in rows: + worksheet.append(row) + buffer = io.BytesIO() + workbook.save(buffer) + return buffer.getvalue() + + +# Workbook fixtures are content-hashed for dedupe, so every helper produces +# distinct bytes per call (each test uploads a genuinely new file). +_counter = itertools.count() + + +def multi_sheet_bytes(): + """Valid CCB sheet + unknown-header sheet + empty sheet in one workbook.""" + return workbook_bytes( + [ + ("正常流水", [CCB_HEADER, *CCB_DATA]), + ("未知模板", [["日期", "金额", "备注"], ["2026-01-01", next(_counter), "x"]]), + ("空表", []), + ] + ) + + +def empty_workbook_bytes(): + return workbook_bytes([("Sheet1", [])]) + + +def single_cell_bytes(): + return workbook_bytes([("候选", [[f"只有一个单元格 {next(_counter)}"]])]) + + +def unknown_header_bytes(): + return workbook_bytes([("流水", [["日期", "金额", "备注"], ["2026-01-01", next(_counter), "x"]])]) + + +# --------------------------------------------------------------------------- +# Test server +# --------------------------------------------------------------------------- + + +class ImportApiServerTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.temp_dir = tempfile.TemporaryDirectory() + root = Path(cls.temp_dir.name) + cls.db_path = root / "app.db" + cls.storage = root / "files" + + cls._old_db_path = server.DB_PATH + cls._old_storage = server.STORAGE_DIR + server.DB_PATH = cls.db_path + server.STORAGE_DIR = cls.storage + + os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD + connection = connect(cls.db_path) + migrate(connection) + generated = server.ensure_bootstrap_admin(connection) + assert generated is None + connection.close() + + class QuietHandler(server.AppHandler): + def log_message(self, *args) -> None: + pass + + cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler) + cls.port = cls.httpd.server_address[1] + cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) + cls.thread.start() + + cls.admin = Client("127.0.0.1", cls.port) + status, _, data = cls.admin.post_json( + "/api/login", + {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"}, + ) + assert status == 200, data + status, _, data = cls.admin.post_json( + "/api/password/change", + {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD}, + ) + assert status == 200, data + + status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "甲公司"}) + assert status == 200, data + cls.company_a = as_json(data)["company_id"] + status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"}) + assert status == 200, data + cls.company_b = as_json(data)["company_id"] + + cls.cashier_a, cls.cashier_a_password = cls.create_company_user( + cls.admin, cls.port, "cashier-a" + ) + cls.cashier_b, cls.cashier_b_password = cls.create_company_user( + cls.admin, cls.port, "cashier-b" + ) + + @classmethod + def create_company_user(cls, admin, port, username): + status, _, data = admin.post_json( + "/api/admin/users", {"username": username, "company_id": cls.company_a} + ) + assert status == 200, data + initial = as_json(data)["initial_password"] + client = Client("127.0.0.1", port) + status, _, data = client.post_json( + "/api/login", + {"username": username, "password": initial, "portal": "company"}, + ) + assert status == 200, data + new_password = "Changed456" + status, _, data = client.post_json( + "/api/password/change", + {"old_password": initial, "new_password": new_password}, + ) + assert status == 200, data + return client, new_password + + @classmethod + def tearDownClass(cls) -> None: + cls.httpd.shutdown() + cls.httpd.server_close() + server.DB_PATH = cls._old_db_path + server.STORAGE_DIR = cls._old_storage + os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None) + cls.temp_dir.cleanup() + + def fresh_client(self) -> Client: + return Client("127.0.0.1", self.port) + + def upload(self, client, content: bytes, filename: str = "语句.xlsx", fields=None): + return client.post_multipart( + "/api/parse", fields or {}, filename, content + ) + + def upload_for_company(self, content: bytes, company_id: int, filename: str = "语句.xlsx"): + return self.admin.post_multipart( + "/api/parse", {"company_id": str(company_id)}, filename, content + ) + + def upload_for_a(self, content: bytes, filename: str = "语句.xlsx"): + return self.cashier_a.post_multipart("/api/parse", {}, filename, content) + + def batch_sheets(self, client, batch_id: int): + status, _, data = client.get(f"/api/batches/{batch_id}/sheets") + self.assertEqual(200, status, data) + return as_json(data)["sheets"] + + def confirm(self, client, batch_id: int, sheets, reason=None): + payload = {"sheets": sheets} + if reason is not None: + payload["reason"] = reason + return client.post_json(f"/api/batches/{batch_id}/confirm", payload) + + def ignore(self, client, batch_id: int, sheets, reason): + return client.post_json( + f"/api/batches/{batch_id}/ignore", {"sheets": sheets, "reason": reason} + ) + + # ------------------------------------------------------------------ + # Multipart validation + # ------------------------------------------------------------------ + + def test_oversized_body_returns_stable_error(self) -> None: + content = b"x" * (server.MAX_UPLOAD_BYTES + 1) + status, _, data = self.upload_for_a(content) + self.assertEqual(422, status) + payload = as_json(data) + self.assertIn("20 MB", payload["message"]) + self.assertNotIn(server.STORAGE_DIR.as_posix(), payload["message"]) + self.assertNotIn(str(self.storage), payload["message"]) + + def test_missing_boundary_returns_stable_error(self) -> None: + status, _, data = self.cashier_a.request( + "POST", + "/api/parse", + body=b"whatever", + headers={"Content-Type": "multipart/form-data"}, + ) + self.assertEqual(422, status) + self.assertIn("边界", as_json(data)["message"]) + + def test_wrong_content_type_returns_stable_error(self) -> None: + status, _, data = self.cashier_a.request( + "POST", + "/api/parse", + body=CCB_SAMPLE.read_bytes(), + headers={"Content-Type": "application/octet-stream"}, + ) + self.assertEqual(422, status) + + def test_missing_content_length_returns_stable_error(self) -> None: + boundary = "----missingcl" + body = f"--{boundary}--\r\n".encode() + status, _, data = self.cashier_a.request( + "POST", + "/api/parse", + body=body, + headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, + ) + self.assertEqual(422, status) + + def test_bad_extension_returns_stable_error(self) -> None: + status, _, data = self.upload_for_a(CCB_SAMPLE.read_bytes(), filename="statement.pdf") + self.assertEqual(422, status) + self.assertIn(".xls", as_json(data)["message"]) + + def test_signature_mismatch_returns_stable_error(self) -> None: + # Valid .xls content renamed with a .xlsx extension must be rejected. + status, _, data = self.upload_for_a(CCB_SAMPLE.read_bytes(), filename="fake.xlsx") + self.assertEqual(422, status) + self.assertIn("损坏", as_json(data)["message"]) + + def test_corrupt_file_returns_stable_error_without_temp_path(self) -> None: + status, _, data = self.upload_for_a(b"PK\x03\x04 not a real zip at all") + self.assertEqual(422, status) + payload = as_json(data) + text = json.dumps(payload, ensure_ascii=False) + self.assertNotIn(str(self.storage), text) + self.assertNotIn("temp", text.lower()) + self.assertNotIn("data/files", text) + + def test_two_file_parts_rejected(self) -> None: + boundary = "----twofilesboundary" + content = CCB_SAMPLE.read_bytes() + crlf = b"\r\n" + body = ( + b"--" + boundary.encode() + crlf + + b'Content-Disposition: form-data; name="file"; filename="a.xls"' + crlf + + b"Content-Type: application/octet-stream" + crlf + crlf + + content + crlf + + b"--" + boundary.encode() + crlf + + b'Content-Disposition: form-data; name="file"; filename="b.xls"' + crlf + + b"Content-Type: application/octet-stream" + crlf + crlf + + content + crlf + + b"--" + boundary.encode() + b"--" + crlf + ) + status, _, data = self.cashier_a.request( + "POST", + "/api/parse", + body=body, + headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, + ) + self.assertEqual(422, status) + + def test_no_file_part_rejected(self) -> None: + boundary = "----nofile" + crlf = b"\r\n" + body = ( + b"--" + boundary.encode() + crlf + + b'Content-Disposition: form-data; name="company_id"' + crlf + crlf + + b"1" + crlf + + b"--" + boundary.encode() + b"--" + crlf + ) + status, _, data = self.cashier_a.request( + "POST", + "/api/parse", + body=body, + headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, + ) + self.assertEqual(422, status) + self.assertIn("文件", as_json(data)["message"]) + + def test_upload_trailing_bytes_preserved_exactly(self) -> None: + # The stored source file must be byte-identical to what was uploaded + # even when it ends with CRLF/LF; the parser must never trim tail + # bytes, and the upload temp must not appear anywhere in the payload. + content = CCB_SAMPLE.read_bytes() + uploaded = content + b"\r\n\r\n\x00\x01" + status, _, data = self.upload_for_a(uploaded, filename="尾部字节.xls") + self.assertEqual(200, status, data) + batch_id = as_json(data)["batch_id"] + connection = connect(self.db_path) + try: + row = connection.execute( + """ + SELECT f.storage_path FROM source_files f + JOIN import_batches b ON b.source_file_id = f.id + WHERE b.id = ? + """, + (batch_id,), + ).fetchone() + finally: + connection.close() + self.assertIsNotNone(row) + self.assertEqual(uploaded, Path(row["storage_path"]).read_bytes()) + self.assertNotIn("upload-", json.dumps(as_json(data), ensure_ascii=False)) + + # ------------------------------------------------------------------ + # Per-sheet results and four-item evidence + # ------------------------------------------------------------------ + + def test_multi_sheet_workbook_returns_every_sheet_result(self) -> None: + status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="多工作表.xlsx") + self.assertEqual(200, status, data) + payload = as_json(data) + self.assertEqual("parsed", payload["status"]) + self.assertEqual("多工作表.xlsx", payload.get("original_filename")) + sheets = payload["sheets"] + self.assertEqual(3, len(sheets)) + + by_name = {sheet["sheet_name"]: sheet for sheet in sheets} + self.assertEqual("parsed", by_name["正常流水"]["outcome"]) + self.assertEqual("pending", by_name["正常流水"]["review_status"]) + self.assertEqual("中国建设银行", by_name["正常流水"]["bank"]) + self.assertEqual(2, by_name["正常流水"]["transactions"]) + + unknown = by_name["未知模板"] + self.assertEqual("exception", unknown["outcome"]) + self.assertIn("未识别到受支持的银行表头", unknown["message"]) + self.assertGreaterEqual(unknown["scanned_rows"], 2) + self.assertTrue(unknown["candidate_headers"]) + self.assertIn("日期、金额、备注", "；".join(unknown["candidate_headers"])) + + empty = by_name["空表"] + self.assertEqual("ignored", empty["outcome"]) + self.assertIn("为空", empty["message"]) + self.assertEqual(0, empty["scanned_rows"]) + + def test_batch_detail_matches_parse_response(self) -> None: + status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="详情.xlsx") + batch_id = as_json(data)["batch_id"] + sheets = self.batch_sheets(self.cashier_a, batch_id) + self.assertEqual(3, len(sheets)) + status, _, data = self.cashier_a.get(f"/api/batches/{batch_id}/sheets") + self.assertEqual(200, status) + detail = as_json(data) + self.assertEqual("详情.xlsx", detail["original_filename"]) + + def test_unknown_header_file_returns_four_evidence_items(self) -> None: + status, _, data = self.upload_for_a(unknown_header_bytes(), filename="未知银行.xlsx") + self.assertEqual(422, status, data) + payload = as_json(data) + self.assertEqual("exception", payload["status"]) + self.assertEqual("未知银行.xlsx", payload.get("original_filename")) + sheet = payload["sheets"][0] + self.assertEqual("exception", sheet["outcome"]) + self.assertEqual("流水", sheet["sheet_name"]) + self.assertGreaterEqual(sheet["scanned_rows"], 1) + self.assertTrue(sheet["candidate_headers"]) + text = json.dumps(payload, ensure_ascii=False) + for evidence in (str(self.storage), "data/files", ".uploads", "upload-"): + self.assertNotIn(evidence, text) + + def test_empty_workbook_returns_evidence(self) -> None: + status, _, data = self.upload_for_a(empty_workbook_bytes(), filename="空工作簿.xlsx") + self.assertEqual(422, status, data) + payload = as_json(data) + self.assertEqual("空工作簿.xlsx", payload.get("original_filename")) + sheet = payload["sheets"][0] + self.assertEqual("ignored", sheet["outcome"]) + self.assertEqual("Sheet1", sheet["sheet_name"]) + self.assertEqual(0, sheet["scanned_rows"]) + + def test_single_cell_sheet_returns_candidate_evidence(self) -> None: + status, _, data = self.upload_for_a(single_cell_bytes(), filename="单格.xlsx") + self.assertEqual(422, status, data) + sheet = as_json(data)["sheets"][0] + self.assertEqual("exception", sheet["outcome"]) + self.assertEqual("候选", sheet["sheet_name"]) + self.assertEqual(1, sheet["scanned_rows"]) + self.assertTrue(sheet["candidate_headers"]) + self.assertIn("只有一个单元格", "；".join(sheet["candidate_headers"])) + + # ------------------------------------------------------------------ + # Confirm / ignore lifecycle + # ------------------------------------------------------------------ + + def test_confirm_and_ignore_lifecycle(self) -> None: + status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="确认.xlsx") + payload = as_json(data) + batch_id = payload["batch_id"] + names = [s["sheet_name"] for s in payload["sheets"]] + self.assertIn("正常流水", names) + + # Confirm the parsed sheet; exception/ignored cannot be confirmed. + status, _, data = self.confirm(self.cashier_a, batch_id, ["正常流水"]) + self.assertEqual(200, status, data) + result = as_json(data) + self.assertEqual(["正常流水"], result["updated"]) + sheet = next(s for s in result["sheets"] if s["sheet_name"] == "正常流水") + self.assertEqual("confirmed", sheet["review_status"]) + + # Confirming an exception sheet is a conflict. + status, _, data = self.confirm(self.cashier_a, batch_id, ["未知模板"]) + self.assertEqual(409, status, data) + + # Ignore the exception sheet with a reason. + status, _, data = self.ignore(self.cashier_a, batch_id, ["未知模板"], "模板待补充") + self.assertEqual(200, status, data) + sheet = next(s for s in as_json(data)["sheets"] if s["sheet_name"] == "未知模板") + self.assertEqual("ignored", sheet["review_status"]) + self.assertEqual("模板待补充", sheet["review_reason"]) + + # Idempotent repeat of an already-applied decision. + status, _, data = self.confirm(self.cashier_a, batch_id, ["正常流水"]) + self.assertEqual(200, status, data) + result = as_json(data) + self.assertEqual([], result["updated"]) + self.assertEqual(["正常流水"], result["already"]) + + # Changing a settled decision is a conflict. + status, _, data = self.ignore(self.cashier_a, batch_id, ["正常流水"], "改主意") + self.assertEqual(409, status, data) + + def test_ignore_requires_reason(self) -> None: + status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="忽略原因.xlsx") + batch_id = as_json(data)["batch_id"] + status, _, data = self.cashier_a.post_json( + f"/api/batches/{batch_id}/ignore", {"sheets": ["未知模板"], "reason": " "} + ) + self.assertEqual(400, status, data) + self.assertIn("原因", as_json(data)["message"]) + + def test_unknown_sheet_name_is_rejected_and_atomic(self) -> None: + status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="原子.xlsx") + batch_id = as_json(data)["batch_id"] + # One valid + one unknown sheet in the same request must not leave a + # half-applied confirmation behind. + status, _, data = self.confirm(self.cashier_a, batch_id, ["正常流水", "不存在"]) + self.assertEqual(400, status, data) + sheet = next( + s for s in self.batch_sheets(self.cashier_a, batch_id) + if s["sheet_name"] == "正常流水" + ) + self.assertEqual("pending", sheet["review_status"]) + + def test_company_cannot_confirm_another_companys_batch(self) -> None: + status, _, data = self.upload_for_company(multi_sheet_bytes(), self.company_b) + batch_b = as_json(data)["batch_id"] + status, _, data = self.confirm(self.cashier_a, batch_b, ["正常流水"]) + self.assertEqual(404, status, data) + status, _, data = self.cashier_a.get(f"/api/batches/{batch_b}/sheets") + self.assertEqual(404, status, data) + + def test_export_excludes_unconfirmed_and_ignore_keeps_excluded(self) -> None: + def export_line_count() -> int: + status, _, data = self.cashier_a.get("/api/export.csv") + self.assertEqual(200, status) + return len([line for line in data.decode("utf-8-sig").splitlines() if line]) + + baseline = export_line_count() + + # Unconfirmed upload: export stays unchanged (no downstream leak). + status, _, data = self.upload_for_a(CCB_SAMPLE.read_bytes(), filename="导出.xls") + batch_id = as_json(data)["batch_id"] + self.assertEqual("parsed", as_json(data)["status"]) + self.assertEqual(baseline, export_line_count()) + + # Confirm → the sheet's rows enter the export. + sheets = self.batch_sheets(self.cashier_a, batch_id) + name = sheets[0]["sheet_name"] + status, _, data = self.confirm(self.cashier_a, batch_id, [name]) + self.assertEqual(200, status, data) + self.assertGreater(export_line_count(), baseline) + + # A confirmed sheet cannot be silently revoked: the settled decision + # is a conflict, not a silent data removal. + status, _, data = self.ignore(self.cashier_a, batch_id, [name], "取消") + self.assertEqual(409, status, data) + self.assertGreater(export_line_count(), baseline) + + # A sheet that was only ever ignored never enters the export. + status, _, data = self.upload_for_a(multi_sheet_bytes(), filename="忽略.xlsx") + batch2 = as_json(data)["batch_id"] + parsed = [ + sheet for sheet in self.batch_sheets(self.cashier_a, batch2) + if sheet["outcome"] == "parsed" + ] + self.assertTrue(parsed) + before_ignore = export_line_count() + status, _, data = self.ignore( + self.cashier_a, batch2, [parsed[0]["sheet_name"]], "该表复核后不采用" + ) + self.assertEqual(200, status, data) + self.assertEqual(before_ignore, export_line_count()) + + # ------------------------------------------------------------------ + # Concurrency + # ------------------------------------------------------------------ + + def test_concurrent_duplicate_upload_keeps_one_fact_set(self) -> None: + content = CITIC_SAMPLE.read_bytes() + results: list[tuple[int, dict]] = [] + errors: list[Exception] = [] + barrier = threading.Barrier(2) + + def upload() -> None: + try: + client = self.fresh_client() + client.cookies.update(self.cashier_a.cookies) + barrier.wait(timeout=10) + status, _, data = client.post_multipart( + "/api/parse", {}, "并发.xlsx", content + ) + results.append((status, as_json(data))) + except Exception as exc: # pragma: no cover + errors.append(exc) + + threads = [threading.Thread(target=upload) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + self.assertEqual([], errors) + self.assertEqual(2, len(results)) + statuses = sorted(payload["status"] for _, payload in results) + self.assertEqual(["duplicate", "parsed"], statuses) + + connection = connect(self.db_path) + try: + # The concurrent upload adds exactly one source file: identical + # bytes never create a second fact set. + count = connection.execute( + "SELECT COUNT(*) AS n FROM source_files WHERE original_filename = '并发.xlsx'" + ).fetchone()["n"] + self.assertEqual(1, count) + duplicates = connection.execute( + "SELECT COUNT(*) AS n FROM import_batches WHERE status = 'duplicate'" + ).fetchone()["n"] + self.assertGreaterEqual(duplicates, 1) + finally: + connection.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_parser.py b/tests/test_parser.py index f71ae09..a229e48 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -1,10 +1,19 @@ from __future__ import annotations from datetime import date +import io from pathlib import Path +import tempfile import unittest -from bank_importer.parser import UnknownTemplateError, detect_header, parse_directory +from openpyxl import Workbook + +from bank_importer.parser import ( + UnknownTemplateError, + analyze_workbook, + detect_header, + parse_directory, +) ROOT = Path(__file__).resolve().parents[1] @@ -83,5 +92,62 @@ class StatementParserTests(unittest.TestCase): 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() diff --git a/tests/test_persistence.py b/tests/test_persistence.py index 7cd60f4..a061702 100644 --- a/tests/test_persistence.py +++ b/tests/test_persistence.py @@ -41,7 +41,7 @@ class PersistenceTestCase(unittest.TestCase): class MigrationTests(PersistenceTestCase): def test_migrate_creates_schema_and_is_idempotent(self) -> None: first = applied_versions(self.connection) - self.assertEqual([1, 2, 3], first) + self.assertEqual([1, 2, 3, 4], first) self.assertEqual([], migrate(self.connection)) self.assertEqual(first, applied_versions(self.connection)) tables = { @@ -58,6 +58,7 @@ class MigrationTests(PersistenceTestCase): "source_files", "import_batches", "sheet_batches", + "sheet_reviews", "source_rows", "import_exceptions", "users", @@ -69,14 +70,14 @@ class MigrationTests(PersistenceTestCase): self.assertIn(table, tables) def test_rollback_removes_schema_and_forward_rebuilds_it(self) -> None: - self.assertEqual([3, 2, 1], rollback(self.connection, 0)) + self.assertEqual([4, 3, 2, 1], rollback(self.connection, 0)) self.assertEqual([], applied_versions(self.connection)) remaining = self.connection.execute( "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'source_rows'" ).fetchone() self.assertIsNone(remaining) - self.assertEqual([1, 2, 3], migrate(self.connection)) - self.assertEqual([1, 2, 3], applied_versions(self.connection)) + self.assertEqual([1, 2, 3, 4], migrate(self.connection)) + self.assertEqual([1, 2, 3, 4], applied_versions(self.connection)) class ImportPersistenceTests(PersistenceTestCase): diff --git a/tests/test_server_auth.py b/tests/test_server_auth.py index a0c9200..36039fe 100644 --- a/tests/test_server_auth.py +++ b/tests/test_server_auth.py @@ -209,12 +209,30 @@ class ServerAuthMatrixTests(unittest.TestCase): ) assert status == 200, data cls.b_batch_id = as_json(data)["batch_id"] + cls.confirm_all_sheets(cls.admin, cls.b_batch_id) status, _, data = cls.cashier_a.post_multipart( "/api/parse", {}, CITIC_SAMPLE.name, CITIC_SAMPLE.read_bytes() ) assert status == 200, data cls.a_batch_id = as_json(data)["batch_id"] + cls.confirm_all_sheets(cls.cashier_a, cls.a_batch_id) + + @classmethod + def confirm_all_sheets(cls, client, batch_id: int) -> None: + status, _, data = client.get(f"/api/batches/{batch_id}/sheets") + assert status == 200, data + names = [ + sheet["sheet_name"] + for sheet in as_json(data)["sheets"] + if sheet["outcome"] == "parsed" + ] + if not names: + return + status, _, data = client.post_json( + f"/api/batches/{batch_id}/confirm", {"sheets": names} + ) + assert status == 200, data @classmethod def tearDownClass(cls) -> None: @@ -459,10 +477,11 @@ class ServerAuthMatrixTests(unittest.TestCase): payload = as_json(data) self.assertEqual("duplicate", payload["status"]) self.assertEqual(self.a_batch_id, payload["batch_id"]) - self.assertEqual("中信银行", payload.get("bank")) - self.assertTrue(payload.get("transactions", 0) > 0) - self.assertIn("period_start", payload) - self.assertIn("warnings", payload) + self.assertTrue(payload.get("sheets")) + self.assertEqual("中信银行", payload["sheets"][0].get("bank")) + self.assertTrue(payload["sheets"][0].get("transactions", 0) > 0) + self.assertIn("period_start", payload["sheets"][0]) + self.assertIn("warnings", payload["sheets"][0]) # ------------------------------------------------------------------ # Disable / reset flows diff --git a/web/app.js b/web/app.js index a265b18..7590070 100644 --- a/web/app.js +++ b/web/app.js @@ -1151,6 +1151,8 @@ function resetUpload() { $("#uploadForm")?.reset(); if ($("#filePreview")) $("#filePreview").hidden = true; if ($("#parseResult")) $("#parseResult").hidden = true; + if ($("#sheetReview")) $("#sheetReview").hidden = true; + if ($("#sheetList")) $("#sheetList").replaceChildren(); if ($("#dropzone")) $("#dropzone").hidden = false; if ($("#parseButton")) { $("#parseButton").disabled = true; @@ -1203,41 +1205,265 @@ async function parseFile() { result = { status: "error", message: "解析服务暂时不可用，请稍后重试。" }; } state.parseResult = result; - const duplicated = result.status === "duplicate"; - const opaqueDuplicate = duplicated && !result.bank; - const panel = $("#parseResult"); - panel.classList.toggle("is-exception", !parsed); - $("use", panel).setAttribute("href", parsed ? "icons.svg#circle-check" : "icons.svg#circle-alert"); - $("strong", panel).textContent = duplicated ? "文件已导入过" : parsed ? "文件解析完成" : "未识别到银行模板"; - $("#parseSummary").textContent = parsed - ? opaqueDuplicate - ? "相同内容的文件已由其他公司导入，仅记录重复状态，不重复入账。" - : `${result.bank} · 表头第 ${result.header_row} 行 · ${result.transactions} 条明细 · ${result.warnings.length ? `${result.warnings.length} 项提示` : "校验通过"}${duplicated ? " · 重复上传，复用已有批次" : ""}` - : `${result.message} 系统不会猜测模板或自动入账。`; - panel.hidden = false; - $("#parseButton span").textContent = parsed ? "确认导入" : "提交异常"; - $("#parseButton").disabled = false; - $("#parseButton").dataset.stage = "confirm"; + renderParseResult(result, parsed); } -function confirmImport() { - const row = document.createElement("tr"); - row.innerHTML = `IMP-260806-019${$("#accountSelect").value}${state.parseResult?.period_start || "待确认"}—${state.parseResult?.period_end || "待确认"}${state.parseResult?.transactions ?? "—"}待计算已解析刚刚`; - $("#importRows")?.prepend(row); +function renderParseResult(result, parsed) { + const panel = $("#parseResult"); + if (!panel) return; + const sheets = Array.isArray(result.sheets) ? result.sheets : []; + const duplicated = result.status === "duplicate"; + const opaque = duplicated && !sheets.length; + panel.classList.toggle("is-exception", !parsed); + $("use", panel).setAttribute("href", parsed ? "icons.svg#circle-check" : "icons.svg#circle-alert"); + $("#parseTitle").textContent = duplicated ? "文件已导入过" : parsed ? "文件解析完成" : "未识别到银行模板"; + const pendingCount = sheets.filter((s) => s.outcome === "parsed" && s.review_status === "pending").length; + const exceptionCount = sheets.filter((s) => s.outcome === "exception").length; + const ignoredCount = sheets.filter((s) => s.outcome === "ignored" || s.review_status === "ignored").length; + let summary; + if (opaque) { + summary = "相同内容的文件已由其他公司导入，仅记录重复状态，不重复入账。"; + } else if (sheets.length) { + summary = `${sheets.length} 个工作表：${pendingCount} 个待确认${exceptionCount ? `、${exceptionCount} 个异常` : ""}${ignoredCount ? `、${ignoredCount} 个忽略` : ""}。解析成功不等于业务确认。`; + } else { + summary = `${result.message || ""} 系统不会猜测模板或自动入账。`; + } + $("#parseSummary").textContent = summary; + panel.hidden = false; + renderSheetList(sheets, result.batch_id); + + const button = $("#parseButton"); + button.dataset.stage = "confirm"; + if (!parsed) { + button.querySelector("span").textContent = "关闭"; + } else if (opaque) { + button.querySelector("span").textContent = "完成"; + } else if (sheets.length) { + button.querySelector("span").textContent = pendingCount ? `确认全部（${pendingCount} 个待确认）` : "完成"; + } else { + button.querySelector("span").textContent = "完成"; + } +} + +function sheetStatusMeta(sheet) { + if (sheet.review_status === "confirmed") return { className: "success", label: "已确认" }; + if (sheet.review_status === "ignored") return { className: "neutral", label: "已忽略" }; + if (sheet.outcome === "exception") return { className: "danger", label: "异常待处理" }; + if (sheet.outcome === "ignored") return { className: "neutral", label: "空表忽略" }; + return { className: "warning", label: "待确认" }; +} + +function renderSheetList(sheets, batchId) { + const wrap = $("#sheetReview"); + const list = $("#sheetList"); + if (!wrap || !list || !sheets.length) { + if (wrap) wrap.hidden = true; + return; + } + wrap.hidden = false; + list.replaceChildren(...sheets.map((sheet) => buildSheetItem(sheet, batchId))); +} + +function buildSheetItem(sheet, batchId) { + const item = document.createElement("article"); + item.className = "sheet-item"; + if (sheet.review_status === "pending") item.classList.add("is-pending"); + const meta = sheetStatusMeta(sheet); + const head = document.createElement("div"); + head.className = "sheet-item-head"; + const name = document.createElement("strong"); + name.textContent = sheet.sheet_name; + const badge = document.createElement("em"); + badge.className = `status ${meta.className}`; + badge.textContent = meta.label; + head.append(name, badge); + const details = document.createElement("p"); + details.className = "sheet-item-meta"; + if (sheet.outcome === "parsed" && sheet.bank) { + const period = sheet.period_start ? ` · ${sheet.period_start}—${sheet.period_end}` : ""; + details.textContent = `${sheet.bank} · ${sheet.transactions} 条明细${period}`; + } else if (sheet.message) { + details.textContent = sheet.message; + } else { + details.textContent = "空工作表。"; + } + if (sheet.review_reason) { + details.textContent += ` · 原因：${sheet.review_reason}`; + } + const body = document.createElement("div"); + body.append(head, details); + + const actions = document.createElement("div"); + actions.className = "sheet-item-actions"; + if (sheet.review_status === "pending") { + if (sheet.outcome === "parsed") { + const confirmButton = document.createElement("button"); + confirmButton.type = "button"; + confirmButton.className = "text-button"; + confirmButton.textContent = "确认"; + confirmButton.addEventListener("click", () => sheetReviewAction(batchId, sheet.sheet_name, "confirm")); + actions.append(confirmButton); + } + const ignoreButton = document.createElement("button"); + ignoreButton.type = "button"; + ignoreButton.className = "text-button"; + ignoreButton.textContent = "忽略"; + ignoreButton.addEventListener("click", () => sheetReviewAction(batchId, sheet.sheet_name, "ignore")); + actions.append(ignoreButton); + } + item.append(body, actions); + return item; +} + +async function sheetReviewAction(batchId, sheetName, decision) { + const payload = { sheets: [sheetName] }; + if (decision === "ignore") { + const reason = (window.prompt("请填写忽略原因（必填）：", "") || "").trim(); + if (!reason) { + showToast("忽略未提交", "必须填写忽略原因"); + return; + } + payload.reason = reason; + } + const response = await fetch(`/api/batches/${batchId}/${decision}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }).catch(() => null); + if (response?.status === 401 || response?.status === 403) { + window.location.href = "index.html"; + return; + } + const result = await response?.json().catch(() => ({})); + if (!response || !response.ok) { + showToast("操作失败", result?.message || "请稍后重试"); + return; + } + showToast(decision === "confirm" ? "工作表已确认" : "工作表已忽略", `${sheetName}`); + await refreshAfterSheetAction(result, batchId); +} + +async function refreshAfterSheetAction(result, batchId) { + if (Array.isArray(result.sheets)) renderSheetList(result.sheets, batchId); + await loadImportBatches(); + const button = $("#parseButton"); + if (!button) return; + const pending = (result.sheets || []).filter((s) => s.outcome === "parsed" && s.review_status === "pending").length; + if (pending === 0 && (result.sheets || []).length) { + button.querySelector("span").textContent = "完成"; + button.dataset.stage = "done"; + } else { + button.querySelector("span").textContent = `确认全部（${pending} 个待确认）`; + } +} + +async function confirmImport() { + const result = state.parseResult; + const batchId = result?.batch_id; + const sheets = Array.isArray(result?.sheets) ? result.sheets : []; + const pending = sheets + .filter((s) => s.outcome === "parsed" && s.review_status === "pending") + .map((s) => s.sheet_name); + if (!pending.length) { + $("#uploadDialog").close(); + await loadImportBatches(); + showView("upload"); + return; + } + const button = $("#parseButton"); + button.disabled = true; + button.querySelector("span").textContent = "正在确认..."; + const response = await fetch(`/api/batches/${batchId}/confirm`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ sheets: pending }), + }).catch(() => null); + if (response?.status === 401 || response?.status === 403) { + window.location.href = "index.html"; + return; + } + const outcome = await response?.json().catch(() => ({})); + if (!response || !response.ok) { + button.disabled = false; + button.querySelector("span").textContent = "确认失败，点击重试"; + showToast("确认失败", outcome?.message || "请稍后重试"); + return; + } + button.disabled = false; + if (Array.isArray(outcome.sheets)) renderSheetList(outcome.sheets, batchId); + await loadImportBatches(); $("#uploadDialog").close(); showView("upload"); - showToast("流水已进入归集队列", "原始文件和解析结果已保留"); + showToast("流水已确认", `${outcome.updated.length} 个工作表已确认；未确认的工作表不参与计算`); } function submitImportException() { $("#uploadDialog").close(); showView("upload"); - showToast("解析异常已提交", `${state.selectedFile?.name || "该文件"} 未入账，等待总账或模板维护人员处理`); + showToast("解析异常未入账", `${state.selectedFile?.name || "该文件"} 不会进入匹配与计算，请核对模板后重新导出`); +} + +function renderBatchRow(batch) { + const row = document.createElement("tr"); + const idCell = document.createElement("td"); + const id = document.createElement("strong"); + id.textContent = `IMP-${String(batch.id).padStart(6, "0")}`; + const file = document.createElement("small"); + file.textContent = batch.original_filename || ""; + idCell.append(id, file); + + const bank = document.createElement("td"); + bank.textContent = batch.bank_name || "—"; + + const period = document.createElement("td"); + period.textContent = batch.period_start && batch.period_end + ? `${batch.period_start}—${batch.period_end}` + : "—"; + + const count = document.createElement("td"); + count.className = "number"; + count.textContent = `${batch.confirmed_transactions ?? 0} 笔`; + + const coverage = document.createElement("td"); + const coverageStatus = batch.status === "exception" + ? { className: "danger", label: "未导入" } + : batch.pending_sheets > 0 + ? { className: "warning", label: `${batch.pending_sheets} 个待确认` } + : batch.confirmed_sheets > 0 + ? { className: "success", label: "已确认" } + : { className: "neutral", label: "待处理" }; + coverage.innerHTML = `${coverageStatus.label}`; + + const parseState = document.createElement("td"); + const statusParts = []; + if (batch.exception_sheets > 0) statusParts.push(`${batch.exception_sheets} 个异常`); + if (batch.ignored_sheets > 0) statusParts.push(`${batch.ignored_sheets} 个忽略`); + parseState.textContent = statusParts.length ? statusParts.join("、") : "解析成功"; + + const time = document.createElement("td"); + time.textContent = String(batch.created_at || "").slice(0, 16).replace("T", " "); + + row.append(idCell, bank, period, count, coverage, parseState, time); + return row; +} + +async function loadImportBatches() { + const tbody = $("#importRows"); + if (!tbody) return; + const response = await fetch("/api/batches").catch(() => null); + if (response?.status === 401 || response?.status === 403) { + window.location.href = "index.html"; + return; + } + const result = await response?.json().catch(() => ({})); + const batches = Array.isArray(result?.batches) ? result.batches : []; + if (batches.length) tbody.replaceChildren(...batches.map(renderBatchRow)); } function initCompany() { renderCompanyManualRecords(); loadCompanyAccounts(); + loadImportBatches(); const uploadDialog = $("#uploadDialog"); $$('[data-open-upload]').forEach((button) => button.addEventListener("click", () => uploadDialog.showModal())); $$('[data-close-upload]').forEach((button) => button.addEventListener("click", () => uploadDialog.close())); @@ -1260,9 +1486,19 @@ function initCompany() { } $("#uploadForm")?.addEventListener("submit", async (event) => { event.preventDefault(); - if ($("#parseButton").dataset.stage === "confirm") { - if (["parsed", "duplicate"].includes(state.parseResult?.status)) confirmImport(); - else submitImportException(); + const stage = $("#parseButton").dataset.stage; + if (stage === "confirm" || stage === "done") { + const result = state.parseResult; + const parsed = result && ["parsed", "duplicate"].includes(result.status); + if (parsed && stage === "confirm") { + await confirmImport(); + } else if (parsed) { + $("#uploadDialog").close(); + showView("upload"); + await loadImportBatches(); + } else { + submitImportException(); + } return; } $("#parseButton").disabled = true; diff --git a/web/company.html b/web/company.html index 0f75241..8668bb2 100644 --- a/web/company.html +++ b/web/company.html @@ -129,7 +129,7 @@

流水导入

上传 A公司银行账户流水，系统按表头识别银行模板

-
最近导入批次原始文件与解析结果将永久保留
批次银行账户流水期间明细数覆盖状态解析状态上传时间
IMP-260806-018中信银行 · 531607.01—07.31128连续已确认今天 09:42
IMP-260731-012工商银行 · 948106.01—06.3096后续断档已确认07.31 16:18
+
最近导入批次仅已确认工作表参与匹配与计算
批次银行账户流水期间明细数覆盖状态解析状态上传时间
@@ -184,7 +184,7 @@

上传银行流水

A公司 · 系统将识别表头与银行模板

-
+
diff --git a/web/styles.css b/web/styles.css index 8543cb1..4f20d46 100644 --- a/web/styles.css +++ b/web/styles.css @@ -412,6 +412,19 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; } .parse-result p { color: #9bd7b6; } .parse-result.is-exception { border-color: rgba(255, 188, 82, 0.25); background: var(--color-warning-wash); } .parse-result.is-exception p { color: #e7bd78; } +.sheet-review { margin-top: 12px; padding-top: 12px; border-top: 1px solid var(--color-line); } +.sheet-review h3 { font-size: 14px; } +.sheet-review-hint { margin-top: 4px; color: var(--color-ink-muted); font-size: 12px; line-height: 1.5; } +.sheet-list { display: grid; gap: 8px; margin-top: 10px; max-height: 220px; overflow-y: auto; } +.sheet-item { display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: 8px; padding: 10px; border: 1px solid var(--color-line); border-radius: var(--radius-md); background: var(--color-surface-muted); } +.sheet-item.is-pending { border-color: rgba(255, 188, 82, 0.45); } +.sheet-item-head { display: flex; align-items: center; gap: 8px; flex-wrap: wrap; } +.sheet-item-head strong { font-size: 13px; } +.sheet-item-head small { color: var(--color-ink-muted); } +.sheet-item-meta { margin-top: 5px; color: var(--color-ink-muted); font-size: 12px; line-height: 1.5; } +.sheet-item-actions { display: flex; align-items: center; gap: 6px; } +.sheet-item-actions .text-button { font-size: 12px; padding: 4px 8px; } +.sheet-item-reason { margin-top: 6px; padding: 6px 8px; border: 1px solid var(--color-line); border-radius: var(--radius-sm); color: var(--color-ink-muted); font-size: 12px; } .toast-region { position: fixed; right: 20px; bottom: 20px; z-index: 200; display: grid; gap: 8px; } .toast { min-width: 280px; max-width: 390px; padding: 13px 15px; border: 1px solid var(--color-line-strong); border-radius: var(--radius-md); background: rgba(20, 20, 20, 0.96); color: var(--color-ink); box-shadow: var(--shadow-panel); backdrop-filter: blur(20px); } .toast strong, .toast small { display: block; }