import os import base64 from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, BackgroundTasks import httpx from contextlib import asynccontextmanager from dotenv import load_dotenv from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from auth import get_current_active_user, User, supabase, get_supabase_token, verify_supabase_token, create_backend_jwt_token import logging import random from io import BytesIO from urllib.parse import unquote from fastapi.responses import StreamingResponse, JSONResponse from typing import Optional from routers import user_models, payments import stripe load_dotenv() dev_mode = False class Settings: def __init__(self): self.mesh_api_key = os.getenv("MESHY_API_KEY") self.openai_api_key = os.getenv("OPENAI_API_KEY") self.stripe_secret_key = os.getenv("STRIPE_SECRET_KEY") if not self.stripe_secret_key: raise RuntimeError("STRIPE_SECRET_KEY environment variable not set") # Initialize Stripe stripe.api_key = self.stripe_secret_key settings = Settings() def get_settings(): return settings @asynccontextmanager async def lifespan(app: FastAPI): app.state.client = httpx.AsyncClient(timeout=10.0) yield await app.state.client.aclose() app = FastAPI( title="3DAI API Module", version="1.0.0", description="Proxy endpoints for the APIs", lifespan=lifespan ) # Include routers app.include_router(user_models.router) app.include_router(payments.router) origins = [ "https://www.3dai.co.uk", "https://3dai.co.uk", "http://localhost:3000", ] # Add CORS for all origins (for testing) app.add_middleware( CORSMiddleware, allow_origins=origins, # never "*" when credentials=True allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["authorization", "content-type"], ) # Pydantic models class TextPrompt(BaseModel): text: str class SignUpRequest(BaseModel): email: str password: str class SignInRequest(BaseModel): email: str password: str class SyncUserRequest(BaseModel): supabase_user_id: str email: str full_name: Optional[str] = None avatar_url: Optional[str] = None class CompleteProfileRequest(BaseModel): address: str fullname: str phone_number: str class PurchaseCreditsRequest(BaseModel): number_of_generations: int price: float transaction_id: str class PlaceOrderRequest(BaseModel): generated_model_id: str price: float packaging_price: float material_price: float delivery_address: str order_status: str = "pending" estimated_delivery_date: str = None shipping_tracking_id: str = None payment_status: str = "pending" transaction_id: str = None # Auth endpoints @app.post("/auth/signup", tags=["Authentication"]) async def signup(request: SignUpRequest): """ Sign up a new user and create initial profile """ try: response = supabase.auth.sign_up({ "email": request.email, "password": request.password }) # Create user profile entry if user creation was successful if response.user and response.user.id: try: supabase.from_("User").insert({ "user_id": response.user.id, "email": response.user.email, "address": None, "fullname": None, "phone_number": None }).execute() # Initialize credits supabase.from_("User_Credit_Account").insert({ "user_id": response.user.id, "num_of_available_gens": 0 }).execute() print(f"User created successfully: {response.user}") except Exception as profile_error: logging.error(f"Failed to create user profile: {str(profile_error)}") return {"message": "User created successfully", "user": response.user} except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.post("/auth/signin", tags=["Authentication"]) async def signin(request: SignInRequest): """ Sign in an existing user """ #logging.basicConfig(level=logging.DEBUG) logging.debug(f"SignIn request: email={request.email}") try: response = supabase.auth.sign_in_with_password({ "email": request.email, "password": request.password }) logging.debug(f"Supabase sign_in_with_password response: {response}") return { "access_token": response.session.access_token, "token_type": "bearer", "user": response.user } except Exception as e: logging.error(f"SignIn error: {str(e)}") raise HTTPException(status_code=401, detail=f"Invalid credentials: {str(e)}") #fale @app.post("/auth/sync-supabase-user", tags=["Authentication"]) async def sync_supabase_user(request: SyncUserRequest, supabase_token: str = Depends(get_supabase_token)): """ Sync a Supabase OAuth user (e.g., Google OAuth) with the backend database and return a backend JWT token. """ try: # Check environment variables first required_env_vars = ["SUPABASE_URL", "SUPABASE_KEY", "JWT_SECRET_KEY"] missing_vars = [var for var in required_env_vars if not os.getenv(var)] if missing_vars: error_msg = f"Missing required environment variables: {', '.join(missing_vars)}" logging.error(error_msg) raise HTTPException(status_code=500, detail=error_msg) # Verify the Supabase token logging.info(f"Verifying Supabase token for user sync") supabase_user = verify_supabase_token(supabase_token) logging.info(f"Token verified for user: {supabase_user.get('email', 'unknown')}") # Check if user already exists in our database try: existing_user = supabase.from_("User").select("*").eq("user_id", supabase_user["id"]).execute() logging.info(f"Database query completed. Existing user found: {bool(existing_user.data)}") except Exception as db_error: logging.error(f"Database query failed: {str(db_error)}") raise HTTPException(status_code=500, detail=f"Database error: {str(db_error)}") user_data = { "user_id": supabase_user["id"], "email": supabase_user["email"], "address": None, "fullname": request.full_name, "phone_number": None } if existing_user.data: logging.info("Updating existing user profile") # Update existing user with any new information if request.full_name: try: supabase.from_("User").update({ "fullname": request.full_name }).eq("user_id", supabase_user["id"]).execute() logging.info("User profile updated successfully") except Exception as update_error: logging.error(f"Failed to update user profile: {str(update_error)}") # Don't fail the entire operation for profile update errors else: logging.info("Creating new user profile") try: # Create new user profile supabase.from_("User").insert(user_data).execute() logging.info("User profile created successfully") # Initialize credits for new user supabase.from_("User_Credit_Account").insert({ "user_id": supabase_user["id"], "num_of_available_gens": 0 }).execute() logging.info("User credits initialized successfully") except Exception as create_error: logging.error(f"Failed to create user profile or credits: {str(create_error)}") raise HTTPException(status_code=500, detail=f"Failed to create user profile: {str(create_error)}") # Generate backend JWT token try: backend_token = create_backend_jwt_token({ "id": supabase_user["id"], "email": supabase_user["email"] }) logging.info("Backend JWT token generated successfully") except Exception as token_error: logging.error(f"Failed to generate backend token: {str(token_error)}") raise HTTPException(status_code=500, detail=f"Token generation failed: {str(token_error)}") # Get updated user profile try: user_profile = supabase.from_("User").select("*").eq("user_id", supabase_user["id"]).single().execute() profile_data = user_profile.data if user_profile.data else user_data except Exception as profile_error: logging.warning(f"Failed to fetch updated profile: {str(profile_error)}") profile_data = user_data # Use the original user_data as fallback response_data = { "access_token": backend_token, "token_type": "bearer", "user": { "id": supabase_user["id"], "email": supabase_user["email"], "profile": profile_data }, "message": "User synced successfully" } logging.info("User sync completed successfully") return response_data except HTTPException: # Re-raise HTTP exceptions as-is raise except Exception as e: logging.error(f"Unexpected error in sync_supabase_user: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"Internal server error during user sync: {str(e)}") # TODO: The signup also creates a profile, so this endpoint should either be removed or /signup updated. @app.post("/auth/complete-profile", tags=["Authentication"]) async def complete_profile(request: CompleteProfileRequest, current_user: User = Depends(get_current_active_user)): """ Complete user profile after signup and initialize credits. """ # Check if profile already exists profile = supabase.from_("User").select("*").eq("user_id", current_user.id).single().execute() if profile.data: return {"message": "Profile already completed."} # Insert into User table supabase.from_("User").insert({ "user_id": current_user.id, "email": current_user.email, "address": request.address, "fullname": request.fullname, "phone_number": request.phone_number }).execute() # Initialize credits if not present credit = supabase.from_("User_Credit_Account").select("*").eq("user_id", current_user.id).single().execute() if not credit.data: supabase.from_("User_Credit_Account").insert({ "user_id": current_user.id, "num_of_available_gens": 0 }).execute() return {"message": "Profile completed and credits initialized."} # Protected endpoints async def check_and_decrement_credits(user_id: str): # Get current credits credit = supabase.from_("User_Credit_Account").select("num_of_available_gens").eq("user_id", user_id).execute() if not credit.data: raise HTTPException(status_code=403, detail="No credit account found. Please complete your profile.") credit_data = credit.data[0] if credit_data["num_of_available_gens"] is None or credit_data["num_of_available_gens"] <= 0: raise HTTPException(status_code=402, detail="No credits left. Please purchase more to generate models.") # Decrement credits atomically new_credits = credit_data["num_of_available_gens"] - 1 supabase.from_("User_Credit_Account").update({"num_of_available_gens": new_credits}).eq("user_id", user_id).execute() @app.post("/req-img-to-3d", tags=["Image-to-3D"]) async def req_img_to_3d( image: UploadFile = File(...), settings: Settings = Depends(get_settings), current_user: User = Depends(get_current_active_user) ): """ Create a new Image-to-3D task by uploading an image. """ # Credit check and decrement await check_and_decrement_credits(current_user.id) content = await image.read() if not content: raise HTTPException(status_code=400, detail="Empty file upload") data_uri = (f"data:{image.content_type};base64," + base64.b64encode(content).decode()) payload = {"image_url": data_uri} headers = {"Authorization": f"Bearer {settings.mesh_api_key}"} if dev_mode: try: supabase.from_("Generated_Models").insert({ "user_id": current_user.id, "meshy_api_job_id": "0197458e-d2f6-7ff0-a211-b660ad057140", "model_name": f"Image to 3D - {image.filename}", "prompts_and_models_config": { "generation_type": "image_to_3d", "source_filename": image.filename, "content_type": image.content_type, "status": "processing", "meshy_response": "dev_mode" } }).execute() except Exception as e: logging.error(f"Failed to save model to database: {str(e)}") # Don't fail the request if database save fails return return {"id": "0197458e-d2f6-7ff0-a211-b660ad057140", "status": "processing", "meshy_response": "dev_mode"} resp = await app.state.client.post( "https://api.meshy.ai/openapi/v1/image-to-3d", json=payload, headers=headers, ) if resp.status_code not in (200, 201, 202): raise HTTPException(status_code=resp.status_code, detail=resp.text) meshy_response = resp.json() meshy_job_id = meshy_response.get("id") or meshy_response.get("task_id") if not meshy_job_id: raise HTTPException(status_code=500, detail="No task ID received from Meshy API") # Save to database try: supabase.from_("Generated_Models").insert({ "user_id": current_user.id, "meshy_api_job_id": meshy_job_id, "model_name": f"Image to 3D - {image.filename}", "prompts_and_models_config": { "generation_type": "image_to_3d", "source_filename": image.filename, "content_type": image.content_type, "status": "processing", "meshy_response": meshy_response } }).execute() except Exception as e: logging.error(f"Failed to save model to database: {str(e)}") # Don't fail the request if database save fails return {"id": meshy_job_id, "status": "processing", "meshy_response": meshy_response} @app.get("/req-result-img-to-3d/{task_id}", tags=["Image-to-3D"]) async def req_result_img_to_3d( task_id: str, settings: Settings = Depends(get_settings), current_user: User = Depends(get_current_active_user) ): """ Retrieve the status and result of a Meshy Image-to-3D task by ID. """ headers = {"Authorization": f"Bearer {settings.mesh_api_key}"} url = f"https://api.meshy.ai/openapi/v1/image-to-3d/{task_id}" resp = await app.state.client.get(url, headers=headers) if resp.status_code != 200: raise HTTPException(status_code=resp.status_code, detail=resp.text) result = resp.json() # Update database with current status try: # Get existing config to merge with new data existing_model = supabase.from_("Generated_Models").select("prompts_and_models_config").eq("meshy_api_job_id", task_id).eq("user_id", current_user.id).single().execute() if existing_model.data: config = existing_model.data.get("prompts_and_models_config", {}) config.update({ "status": result.get("status", "unknown"), "meshy_response": result }) # If completed, add the model URLs if result.get("status") == "SUCCEEDED" and "model_urls" in result: config["model_urls"] = result["model_urls"] config["thumbnail_url"] = result.get("thumbnail_url") supabase.from_("Generated_Models").update({ "prompts_and_models_config": config }).eq("meshy_api_job_id", task_id).eq("user_id", current_user.id).execute() except Exception as e: logging.error(f"Failed to update model status in database: {str(e)}") return result @app.get("/list-img-to-3d-req", tags=["Image-to-3D"]) async def list_img_to_3d_req( settings: Settings = Depends(get_settings), current_user: User = Depends(get_current_active_user) ): """ List all Image-to-3D tasks. """ headers = {"Authorization": f"Bearer {settings.mesh_api_key}"} resp = await app.state.client.get( "https://api.meshy.ai/openapi/v1/image-to-3d", headers=headers ) if resp.status_code != 200: raise HTTPException(status_code=resp.status_code, detail=resp.text) return resp.json() @app.post("/req-text-to-3d", tags=["Text-to-3D"]) async def req_text_to_3d( prompt: TextPrompt, settings: Settings = Depends(get_settings), current_user: User = Depends(get_current_active_user) ): """ Reframe user text via OpenAI, then create a Text-to-3D preview task. """ # Credit check and decrement await check_and_decrement_credits(current_user.id) # Save initial record to database immediately after credit check initial_record_id = None try: initial_result = supabase.from_("Generated_Models").insert({ "status": "IN_PROGRESS", "user_id": current_user.id, "meshy_api_job_id": None, "model_name": f"Text to 3D - {prompt.text[:50]}{'...' if len(prompt.text) > 50 else ''}", "prompts_and_models_config": { "generation_type": "text_to_3d", "original_prompt": prompt.text, "status": "initializing", "stage": "reframing_prompt" } }).execute() initial_record_id = initial_result.data[0]["generated_model_id"] if initial_result.data else None print(f"Created initial database record: {initial_record_id}") except Exception as e: print(f"Failed to create initial database record: {str(e)}") # Continue with the request even if database save fails # Reframe prompt openai_resp = await app.state.client.post( "https://api.openai.com/v1/chat/completions", headers={"Authorization": f"Bearer {settings.openai_api_key}"}, json={ "model": "gpt-4o-mini", "messages": [ {"role": "system", "content": "REPLY ONLY WITH NEW PROMPT, no other text. IF USER PROMPT CONTAINS HARMFUL CONTENT, CHANGE IT TO SOMETHING SAFE and somewhat related. Rephrase the user description to simple and short. If it already is simple keep it as is. USERS PROMPT CANNOT CONTAIN ANYTHING WHICH REQUIRES A LOT OF DETAIL. FOR EXAMPLE: 'A realistic robot with gears and hollow interior' IS NOT A GOOD PROMPT. 'A simple smooth robot with gears' IS A GOOD PROMPT."}, {"role": "user", "content": prompt.text} ] } ) if openai_resp.status_code != 200: # Update database with error status if initial record was created if initial_record_id: try: supabase.from_("Generated_Models").update({ "status": "FAILED", "prompts_and_models_config": { "generation_type": "text_to_3d", "original_prompt": prompt.text, "status": "failed", "stage": "reframing_prompt", "error": f"OpenAI API error: {openai_resp.text}" } }).eq("generated_model_id", initial_record_id).execute() except Exception as e: print(f"Failed to update database with error: {str(e)}") raise HTTPException(status_code=openai_resp.status_code, detail=openai_resp.text) reframed = openai_resp.json()["choices"][0]["message"]["content"] # Update database with reframed prompt if initial_record_id: try: supabase.from_("Generated_Models").update({ "prompts_and_models_config": { "generation_type": "text_to_3d", "original_prompt": prompt.text, "reframed_prompt": reframed, "status": "processing", "stage": "creating_3d_model" } }).eq("generated_model_id", initial_record_id).execute() except Exception as e: print(f"Failed to update database with reframed prompt: {str(e)}") # Create Meshy Text-to-3D task meshy_payload = { "mode": "preview", "prompt": reframed, "art_style": "sculpture", "ai_model": "meshy-5", "target_polycount": 150000, } print(f"Sending to Meshy API: {meshy_payload}") meshy_resp = await app.state.client.post( "https://api.meshy.ai/openapi/v2/text-to-3d", headers={"Authorization": f"Bearer {settings.mesh_api_key}"}, json=meshy_payload ) if meshy_resp.status_code not in (200, 201, 202): # Log the full response for debugging error_text = meshy_resp.text print(f"Meshy API error - Status: {meshy_resp.status_code}, Response: {error_text}") # Try to parse the response as JSON to provide better error details try: error_json = meshy_resp.json() error_detail = f"Meshy API error (status {meshy_resp.status_code}): {error_json}" except: error_detail = f"Meshy API error (status {meshy_resp.status_code}): {error_text}" # Update database with error status if initial record was created if initial_record_id: try: supabase.from_("Generated_Models").update({ "prompts_and_models_config": { "generation_type": "text_to_3d", "original_prompt": prompt.text, "reframed_prompt": reframed, "status": "failed", "stage": "creating_3d_model", "error": error_detail } }).eq("generated_model_id", initial_record_id).execute() except Exception as e: print(f"Failed to update database with error: {str(e)}") raise HTTPException(status_code=meshy_resp.status_code, detail=error_detail) meshy_response = meshy_resp.json() task_id = meshy_response.get("result") or meshy_response.get("id") or meshy_response.get("task_id") if not task_id: # Update database with error status if initial record was created if initial_record_id: try: supabase.from_("Generated_Models").update({ "prompts_and_models_config": { "generation_type": "text_to_3d", "original_prompt": prompt.text, "reframed_prompt": reframed, "status": "failed", "stage": "creating_3d_model", "error": "No task ID received from Meshy API" } }).eq("generated_model_id", initial_record_id).execute() except Exception as e: print(f"Failed to update database with error: {str(e)}") raise HTTPException(status_code=500, detail="No task ID received from Meshy API") # Update database with final successful response print(f"Updating database record with task ID: {task_id}") try: supabase.from_("Generated_Models").update({ "meshy_api_job_id": task_id, "prompts_and_models_config": { "generation_type": "text_to_3d", "original_prompt": prompt.text, "reframed_prompt": reframed, "status": "processing", "stage": "generating", "meshy_response": meshy_response } }).eq("generated_model_id", initial_record_id).execute() except Exception as e: print(f"Failed to update database with final response: {str(e)}") # Don't fail the request if database save fails return {"id": initial_record_id, "meshy_task_id": task_id, "status": "processing", "original_prompt": prompt.text, "reframed_prompt": reframed, "meshy_response": meshy_response} @app.get("/req-result-text-to-3d/{task_id}", tags=["Text-to-3D"]) async def req_result_text_to_3d( task_id: str, settings: Settings = Depends(get_settings), current_user: User = Depends(get_current_active_user) ): """ Retrieve the status/result of a Text-to-3D task by ID. """ resp = await app.state.client.get( f"https://api.meshy.ai/openapi/v2/text-to-3d/{task_id}", headers={"Authorization": f"Bearer {settings.mesh_api_key}"} ) if resp.status_code != 200: raise HTTPException(status_code=resp.status_code, detail=resp.text) result = resp.json() # Update database with current status try: # Get existing config to merge with new data existing_model = supabase.from_("Generated_Models").select("prompts_and_models_config").eq("meshy_api_job_id", task_id).eq("user_id", current_user.id).single().execute() if existing_model.data: config = existing_model.data.get("prompts_and_models_config", {}) config.update({ "status": result.get("status", "unknown"), "meshy_response": result }) # If completed, add the model URLs if result.get("status") == "SUCCEEDED" and "model_urls" in result: config["model_urls"] = result["model_urls"] config["thumbnail_url"] = result.get("thumbnail_url") supabase.from_("Generated_Models").update({ "prompts_and_models_config": config }).eq("meshy_api_job_id", task_id).eq("user_id", current_user.id).execute() except Exception as e: logging.error(f"Failed to update model status in database: {str(e)}") return result @app.get("/list-text-to-3d-req", tags=["Text-to-3D"]) async def list_text_to_3d_req( settings: Settings = Depends(get_settings), current_user: User = Depends(get_current_active_user) ): """ List all Text-to-3D tasks. """ resp = await app.state.client.get( "https://api.meshy.ai/openapi/v2/text-to-3d", headers={"Authorization": f"Bearer {settings.mesh_api_key}"} ) if resp.status_code != 200: raise HTTPException(status_code=resp.status_code, detail=resp.text) return resp.json() @app.post("/credits/purchase", tags=["Credits"]) async def purchase_credits(request: PurchaseCreditsRequest, current_user: User = Depends(get_current_active_user)): """ Purchase more credits: record in Credit_Order_History and update User_Credit_Account. """ # Record the purchase supabase.from_("Credit_Order_History").insert({ "user_id": current_user.id, "price": request.price, "number_of_generations": request.number_of_generations, "order_date": "now()", "payment_status": "paid", "transaction_id": request.transaction_id }).execute() # Update credits credit = supabase.from_("User_Credit_Account").select("num_of_available_gens").eq("user_id", current_user.id).single().execute() new_credits = (credit.data["num_of_available_gens"] if credit.data else 0) + request.number_of_generations if credit.data: supabase.from_("User_Credit_Account").update({"num_of_available_gens": new_credits}).eq("user_id", current_user.id).execute() else: supabase.from_("User_Credit_Account").insert({"user_id": current_user.id, "num_of_available_gens": new_credits}).execute() return {"message": "Credits purchased successfully.", "total_credits": new_credits} # User dashboard endpoints @app.get("/user/profile", tags=["User"]) async def get_profile(current_user: User = Depends(get_current_active_user)): profile = supabase.from_("User").select("*").eq("user_id", current_user.id).single().execute() return profile.data @app.get("/user/credits", tags=["User"]) async def get_credits(current_user: User = Depends(get_current_active_user)): credit = supabase.from_("User_Credit_Account").select("num_of_available_gens").eq("user_id", current_user.id).single().execute() return {"num_of_available_gens": credit.data["num_of_available_gens"] if credit.data else 0} @app.get("/user/orders", tags=["User"]) async def get_orders(current_user: User = Depends(get_current_active_user)): orders = supabase.from_("Model_Order_History").select("*").eq("user_id", current_user.id).execute() return orders.data @app.get("/user/models", tags=["User"]) async def get_models(current_user: User = Depends(get_current_active_user)): models = supabase.from_("Generated_Models").select("*").eq("user_id", current_user.id).order("created_at", desc=True).execute() return models.data @app.get("/user/models/{task_id}", tags=["User"]) async def get_model_by_task_id(task_id: str, current_user: User = Depends(get_current_active_user)): """ Get a specific model by task ID for the current user. """ model = supabase.from_("Generated_Models").select("*").eq("meshy_api_job_id", task_id).eq("user_id", current_user.id).single().execute() if not model.data: raise HTTPException(status_code=404, detail="Model not found") return model.data @app.post("/orders/place", tags=["Orders"]) async def place_order(request: PlaceOrderRequest, current_user: User = Depends(get_current_active_user)): """ Place an order for a 3D model print. """ supabase.from_("Model_Order_History").insert({ "user_id": current_user.id, "generated_model_id": request.generated_model_id, "price": request.price, "packaging_price": request.packaging_price, "material_price": request.material_price, "delivery_address": request.delivery_address, "order_status": request.order_status, "order_date": "now()", "estimated_delivery_date": request.estimated_delivery_date, "shipping_tracking_id": request.shipping_tracking_id, "payment_status": request.payment_status, "transaction_id": request.transaction_id }).execute() return {"message": "Order placed successfully."} @app.get("/", tags=["Health"]) async def health_check(): return {"status": "ok", "message": "Service is running"} # Add debug endpoint to check environment and connections @app.get("/debug/env", tags=["Debug"]) async def debug_environment(): """Debug endpoint to check environment variables and connections""" debug_info = { "environment_variables": { "SUPABASE_URL": "✅ Set" if os.getenv("SUPABASE_URL") else "❌ Missing", "SUPABASE_KEY": "✅ Set" if os.getenv("SUPABASE_KEY") else "❌ Missing", "JWT_SECRET_KEY": "✅ Set" if os.getenv("JWT_SECRET_KEY") else "❌ Missing", "MESHY_API_KEY": "✅ Set" if os.getenv("MESHY_API_KEY") else "❌ Missing", "OPENAI_API_KEY": "✅ Set" if os.getenv("OPENAI_API_KEY") else "❌ Missing", "STRIPE_SECRET_KEY": "✅ Set" if os.getenv("STRIPE_SECRET_KEY") else "❌ Missing" }, "supabase_connection": "Unknown" } # Test Supabase connection try: # Simple test query to check connection test_response = supabase.from_("User").select("user_id").limit(1).execute() debug_info["supabase_connection"] = "✅ Connected" except Exception as e: debug_info["supabase_connection"] = f"❌ Connection failed: {str(e)}" return debug_info @app.post("/debug/test-background", tags=["Debug"]) async def test_background_response(background_tasks: BackgroundTasks): """Test endpoint to debug background task response issues""" from fastapi.responses import JSONResponse def dummy_background_task(): import time time.sleep(1) # Simulate some work logging.info("Background task completed") # Add a background task background_tasks.add_task(dummy_background_task) response_data = { "message": "Background task started", "test_id": "test-123", "status": "processing" } logging.info(f"Returning test response: {response_data}") response = JSONResponse(content=response_data, status_code=200) response.headers["Cache-Control"] = "no-cache" return response @app.get("/proxy/model", tags=["Proxy"]) async def proxy_model(url: str): """ Simple CORS-friendly proxy for GLB (or any binary) files. The front-end should call this endpoint with the target URL percent-encoded, e.g. /proxy/model?url=https%3A%2F%2Fassets.meshy.ai%2F....glb """ try: # Decode once in case the URL is already percent-encoded target_url = unquote(url) upstream_resp = await app.state.client.get(target_url) if upstream_resp.status_code != 200: raise HTTPException(status_code=upstream_resp.status_code, detail=f"Failed to fetch remote model: {upstream_resp.text}") content_type = upstream_resp.headers.get("Content-Type", "model/gltf-binary") headers = {"Access-Control-Allow-Origin": "*"} # Stream the content back to the client return StreamingResponse(BytesIO(upstream_resp.content), media_type=content_type, headers=headers) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": import uvicorn port = int(os.getenv("PORT", 8000)) uvicorn.run("main:app", host="0.0.0.0", port=port, log_level="info")