B-44: 返修——余额目录视觉、八列事件表、抽屉键盘与审核三项决定

Co-authored-by: Cursor <cursoragent@cursor.com>

src/bank_importer/manual_records.py

@@ -683,12 +683,14 @@ def list_records(
         SELECT m.*, d.id AS decision_id, d.state AS state, d.revision AS decision_revision,
                d.action AS action, d.reason AS decision_reason,
                d.actor_username AS decision_actor, d.created_at AS decision_at,
-               c.name AS company_name, cc.name AS counterparty_company_name
+               c.name AS company_name, cc.name AS counterparty_company_name,
+               u.username AS submitted_by_username
         FROM manual_records m
         JOIN current_manual_record_decisions c ON c.record_id = m.id
         JOIN manual_record_decisions d ON d.id = c.decision_id
         LEFT JOIN companies c ON c.id = m.company_id
         LEFT JOIN companies cc ON cc.id = m.counterparty_company_id
+        LEFT JOIN users u ON u.id = m.submitted_by
         {where}
         ORDER BY m.id DESC
         LIMIT ?
@@ -752,6 +754,7 @@ def _record_payload(connection: sqlite3.Connection, record_id: int, *, idempoten
 
 
 def _row_payload(connection: sqlite3.Connection, row: sqlite3.Row) -> dict[str, object]:
+    evidence = json.loads(row["evidence_json"] or "{}") if row["evidence_json"] else {}
     return {
         "id": row["id"],
         "company_id": row["company_id"],
@@ -772,6 +775,8 @@ def _row_payload(connection: sqlite3.Connection, row: sqlite3.Row) -> dict[str,
         "request_key": row["request_key"],
         "supersedes_record_id": row["supersedes_record_id"],
         "submitted_by": row["submitted_by"],
+        "submitted_by_username": row["submitted_by_username"] if "submitted_by_username" in row.keys() else None,
+        "attachment_name": evidence.get("attachment_name"),
         "created_at": row["created_at"],
         "state": row["state"],
         "decision_id": row["decision_id"],

src/bank_importer/positions.py

@@ -832,9 +832,106 @@ def event_payload(
             if int(event["payer_company_id"]) == int(viewer_company_id)
             else event["payer_company_name"]
         )
+    item.update(_event_line_display(connection, event, viewer_company_id))
     return item
 
 
+_REPAY_MARKERS = ("还款", "归还借款", "归还往来款")
+
+
+def bank_short_name(name: str | None) -> str:
+    text = str(name or "").strip()
+    if text.startswith("中国"):
+        text = text[2:]
+    if text.endswith("银行"):
+        text = text[:-2]
+    return text or "银行"
+
+
+def _account_chip(visibility: str, bank_name: str | None, account: str | None) -> dict[str, object]:
+    if visibility == "missing":
+        return {"visibility": "missing", "label": None}
+    if visibility == "masked":
+        return {"visibility": "masked", "label": "按对方授权不可见"}
+    number = str(account or "")
+    tail = number[-4:] if number else ""
+    short = bank_short_name(bank_name)
+    label = f"{short} {tail}".strip() if tail else short
+    return {"visibility": "visible", "label": label}
+
+
+def _event_line_display(
+    connection: sqlite3.Connection,
+    event: sqlite3.Row,
+    viewer_company_id: int | None,
+) -> dict[str, object]:
+    """Account chips, summary and repayment flag for the event table."""
+    missing = _account_chip("missing", None, None)
+    payer_chip, payee_chip = missing, missing
+    summary = None
+    texts: list[str] = []
+    sides = _event_source_sides(connection, event["ledger_event_id"])
+    for side in sides:
+        company_id = side["company_id"]
+        own = viewer_company_id is None or int(company_id) == int(viewer_company_id)
+        visibility = "visible" if own else "masked"
+        chip = _account_chip(visibility, side.get("bank_name"), side.get("own_account"))
+        if int(company_id) == int(event["payer_company_id"]):
+            payer_chip = chip
+        elif int(company_id) == int(event["payee_company_id"]):
+            payee_chip = chip
+        if own or viewer_company_id is None:
+            if side.get("summary"):
+                texts.append(str(side["summary"]))
+            if side.get("reason"):
+                texts.append(str(side["reason"]))
+    if texts:
+        summary = texts[0]
+    blob = " ".join(texts)
+    is_repayment = any(marker in blob for marker in _REPAY_MARKERS)
+    return {
+        "payer_account": payer_chip,
+        "payee_account": payee_chip,
+        "summary": summary,
+        "is_repayment": is_repayment,
+    }
+
+
+def _event_source_sides(
+    connection: sqlite3.Connection, ledger_event_id: int
+) -> list[dict[str, object]]:
+    rows = connection.execute(
+        """
+        SELECT b.company_id AS company_id, s.bank_name AS bank_name,
+               r.own_account AS own_account, r.summary AS summary,
+               r.purpose AS purpose, NULL AS reason
+        FROM ledger_event_bank_sources bs
+        JOIN current_transfer_decisions cur ON cur.event_id = bs.bank_event_id
+        JOIN transfer_decision_observations o ON o.decision_id = cur.decision_id
+        JOIN source_rows r ON r.id = o.source_row_id
+        JOIN sheet_batches s ON s.id = r.sheet_batch_id
+        JOIN import_batches b ON b.id = s.import_batch_id
+        WHERE bs.ledger_event_id = ?
+        """,
+        (ledger_event_id,),
+    ).fetchall()
+    if rows:
+        return [_row(row) for row in rows]
+    manuals = connection.execute(
+        """
+        SELECT m.company_id AS company_id, ba.bank_name AS bank_name,
+               ba.account_number AS own_account, m.summary AS summary,
+               NULL AS purpose, m.reason AS reason
+        FROM ledger_event_manual_sources ms
+        JOIN manual_records m ON m.id = ms.manual_record_id
+        LEFT JOIN bank_accounts ba ON ba.id = m.bank_account_id
+        WHERE ms.ledger_event_id = ?
+        """,
+        (ledger_event_id,),
+    ).fetchall()
+    return [_row(row) for row in manuals]
+
+
 _DETAIL_SELECT = """
     SELECT p.ledger_event_id, p.id AS ledger_revision_id, p.effective_at, p.amount,
            p.amount_scale, p.currency, p.payer_company_id, p.payee_company_id,
@@ -939,7 +1036,7 @@ def subject_review_queue(
         SELECT r.ledger_event_id, r.effective_at, r.amount, r.currency,
                r.payer_company_id, r.payee_company_id,
                cpayer.name AS payer_company_name, cpayee.name AS payee_company_name,
-               r.id AS revision_id
+               r.id AS revision_id, r.evidence_json AS evidence_json
         FROM current_ledger_event_revisions cur
         JOIN ledger_event_revisions r ON r.id = cur.revision_id
         JOIN companies cpayer ON cpayer.id = r.payer_company_id
@@ -957,6 +1054,9 @@ def subject_review_queue(
         cursor_date, cursor_id = None, None
     filtered = []
     for row in rows:
+        evidence = json.loads(row["evidence_json"] or "{}") if row["evidence_json"] else {}
+        if evidence.get("admin_disposition") == "exception":
+            continue
         if cursor_date is not None:
             if row["effective_at"] > cursor_date or (
                 row["effective_at"] == cursor_date and row["ledger_event_id"] >= cursor_id
@@ -982,6 +1082,7 @@ def subject_review_queue(
             """,
             (row["ledger_event_id"],),
         ).fetchall()
+        display = _event_line_display(connection, row, None)
         items.append(
             {
                 "ledger_event_id": row["ledger_event_id"],
@@ -993,6 +1094,8 @@ def subject_review_queue(
                 "payer_company_name": row["payer_company_name"],
                 "payee_company_id": row["payee_company_id"],
                 "payee_company_name": row["payee_company_name"],
+                "summary": display.get("summary"),
+                "is_repayment": display.get("is_repayment"),
                 "suggestions": [
                     {
                         "suggested_perspective_company_id": sug["suggested_perspective_company_id"],

src/bank_importer/subjects.py

@@ -281,6 +281,96 @@ def confirm_subject(
         return _revision_payload(connection, row)
 
 
+def park_subject(
+    connection: sqlite3.Connection,
+    ledger_event_id: int,
+    *,
+    disposition: str,
+    reason: str,
+    expected_revision: int | None,
+    request_key: str | None,
+    actor: sqlite3.Row,
+) -> dict[str, object]:
+    """Record 退回/转异常 without confirming a statutory subject.
+
+    The event stays ``pending_subject`` so it never enters confirmed balances.
+    ``exception`` is hidden from the active review queue; ``return`` remains
+    visible so the company can supplement materials.
+    """
+    from .ledger_events import append_revision, current_revision
+
+    if disposition not in ("return", "exception"):
+        raise SubjectInputError("科目处理只能是退回或转异常。")
+    reason = (reason or "").strip()
+    if not reason:
+        raise SubjectInputError("必须填写处理依据。")
+
+    began = False
+    if not connection.in_transaction:
+        connection.execute("BEGIN IMMEDIATE")
+        began = True
+    try:
+        if request_key:
+            existing = connection.execute(
+                """
+                SELECT * FROM ledger_event_revisions
+                WHERE ledger_event_id = ? AND idempotency_key = ?
+                ORDER BY id LIMIT 1
+                """,
+                (ledger_event_id, request_key),
+            ).fetchone()
+            if existing is not None:
+                if began:
+                    connection.commit()
+                return _revision_payload(connection, existing)
+
+        current = current_revision(connection, ledger_event_id)
+        if current is None:
+            raise SubjectConflictError("该事件不存在或没有当前修订。")
+        if current["state"] != "pending_subject":
+            raise SubjectConflictError("只有待确认科目的事件可以退回或转异常。")
+        if expected_revision is not None and int(expected_revision) not in (
+            current["id"], current["revision"],
+        ):
+            raise SubjectConflictError("事件已发生变更，请刷新后重试。")
+        evidence = json.loads(current["evidence_json"] or "{}") if current["evidence_json"] else {}
+        evidence["admin_disposition"] = disposition
+        revision_id = append_revision(
+            connection,
+            ledger_event_id,
+            state="pending_subject",
+            effective_at=current["effective_at"],
+            amount=current["amount"],
+            currency=current["currency"],
+            payer_company_id=current["payer_company_id"],
+            payee_company_id=current["payee_company_id"],
+            perspective_company_id=None,
+            subject_code=None,
+            source_kind=current["source_kind"],
+            source_revision_token=current["source_revision_token"],
+            posting_kind=current["posting_kind"],
+            reverses_ledger_event_id=current["reverses_ledger_event_id"],
+            adjusts_ledger_event_id=current["adjusts_ledger_event_id"],
+            rule_version=current["rule_version"] or SUBJECT_RULE_VERSION,
+            evidence_json=json.dumps(evidence, ensure_ascii=False),
+            idempotency_key=request_key,
+            actor=actor,
+            reason=reason,
+            supersedes_revision_id=current["id"],
+        )
+        row = connection.execute(
+            "SELECT * FROM ledger_event_revisions WHERE id = ?", (revision_id,)
+        ).fetchone()
+    except Exception:
+        if began:
+            connection.rollback()
+        raise
+    else:
+        if began:
+            connection.commit()
+        return _revision_payload(connection, row)
+
+
 def _revision_payload(connection: sqlite3.Connection, revision: sqlite3.Row) -> dict[str, object]:
     company = connection.execute(
         "SELECT name FROM companies WHERE id = ?", (revision["perspective_company_id"],)