fix: process.py — use apply_ops() instead of phantom add_row/update_row/delete_row

site/services/process.py

@@ -86,27 +86,21 @@ def run_pipeline(contract_id, order_ids, build_prompt_fn):
                 "time_s": round(time.time() - t1, 1),
             }
 
-            # Apply ops to DB
+            # Apply ops to DB via apply_ops
-            applied_ops = []
-            summary = {"added": 0, "updated": 0, "deleted": 0, "unresolved": 0}
-            for op in ops:
-                action = op.get("action", "UNRESOLVED")
-                summary[action.lower()] = summary.get(action.lower(), 0) + 1
             try:
-                    if action == "ADD":
+                summary = spec_events.apply_ops(
-                        nr = op.get("new_row", {})
+                    contract_id, sid, s["document_id"], ops, prompt_id, result
-                        spec_events.add_row(contract_id, sid, nr)
+                )
-                    elif action == "UPDATE":
+                applied_ops = ops
-                        nr = op.get("new_row", {})
+            except Exception as e:
-                        nv = op.get("new_values", {})
+                summary = {"added": 0, "updated": 0, "deleted": 0, "unresolved": len(ops)}
-                        target = op.get("target_hash", "")
+                applied_ops = []
-                        spec_events.update_row(contract_id, sid, target, nr, nv)
+                yield {
-                    elif action == "DELETE":
+                    "type": "extract_error",
-                        target = op.get("target_hash", "")
+                    "supplement_id": sid,
-                        spec_events.delete_row(contract_id, sid, target)
+                    "filename": filename,
-                    applied_ops.append(op)
+                    "error": str(e),
-                except Exception:
+                }
-                    summary["unresolved"] = summary.get("unresolved", 0) + 1
 
             # Arithmetic check
             check_arithmetic(contract_id)