9193a3fce0
同类未提交即 close 回滚、业务与审计拆成两笔事务的路径一并收进可嵌套事务边界。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
672 lines
24 KiB
Python
672 lines
24 KiB
Python
"""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
|
||
|
||
from dataclasses import dataclass
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import shutil
|
||
import sqlite3
|
||
import tempfile
|
||
|
||
from . import auth
|
||
from . import ledger_events
|
||
from . import matching
|
||
from .db import utc_now
|
||
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)
|
||
class ImportResult:
|
||
batch_id: int
|
||
status: str # parsed | duplicate | exception
|
||
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'
|
||
|
||
|
||
def import_statement(
|
||
connection: sqlite3.Connection,
|
||
storage_dir: str | Path,
|
||
original_filename: str,
|
||
content: bytes,
|
||
company_id: int | None = None,
|
||
upload_bank_account_id: int | None = None,
|
||
) -> ImportResult:
|
||
sha256 = hashlib.sha256(content).hexdigest()
|
||
existing_file = connection.execute(
|
||
"SELECT id FROM source_files WHERE sha256 = ?", (sha256,)
|
||
).fetchone()
|
||
|
||
if existing_file is not None:
|
||
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, upload_bank_account_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,
|
||
upload_bank_account_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. The approved ``upload_bank_account_id`` used at upload time
|
||
is persisted on the batch so ownership can be resolved later even when the
|
||
source rows carry no own account.
|
||
"""
|
||
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, upload_bank_account_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,
|
||
upload_bank_account_id: int | None,
|
||
) -> tuple[int, int]:
|
||
now = utc_now()
|
||
with connection:
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(sha256, original_filename, size_bytes, str(stored_path), now),
|
||
)
|
||
source_file_id = cursor.lastrowid
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO import_batches (source_file_id, status, company_id, upload_bank_account_id, created_at, updated_at)
|
||
VALUES (?, 'parsing', ?, ?, ?, ?)
|
||
""",
|
||
(source_file_id, company_id, upload_bank_account_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:
|
||
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,
|
||
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,
|
||
diagnostics={"original_filename": original_filename},
|
||
)
|
||
_set_batch_status(connection, batch_id, "failed")
|
||
raise
|
||
|
||
with connection:
|
||
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=tuple(parsed_batches), sheets=sheets,
|
||
)
|
||
|
||
|
||
def _record_duplicate(
|
||
connection: sqlite3.Connection,
|
||
source_file_id: int,
|
||
sha256: str,
|
||
company_id: int | None = None,
|
||
) -> ImportResult:
|
||
original = connection.execute(
|
||
"""
|
||
SELECT id, company_id FROM import_batches
|
||
WHERE source_file_id = ? AND status = 'parsed'
|
||
ORDER BY id LIMIT 1
|
||
""",
|
||
(source_file_id,),
|
||
).fetchone()
|
||
if original is None:
|
||
original = connection.execute(
|
||
"""
|
||
SELECT id, company_id FROM import_batches
|
||
WHERE source_file_id = ? AND status != 'duplicate'
|
||
ORDER BY id LIMIT 1
|
||
""",
|
||
(source_file_id,),
|
||
).fetchone()
|
||
now = utc_now()
|
||
with connection:
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO import_batches
|
||
(source_file_id, status, duplicate_of_id, company_id, diagnostics, created_at, updated_at)
|
||
VALUES (?, 'duplicate', ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
source_file_id,
|
||
original["id"],
|
||
company_id,
|
||
json.dumps({"note": "内容哈希相同，复用已有批次，不产生第二份事实。"}, ensure_ascii=False),
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
# A duplicate is only "same company" when the uploader belongs to the
|
||
# same tenant that owns the original batch. Cross-company duplicates
|
||
# must stay opaque: the caller gets the new duplicate batch id (which
|
||
# belongs to its own company) but never the other company's batch id.
|
||
same_company = original["company_id"] == company_id
|
||
if same_company:
|
||
batch_id = original["id"]
|
||
else:
|
||
batch_id = int(cursor.lastrowid)
|
||
return ImportResult(
|
||
batch_id,
|
||
"duplicate",
|
||
sha256,
|
||
source_file_id,
|
||
message="相同内容的文件已导入，本次按重复上传处理。",
|
||
duplicate_same_company=same_company,
|
||
)
|
||
|
||
|
||
def _store_immutable(
|
||
storage_dir: Path, original_filename: str, content: bytes, 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}"
|
||
# Publish via a unique temporary file + hard link: the content-addressed
|
||
# target either appears complete or not at all, and is never overwritten.
|
||
# The temp name is unique per writer, so concurrent uploads of the same
|
||
# bytes cannot corrupt each other's staging file (B-43 baseline fix).
|
||
fd, temp_path = tempfile.mkstemp(prefix=f".{sha256}.", suffix=".tmp", dir=target_dir)
|
||
try:
|
||
with os.fdopen(fd, "wb") as handle:
|
||
handle.write(content)
|
||
try:
|
||
os.link(temp_path, target)
|
||
except FileExistsError:
|
||
# Content-addressed name means identical bytes; never overwrite.
|
||
pass
|
||
finally:
|
||
try:
|
||
os.unlink(temp_path)
|
||
except FileNotFoundError:
|
||
pass
|
||
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}"
|
||
fd, temp_path = tempfile.mkstemp(prefix=f".{sha256}.", suffix=".tmp", dir=target_dir)
|
||
try:
|
||
with os.fdopen(fd, "wb") as handle:
|
||
# Copy keeps the same-filesystem guarantee even if the upload temp
|
||
# lives elsewhere; the hard link then publishes atomically. The
|
||
# staging file is unique per writer, so two concurrent uploads of
|
||
# the same bytes never corrupt each other's copy.
|
||
with upload_path.open("rb") as source:
|
||
shutil.copyfileobj(source, handle, length=1024 * 1024)
|
||
try:
|
||
os.link(temp_path, target)
|
||
except FileExistsError:
|
||
pass
|
||
finally:
|
||
try:
|
||
os.unlink(temp_path)
|
||
except FileNotFoundError:
|
||
pass
|
||
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
|
||
) -> int:
|
||
now = utc_now()
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO sheet_batches (
|
||
import_batch_id, sheet_name, bank_name, template_id, template_version,
|
||
header_row, own_account, own_name, period_start, period_end,
|
||
transaction_count, warnings, created_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
batch_id,
|
||
batch.sheet_name,
|
||
batch.bank_name,
|
||
batch.template_id,
|
||
batch.template_version,
|
||
batch.header_row,
|
||
batch.own_account,
|
||
batch.own_name,
|
||
batch.period_start.isoformat() if batch.period_start else None,
|
||
batch.period_end.isoformat() if batch.period_end else None,
|
||
len(batch.transactions),
|
||
json.dumps(list(batch.warnings), ensure_ascii=False),
|
||
now,
|
||
),
|
||
)
|
||
sheet_batch_id = cursor.lastrowid
|
||
for transaction in batch.transactions:
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO source_rows (
|
||
sheet_batch_id, source_row, transaction_at, income, expense, balance,
|
||
own_account, own_name, counterparty_account, counterparty_name,
|
||
counterparty_bank, summary, purpose, reference, currency, created_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
sheet_batch_id,
|
||
transaction.source_row,
|
||
transaction.transaction_at.isoformat(),
|
||
str(transaction.income),
|
||
str(transaction.expense),
|
||
str(transaction.balance) if transaction.balance is not None else None,
|
||
transaction.own_account,
|
||
transaction.own_name,
|
||
transaction.counterparty_account,
|
||
transaction.counterparty_name,
|
||
transaction.counterparty_bank,
|
||
transaction.summary,
|
||
transaction.purpose,
|
||
transaction.reference,
|
||
transaction.currency,
|
||
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(
|
||
connection: sqlite3.Connection,
|
||
batch_id: int,
|
||
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)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
batch_id,
|
||
stage,
|
||
message,
|
||
json.dumps(payload, ensure_ascii=False),
|
||
utc_now(),
|
||
),
|
||
)
|
||
|
||
|
||
def _set_batch_status(connection: sqlite3.Connection, batch_id: int, status: str) -> None:
|
||
connection.execute(
|
||
"UPDATE import_batches SET status = ?, updated_at = ? WHERE id = ?",
|
||
(status, utc_now(), batch_id),
|
||
)
|
||
|
||
|
||
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] = []
|
||
matching_result: dict[str, object] | None = None
|
||
# BEGIN IMMEDIATE serializes concurrent confirmations: the write lock is
|
||
# taken before the read, so two cashiers confirming matching worksheets can
|
||
# never derive their match decisions from stale snapshots.
|
||
began = False
|
||
if not connection.in_transaction:
|
||
connection.execute("BEGIN IMMEDIATE")
|
||
began = True
|
||
try:
|
||
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"])
|
||
|
||
# Confirming a worksheet and reconciling its rows must succeed or fail
|
||
# together: any matching failure rolls the confirmation back, so there
|
||
# is never a "confirmed but unmatched" half state.
|
||
if updated and decision == "confirm":
|
||
placeholders = ",".join("?" for _ in updated)
|
||
confirmed_rows = connection.execute(
|
||
f"""
|
||
SELECT r.id 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 rv.sheet_name IN ({placeholders})
|
||
AND rv.import_batch_id = ?
|
||
ORDER BY r.id
|
||
""",
|
||
(*updated, batch_id),
|
||
).fetchall()
|
||
matching_result = matching.reconcile_rows(
|
||
connection,
|
||
[item["id"] for item in confirmed_rows],
|
||
actor=actor,
|
||
)
|
||
ledger_events.reconcile_bank_events(connection, actor=actor)
|
||
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 ""),
|
||
)
|
||
if began:
|
||
connection.commit()
|
||
except Exception:
|
||
if began:
|
||
connection.rollback()
|
||
raise
|
||
|
||
payload: dict[str, object] = {"updated": updated, "already": already}
|
||
if matching_result is not None:
|
||
payload["matching"] = matching_result
|
||
return payload
|