""" Sahih Islam SuperAdmin API (Hugging Face Space - Gradio SDK) FastAPI Backend running natively on Hugging Face Spaces without Docker. Configured for ZeroGPU and CPU hardware compatibility. """ import os os.environ["GRADIO_SSR_MODE"] = "False" import logging from fastapi.middleware.cors import CORSMiddleware # ------------------------------------------------------------- # ZeroGPU Support for Hugging Face Spaces # ------------------------------------------------------------- # When a Space is assigned ZeroGPU hardware (zero-a10g), Hugging Face # requires at least one module-level function decorated with @spaces.GPU # during startup. We provide a safe, high-availability wrapper here. # ------------------------------------------------------------- try: import spaces HAS_SPACES = True except ImportError: HAS_SPACES = False try: import torch except ImportError: torch = None if HAS_SPACES: @spaces.GPU def zero_gpu_compute_node(query: str = "ping") -> str: """ ZeroGPU Compute Worker. Decorated with @spaces.GPU to satisfy Hugging Face ZeroGPU runtime supervisor. """ hardware_info = "ZeroGPU Cluster Attached" if torch is not None and torch.cuda.is_available(): try: hardware_info = f"NVIDIA GPU: {torch.cuda.get_device_name(0)}" except Exception: hardware_info = "NVIDIA CUDA Device Active" return f"Operational: {hardware_info} | Query: {query}" else: def zero_gpu_compute_node(query: str = "ping") -> str: return f"Operational: Standard CPU Host | Query: {query}" # ------------------------------------------------------------- # Hugging Face Hub Compatibility Shim # Fixes 'cannot import name HfFolder from huggingface_hub' # ------------------------------------------------------------- try: import huggingface_hub if not hasattr(huggingface_hub, "HfFolder"): class _HfFolderShim: @staticmethod def get_token(): return os.environ.get("HF_TOKEN") @staticmethod def save_token(token): pass @staticmethod def delete_token(): pass huggingface_hub.HfFolder = _HfFolderShim except Exception: pass import gradio as gr # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Import routes and collections from app.superadmin_routes import router as superadmin_router from app.auth import router as auth_router try: from app.masjid_routes import router as masjid_router except Exception: masjid_router = None try: from app.hadith import router as hadith_router except Exception: hadith_router = None try: from app.ads_routes import router as ads_router except Exception: ads_router = None try: from app.payment_routes import router as payment_router except Exception: payment_router = None try: from app.tafsir_routes import router as tafsir_router, quran_pdf_router except Exception: tafsir_router = None quran_pdf_router = None try: from app.mufti_routes import router as mufti_router except Exception: mufti_router = None try: from app.events_routes import router as events_router except Exception: events_router = None from app.database import sendads_collection def _register_routes(fastapi_app): """Mounts CORS, SuperAdmin routes, auth routes, health check, and ad config onto the FastAPI app.""" if getattr(fastapi_app.state, "superadmin_routes_mounted", False): return for route in getattr(fastapi_app, "routes", []): if getattr(route, "path", None) == "/health": fastapi_app.state.superadmin_routes_mounted = True return fastapi_app.state.superadmin_routes_mounted = True logger.info("Mounting CORS, SuperAdmin routes, Auth routes, /health, and /ads/config onto FastAPI application...") fastapi_app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 1. SuperAdmin Control Panel endpoints (/admin/...) fastapi_app.include_router(superadmin_router, prefix="/admin", tags=["SuperAdmin Control Panel"]) # 2. Authentication & Profile endpoints (/auth/profile/{uid} and /profile/{uid}) fastapi_app.include_router(auth_router, prefix="/auth", tags=["Authentication & Profile"]) fastapi_app.include_router(auth_router, tags=["Auth Direct"]) # 3. Masjid endpoints (/masjid/...) if masjid_router is not None: fastapi_app.include_router(masjid_router, prefix="/masjid", tags=["Masjids"]) # 4. Hadith & Daily content endpoints (/hadith/..., /api/..., and direct) if hadith_router is not None: fastapi_app.include_router(hadith_router, prefix="/hadith", tags=["Hadiths"]) fastapi_app.include_router(hadith_router, prefix="/api", tags=["Hadith & AI Engine"]) fastapi_app.include_router(hadith_router, tags=["Hadiths Direct"]) # 5. Ad delivery endpoints (/ads/...) if ads_router is not None: fastapi_app.include_router(ads_router) # 6. Payment & Subscription endpoints (/api/v1/payments/... and /payment/...) if payment_router is not None: fastapi_app.include_router(payment_router, prefix="/api/v1/payments", tags=["Payment Gateway"]) fastapi_app.include_router(payment_router, prefix="/payment", tags=["Payment Direct"]) # 7. Tafsir & Quran PDF Library endpoints (/api/...) if tafsir_router is not None: fastapi_app.include_router(tafsir_router, prefix="/api", tags=["Tafsir Engine"]) if quran_pdf_router is not None: fastapi_app.include_router(quran_pdf_router, prefix="/api", tags=["Quran PDF Library"]) # 8. Events & Mufti if events_router is not None: fastapi_app.include_router(events_router, tags=["Events"]) if mufti_router is not None: fastapi_app.include_router(mufti_router, prefix="/mufti", tags=["Mufti Verification"]) @fastapi_app.get("/health") async def health_check(): return { "status": "healthy", "service": "superadmin-backend", "version": "1.0.0" } @fastapi_app.get("/ads/config") async def get_public_ads_config(): try: config = await sendads_collection.find_one({"_id": "global_ad_config"}) if not config: return { "status": "ok", "ads_enabled": True, "real_ads_enabled": False, "min_seconds_between_ads": 180, "content_rating": "G", "banner_ads_enabled": False, "home_ads": True, "chat_ads": True, "hadith_ads": True, "tafsir_ads": True, "qaza_ads": True, } return { "status": "ok", "ads_enabled": config.get("ads_enabled", True), "real_ads_enabled": config.get("real_ads_enabled", False), "min_seconds_between_ads": config.get("min_seconds_between_ads", 180), "content_rating": config.get("content_rating", "G"), "banner_ads_enabled": config.get("banner_ads_enabled", False), "home_ads": config.get("home_ads", True), "chat_ads": config.get("chat_ads", True), "hadith_ads": config.get("hadith_ads", True), "tafsir_ads": config.get("tafsir_ads", True), "qaza_ads": config.get("qaza_ads", True), "instant_ad": config.get("instant_ad"), "updated_at": config.get("updated_at") } except Exception: return { "status": "ok", "ads_enabled": True, "real_ads_enabled": False, "min_seconds_between_ads": 180, "content_rating": "G", "banner_ads_enabled": False, "home_ads": True, "chat_ads": True, "hadith_ads": True, "tafsir_ads": True, "qaza_ads": True, } # Hook gr.routes.App.create_app so our routes are embedded inside demo.launch() try: if hasattr(gr, "routes") and hasattr(gr.routes, "App") and hasattr(gr.routes.App, "create_app"): _original_create_app = gr.routes.App.create_app def _custom_create_app(*args, **kwargs): app = _original_create_app(*args, **kwargs) _register_routes(app) return app gr.routes.App.create_app = _custom_create_app logger.info("Successfully hooked gr.routes.App.create_app for seamless FastAPI route injection.") except Exception as e: logger.warning(f"Could not hook gr.routes.App.create_app: {e}") # Build Gradio Status Dashboard for visual monitoring on Hugging Face with gr.Blocks(title="Sahih Islam SuperAdmin API", theme=gr.themes.Soft()) as demo: gr.Markdown("# 🕌 Sahih Islam SuperAdmin API Server") gr.Markdown("### ✅ Status: **ONLINE & OPERATIONAL** (Hugging Face Gradio SDK)") gr.Markdown(""" This Space hosts the SuperAdmin backend for **Sahih Islam**. #### 🚀 Quick Links: - **Interactive API Docs (Swagger)**: [/docs](/docs) - **ReDoc Documentation**: [/redoc](/redoc) - **Health Probe**: [/health](/health) - **Remote Ad Config Endpoint**: [/ads/config](/ads/config) - **SuperAdmin Routes**: `/admin/...` #### 🛡️ Authentication: All `/admin/*` operations require a valid Firebase ID Token with SuperAdmin privileges. """) # Interactive diagnostics connected to @spaces.GPU with gr.Accordion("⚡ Compute & ZeroGPU Hardware Status", open=False): status_input = gr.Textbox(label="Test Input", value="Ping") status_btn = gr.Button("Inspect GPU Compute Node") status_output = gr.Textbox(label="Compute Node Response") status_btn.click(fn=zero_gpu_compute_node, inputs=status_input, outputs=status_output) # Also attempt to register on demo.app if already instantiated if hasattr(demo, "app") and demo.app: _register_routes(demo.app) # Launch the Gradio app and keep the main process alive permanently if __name__ == "__main__": logger.info("Starting SuperAdmin API Server via demo.launch()...") demo.launch(server_name="0.0.0.0", server_port=7860) # Keep the server thread active permanently on Hugging Face Spaces (prevents exit) import time while True: time.sleep(3600)