import os from pathlib import Path from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from dotenv import load_dotenv # Load environment variables from root .env file env_path = Path(__file__).parent.parent.parent / ".env" load_dotenv(dotenv_path=env_path) from app.core.database import Base, engine from app.core.capabilities import assert_required_features, get_capabilities, log_capabilities_summary import importlib import logging class HTTPSRedirectMiddleware(BaseHTTPMiddleware): """ Middleware to ensure redirects use HTTPS in production. When deployed behind a reverse proxy the request arrives as HTTP but should redirect to HTTPS. Starlette's redirect_slashes uses the request scheme, so we wrap the scope to force HTTPS redirects in production. Activated by setting DEPLOY_ENV=production or NODE_ENV=production. """ async def dispatch(self, request: Request, call_next): # In production, ensure the scheme seen by Starlette is HTTPS # by checking X-Forwarded-Proto header (set by reverse proxies) if (os.getenv("NODE_ENV") == "production" or os.getenv("DEPLOY_ENV") == "production"): forwarded_proto = request.headers.get("x-forwarded-proto", "").lower() if forwarded_proto == "https": request.scope["scheme"] = "https" return await call_next(request) # Initialize FastAPI app early so lightweight endpoints work even if heavy # ML-related dependencies fail to import. Routers are added conditionally. app = FastAPI( title="AI Talent Finder", version="1.0.0", # Allow automatic redirect from paths without trailing slash to their # canonical route with trailing slash. This prevents 404s for clients # that omit the trailing slash while endpoints require it. redirect_slashes=True, ) # Add HTTPS redirect middleware BEFORE CORS to catch all requests if os.getenv("ENABLE_HTTPS_REDIRECT", "false").lower() == "true": app.add_middleware(HTTPSRedirectMiddleware) # Configure CORS allowed_origins = [ "https://ai-talent-finder-flame.vercel.app", "http://localhost:3000", ] app.add_middleware( CORSMiddleware, allow_origins=allowed_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def include_optional_router(module_path: str, attr_name: str = "router"): try: module = importlib.import_module(module_path) router = getattr(module, attr_name) app.include_router(router) logging.info(f"Included router {module_path}.{attr_name}") except Exception as e: logging.warning(f"Skipping router {module_path}.{attr_name}: {e}") @app.on_event("startup") def on_startup(): # Alias HF_TOKEN_CHATBOT → HF_TOKEN so GLiNER/HF downloads are authenticated. # The Space secret is named HF_TOKEN_CHATBOT; the HF Hub SDK reads HF_TOKEN. if not os.getenv("HF_TOKEN") and os.getenv("HF_TOKEN_CHATBOT"): os.environ["HF_TOKEN"] = os.environ["HF_TOKEN_CHATBOT"] logging.info("HF_TOKEN aliased from HF_TOKEN_CHATBOT") # Ensure database tables exist (best-effort). # WARNING: create_all() creates missing tables but does NOT add new columns to # existing tables. Run `alembic upgrade head` to apply schema migrations properly. try: Base.metadata.create_all(bind=engine) except Exception as e: logging.exception("Failed to create database tables: %s", e) capabilities = log_capabilities_summary() assert_required_features(capabilities) # Load admin-configurable AI pipeline parameters into the runtime cache. try: from app.core.database import SessionLocal from app.core.settings_store import load_pipeline_config _db = SessionLocal() try: load_pipeline_config(_db) finally: _db.close() except Exception as e: logging.warning("Could not preload pipeline config: %s", e) # Seed the admin account from environment variables. # If ADMIN_EMAIL and ADMIN_PASSWORD are set and no account with that email # exists yet, create it automatically with the admin role. admin_email = os.getenv("ADMIN_EMAIL", "").strip() admin_password = os.getenv("ADMIN_PASSWORD", "").strip() admin_name = os.getenv("ADMIN_FULL_NAME", "Admin").strip() if admin_email and admin_password: try: from app.core.database import SessionLocal from app.core.security import get_password_hash from app.models.models import User as UserModel, UserRole as DBUserRole _db = SessionLocal() try: existing = _db.query(UserModel).filter(UserModel.email == admin_email).first() if existing: if existing.role != DBUserRole.admin: existing.role = DBUserRole.admin _db.commit() logging.info("Admin role enforced for: %s", admin_email) else: admin_user = UserModel( email=admin_email, hashed_password=get_password_hash(admin_password), full_name=admin_name, role=DBUserRole.admin, ) _db.add(admin_user) _db.commit() logging.info("Admin account created: %s", admin_email) finally: _db.close() except Exception as e: logging.warning("Could not seed admin account: %s", e) else: logging.warning("ADMIN_EMAIL or ADMIN_PASSWORD not set — no admin account seeded.") # Conditionally include API routers. If a router import fails (e.g. heavy # ML dependencies missing), the app still starts and exposes /health. include_optional_router("app.api.auth") include_optional_router("app.api.admin") include_optional_router("app.api.candidates") include_optional_router("app.api.skills") include_optional_router("app.api.jobs") include_optional_router("app.api.scoring") include_optional_router("app.api.criteria", "criteria_router") include_optional_router("app.api.criteria", "matching_router") include_optional_router("app.api.favorites") include_optional_router("app.api.experiences") include_optional_router("app.api.educations") include_optional_router("app.api.match_results") include_optional_router("app.api.chat", "router") include_optional_router("app.api.export", "router") # Ensure the full matching API (rich endpoints like /predict) is included when available include_optional_router("app.api.matching", "router") # Phase 3: Feedback loop, recommendations, bias detection include_optional_router("app.api.feedback", "router") # Admin panel include_optional_router("app.api.admin", "router") @app.get("/") def root(): return {"status": "ok", "service": "AI Talent Finder"} # Health check endpoint (always available) @app.get("/health") def health(): return {"status": "ok", "version": "1.0.0"} @app.get("/health/deps") def health_deps(): return get_capabilities()