contracts-flask/site/services/process.py

"""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")

            if mode == "full_replace" and not ops:
                yield {
                    "type": "extract_error",
                    "supplement_id": sid,
                    "filename": filename,
                    "error": "full_replace returned empty ops",
                }
                continue

            # Трансляция target_id ("r1","r2"...) → target_hash (name_hash из current_spec).
            # LLM возвращает target_id, а apply_ops() читает target_hash — без этого UPDATE/DELETE уходят в UNRESOLVED.
            for _op in ops:
                _tid = _op.get("target_id", "")
                if _tid and isinstance(_tid, str) and _tid.startswith("r"):
                    try:
                        _idx = int(_tid[1:]) - 1
                        if 0 <= _idx < len(current_spec):
                            _op["target_hash"] = current_spec[_idx]["hash"]
                    except ValueError:
                        pass

            yield {
                "type": "llm_done",
                "supplement_id": sid,
                "filename": filename,
                "ops_count": len(ops),
                "mode": mode,
                "time_s": round(time.time() - t1, 1),
            }

            # full_replace: очистить текущее состояние, чтобы ADD-строки новой редакции
            # не дублировали старые (reset на старте пайплайна это не покрывает).
            if mode == "full_replace":
                spec_events.clear_current(contract_id)

            # Apply ops to DB via apply_ops
            try:
                summary = spec_events.apply_ops(
                    contract_id, sid, s["document_id"], ops, prompt_id, result
                )
                applied_ops = ops
            except Exception as e:
                summary = {"added": 0, "updated": 0, "deleted": 0, "unresolved": len(ops)}
                applied_ops = []
                yield {
                    "type": "extract_error",
                    "supplement_id": sid,
                    "filename": filename,
                    "error": str(e),
                }

            # Arithmetic check
            arithmetic_warnings = check_arithmetic(ops)

            yield {
                "type": "applied",
                "supplement_id": sid,
                "filename": filename,
                "summary": summary,
                "ops": applied_ops,
                "arithmetic_warnings": arithmetic_warnings,
            }

        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": "done", "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)