142 lines
No EOL
4.7 KiB
Python
142 lines
No EOL
4.7 KiB
Python
# api streamlit de démo
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import FastAPI, Query
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.responses import FileResponse
|
|
|
|
|
|
app = FastAPI(title="Demo API", version="0.1.0")
|
|
|
|
|
|
# CORS: autoriser l'UI Angular en local et depuis docker
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=[
|
|
"http://localhost:4200",
|
|
"http://127.0.0.1:4200",
|
|
"http://0.0.0.0:4200",
|
|
"http://localhost:8081",
|
|
"http://127.0.0.1:8081",
|
|
"*", # démo seulement
|
|
],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.get("/")
|
|
def root() -> Dict[str, Any]:
|
|
return {
|
|
"service": "demo-api",
|
|
"message": "Bienvenue sur l'API de démo",
|
|
"endpoints": [
|
|
"/app",
|
|
"/app/summarize_questions",
|
|
"/app/get_keyword_from_email",
|
|
"/app/search_engine",
|
|
"/app/query_simple",
|
|
"/app/dica_detect",
|
|
"/app/esm_detect",
|
|
"/app/hardware_detect",
|
|
"/app/esm_content_analyse",
|
|
"/app/ai_filter",
|
|
],
|
|
}
|
|
|
|
|
|
|
|
# ----------------------
|
|
# Endpoints /app/...
|
|
# Tous en POST, reçoivent un objet JSON libre ("event") et renvoient une réponse de démo
|
|
# ----------------------
|
|
from fastapi import Body
|
|
|
|
|
|
@app.post("/app")
|
|
def app_root(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
# Simple log-like echo
|
|
return {"received": event, "info": "/app"}
|
|
|
|
|
|
@app.post("/app/summarize_questions")
|
|
def summarize_questions(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
questions = event.get("questions", [])
|
|
count = len(questions) if isinstance(questions, list) else 0
|
|
return {"summary": f"{count} question(s) résumée(s)", "input": questions}
|
|
|
|
|
|
@app.post("/app/get_keyword_from_email")
|
|
def get_keyword_from_email(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
subject = event.get("subject", "")
|
|
keyword = subject.split()[0] if isinstance(subject, str) and subject else None
|
|
return {"keyword": keyword, "subject": subject}
|
|
|
|
|
|
@app.post("/app/search_engine")
|
|
def search_engine(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
query = event.get("query", "")
|
|
return {"query": query, "results": [{"title": "demo", "score": 0.42}]}
|
|
|
|
|
|
@app.post("/app/query_simple")
|
|
def get_query_simple(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
return {"echo": event, "note": "simple query demo"}
|
|
|
|
|
|
@app.post("/app/dica_detect")
|
|
def dica_avis_detector(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
text = event.get("text", "")
|
|
has_flag = "DICA" in text.upper() if isinstance(text, str) else False
|
|
return {"detected": has_flag, "label": "DICA"}
|
|
|
|
|
|
@app.post("/app/esm_detect")
|
|
def esm_detecter(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
text = event.get("text", "")
|
|
detected = any(token in text.lower() for token in ["esm", "maintenance"]) if isinstance(text, str) else False
|
|
return {"detected": detected, "label": "ESM"}
|
|
|
|
|
|
@app.post("/app/hardware_detect")
|
|
def detect_hardware(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
text = event.get("text", "")
|
|
detected = any(token in text.lower() for token in ["cpu", "gpu", "ram"]) if isinstance(text, str) else False
|
|
return {"detected": detected, "label": "HARDWARE"}
|
|
|
|
|
|
@app.post("/app/esm_content_analyse")
|
|
def process_esm_documents(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
docs = event.get("docs", [])
|
|
count = len(docs) if isinstance(docs, list) else 0
|
|
return {"processed": count, "details": [{"doc": i, "status": "ok"} for i in range(count)]}
|
|
|
|
|
|
@app.post("/app/ai_filter")
|
|
def ai_filter(event: Dict[str, Any] = Body(default_factory=dict)) -> Dict[str, Any]:
|
|
threshold = event.get("threshold", 0.5)
|
|
score = 0.42
|
|
return {"allowed": score >= threshold, "score": score, "threshold": threshold}
|
|
|
|
|
|
# ----------------------
|
|
# Static UI (Angular build) monté sous /ui
|
|
# ----------------------
|
|
try:
|
|
app.mount("/ui", StaticFiles(directory="/app/ui", html=True), name="ui")
|
|
except Exception:
|
|
# En build local sans assets, ignorer le montage
|
|
pass
|
|
|
|
|
|
@app.get("/ui/{full_path:path}")
|
|
def ui_spa_fallback(full_path: str) -> Any:
|
|
# Fallback SPA: renvoyer index.html pour toute route interne Angular
|
|
index_path = "/app/ui/index.html"
|
|
try:
|
|
return FileResponse(index_path)
|
|
except Exception:
|
|
return {"error": "UI not available", "detail": full_path} |