diff --git a/app/api/screen/route.ts b/app/api/screen/route.ts new file mode 100644 index 0000000..3dac028 --- /dev/null +++ b/app/api/screen/route.ts @@ -0,0 +1,27 @@ +import { NextRequest, NextResponse } from "next/server"; +import { engineFetch } from "@/lib/engine"; + +// GET /api/screen?q=Name — OpenSanctions screening (sanctions/PEP). VERIFY +// commercial terms before production volume. Defers to engine when running. +export async function GET(req: NextRequest) { + const q = req.nextUrl.searchParams.get("q") ?? ""; + if (!q) return NextResponse.json({ matches: [] }); + + const eng = await engineFetch(`/screen?q=${encodeURIComponent(q)}`); + if (eng && eng.ok) return NextResponse.json(await eng.json()); + + try { + const r = await fetch(`https://api.opensanctions.org/search/default?q=${encodeURIComponent(q)}&limit=5`, { + headers: { "User-Agent": "0pt/0.1 (+https://0pt.app; SIAX)" }, + cache: "no-store", + }); + if (!r.ok) return NextResponse.json({ matches: [], source: "opensanctions" }); + const data = (await r.json()) as { results?: { caption?: string; score?: number; schema?: string }[] }; + return NextResponse.json({ + source: "opensanctions", + matches: (data.results ?? []).map((x) => ({ name: x.caption ?? null, score: x.score ?? null, schema: x.schema ?? null })), + }); + } catch { + return NextResponse.json({ matches: [], source: "opensanctions" }); + } +} diff --git a/app/api/vehicles/route.ts b/app/api/vehicles/route.ts new file mode 100644 index 0000000..ee0dc1a --- /dev/null +++ b/app/api/vehicles/route.ts @@ -0,0 +1,34 @@ +import { NextRequest, NextResponse } from "next/server"; +import { engineFetch } from "@/lib/engine"; + +// GET /api/vehicles?plate=XX-99-YY — RDW (NL) open vehicle data (technical/fleet +// only, no owner PII). Defers to the engine when running; else calls RDW directly. +export async function GET(req: NextRequest) { + const plate = (req.nextUrl.searchParams.get("plate") ?? "").toUpperCase().replace(/[-\s]/g, ""); + if (!plate) return NextResponse.json({ error: { code: "bad_request", message: "plate krävs" } }, { status: 400 }); + + const eng = await engineFetch(`/vehicles/${plate}`); + if (eng && eng.ok) return NextResponse.json(await eng.json()); + + try { + const r = await fetch(`https://opendata.rdw.nl/resource/m9d7-ebf2.json?kenteken=${plate}`, { + headers: { "User-Agent": "0pt/0.1 (+https://0pt.app; SIAX)" }, + cache: "no-store", + }); + if (!r.ok) return NextResponse.json({ vehicle: null }, { status: 200 }); + const arr = (await r.json()) as Record[]; + if (!arr.length) return NextResponse.json({ vehicle: null }); + const d = arr[0]; + return NextResponse.json({ + source: "rdw", + vehicle: { + reg_plate: plate, market: "NL", + make: d.merk ?? null, model: d.handelsbenaming ?? null, + year: (d.datum_eerste_toelating ?? "").slice(0, 4) || null, + }, + provenance: { level: 3, source: "rdw" }, + }); + } catch { + return NextResponse.json({ vehicle: null, error: "fetch_failed" }, { status: 200 }); + } +} diff --git a/services/api/orchestrator/__pycache__/main.cpython-314.pyc b/services/api/orchestrator/__pycache__/main.cpython-314.pyc index 307123e..ef0a4a7 100644 Binary files a/services/api/orchestrator/__pycache__/main.cpython-314.pyc and b/services/api/orchestrator/__pycache__/main.cpython-314.pyc differ diff --git a/services/api/orchestrator/main.py b/services/api/orchestrator/main.py index e4f77f1..321941e 100644 --- a/services/api/orchestrator/main.py +++ b/services/api/orchestrator/main.py @@ -16,6 +16,7 @@ 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") @@ -87,6 +88,17 @@ 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): + v = await rdw_vehicle(plate) + return v or {"vehicle": None} + + +@app.get("/screen") +async def screen(q: str): + return await opensanctions_screen(q) + + @app.post("/estimates") async def estimates(body: EstimateIn): return compute_estimate(body) diff --git a/services/api/sources/__pycache__/financials.cpython-314.pyc b/services/api/sources/__pycache__/financials.cpython-314.pyc new file mode 100644 index 0000000..908852f Binary files /dev/null and b/services/api/sources/__pycache__/financials.cpython-314.pyc differ diff --git a/services/api/sources/financials.py b/services/api/sources/financials.py new file mode 100644 index 0000000..188ee3f --- /dev/null +++ b/services/api/sources/financials.py @@ -0,0 +1,34 @@ +""" +Structured financials (docs/06). XBRL/iXBRL is the fast, exact lane (not OCR). +Companies House publishes accounts as iXBRL; this fetches the accounts filing +history (real, key-gated) and documents the XBRL extraction step. +""" +from __future__ import annotations + +import os +from typing import Any + +import httpx + +USER_AGENT = "0pt-data-engine/0.1 (+https://0pt.app; SIAX)" + + +async def companies_house_filings(number: str) -> dict[str, Any]: + key = os.getenv("COMPANIES_HOUSE_API_KEY") + if not key: + return {"company": number, "filings": [], "note": "level0: kräver COMPANIES_HOUSE_API_KEY"} + url = f"https://api.company-information.service.gov.uk/company/{number}/filing-history?category=accounts" + async with httpx.AsyncClient(headers={"User-Agent": USER_AGENT}, timeout=15) as c: + r = await c.get(url, auth=(key, "")) + if r.status_code != 200: + return {"company": number, "filings": []} + items = r.json().get("items", []) + return { + "company": number, + "source": "companies_house", + "filings": [{"date": i.get("date"), "type": i.get("type"), "description": i.get("description")} for i in items[:10]], + } + +# XBRL extraction (extension): with `arelle` installed, download each accounts +# filing's iXBRL document and pull tagged facts (turnover, profit, employees) +# into fact_financials with provenance. Structured, exact — never OCR for XBRL. diff --git a/services/api/workers/__pycache__/pipeline.cpython-314.pyc b/services/api/workers/__pycache__/pipeline.cpython-314.pyc index 77e79ea..30cbb05 100644 Binary files a/services/api/workers/__pycache__/pipeline.cpython-314.pyc and b/services/api/workers/__pycache__/pipeline.cpython-314.pyc differ diff --git a/services/api/workers/__pycache__/queue.cpython-314.pyc b/services/api/workers/__pycache__/queue.cpython-314.pyc new file mode 100644 index 0000000..55557b1 Binary files /dev/null and b/services/api/workers/__pycache__/queue.cpython-314.pyc differ diff --git a/services/api/workers/__pycache__/resolve.cpython-314.pyc b/services/api/workers/__pycache__/resolve.cpython-314.pyc new file mode 100644 index 0000000..1996864 Binary files /dev/null and b/services/api/workers/__pycache__/resolve.cpython-314.pyc differ diff --git a/services/api/workers/pipeline.py b/services/api/workers/pipeline.py index 4e279e9..b30ebad 100644 --- a/services/api/workers/pipeline.py +++ b/services/api/workers/pipeline.py @@ -7,7 +7,7 @@ from __future__ import annotations from typing import Any -from rapidfuzz import fuzz +from workers.resolve import resolve from sources.connectors import ( CompanyResult, @@ -70,14 +70,8 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]: break # first compliant source that answers wins per_market[market] = got - # 3) structuring — dedup by fuzzy name within market - deduped: list[dict[str, Any]] = [] - for r in records: - dup = next((d for d in deduped if d["market"] == r["market"] and fuzz.ratio(d["name"], r["name"]) > 92), None) - if dup: - dup["confidence"] = min(0.99, dup["confidence"] + 0.1) # corroboration bump - else: - deduped.append(r) + # 3) structuring — entity resolution (blocking + fuzzy match + corroboration) + deduped = resolve(records) # 4) enrichment/fusion already folded into confidence above. # 5) QA — drop rows below the order's confidence threshold. diff --git a/services/api/workers/queue.py b/services/api/workers/queue.py new file mode 100644 index 0000000..8aa79fe --- /dev/null +++ b/services/api/workers/queue.py @@ -0,0 +1,31 @@ +""" +Lightweight in-process job queue (docs/02). Adequate for single-worker dev; swap +the backend for Redis/Celery at scale without changing callers. +""" +from __future__ import annotations + +import asyncio +from typing import Any, Awaitable, Callable + +Job = Callable[..., Awaitable[Any]] + + +class JobQueue: + def __init__(self) -> None: + self._q: "asyncio.Queue[tuple[Job, tuple[Any, ...]]]" = asyncio.Queue() + + async def enqueue(self, fn: Job, *args: Any) -> None: + await self._q.put((fn, args)) + + async def worker(self) -> None: + while True: + fn, args = await self._q.get() + try: + await fn(*args) + except Exception as e: # keep the worker alive on a failed job + print("job failed:", repr(e)) + finally: + self._q.task_done() + + +queue = JobQueue() diff --git a/services/api/workers/resolve.py b/services/api/workers/resolve.py new file mode 100644 index 0000000..92330fd --- /dev/null +++ b/services/api/workers/resolve.py @@ -0,0 +1,33 @@ +""" +Entity resolution (docs/06): block, score, and link records across sources so the +same company from GLEIF + Brreg + CVR collapses to one record with corroborated +confidence. rapidfuzz only; deterministic; no external calls. +""" +from __future__ import annotations + +from typing import Any + +from rapidfuzz import fuzz + + +def block_key(rec: dict[str, Any]) -> tuple[str, str]: + """Cheap blocking key — same market + first initial — to avoid O(n^2).""" + return (rec.get("market") or "", (rec.get("name") or "")[:1].upper()) + + +def resolve(records: list[dict[str, Any]], threshold: int = 92) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for r in records: + match = None + for o in out: + if block_key(o) == block_key(r) and fuzz.token_sort_ratio(o.get("name", ""), r.get("name", "")) >= threshold: + match = o + break + if match: + # corroboration: independent source agreeing bumps confidence + level + match["confidence"] = min(0.99, match.get("confidence", 0.0) + 0.1) + match["agreed_sources"] = match.get("agreed_sources", 1) + 1 + match.setdefault("merged_from", []).append(r.get("source")) + else: + out.append({**r, "agreed_sources": 1}) + return out