app.py

from __future__ import annotations

import os
import sys
import time
import traceback
from typing import Any

import gradio as gr
from fastapi import File, UploadFile
from fastapi.responses import RedirectResponse
from starlette.middleware import Middleware
from fastapi.middleware.cors import CORSMiddleware

from src.api.main import detect_image as api_detect_image
from src.api.main import detect_video as api_detect_video
from src.api.main import health as api_health
from src.api.main import health_models as api_health_models
from src.types import DetectionResponse


_default_unraisablehook = sys.unraisablehook


def _suppress_asyncio_fd_shutdown_noise(unraisable) -> None:
    exc = unraisable.exc_value
    if (
        isinstance(exc, ValueError)
        and str(exc) == "Invalid file descriptor: -1"
        and getattr(unraisable.object, "__name__", "") == "__del__"
    ):
        return
    _default_unraisablehook(unraisable)


sys.unraisablehook = _suppress_asyncio_fd_shutdown_noise


def _install_excepthook() -> None:
    def handle_exception(exc_type, exc_value, exc_tb):
        traceback.print_exception(exc_type, exc_value, exc_tb)

    sys.excepthook = handle_exception


def _normalize_gradio_env() -> None:
    os.environ["GRADIO_SSR_MODE"] = "False"
    os.environ.pop("GRADIO_NODE_SERVER_PORT", None)


def _build_demo() -> gr.Blocks:
    with gr.Blocks(title="GenAI-DeepDetect") as demo:
        gr.Markdown(
            """
            # GenAI-DeepDetect

            Gradio frontend with API routes attached to the same app for your external frontend.

            Available API endpoints:
            - `GET /api/health`
            - `GET /api/health/models`
            - `POST /api/detect/image`
            - `POST /api/detect/video`
            """
        )
    return demo


def _attach_api_routes(app: Any) -> None:
    async def health() -> dict:
        return await api_health()

    async def api_health_route() -> dict:
        return await health()

    async def health_models() -> dict[str, object]:
        return await api_health_models()

    async def api_health_models_route() -> dict[str, object]:
        return await health_models()

    async def detect_image(file: UploadFile = File(...)) -> DetectionResponse:
        return await api_detect_image(file)

    async def api_detect_image_route(file: UploadFile = File(...)) -> DetectionResponse:
        return await detect_image(file)

    async def detect_video(file: UploadFile = File(...)) -> DetectionResponse:
        return await api_detect_video(file)

    async def api_detect_video_route(file: UploadFile = File(...)) -> DetectionResponse:
        return await detect_video(file)

    async def gradio_compat_redirect() -> RedirectResponse:
        return RedirectResponse(url="/", status_code=307)

    app.add_api_route("/health", health, methods=["GET"])
    app.add_api_route("/api/health", api_health_route, methods=["GET"])
    app.add_api_route("/health/models", health_models, methods=["GET"])
    app.add_api_route("/api/health/models", api_health_models_route, methods=["GET"])
    app.add_api_route(
        "/detect/image",
        detect_image,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route(
        "/api/detect/image",
        api_detect_image_route,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route(
        "/detect/video",
        detect_video,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route(
        "/api/detect/video",
        api_detect_video_route,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route("/gradio", gradio_compat_redirect, methods=["GET"])


demo = _build_demo().queue()


if __name__ == "__main__":
    _install_excepthook()
    _normalize_gradio_env()
    try:
        app, _, _ = demo.launch(
            prevent_thread_lock=True,
            show_error=True,
            ssr_mode=False,
            app_kwargs={
                "docs_url": "/docs",
                "redoc_url": "/redoc",
                "middleware": [
                    Middleware(
                        CORSMiddleware,
                        allow_origins=["*"],
                        allow_methods=["*"],
                        allow_headers=["*"],
                    )
                ],
            },
        )
        _attach_api_routes(app)
        while True:
            time.sleep(60)
    except Exception:
        traceback.print_exc()
        sys.exit(1)