Spaces:
Sleeping
Sleeping
File size: 3,465 Bytes
a1933cb | 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """Application factory for the Flask service."""
from __future__ import annotations
from pathlib import Path
from time import perf_counter
from flask import Flask, Response, g, jsonify, request
from flask_cors import CORS
from werkzeug.exceptions import HTTPException
from api_layer.routes import build_routes
from brain.decision_maker import DecisionEngine
from config.settings import AppSettings, configure_logging, load_settings
from core_engine.analytics import AnalyticsTracker
from core_engine.leaderboard import LeaderboardStore
from core_engine.simulator import SimulationEngine
from utils.logger import get_logger
LOGGER = get_logger(__name__)
def create_server(settings: AppSettings | None = None) -> Flask:
"""Create and configure the Flask application."""
app_settings = settings or load_settings()
configure_logging(app_settings)
frontend_dir = Path(__file__).resolve().parent.parent / "frontend"
app = Flask(
__name__,
static_folder=str(frontend_dir),
static_url_path="",
)
CORS(app, resources={r"/*": {"origins": app_settings.cors_origins}})
engine = SimulationEngine(
batch_size=app_settings.default_mail_count,
random_seed=app_settings.random_seed,
simulation_mode=app_settings.simulation_mode,
)
decision_engine = DecisionEngine(
model_type=app_settings.model_type,
hf_model=app_settings.hf_model,
hf_token=app_settings.hf_token,
hf_timeout=app_settings.hf_timeout,
)
analytics = AnalyticsTracker()
leaderboard = LeaderboardStore()
@app.before_request
def log_request_start() -> None:
g.request_started_at = perf_counter()
@app.after_request
def log_request_end(response: Response) -> Response:
started_at = getattr(g, "request_started_at", perf_counter())
elapsed_ms = (perf_counter() - started_at) * 1000
LOGGER.info(
"%s %s -> %s %.2fms",
request.method,
request.path,
response.status_code,
elapsed_ms,
)
return response
@app.get("/")
def dashboard() -> object:
try:
return app.send_static_file("index.html")
except Exception:
LOGGER.exception("Failed to serve dashboard.")
return jsonify({"success": False, "error": "Unable to serve dashboard."}), 500
@app.errorhandler(HTTPException)
def handle_http_error(error: HTTPException) -> tuple[Response, int]:
return (
jsonify(
{
"success": False,
"error": error.description or "HTTP request failed.",
}
),
error.code or 500,
)
@app.errorhandler(Exception)
def handle_unexpected_error(error: Exception) -> tuple[Response, int]:
LOGGER.exception("Unhandled application error: %s", error)
return jsonify({"success": False, "error": "Internal server error."}), 500
app.register_blueprint(
build_routes(
engine,
decision_engine=decision_engine,
analytics=analytics,
leaderboard=leaderboard,
max_email_count=app_settings.max_emails,
)
)
app.config["simulation_engine"] = engine
app.config["decision_engine"] = decision_engine
app.config["analytics"] = analytics
app.config["leaderboard"] = leaderboard
return app
|