""" app.py — CineAI Gradio 5.x compatible. CPU-only. HF Spaces ready. """ from __future__ import annotations import gc import os import sys from pathlib import Path from typing import Dict, Generator, List, Optional # ── Paths ───────────────────────────────────────────────────────────────────── ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT)) for _d in ["/tmp/cineai", "/tmp/cineai_models"]: os.makedirs(_d, exist_ok=True) # ── Env (before any model import) ──────────────────────────────────────────── try: from dotenv import load_dotenv load_dotenv(ROOT / ".env", override=False) except ImportError: pass os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") os.environ.setdefault("OMP_NUM_THREADS", "4") os.environ.setdefault("MKL_NUM_THREADS", "4") os.environ.setdefault("TRANSFORMERS_CACHE", "/tmp/cineai_models") os.environ.setdefault("HF_HOME", "/tmp/cineai_models") os.environ.setdefault("TORCH_HOME", "/tmp/cineai_models/torch") os.environ.setdefault("CINEAI_TEMP_DIR", "/tmp/cineai") # ── Torch first ─────────────────────────────────────────────────────────────── try: import torch torch.set_num_threads(int(os.environ.get("OMP_NUM_THREADS", "4"))) _CUDA = torch.cuda.is_available() except ImportError: _CUDA = False # ── Silence noise ──────────────────────────────────────────────────────────── import logging as _lg for _n in ["transformers", "sentence_transformers", "httpx", "httpcore"]: _lg.getLogger(_n).setLevel(_lg.ERROR) # ── Logger ──────────────────────────────────────────────────────────────────── from loguru import logger logger.remove() logger.add(sys.stderr, format="{time:HH:mm:ss} | {level:<8} | {message}", level=os.environ.get("LOG_LEVEL", "INFO"), colorize=True) logger.add("/tmp/cineai/app.log", rotation="5 MB", retention=1, level="DEBUG", enqueue=True) # ── Gradio ──────────────────────────────────────────────────────────────────── import gradio as gr logger.info(f"Gradio {gr.__version__} | CUDA={_CUDA} | Python {sys.version.split()[0]}") # ── Project modules ─────────────────────────────────────────────────────────── from core.model_manager import ModelManager, ModelKey # noqa: E402 from core.memory import session_registry # noqa: E402 from core.video_processor import VideoProcessor # noqa: E402 from core.speech_to_text import SpeechToText # noqa: E402 from core.vision_analyzer import VisionAnalyzer # noqa: E402 from core.translator import Translator # noqa: E402 from core.chatbot import CineAIChatbot # noqa: E402 from core.tts_engine import TTSEngine # noqa: E402 from utils.session_manager import SessionManager, new_session_id # noqa: E402 from utils.utils import ( # noqa: E402 check_environment, extract_video_title, file_size_mb, format_duration, format_system_status, free_memory, memory_status_card, video_info_card, ) _vp = VideoProcessor() _stt = SpeechToText() _vision = VisionAnalyzer() _xlate = Translator() _chatbot = CineAIChatbot() _tts = TTSEngine() # ============================================================================= # Pipeline # ============================================================================= def _run_process(video_file, session_id: str, progress: gr.Progress = gr.Progress()) -> tuple: if video_file is None: return "⚠️ Upload a video first.", "", {}, "" video_path = video_file if isinstance(video_file, str) else str(video_file) if not Path(video_path).exists(): return f"❌ File not found: `{video_path}`", "", {}, "" mgr = SessionManager(session_id) mgr.get_dir(); mgr.write_marker() memory = session_registry.get_or_create(session_id) memory.clear() title = extract_video_title(Path(video_path).name) def prog(msg: str, pct: float): progress(pct, desc=msg) try: prog("📦 Processing video…", 0.05) meta = _vp.process(video_path, session_id, prog) prog("🎤 Transcribing…", 0.35) transcript = _stt.transcribe(meta.audio_path, memory=memory, progress_cb=prog) if os.environ.get("ENABLE_VISION", "true").lower() == "true": prog("👁️ Analysing scenes…", 0.62) step = max(1, len(meta.frames) // 50) _vision.analyze_frames(meta.frames[::step][:50], memory=memory, progress_cb=prog) prog("📝 Building index…", 0.87) auto_sum = "" if len(transcript.full_text) > 100: auto_sum = _chatbot.summarize_scene(transcript.full_text[:2500], memory=memory) prog("✅ Complete!", 1.0) meta_dict = { "title": title, "video_path": meta.video_path, "audio_path": meta.audio_path, "duration_sec": meta.duration_sec, "fps": meta.fps, "width": meta.width, "height": meta.height, "frame_count": len(meta.frames), "transcript_lang": transcript.language, "transcript_full": transcript.full_text[:8000], "auto_summary": auto_sum, "session_id": session_id, } stats = memory.stats() status = (f"✅ **{title}** processed!\n\n" f"- 🎤 Language: `{transcript.language}`\n" f"- 🧠 Memory chunks: `{stats['total_chunks']}`\n" f"- ⏱️ Duration: `{format_duration(meta.duration_sec)}`\n\n" f"*Ask me anything below ↓*") info = video_info_card(title, meta.duration_sec, meta.fps, (meta.width, meta.height), len(meta.frames), file_size_mb(meta.video_path)) free_memory() return info, status, meta_dict, transcript.full_text[:6000] except Exception as e: logger.exception(f"Processing failed: {e}") return f"❌ **Failed**: `{e}`", "Check the file and try again.", {}, "" def _chat_stream(user_msg: str, history: List[Dict], session_id: str, meta_dict: dict, hindi_mode: bool) -> Generator[tuple, None, None]: if not user_msg.strip(): yield history, ""; return if not meta_dict: yield history + [ {"role": "user", "content": user_msg}, {"role": "assistant", "content": "⚠️ Upload and process a video first."}, ], ""; return memory = session_registry.get_or_create(session_id) history = history + [{"role": "user", "content": user_msg}, {"role": "assistant", "content": ""}] yield history, "" resp = "" try: for token in _chatbot.stream_response( user_msg, memory, meta_dict.get("title", ""), meta_dict.get("duration_sec", 0.0), hindi_mode, ): resp += token history[-1]["content"] = resp yield history, "" except Exception as e: history[-1]["content"] = f"❌ Error: {e}" logger.exception(e) yield history, "" return try: SessionManager(session_id).write_marker() except Exception: pass def _speak(history: List[Dict], session_id: str) -> Optional[str]: if not history: return None last = next((m["content"] for m in reversed(history) if m["role"] == "assistant"), None) if not last: return None try: out = str(SessionManager(session_id).get_path("tts_response.wav")) return _tts.synthesize(last, out) except Exception as e: logger.warning(f"TTS: {e}") return None def _translate_last(history: List[Dict]) -> str: last = next((m["content"] for m in reversed(history) if m["role"] == "assistant"), "") return _xlate.translate_to_hindi(last) if last else "No response to translate yet." def _delete_session(session_id: str) -> tuple: try: session_registry.destroy(session_id) SessionManager(session_id).delete() msg = "✅ All session data deleted." except Exception as e: msg = f"⚠️ Error: {e}" return msg, new_session_id(), {}, [], "" # ============================================================================= # UI # ============================================================================= CSS = """ .gradio-container{max-width:1380px!important;margin:0 auto!important} #hdr{background:linear-gradient(135deg,#1a0533 0%,#0d0c22 50%,#1e1b4b 100%); border-radius:14px;padding:22px 28px;margin-bottom:12px; border:1px solid rgba(139,92,246,.35); box-shadow:0 0 40px rgba(109,40,217,.18);text-align:center} #hdr h1{font-size:clamp(1.4rem,4vw,2.1rem);font-weight:800; background:linear-gradient(90deg,#a78bfa 0%,#60a5fa 50%,#34d399 100%); -webkit-background-clip:text;-webkit-text-fill-color:transparent; background-clip:text;margin:0 0 6px} #hdr p{color:#94a3b8;margin:0;font-size:.93rem} #up{border:2px dashed #7c3aed!important;border-radius:10px!important;min-height:100px!important} @media(max-width:700px){.gradio-row{flex-direction:column!important}} """ PRESETS = [ "What happened in this video? Give a complete overview.", "Who are all the main characters? Describe each one.", "Explain the ending in detail. What did it mean?", "What are the hidden meanings, symbolism, and themes?", "Summarize the last 10 minutes of the video.", "Translate the main dialogue to Hindi.", "What is the IMDb rating and critical reception?", "Explain the most important scene in this video.", ] PRESET_LABELS = [ "📋 What happened?", "👥 Characters?", "🎭 Explain ending", "💡 Hidden meanings", "⏱️ Last 10 min", "🌐 Translate", "⭐ IMDb & rating", "😮 Key scene", ] def build_ui() -> gr.Blocks: theme = gr.themes.Base( primary_hue="violet", secondary_hue="indigo", neutral_hue="slate", font=["Inter", "ui-sans-serif", "system-ui"], ) with gr.Blocks(title="CineAI", theme=theme, css=CSS, analytics_enabled=False) as demo: session_id_st = gr.State(value=new_session_id) meta_dict_st = gr.State(value={}) gr.HTML('

