app.py

#!/usr/bin/env python3
"""
BackgroundFX Pro - CSP-Safe Application Entry Point
Built for Hugging Face Spaces (strict Content Security Policy)
No inline JavaScript, CSP-compliant Gradio, safe environment vars, fallback AI models
"""

import early_env  # <<< must be FIRST to sanitize threading/env before anything else

import os
import logging
from pathlib import Path
from typing import Optional, Tuple, Dict, Any, Callable

# 1️⃣ Set CSP-safe environment variables BEFORE any imports (Gradio will see these)
os.environ['GRADIO_ALLOW_FLAGGING'] = 'never'
os.environ['GRADIO_ANALYTICS_ENABLED'] = 'False'
os.environ['GRADIO_SERVER_NAME'] = '0.0.0.0'
os.environ['GRADIO_SERVER_PORT'] = '7860'

# 2️⃣ Patch Gradio schema early for Hugging Face bug compatibility
try:
    import gradio_client.utils as gc_utils
    orig_get_type = gc_utils.get_type
    def patched_get_type(schema):
        if not isinstance(schema, dict):
            if isinstance(schema, bool):
                return "boolean"
            if isinstance(schema, str):
                return "string"
            if isinstance(schema, (int, float)):
                return "number"
            return "string"
        return orig_get_type(schema)
    gc_utils.get_type = patched_get_type
except Exception:
    pass  # No fatal error if Gradio patch fails

# 3️⃣ Set up logging early for debugging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"
)
logger = logging.getLogger("BackgroundFX")

# 4️⃣ Import your modular code (assuming your project structure)
from core.exceptions import ModelLoadingError, VideoProcessingError
from config.app_config import get_config
from utils.hardware.device_manager import DeviceManager
from utils.system.memory_manager import MemoryManager
from models.loaders.model_loader import ModelLoader
from processing.video.video_processor import CoreVideoProcessor, ProcessorConfig
from processing.audio.audio_processor import AudioProcessor
from utils.cv_processing import PROFESSIONAL_BACKGROUNDS, validate_video_file
# If you have TwoStage: from processing.two_stage.two_stage_processor import TwoStageProcessor, CHROMA_PRESETS

# 5️⃣ CSP-safe fallback model stubs
class CSPSafeSAM2:
    def set_image(self, image):
        self.shape = getattr(image, 'shape', (512, 512, 3))
    def predict(self, point_coords=None, point_labels=None, box=None, multimask_output=True, **kwargs):
        import numpy as np
        h, w = self.shape[:2] if hasattr(self, 'shape') else (512, 512)
        n = 3 if multimask_output else 1
        return np.ones((n, h, w), dtype=bool), np.array([0.9, 0.8, 0.7][:n]), np.ones((n, h, w), dtype=np.float32)

class CSPSafeMatAnyone:
    def step(self, image_tensor, mask_tensor=None, objects=None, first_frame_pred=False):
        import torch
        shape = getattr(image_tensor, 'shape', (1, 3, 256, 256))
        return torch.ones((shape[0], 1, shape[2], shape[3]))
    def output_prob_to_mask(self, output_prob):
        return (output_prob > 0.5).float()
    def process(self, image, mask):
        return mask

# 6️⃣ Application main processor object
class VideoBackgroundApp:
    def __init__(self):
        self.config = get_config()
        self.device_mgr = DeviceManager()
        self.memory_mgr = MemoryManager(self.device_mgr.get_optimal_device())
        self.model_loader = ModelLoader(self.device_mgr, self.memory_mgr)
        self.audio_proc = AudioProcessor()
        self.models_loaded = False
        self.core_processor = None

    def load_models(self, progress_callback: Optional[Callable]=None) -> str:
        logger.info("Loading models (CSP-safe)...")
        try:
            sam2, matanyone = self.model_loader.load_all_models(progress_callback=progress_callback)
        except Exception as e:
            logger.warning(f"Model loading failed ({e}) - Using CSP-safe fallbacks")
            sam2, matanyone = None, None
        sam2_model = getattr(sam2, "model", sam2) if sam2 else CSPSafeSAM2()
        matanyone_model = getattr(matanyone, "model", matanyone) if matanyone else CSPSafeMatAnyone()
        self.core_processor = CoreVideoProcessor(config=ProcessorConfig(), models=None)
        self.core_processor.models = type('FakeModelManager', (), {
            'get_sam2': lambda self: sam2_model,
            'get_matanyone': lambda self: matanyone_model
        })()
        self.models_loaded = True
        return "Models loaded (CSP-safe, fallback mode if no AI models loaded)."

    def process_video(self, video, bg_style, custom_bg_file):
        if not self.models_loaded:
            return None, "Models not loaded yet"
        import time
        output_path = f"/tmp/output_{int(time.time())}.mp4"
        cfg = PROFESSIONAL_BACKGROUNDS.get(bg_style, PROFESSIONAL_BACKGROUNDS["minimalist"])
        if custom_bg_file:
            cfg = {"custom_path": custom_bg_file.name}
        ok, msg = validate_video_file(video)
        if not ok:
            return None, f"Invalid video: {msg}"
        try:
            result = self.core_processor.process_video(
                input_path=video,
                output_path=output_path,
                bg_config=cfg
            )
            output_with_audio = self.audio_proc.add_audio_to_video(video, output_path)
            return output_with_audio, f"Processing complete ({result.get('frames', 'n/a')} frames, {bg_style})"
        except Exception as e:
            return None, f"Processing failed: {e}"

# 7️⃣ Gradio interface CSP-safe
def create_csp_safe_gradio():
    import gradio as gr
    app = VideoBackgroundApp()
    with gr.Blocks(
        title="BackgroundFX Pro - CSP Safe",
        analytics_enabled=False,
        css="""
        .gradio-container { max-width: 1000px; margin: auto; }
        """
    ) as demo:
        gr.Markdown("# 🎬 BackgroundFX Pro (CSP-Safe)")
        gr.Markdown("Replace your video background with cinema-quality AI matting. Built for Hugging Face Spaces CSP.")

        with gr.Row():
            with gr.Column():
                video = gr.Video(label="Upload Video")
                bg_style = gr.Dropdown(
                    choices=list(PROFESSIONAL_BACKGROUNDS.keys()),
                    value="minimalist",
                    label="Background Style"
                )
                custom_bg = gr.File(label="Custom Background (Optional)", file_types=["image"])
                btn_load = gr.Button("🔄 Load Models", variant="secondary")
                btn_run = gr.Button("🎬 Process Video", variant="primary")

            with gr.Column():
                status = gr.Textbox(label="Status", lines=4)
                out_video = gr.Video(label="Processed Video")

        def safe_load():
            return app.load_models()
        def safe_process(vid, style, custom_bg_file):
            return app.process_video(vid, style, custom_bg_file)
        btn_load.click(fn=safe_load, outputs=[status])
        btn_run.click(fn=safe_process, inputs=[video, bg_style, custom_bg], outputs=[out_video, status])

    return demo

# 8️⃣ Main entry point (for Hugging Face Spaces)
if __name__ == "__main__":
    logger.info("Launching CSP-safe Gradio interface for Hugging Face Spaces")
    demo = create_csp_safe_gradio()
    demo.queue().launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True,
        debug=False,
        inbrowser=False
    )