Add FastAPI endpoint and catalog loader

app/__init__.py

@@ -0,0 +1 @@
+"""Offline product matching service."""

app/__main__.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+import uvicorn
+
+
+def main() -> None:
+    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=False)
+
+
+if __name__ == "__main__":
+    main()

app/catalog.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import csv
+from dataclasses import dataclass
+from decimal import Decimal, InvalidOperation
+from pathlib import Path
+
+
+@dataclass(frozen=True, slots=True)
+class RawCatalogItem:
+    sku: str
+    name: str
+    unit: str
+    price: Decimal
+
+
+def load_raw_catalog(path: Path) -> tuple[RawCatalogItem, ...]:
+    if not path.is_file():
+        raise FileNotFoundError(f"Catalog file does not exist: {path}")
+
+    items: list[RawCatalogItem] = []
+    with path.open("r", encoding="utf-8-sig", newline="") as file:
+        reader = csv.DictReader(file)
+        expected = {"sku", "name", "unit", "price"}
+        if set(reader.fieldnames or ()) != expected:
+            raise ValueError(
+                f"Unexpected catalog columns: {reader.fieldnames!r}; expected {sorted(expected)!r}"
+            )
+
+        seen_skus: set[str] = set()
+        for line_number, row in enumerate(reader, start=2):
+            sku = (row.get("sku") or "").strip()
+            name = (row.get("name") or "").strip()
+            unit = (row.get("unit") or "").strip()
+            price_text = (row.get("price") or "").strip()
+
+            if not sku or not name or not unit or not price_text:
+                raise ValueError(f"Catalog line {line_number} contains an empty required field")
+            if sku in seen_skus:
+                raise ValueError(f"Duplicate SKU at catalog line {line_number}: {sku}")
+
+            try:
+                price = Decimal(price_text)
+            except InvalidOperation as exc:
+                raise ValueError(
+                    f"Invalid price at catalog line {line_number}: {price_text!r}"
+                ) from exc
+
+            seen_skus.add(sku)
+            items.append(RawCatalogItem(sku=sku, name=name, unit=unit, price=price))
+
+    if not items:
+        raise ValueError("Catalog is empty")
+    return tuple(items)

app/main.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from fastapi import FastAPI
+
+from app.catalog import load_raw_catalog
+from app.models import MatchRequest, MatchResponse, MatchResult
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+CATALOG_PATH = Path(os.environ.get("CATALOG_PATH", BASE_DIR / "catalog_excel.csv"))
+RAW_CATALOG = load_raw_catalog(CATALOG_PATH)
+
+app = FastAPI(title="Offline catalog matcher", version="0.1.0")
+
+
+@app.get("/health")
+def health() -> dict[str, object]:
+    return {"status": "ok", "catalog_items": len(RAW_CATALOG)}
+
+
+@app.post("/match", response_model=MatchResponse)
+def match(request: MatchRequest) -> MatchResponse:
+    return MatchResponse(
+        results=[
+            MatchResult(message=message, status="not_found", candidates=[])
+            for message in request.messages
+        ]
+    )

app/models.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class MatchRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    messages: list[str] = Field(default_factory=list)
+
+
+class Candidate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    sku: str
+    confidence: float = Field(ge=0.0, le=1.0)
+
+
+class MatchResult(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    message: str
+    status: Literal["matched", "ambiguous", "not_found"]
+    candidates: list[Candidate]
+
+
+class MatchResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    results: list[MatchResult]