""" LandveX Fas 1 — Object API (förenklad) 3 endpoints: 1. GET /v0/objects/{lvx_id} — hämta objekt från SQLite 2. GET /v0/search?q=... — enkel fritextsökning 3. POST /v0/feedback — ta emot feedback (spara till fil) """ import os import json import sqlite3 from datetime import datetime from typing import Optional from fastapi import FastAPI, Query, Depends from fastapi.responses import JSONResponse from pydantic import BaseModel from auth import verify_api_key from rate_limit import check_rate_limit # ── Config ────────────────────────────────────────────────────────── DB_PATH = os.environ.get( "LANDVEX_DB", "/home/bernt/.openclaw/workspace/landvex-paket/landvex.db", ) FEEDBACK_DIR = os.environ.get( "LANDVEX_FEEDBACK_DIR", "/home/bernt/.openclaw/workspace/landvex-paket/api/feedback", ) os.makedirs(FEEDBACK_DIR, exist_ok=True) # ── FastAPI app ───────────────────────────────────────────────────── app = FastAPI( title="LandveX Object API", version="0.1.0", description="Fas 1 — förenklad Object API över landvex.db", ) # ── Helpers ───────────────────────────────────────────────────────── def _db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def _row_to_dict(row: sqlite3.Row) -> dict: return {key: row[key] for key in row.keys()} # ── Endpoints ─────────────────────────────────────────────────────── @app.get("/v0/objects/{lvx_id}") def get_object(lvx_id: str, auth=Depends(verify_api_key), rate=Depends(check_rate_limit)): """Hämta ett objekt (klass eller modell) via dess LVX-ID.""" conn = _db() cursor = conn.cursor() # Sök i klasser först cursor.execute("SELECT * FROM objects WHERE lvx_id = ?", (lvx_id,)) row = cursor.fetchone() if row: conn.close() data = _row_to_dict(row) data["typ"] = "klass" return {"status": "ok", "data": data} # Sök i modeller cursor.execute("SELECT * FROM objects WHERE lvx_id = ?", (lvx_id,)) row = cursor.fetchone() if row: conn.close() data = _row_to_dict(row) data["typ"] = "modell" return {"status": "ok", "data": data} conn.close() return JSONResponse( status_code=404, content={"status": "error", "message": f"Objekt '{lvx_id}' hittades inte."}, ) @app.get("/v0/search") def search(q: str = Query(..., min_length=1, description="Sökfras"), auth=Depends(verify_api_key), rate=Depends(check_rate_limit)): """Enkel fritextsökning över namn, slug och LVX-ID i klasser och modeller.""" conn = _db() cursor = conn.cursor() pattern = f"%{q}%" results = [] # Sök i klasser cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, tier, verifieringsniva, konfidens FROM objects WHERE posttyp = 'objektklass' WHERE lvx_id LIKE ? OR slug LIKE ? OR namn_sv LIKE ? OR namn_en LIKE ? """, (pattern, pattern, pattern, pattern), ) for row in cursor.fetchall(): d = _row_to_dict(row) d["typ"] = "klass" results.append(d) # Sök i modeller cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, parent, verifieringsniva, konfidens FROM objects WHERE posttyp = 'produktmodell' WHERE lvx_id LIKE ? OR slug LIKE ? OR namn_sv LIKE ? OR namn_en LIKE ? """, (pattern, pattern, pattern, pattern), ) for row in cursor.fetchall(): d = _row_to_dict(row) d["typ"] = "modell" results.append(d) conn.close() return { "status": "ok", "query": q, "count": len(results), "results": results, } # ── Identify ──────────────────────────────────────────────────────── class IdentifyPayload(BaseModel): bild: Optional[str] = None # base64 eller URL — stubb för nu position: Optional[dict] = None ocr_extraherat: list[str] = [] @app.post("/v0/identify") def identify(payload: IdentifyPayload, auth=Depends(verify_api_key), rate=Depends(check_rate_limit)): """Identifiera objekt från OCR-text. Förenklad version — matchar mot databasen.""" conn = _db() cursor = conn.cursor() kandidater = [] bounty = None # Om OCR finns — sök på nyckelord if payload.ocr_extraherat: for term in payload.ocr_extraherat: pattern = f"%{term}%" # Sök i modeller först (högre precision) cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, parent, verifieringsniva, konfidens FROM objects WHERE posttyp = 'produktmodell' WHERE lvx_id LIKE ? OR slug LIKE ? OR namn_sv LIKE ? OR namn_en LIKE ? """, (pattern, pattern, pattern, pattern), ) for row in cursor.fetchall(): d = _row_to_dict(row) d["niva"] = "produktmodell" d["konfidens"] = min(0.95, 0.7 + len(term) * 0.05) # Förenklad konfidens if not any(k["lvx_id"] == d["lvx_id"] for k in kandidater): kandidater.append(d) # Sök i klasser cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, tier, verifieringsniva, konfidens FROM objects WHERE posttyp = 'objektklass' WHERE lvx_id LIKE ? OR slug LIKE ? OR namn_sv LIKE ? OR namn_en LIKE ? """, (pattern, pattern, pattern, pattern), ) for row in cursor.fetchall(): d = _row_to_dict(row) d["niva"] = "objektklass" d["konfidens"] = min(0.85, 0.6 + len(term) * 0.05) if not any(k["lvx_id"] == d["lvx_id"] for k in kandidater): kandidater.append(d) # Om inga träffar — föreslå närmaste klass baserat på domän/land if not kandidater: # Förenklad: returnera vanliga klasser cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, tier, verifieringsniva, konfidens FROM objects WHERE posttyp = 'objektklass' WHERE tier = 1 ORDER BY konfidens DESC LIMIT 3 """ ) for row in cursor.fetchall(): d = _row_to_dict(row) d["niva"] = "objektklass" d["konfidens"] = 0.5 # Låg konfidens — osäkert kandidater.append(d) # Skapa bounty-förslag bounty = { "skapad": True, "orsak": "Ingen modell kunde bestämmas från OCR", "position": payload.position, } conn.close() # Sortera efter konfidens kandidater.sort(key=lambda x: x.get("konfidens", 0), reverse=True) return { "status": "ok", "kandidater": kandidater[:5], # Max 5 kandidater "bounty": bounty, "ocr_mottagen": payload.ocr_extraherat, } # ── Feedback ──────────────────────────────────────────────────────── class FeedbackPayload(BaseModel): lvx_id: Optional[str] = None typ: Optional[str] = None # t.ex. "felklassning", "saknas", "annat" kommentar: str avsandare: Optional[str] = None @app.get("/v0/bounties") def list_bounties(status: str = None): """Lista aktiva bounties""" conn = _db() cursor = conn.cursor() cursor.execute("SELECT id, title, description, reward, status, created_at FROM bounties ORDER BY id DESC") rows = cursor.fetchall() conn.close() bounties = [] for row in rows: bounties.append({ "id": row[0], "title": row[1], "description": row[2], "reward": row[3], "status": row[4], "created_at": row[5] }) return { "status": "ok", "count": len(bounties), "bounties": bounties } @app.post("/v0/feedback") def post_feedback(payload: FeedbackPayload, auth=Depends(verify_api_key), rate=Depends(check_rate_limit)): """Ta emot feedback och spara till en JSONL-fil.""" timestamp = datetime.utcnow().isoformat() + "Z" record = { "timestamp": timestamp, "lvx_id": payload.lvx_id, "typ": payload.typ, "kommentar": payload.kommentar, "avsandare": payload.avsandare, } # Spara till dagens fil date_str = datetime.utcnow().strftime("%Y-%m-%d") filepath = os.path.join(FEEDBACK_DIR, f"feedback_{date_str}.jsonl") with open(filepath, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") return { "status": "ok", "message": "Feedback mottagen. Tack!", "received_at": timestamp, } # ── Succession ────────────────────────────────────────────────────── @app.get("/v0/objects/{lvx_id}/succession") def get_succession(lvx_id: str, auth=Depends(verify_api_key), rate=Depends(check_rate_limit)): """Hämta ersättningskedja för en modell.""" conn = _db() cursor = conn.cursor() # Hämta modellen cursor.execute("SELECT * FROM objects WHERE lvx_id = ?", (lvx_id,)) row = cursor.fetchone() if not row: conn.close() return JSONResponse( status_code=404, content={"status": "error", "message": f"Modell '{lvx_id}' hittades inte."}, ) modell = _row_to_dict(row) # Hämta ersättare (äldre modeller som denna ersätter) cursor.execute( "SELECT lvx_id, namn_sv, namn_en FROM objects WHERE posttyp = 'produktmodell' AND lvx_id IN (SELECT parent FROM objects WHERE posttyp = 'produktmodell' AND lvx_id = ?)", (lvx_id,), ) ersatter = [_row_to_dict(r) for r in cursor.fetchall()] # Hämta ersätts_av (nyare modeller) cursor.execute( "SELECT lvx_id, namn_sv, namn_en FROM objects WHERE posttyp = 'produktmodell' WHERE parent = ? AND lvx_id != ?", (lvx_id, lvx_id), ) ersatts_av = [_row_to_dict(r) for r in cursor.fetchall()] conn.close() return { "status": "ok", "lvx_id": lvx_id, "namn": modell.get("namn_sv"), "status": modell.get("status", "okand"), "succession": { "ersatter": ersatter, "ersatts_av": ersatts_av, }, } # ── Standards ─────────────────────────────────────────────────────── @app.get("/v0/standards/{beteckning}/objects") def get_standards_objects(beteckning: str): """Hämta alla objekt kopplade till en standard.""" conn = _db() cursor = conn.cursor() pattern = f"%{beteckning}%" results = [] # Sök i klasser cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, tier, verifieringsniva, konfidens FROM objects WHERE posttyp = 'objektklass' WHERE standarder LIKE ? """, (pattern,), ) for row in cursor.fetchall(): d = _row_to_dict(row) d["typ"] = "klass" results.append(d) # Sök i modeller cursor.execute( """ SELECT lvx_id, slug, namn_sv, namn_en, doman, parent, verifieringsniva, konfidens FROM objects WHERE posttyp = 'produktmodell' WHERE standarder LIKE ? """, (pattern,), ) for row in cursor.fetchall(): d = _row_to_dict(row) d["typ"] = "modell" results.append(d) conn.close() return { "status": "ok", "standard": beteckning, "count": len(results), "results": results, } # ── Mätetal ───────────────────────────────────────────────────────── @app.get("/v0/metrics") def get_metrics(auth=Depends(verify_api_key), rate=Depends(check_rate_limit)): """Hämta API-mätetal.""" conn = _db() cursor = conn.cursor() klasser = cursor.execute("SELECT COUNT(*) FROM objects WHERE posttyp = 'objektklass'").fetchone()[0] modeller = cursor.execute("SELECT COUNT(*) FROM objects WHERE posttyp = 'produktmodell'").fetchone()[0] # Räkna verifieringsnivåer cursor.execute("SELECT verifieringsniva, COUNT(*) FROM objects WHERE posttyp = 'objektklass' GROUP BY verifieringsniva") klass_verifiering = {row[0]: row[1] for row in cursor.fetchall()} cursor.execute("SELECT verifieringsniva, COUNT(*) FROM objects WHERE posttyp = 'produktmodell' GROUP BY verifieringsniva") modell_verifiering = {row[0]: row[1] for row in cursor.fetchall()} conn.close() return { "status": "ok", "objects": { "total": klasser + modeller, "klasser": klasser, "modeller": modeller, }, "verifiering": { "klasser": klass_verifiering, "modeller": modell_verifiering, }, "api_version": "0.1.0", } # ── Health ────────────────────────────────────────────────────────── @app.get("/health") def health(): return {"status": "ok", "service": "landvex-api", "version": "0.1.0"}