🎬 CineAI — AI Video Assistant

' '

Upload any movie · AI understands scenes, dialogue & characters' ' · Chat in English or Hindi · 100% Free & Private

') with gr.Row(equal_height=False): with gr.Column(scale=4, min_width=290): gr.Markdown("### 📁 Upload Video") video_upload = gr.File( label="Drop video or click to browse", file_types=["video", ".mp4", ".mkv", ".avi", ".mov", ".webm", ".flv", ".wmv", ".m4v"], elem_id="up", ) video_preview = gr.Video(label="Preview", visible=False, height=190, interactive=False) with gr.Row(): process_btn = gr.Button("⚡ Process Video", variant="primary", scale=3) clear_btn = gr.Button("✕ Clear", variant="secondary", scale=1) proc_status = gr.Markdown("*Upload a video and click Process to begin.*") with gr.Accordion("📊 Video Info", open=False): video_info_md = gr.Markdown("*No video processed yet.*") with gr.Accordion("🧠 Memory", open=False): mem_md = gr.Markdown("*No session yet.*") gr.Button("🔄 Refresh", size="sm").click( fn=lambda s: memory_status_card( session_registry.get_or_create(s).stats()), inputs=[session_id_st], outputs=[mem_md]) with gr.Accordion("💾 System", open=False): sys_md = gr.Markdown(format_system_status()) gr.Button("🔄 Refresh", size="sm").click( fn=format_system_status, outputs=[sys_md]) gr.Markdown("---") delete_btn = gr.Button("🔐 Delete All Session Data", variant="stop") delete_msg = gr.Markdown("") with gr.Column(scale=6, min_width=370): with gr.Tabs(): with gr.Tab("💬 Chat"): chatbot = gr.Chatbot( label="CineAI", height=415, type="messages", # Gradio 5 mandatory show_copy_button=True, bubble_full_width=False, placeholder=( "
" "

