Spaces:
Sleeping
Sleeping
File size: 2,617 Bytes
dfdddb1 | 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 | from fastapi import FastAPI, Query
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Optional
import os
import typesense
from utils.config import TYPESENSE_HOST, TYPESENSE_PORT, TYPESENSE_API_KEY
from db.postgres_connector import SessionLocal
from db.models import Document
from functools import lru_cache
app = FastAPI(title="Scrap-Dji Turbo API")
# Configuration CORS pour permettre les requêtes depuis le frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # En production, spécifier les domaines autorisés
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Cache LRU pour les requêtes ultra-fréquentes (Zéro latence)
@lru_cache(maxsize=100)
def get_cached_search(q: str, pays: Optional[str]):
# Cette fonction est appelée par le endpoint async
return None # Placeholder pour la logique interne
ts_client = typesense.Client({
'nodes': [{'host': TYPESENSE_HOST, 'port': TYPESENSE_PORT, 'protocol': 'http'}],
'api_key': TYPESENSE_API_KEY,
'connection_timeout_seconds': 1
})
SEARCH_CACHE = {}
@app.get("/search")
async def search(q: str = Query(...), pays: Optional[str] = None):
# Check simple cache manuel (plus rapide pour l'async)
cache_key = f"{q}_{pays}"
if cache_key in SEARCH_CACHE:
return SEARCH_CACHE[cache_key]
try:
res = ts_client.collections['documents'].documents.search({
'q': q,
'query_by': 'titre,texte',
'filter_by': f'pays:={pays}' if pays else ''
})
hits = res['hits']
SEARCH_CACHE[cache_key] = hits # Mise en cache
return hits
except:
# Fallback final : Recherche dans le fichier JSON local (Mode Test)
local_file = "data/search_index.json"
if os.path.exists(local_file):
import json
with open(local_file, "r", encoding="utf-8") as f:
data = json.load(f)
results = [d for d in data if q.lower() in d['titre'].lower() or q.lower() in d['texte'].lower()]
if pays:
results = [d for d in results if d.get('pays') == pays]
return results[:10]
# Fallback SQL Optimisé (Indexé) - essayera quand même si Postgres est là
try:
session = SessionLocal()
results = session.query(Document).filter(Document.titre.ilike(f"%{q}%")).limit(5).all()
session.close()
return results
except:
return []
@app.get("/health")
def health(): return {"status": "turbo-charged"} |