B-43: 双边流水归并与规范事件层——迁移5、匹配引擎、个人过账映射与事件API

src/bank_importer/importing.py

@@ -22,8 +22,10 @@ import os
 from pathlib import Path
 import shutil
 import sqlite3
+import tempfile
 
 from . import auth
+from . import matching
 from .db import utc_now
 from .models import SheetResult, StatementBatch
 from .parser import StatementParseError, analyze_workbook
@@ -52,6 +54,7 @@ def import_statement(
     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(
@@ -64,7 +67,8 @@ def import_statement(
     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
+            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
@@ -87,13 +91,16 @@ def import_statement_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.
+    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)
@@ -109,7 +116,8 @@ def import_statement_path(
     size = stored_path.stat().st_size
     try:
         source_file_id, batch_id = _create_batch_records(
-            connection, sha256, original_filename, size, stored_path, company_id
+            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
@@ -133,6 +141,7 @@ def _create_batch_records(
     size_bytes: int,
     stored_path: Path,
     company_id: int | None,
+    upload_bank_account_id: int | None,
 ) -> tuple[int, int]:
     now = utc_now()
     with connection:
@@ -146,10 +155,10 @@ def _create_batch_records(
         source_file_id = cursor.lastrowid
         cursor = connection.execute(
             """
-            INSERT INTO import_batches (source_file_id, status, company_id, created_at, updated_at)
-            VALUES (?, 'parsing', ?, ?, ?)
+            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, now, now),
+            (source_file_id, company_id, upload_bank_account_id, now, now),
         )
         batch_id = cursor.lastrowid
     return source_file_id, batch_id
@@ -278,17 +287,24 @@ def _store_immutable(
     target_dir = storage_dir / sha256[:2]
     target_dir.mkdir(parents=True, exist_ok=True)
     target = target_dir / f"{sha256}{suffix}"
-    # Publish via a temporary file + hard link: the content-addressed target
-    # either appears complete or not at all, and is never overwritten.
-    temp = target_dir / f".{sha256}.tmp"
-    temp.write_bytes(content)
+    # 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:
-        os.link(temp, target)
-    except FileExistsError:
-        # Content-addressed name means identical bytes; never overwrite.
-        pass
+        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:
-        temp.unlink(missing_ok=True)
+        try:
+            os.unlink(temp_path)
+        except FileNotFoundError:
+            pass
     return target
 
 
@@ -299,17 +315,24 @@ def _publish_immutable(
     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"
+    fd, temp_path = tempfile.mkstemp(prefix=f".{sha256}.", suffix=".tmp", dir=target_dir)
     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)
+        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, target)
+            os.link(temp_path, target)
         except FileExistsError:
             pass
     finally:
-        temp.unlink(missing_ok=True)
+        try:
+            os.unlink(temp_path)
+        except FileNotFoundError:
+            pass
         upload_path.unlink(missing_ok=True)
     return target
 
@@ -572,7 +595,15 @@ def review_sheets(
     now = utc_now()
     updated: list[str] = []
     already: list[str] = []
-    with connection:
+    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"])
@@ -596,6 +627,34 @@ def review_sheets(
             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,
+            )
+        if began:
+            connection.commit()
+    except Exception:
+        if began:
+            connection.rollback()
+        raise
+
     if updated:
         auth.audit(
             connection,
@@ -604,4 +663,7 @@ def review_sheets(
             target=f"batch:{batch_id}",
             detail=f"sheets:{','.join(updated)}" + (f";reason:{reason}" if reason else ""),
         )
-    return {"updated": updated, "already": already}
+    payload: dict[str, object] = {"updated": updated, "already": already}
+    if matching_result is not None:
+        payload["matching"] = matching_result
+    return payload