app.py

import os
import base64
import json
import zlib
import importlib.machinery
import importlib.util
from pathlib import Path
from typing import Any, Dict
import gradio as gr

# --- CONFIGURATION PROTOCOLS (OMNICFG1) ---
_CONFIG_MARKER = b"OMNICFG1"

def _collect_config_values(namespace: Dict[str, Any]) -> Dict[str, Any]:
    data: Dict[str, Any] = {}
    for key, value in namespace.items():
        if key.isupper() and key != "APP_RUNTIME_CONFIG":
            data[key] = value
    class_obj = namespace.get("APP_RUNTIME_CONFIG")
    if class_obj is not None:
        for key in dir(class_obj):
            if key.isupper() and key not in data:
                data[key] = getattr(class_obj, key)
    return data

def _load_config_from_source(content: str) -> Dict[str, Any]:
    scope: Dict[str, Any] = {"__builtins__": __builtins__, "__name__": "omni_private_config"}
    exec(content, scope, scope)
    return _collect_config_values(scope)

def _write_packed_private_config(path: Path, data: Dict[str, Any]) -> None:
    payload = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    packed = _CONFIG_MARKER + zlib.compress(payload, level=9)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(packed)

def _materialize_private_config_from_env() -> None:
    repo_dir = Path(__file__).resolve().parent
    pyc_target = repo_dir / "src" / "config.pyc"
    raw_b64 = (os.getenv("OMNI_PRIVATE_CONFIG_B64") or "").strip()
    raw_py = (os.getenv("OMNI_PRIVATE_CONFIG_PY") or "").strip()
    content = ""
    if raw_b64:
        try:
            content = base64.b64decode(raw_b64.encode("utf-8")).decode("utf-8")
        except Exception:
            return
    elif raw_py:
        content = raw_py
    if content:
        try:
            data = _load_config_from_source(content.rstrip() + "\n")
            _write_packed_private_config(pyc_target, data)
        except Exception:
            pass

def _load_runtime_module():
    repo_dir = Path(__file__).resolve().parent
    try:
        from src import app_lib as module
        return module
    except ImportError:
        return None

# --- INITIALIZATION ---
_materialize_private_config_from_env()
_RUNTIME = _load_runtime_module()

# --- VIDEO INTERFACE BRIDGE ---
# อัปเดต URL เป็น Direct Link เพื่อป้องกันปัญหาหน้าจอสีเทาจากการบล็อก iframe
SOURCE_VIDEO_URL = "https://saravutw-wan2-2-i2v-lightning-4-8step-custom.hf.space"

def build_video_factory_ui():
    # อัปเดต title ของหน้าต่างแอปพลิเคชันให้สอดคล้องกับ Space ใหม่
    with gr.Blocks(fill_height=True, title="WAN2.2 I2V Lightning Bridge") as demo:
        gr.HTML(f"""
            <div style="width: 100%; height: 100vh;">
                <iframe 
                    src="{SOURCE_VIDEO_URL}" 
                    frameborder="0" 
                    width="100%" 
                    height="100%" 
                    style="min-height: 1000px; border-radius: 4px;"
                    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" 
                    allowfullscreen>
                </iframe>
            </div>
        """)
    return demo

if __name__ == "__main__":
    # ปรับแต่ง Startup Log ให้สื่อความหมายตรงกับโมเดลตัวใหม่
    print("[startup] Launching WAN2.2 I2V Lightning Runtime...")
                
    # Background hooks execution
    if _RUNTIME:
        for hook in ["kickoff_runtime_prepare_background", "kickoff_model_prepare_background"]:
            func = getattr(_RUNTIME, hook, None)
            if callable(func):
                print(f"[startup] {hook}: {func()}")
                        
    # เริ่มต้นการทำงานของระบบ Gradio
    demo = build_video_factory_ui()
    demo.launch(
        share=True,
        server_name="0.0.0.0",
        server_port=int(os.getenv("PORT", "7860")),
        inline=False
    )