import os import subprocess from datetime import datetime from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, JSONResponse, HTMLResponse from dotenv import load_dotenv load_dotenv() from app.config import settings from app.routers import upload, caption, metadata, models from app.routers.images import router as images_router from app.routers.prompts import router as prompts_router from app.routers.admin import router as admin_router from app.routers.schemas import router as schemas_router app = FastAPI(title="PromptAid Vision") @app.middleware("http") async def log_requests(request, call_next): print(f"DEBUG: {request.method} {request.url.path}") response = await call_next(request) print(f"DEBUG: {request.method} {request.url.path} -> {response.status_code}") return response app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000","http://localhost:5173","http://localhost:8000"], allow_origin_regex=r"https://.*\.hf\.space$", allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) app.include_router(caption.router, prefix="/api", tags=["captions"]) app.include_router(metadata.router, prefix="/api", tags=["metadata"]) app.include_router(models.router, prefix="/api", tags=["models"]) app.include_router(upload.router, prefix="/api/images", tags=["images"]) app.include_router(images_router, prefix="/api/contribute", tags=["contribute"]) app.include_router(prompts_router, prefix="/api/prompts", tags=["prompts"]) app.include_router(admin_router, prefix="/api/admin", tags=["admin"]) app.include_router(schemas_router, prefix="/api", tags=["schemas"]) @app.get("/api/images", include_in_schema=False) async def list_images_no_slash(): """Handle /api/images without trailing slash to prevent 307 redirect""" from app.routers.upload import list_images from app.database import SessionLocal db = SessionLocal() try: return list_images(db) finally: db.close() @app.get("/api/prompts", include_in_schema=False) async def list_prompts_no_slash(): """Handle /api/prompts without trailing slash to prevent 307 redirect""" from app.routers.prompts import get_prompts from app.database import SessionLocal db = SessionLocal() try: return get_prompts(db) finally: db.close() @app.post("/api/prompts", include_in_schema=False) async def create_prompt_no_slash(prompt_data: dict): """Handle POST /api/prompts without trailing slash to prevent 307 redirect""" from app.routers.prompts import create_prompt from app.database import SessionLocal from app.schemas import PromptCreate db = SessionLocal() try: # Convert dict to PromptCreate schema prompt_create = PromptCreate(**prompt_data) return create_prompt(prompt_create, db) finally: db.close() @app.get("/health", include_in_schema=False, response_class=JSONResponse) async def health(): return {"status": "ok"} @app.get("/", include_in_schema=False, response_class=HTMLResponse) def root(): return """ PromptAid Vision

PromptAid Vision

OK

Open UI • API Docs

