0pt/services/api/orchestrator/main.py

"""
0pt data engine — FastAPI orchestrator (docs/04). Compliant sourcing only.
Runs on a VPS/worker host, NOT on Vercel (long jobs). Amounts in öre.
Reconciles the Next.js /api routes with the Python data engine: the Next layer
is the customer-facing edge; this service runs the actual pipeline.
"""
from __future__ import annotations

import os
import uuid
from pathlib import Path
from typing import Any

import yaml
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

from sources.connectors import opensanctions_screen, rdw_vehicle
from workers.pipeline import run_job

app = FastAPI(title="0pt data engine", version="0.1")
app.add_middleware(
    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)

_SOURCES = yaml.safe_load((Path(__file__).resolve().parent.parent / "sources" / "sources.yaml").read_text(encoding="utf-8"))["sources"]


def _coerce_int(v: Any, default: int) -> int:
    """int(v) raises on decimal-formatted strings/floats; coerce via float() first
    and fall back to `default` on any unparseable, client-controlled input instead
    of a 500."""
    try:
        return int(float(v))
    except (TypeError, ValueError):
        return default

# ── estimate model (mirrors apps/web lib/data.computeEstimate) ───────────────
FIELD_RATE = 240
ENTITY_MULT = {"Bolag": 1.0, "Beslutsfattare": 1.4, "Fordon": 1.2}
FIELD_WEIGHT = {"identity": 1.0, "lei": 0.8, "financials": 1.6, "ubo": 1.4, "contact": 1.8, "intent": 1.5}
ENRICH_WEIGHT = {"dedup": 0.1, "crm": 0.2, "sanctions": 0.15, "nace": 0.1, "tech": 0.15}


class EstimateIn(BaseModel):
    markets: list[str] = []
    entity_type: str = "Bolag"
    fields: list[str] = []
    enrichment: list[str] = []
    volume: int = 1000
    frequency: str = "Engångs"
    confidence_target: int = 92


def compute_estimate(b: EstimateIn) -> dict[str, Any]:
    markets = max(1, len(b.markets))
    fields = set(b.fields) | {"identity"}
    fw = sum(FIELD_WEIGHT.get(f, 0.5) for f in fields) or 1
    em = ENTITY_MULT.get(b.entity_type, 1.0)
    cm = 1 + max(0, b.confidence_target - 90) * 0.03
    ew = sum(ENRICH_WEIGHT.get(e, 0) for e in b.enrichment)
    base = fw * markets * FIELD_RATE * (b.volume / 1000) * em * cm
    price = round(base * (1 + ew) / 5) * 5
    deposit = round(price / 2)
    days = min(14, max(1, round(1 + markets * 0.4 + b.volume / 6000 + len(fields) * 0.25)))
    avg_level = sum(3 if f in ("identity", "lei") else 2 if f in ("financials", "ubo") else 1 for f in fields) / len(fields)
    ceiling = round(82 + avg_level * 5)
    est_conf = min(b.confidence_target, ceiling)
    return {
        "total_ore": price * 100, "deposit_ore": deposit * 100, "balance_ore": (price - deposit) * 100,
        "est_days": days, "est_confidence": est_conf / 100, "recurring": b.frequency != "Engångs",
    }


# ── in-memory store (swapped for asyncpg when DATABASE_URL is set) ───────────
_ORDERS: dict[str, dict[str, Any]] = {}
_JOBS: dict[str, dict[str, Any]] = {}


class OrderIn(BaseModel):
    flow: str = "custom"
    config: dict[str, Any] = {}


@app.get("/health")
async def health():
    return {"ok": True, "service": "0pt-data-engine", "db": bool(os.getenv("DATABASE_URL")), "sources": len(_SOURCES)}


@app.get("/sources")
async def sources():
    return [{k: s.get(k) for k in ("source_id", "name", "market", "access", "licence", "cadence")} for s in _SOURCES]


@app.get("/sources/{market}")
async def sources_for(market: str):
    return [s for s in _SOURCES if s.get("market") in (market, "global")]


@app.get("/vehicles/{plate}")
async def vehicle(plate: str):
    try:
        v = await rdw_vehicle(plate)
    except Exception:
        v = None
    return v or {"vehicle": None}


@app.get("/screen")
async def screen(q: str):
    try:
        return await opensanctions_screen(q)
    except Exception:
        return {"matches": [], "source": "opensanctions"}


@app.post("/estimates")
async def estimates(body: EstimateIn):
    return compute_estimate(body)


@app.post("/orders", status_code=201)
async def create_order(body: OrderIn):
    est = compute_estimate(EstimateIn(
        markets=body.config.get("markets", []),
        entity_type=body.config.get("type", "Bolag"),
        fields=body.config.get("fieldIds", []),
        enrichment=body.config.get("enrichmentIds", []),
        volume=_coerce_int(body.config.get("volume", 1000), 1000),
        frequency=body.config.get("frequency", "Engångs"),
        confidence_target=_coerce_int(body.config.get("confidence", 92), 92),
    ))
    oid = str(uuid.uuid4())
    _ORDERS[oid] = {"order_id": oid, "flow": body.flow, "status": "draft", "config": body.config, **est}
    return {"order_id": oid, "status": "draft", "deposit_ore": est["deposit_ore"], "total_ore": est["total_ore"]}


@app.get("/orders")
async def list_orders():
    return list(_ORDERS.values())


@app.get("/orders/{oid}")
async def get_order(oid: str):
    o = _ORDERS.get(oid)
    if not o:
        raise HTTPException(404, "order saknas")
    return {**o, "job": _JOBS.get(oid)}


class StatusIn(BaseModel):
    status: str


@app.patch("/orders/{oid}")
async def set_status(oid: str, body: StatusIn):
    o = _ORDERS.get(oid)
    if not o:
        raise HTTPException(404, "order saknas")
    o["status"] = body.status
    return o


@app.post("/orders/{oid}/run")
async def run(oid: str):
    o = _ORDERS.get(oid)
    if not o:
        raise HTTPException(404, "order saknas")
    o["status"] = "in_production"
    _JOBS[oid] = await run_job(oid, o["config"])
    o["status"] = "review"
    return _JOBS[oid]


@app.get("/orders/{oid}/status")
async def job_status(oid: str):
    return _JOBS.get(oid) or {"stage": "queued", "pct": 0, "state": "queued"}