fix(review): 调整审查领取超时时间并优化状态处理逻辑或回滚

@@ -29,7 +29,7 @@ DASHBOARD = ROOT / "l2_review_dashboard.html"
 REPORT_DIR.mkdir(parents=True, exist_ok=True)
 REPORT_DIR.mkdir(parents=True, exist_ok=True)
 GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {}
 GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {}
 load_dotenv(ROOT / ".env")
 load_dotenv(ROOT / ".env")
-REVIEW_CLAIM_TTL_SECONDS = max(60, int(os.getenv("REVIEW_CLAIM_TTL_SECONDS", "600")))
+REVIEW_CLAIM_TTL_SECONDS = max(60, int(os.getenv("REVIEW_CLAIM_TTL_SECONDS", "1800")))
 REVIEW_ACCESS_CODE = os.getenv("REVIEW_ACCESS_CODE", "")
 REVIEW_ACCESS_CODE = os.getenv("REVIEW_ACCESS_CODE", "")
 REVIEW_SESSION_TTL_SECONDS = max(300, int(os.getenv("REVIEW_SESSION_TTL_SECONDS", "43200")))
 REVIEW_SESSION_TTL_SECONDS = max(300, int(os.getenv("REVIEW_SESSION_TTL_SECONDS", "43200")))
 REVIEW_SESSIONS: dict[str, tuple[str, float]] = {}
 REVIEW_SESSIONS: dict[str, tuple[str, float]] = {}
@@ -578,7 +578,7 @@ def _claim_staging_review(staging_id: int, reviewer: str, client_token: str) ->
                 review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND),
                 review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND),
                 review_version = review_version + 1
                 review_version = review_version + 1
             WHERE staging_id = %s
             WHERE staging_id = %s
-              AND staging_status NOT IN ('imported', 'deleted')
+              AND staging_status <> 'imported'
               AND (
               AND (
                 biz_review_status IN ('pending', 'unsure')
                 biz_review_status IN ('pending', 'unsure')
                 OR biz_review_status = 'not_required'
                 OR biz_review_status = 'not_required'
@@ -599,8 +599,6 @@ def _claim_staging_review(staging_id: int, reviewer: str, client_token: str) ->
                 message = f"找不到 staging_id={staging_id}，请刷新页面后重试"
                 message = f"找不到 staging_id={staging_id}，请刷新页面后重试"
             elif current.get("staging_status") == "imported":
             elif current.get("staging_status") == "imported":
                 message = "该记录已经入库，只能浏览，不能再领取审核"
                 message = "该记录已经入库，只能浏览，不能再领取审核"
-            elif current.get("staging_status") == "deleted":
-                message = "该记录已被删除，不能领取审核"
             elif current.get("review_claimed_by"):
             elif current.get("review_claimed_by"):
                 message = f"该记录已由 {current['review_claimed_by']} 领取"
                 message = f"该记录已由 {current['review_claimed_by']} 领取"
             else:
             else:

@@ -96,6 +96,27 @@ def test_stale_review_returns_conflict_without_overwrite():
     assert not conn.committed
     assert not conn.committed
 
 
 
 
+def test_pending_review_restores_a_deleted_staging_row():
+    snapshot = {
+        "staging_id": 42,
+        "biz_review_status": "pending",
+        "staging_status": "staged",
+        "review_version": 6,
+    }
+    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
+    conn = FakeConnection(cursor)
+
+    with patch.object(dashboard, "_target_conn", return_value=conn):
+        dashboard._update_staging_review(
+            42, "pending", "", "alice", 5, "alice-token"
+        )
+
+    update_sql, _params = cursor.executions[0]
+    assert "staging_status = CASE WHEN staging_status = 'deleted' THEN 'staged'" in update_sql
+    assert "biz_review_status = 'pending'" in update_sql
+    assert conn.committed
+
+
 def test_claim_conflict_exposes_current_owner():
 def test_claim_conflict_exposes_current_owner():
     snapshot = {
     snapshot = {
         "staging_id": 42,
         "staging_id": 42,
@@ -115,6 +136,38 @@ def test_claim_conflict_exposes_current_owner():
     assert conn.rolled_back
     assert conn.rolled_back
 
 
 
 
+def test_deleted_review_can_be_claimed_for_undo_but_imported_review_cannot():
+    snapshot = {
+        "staging_id": 42,
+        "biz_review_status": "deleted",
+        "staging_status": "deleted",
+        "review_version": 4,
+    }
+    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
+    conn = FakeConnection(cursor)
+
+    with patch.object(dashboard, "_target_conn", return_value=conn):
+        dashboard._claim_staging_review(42, "alice", "alice-token")
+
+    claim_sql, _params = cursor.executions[0]
+    assert "staging_status <> 'imported'" in claim_sql
+    assert "staging_status NOT IN ('imported', 'deleted')" not in claim_sql
+    assert "biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')" in claim_sql
+    assert conn.committed
+
+    imported_cursor = FakeCursor(
+        update_rowcount=0,
+        snapshot={"staging_id": 43, "staging_status": "imported"},
+    )
+    imported_conn = FakeConnection(imported_cursor)
+    with patch.object(dashboard, "_target_conn", return_value=imported_conn):
+        with pytest.raises(dashboard.ReviewConflictError, match="已经入库"):
+            dashboard._claim_staging_review(43, "alice", "alice-token")
+
+    assert imported_conn.rolled_back
+    assert not imported_conn.committed
+
+
 def test_heartbeat_does_not_increment_review_version():
 def test_heartbeat_does_not_increment_review_version():
     snapshot = {"staging_id": 42, "review_version": 5}
     snapshot = {"staging_id": 42, "review_version": 5}
     cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
     cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)