🎬 Upload & process a video, then ask me anything!

" "
"), ) with gr.Row(): chat_in = gr.Textbox( placeholder="Ask anything… (English or Hindi)", show_label=False, scale=5, container=False, autofocus=True) send_btn = gr.Button("Send ➤", variant="primary", scale=1, min_width=80) with gr.Row(): hindi_cb = gr.Checkbox(label="🇮🇳 Hindi mode", value=False, scale=2) speak_btn = gr.Button("🔊 Speak", variant="secondary", scale=1) clr_btn = gr.Button("🧹 Clear", variant="secondary", scale=1) tts_audio = gr.Audio(label="Voice Response", visible=False, autoplay=True, type="filepath", show_download_button=True) gr.Markdown("**⚡ Quick Questions:**") with gr.Row(): qbtns_a = [gr.Button(PRESET_LABELS[i], size="sm") for i in range(4)] with gr.Row(): qbtns_b = [gr.Button(PRESET_LABELS[i], size="sm") for i in range(4, 8)] with gr.Tab("📋 Summary"): summary_out = gr.Markdown("*Click below to generate a full AI summary.*") gr.Button("📝 Generate Full Summary", variant="primary").click( fn=lambda s, m: ( _chatbot.generate_full_summary( session_registry.get_or_create(s), m.get("title", "")) if m else "⚠️ No video processed yet."), inputs=[session_id_st, meta_dict_st], outputs=[summary_out]) with gr.Tab("🇮🇳 Hindi"): hindi_out = gr.Textbox(label="Hindi Translation", lines=7, interactive=False, show_copy_button=True) gr.Button("🔄 Translate Last Response", variant="primary").click( fn=_translate_last, inputs=[chatbot], outputs=[hindi_out]) custom_txt = gr.Textbox(label="Or type any text:", lines=3) gr.Button("Translate ➤", variant="secondary").click( fn=lambda t: _xlate.translate_to_hindi(t) if t.strip() else "", inputs=[custom_txt], outputs=[hindi_out]) with gr.Tab("📄 Transcript"): transcript_out = gr.Textbox( label="Full Transcript", lines=22, interactive=False, show_copy_button=True, placeholder="Transcript appears here after processing…") # ── Events ──────────────────────────────────────────────────────────── video_upload.change( fn=lambda f: gr.update(value=f, visible=True) if f else gr.update(visible=False), inputs=[video_upload], outputs=[video_preview]) process_btn.click( fn=_run_process, inputs=[video_upload, session_id_st], outputs=[video_info_md, proc_status, meta_dict_st, transcript_out]) clear_btn.click( fn=lambda: (None, gr.update(visible=False), "*Upload a video.*", {}), outputs=[video_upload, video_preview, proc_status, meta_dict_st]) _si = [chat_in, chatbot, session_id_st, meta_dict_st, hindi_cb] _so = [chatbot, chat_in] send_btn.click(fn=_chat_stream, inputs=_si, outputs=_so) chat_in.submit(fn=_chat_stream, inputs=_si, outputs=_so) for btn, question in zip(qbtns_a + qbtns_b, PRESETS): btn.click( fn=lambda h, s, m, hi, _q=question: _chat_stream(_q, h, s, m, hi), inputs=[chatbot, session_id_st, meta_dict_st, hindi_cb], outputs=_so) speak_btn.click( fn=lambda h, s: ( gr.update(value=_speak(h, s), visible=True) if _speak(h, s) else gr.update(visible=False)), inputs=[chatbot, session_id_st], outputs=[tts_audio]) clr_btn.click(fn=lambda: ([], ""), outputs=[chatbot, chat_in]) delete_btn.click( fn=_delete_session, inputs=[session_id_st], outputs=[delete_msg, session_id_st, meta_dict_st, chatbot, chat_in]) return demo # ============================================================================= # Entry point # ============================================================================= if __name__ == "__main__": logger.info("=== CineAI Starting ===") env = check_environment() for k, v in env.items(): logger.info(f" {'✅' if v else '❌'} {k}") demo = build_ui() demo.queue() # Gradio 5: no concurrency_count arg demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True, share=False, max_threads=6)