contracts-flask/app/services/process.py

"""Process service — SSE pipeline: reset → supplements → LLM → apply.
Адаптирован из deploy/compare/process.py: callback → generator.
"""
import json, time
from app.db import supplements, spec_current, spec_events
from app.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 app.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)