Spaces:
Running
Running
File size: 2,388 Bytes
ce8f04a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | #!/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)
|