"""Process service — SSE pipeline: reset → supplements → LLM → apply. Адаптирован из deploy/compare/process.py: callback → generator. """ import json, time from db import supplements, spec_current, spec_events from services.metrics import check_arithmetic def run_pipeline(contract_id, order_ids, build_prompt_fn): """Generator: yield SSE events вместо sse_send callback. Использование: for event in run_pipeline(cid, order_ids, build_prompt): yield f"data: {json.dumps(event)}\\n\\n" """ t0 = time.time() # 0. Reset spec_events.reset(contract_id) # 1. Get supplements with parsed documents supps = supplements.list_by_contract(contract_id) if order_ids: order_list = [x.strip() for x in order_ids.split(",") if x.strip()] order_map = {oid: i for i, oid in enumerate(order_list)} supps.sort(key=lambda s: order_map.get(s["id"], 999999)) else: supps.sort(key=lambda s: (s.get("doc_date") or "9999-99-99", s.get("created_at", ""))) if not supps: yield {"type": "error", "message": "Нет распарсенных файлов"} return from services.llm import call_llm for s in supps: sid = s["id"] filename = s.get("filename", "?") # Current spec cur = spec_current.list_by_contract(contract_id) current_spec = [] for r in cur: current_spec.append({ "hash": r["name_hash"], "name": r["name"], "price": float(r["price"]) if r.get("price") is not None else None, "qty": float(r["qty"]) if r.get("qty") is not None else None, "sum": float(r["sum"]) if r.get("sum") is not None else None, "date_start": r["date_start"], }) # Get elements_json ej = spec_current.get_elements_json(s["document_id"]) if not ej: yield { "type": "extract_error", "supplement_id": sid, "filename": filename, "error": "no elements_json", } continue # Build doc text from elements doc_text = _elements_to_text(ej) yield { "type": "extract_start", "supplement_id": sid, "filename": filename, } # LLM call try: t1 = time.time() result, prompt_id = call_llm(current_spec, doc_text, build_prompt_fn) ops = result.get("ops", []) mode = result.get("mode", "llm") yield { "type": "llm_done", "supplement_id": sid, "filename": filename, "ops_count": len(ops), "mode": mode, "time_s": round(time.time() - t1, 1), } # Apply ops to DB 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": nr = op.get("new_row", {}) spec_events.add_row(contract_id, sid, nr) elif action == "UPDATE": nr = op.get("new_row", {}) nv = op.get("new_values", {}) target = op.get("target_hash", "") spec_events.update_row(contract_id, sid, target, nr, nv) elif action == "DELETE": target = op.get("target_hash", "") spec_events.delete_row(contract_id, sid, target) applied_ops.append(op) except Exception: summary["unresolved"] = summary.get("unresolved", 0) + 1 # Arithmetic check check_arithmetic(contract_id) yield { "type": "applied", "supplement_id": sid, "filename": filename, "summary": summary, "ops": applied_ops, } except Exception as e: yield { "type": "extract_error", "supplement_id": sid, "filename": filename, "error": str(e), } total_time = round(time.time() - t0, 1) yield {"type": "complete", "total_time_s": total_time} def _elements_to_text(ej): """Extract flat text from elements_json for LLM prompt.""" if isinstance(ej, dict) and "Value" in ej: ej = ej["Value"] if isinstance(ej, str): try: ej = json.loads(ej) except (json.JSONDecodeError, TypeError): return ej lines = [] if isinstance(ej, list): for el in ej: if isinstance(el, dict): if el.get("type") == "paragraph": lines.append(el.get("text", "")) elif el.get("type") == "table": for row in el.get("rows", []): lines.append(" | ".join(str(c) for c in row)) elif isinstance(el, str): lines.append(el) return "\n".join(lines)