backend/scripts/test_intelligence_local.py

#!/usr/bin/env python3
"""Lokalny smoke test endpointów intelligence (bez pełnego grants.py / HF)."""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))
SCRATCH = Path(os.environ.get("CATALOG_SCRATCH", "/tmp/grok-goal-935580f22161/implementer"))
SCRATCH.mkdir(parents=True, exist_ok=True)

DB_FILE = SCRATCH / "local_intel.db"
os.environ["DATABASE_URL"] = f"sqlite:///{DB_FILE}"
if DB_FILE.exists():
    DB_FILE.unlink()

from fastapi import FastAPI
from fastapi.testclient import TestClient

from core.subscription.db import SessionLocal, init_models
from scripts.migrate_hgis_schema import migrate
from core.grants.wyszukiwarka_import import import_wyszukiwarka_file, resolve_wyszukiwarka_path

init_models()
migrate()

path = resolve_wyszukiwarka_path()
if path:
    db = SessionLocal()
    try:
        import_wyszukiwarka_file(path, db=db)
    finally:
        db.close()

app = FastAPI()
from endpoints.grants_intelligence import router as intelligence_router

app.include_router(intelligence_router, prefix="/api/grants")
client = TestClient(app)
results = {}

r = client.get("/api/grants/intelligence/dashboard")
results["dashboard"] = {
    "status": r.status_code,
    "body_keys": list(r.json().keys()) if r.status_code == 200 else r.text[:200],
}

r = client.post(
    "/api/grants/intelligence/recommendations",
    json={"size": "mikro", "region": "mazowieckie", "pkd": ["62.01"]},
)
body = r.json() if r.status_code == 200 else {}
results["recommendations"] = {
    "status": r.status_code,
    "count": body.get("count"),
    "top_why": (body.get("recommendations") or [{}])[0].get("why_it_fits", "")[:150],
}

r = client.post("/api/grants/intelligence/hybrid-search", json={"query": "dotacja MŚP", "limit": 5})
hb = r.json() if r.status_code == 200 else {}
grants = hb.get("grants") or []
results["hybrid_search"] = {
    "status": r.status_code,
    "count": hb.get("count"),
    "search_mode": grants[0].get("search_mode") if grants else None,
}

out = SCRATCH / "local_intelligence_endpoints.log"
out.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(results, indent=2))
print("saved", out)
failed = [k for k, v in results.items() if v.get("status") != 200]
raise SystemExit(1 if failed else 0)