""" if os.path.exists("/app"): STATIC_DIR = "/app/static" else: STATIC_DIR = "static" print(f"Looking for static files in: {STATIC_DIR}") if os.path.isdir(STATIC_DIR): print(f"Static directory found: {STATIC_DIR}") app.mount("/app", StaticFiles(directory=STATIC_DIR, html=True), name="static") print(f"Static files mounted at /app from {STATIC_DIR}") else: print(f"Static directory NOT found: {STATIC_DIR}") print(f"Current directory contents: {os.listdir(os.path.dirname(__file__))}") print(f"Parent directory contents: {os.listdir(os.path.dirname(os.path.dirname(__file__)))}") print(f"Attempting to find static directory...") possible_paths = [ "static", "../static", "py_backend/static", "../py_backend/static" ] for path in possible_paths: if os.path.isdir(path): print(f"Found static directory at: {path}") STATIC_DIR = path app.mount("/app", StaticFiles(directory=STATIC_DIR, html=True), name="static") print(f"Static files mounted at /app from {STATIC_DIR}") break else: print("Could not find static directory - static file serving disabled") @app.get("/app/{full_path:path}", include_in_schema=False) def spa_fallback(full_path: str): index = os.path.join(STATIC_DIR, "index.html") if os.path.isfile(index): return FileResponse(index) raise HTTPException(status_code=404, detail="Not Found") @app.get("/debug-routes", include_in_schema=False) async def debug_routes(): """Show all registered routes for debugging""" routes = [] for route in app.routes: if hasattr(route, 'path'): routes.append({ "path": route.path, "name": getattr(route, 'name', 'N/A'), "methods": list(route.methods) if hasattr(route, 'methods') else [] }) return {"routes": routes} @app.get("/debug") async def debug(): return { "message": "Backend is working", "timestamp": datetime.now().isoformat(), "routes": [route.path for route in app.routes] } @app.get("/debug-static") async def debug_static(): import os return { "static_dir": STATIC_DIR, "exists": os.path.exists(STATIC_DIR), "is_dir": os.path.isdir(STATIC_DIR) if os.path.exists(STATIC_DIR) else False, "current_dir": os.getcwd(), "app_dir": os.path.dirname(__file__), "parent_dir": os.path.dirname(os.path.dirname(__file__)) } @app.get("/debug-storage") async def debug_storage(): """Debug storage configuration and files""" import os from app.config import settings storage_dir = settings.STORAGE_DIR storage_exists = os.path.exists(storage_dir) files = [] if storage_exists: try: for root, dirs, filenames in os.walk(storage_dir): for filename in filenames[:10]: rel_path = os.path.relpath(os.path.join(root, filename), storage_dir) files.append(rel_path) except Exception as e: files = [f"Error listing files: {e}"] return { "storage_provider": settings.STORAGE_PROVIDER, "storage_dir": storage_dir, "storage_exists": storage_exists, "storage_is_dir": os.path.isdir(storage_dir) if storage_exists else False, "current_dir": os.getcwd(), "sample_files": files[:10], "total_files": len(files) if isinstance(files, list) else 0 } @app.get("/uploads/{file_path:path}") async def serve_upload(file_path: str): """Serve uploaded files from local storage""" if settings.STORAGE_PROVIDER != "local": raise HTTPException(status_code=404, detail="Local storage not enabled") file_path_full = os.path.join(settings.STORAGE_DIR, file_path) if not os.path.exists(file_path_full): raise HTTPException(status_code=404, detail="File not found") return FileResponse(file_path_full) def run_migrations(): """Run database migrations on startup""" try: print("Running database migrations...") current_dir = os.getcwd() print(f"Current working directory: {current_dir}") print("Checking container environment...") try: result = subprocess.run(["which", "alembic"], capture_output=True, text=True) print(f"Alembic location: {result.stdout.strip() if result.stdout else 'Not found'}") except Exception as e: print(f"Could not check alembic location: {e}") print(f"Checking if /app exists: {os.path.exists('/app')}") if os.path.exists('/app'): print(f"Contents of /app: {os.listdir('/app')}") alembic_paths = [ "alembic.ini", "../alembic.ini", "py_backend/alembic.ini", "/app/alembic.ini", ] alembic_dir = None for path in alembic_paths: if os.path.exists(path): alembic_dir = os.path.dirname(path) print(f"Found alembic.ini at: {path}") break if not alembic_dir: print("Could not find alembic.ini - using current directory") alembic_dir = current_dir try: print(f"Running alembic upgrade head from: {alembic_dir}") result = subprocess.run( ["alembic", "upgrade", "head"], cwd=alembic_dir, capture_output=True, text=True, timeout=60 ) print(f"Alembic return code: {result.returncode}") print(f"Alembic stdout: {result.stdout}") print(f"Alembic stderr: {result.stderr}") if result.returncode == 0: print("Database migrations completed successfully") else: print(f"Database migrations failed") print("Trying fallback: create tables directly...") try: from app.database import engine from app.models import Base Base.metadata.create_all(bind=engine) print("Tables created directly via SQLAlchemy") except Exception as fallback_error: print(f"Fallback also failed: {fallback_error}") except Exception as e: print(f"Error running alembic: {e}") except Exception as e: print(f"Could not run migrations: {e}") def ensure_storage_ready(): """Ensure storage is ready before starting the app""" print(f"Storage provider from settings: '{settings.STORAGE_PROVIDER}'") print(f"S3 endpoint from settings: '{settings.S3_ENDPOINT}'") print(f"S3 bucket from settings: '{settings.S3_BUCKET}'") if settings.STORAGE_PROVIDER == "s3": try: print("Checking S3 storage connection...") from app.storage import _ensure_bucket _ensure_bucket() print("S3 storage ready") except Exception as e: print(f"Warning: S3 storage not ready: {e}") print("Storage operations may fail until S3 is available") elif settings.STORAGE_PROVIDER == "local": print("Using local storage - no external dependencies") else: print(f"Unknown storage provider: {settings.STORAGE_PROVIDER}") run_migrations() ensure_storage_ready() print("PromptAid Vision API server ready") print("Endpoints: /api/images, /api/captions, /api/metadata, /api/models") print(f"Environment: {settings.ENVIRONMENT}") print("CORS: localhost + *.hf.space")