Create app.py

app.py

@@ -0,0 +1,201 @@
+# app.py
+# AI Video Enhancer 4K - Gradio app for Hugging Face Spaces
+
+import os
+import shutil
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+from typing import Tuple
+
+import gradio as gr
+from PIL import Image
+
+# Optional: advanced AI enhancers (Real-ESRGAN, GFPGAN)
+try:
+    import torch
+    from gfpgan import GFPGANer  # type: ignore
+    from realesrgan import RealESRGAN  # type: ignore
+    HAVE_ENHANCERS = True
+except Exception:
+    HAVE_ENHANCERS = False
+
+# Config
+MAX_SECONDS = 30
+TEMP_DIR = Path(tempfile.gettempdir()) / "hf_video_enhancer"
+TEMP_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def run_cmd(cmd):
+    p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if p.returncode != 0:
+        raise RuntimeError(f"Command failed: {cmd}\n{p.stderr.decode()}")
+    return p.stdout.decode()
+
+
+def probe_video(video_path: str) -> Tuple[float, int, int]:
+    cmd = [
+        "ffprobe", "-v", "error",
+        "-select_streams", "v:0",
+        "-show_entries", "stream=width,height,duration",
+        "-of", "default=noprint_wrappers=1:nokey=0",
+        video_path
+    ]
+    p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    out = p.stdout.decode()
+    width = height = 0
+    duration = 0.0
+    for line in out.splitlines():
+        if line.startswith("width="):
+            width = int(line.split("=")[1])
+        if line.startswith("height="):
+            height = int(line.split("=")[1])
+        if line.startswith("duration="):
+            try:
+                duration = float(line.split("=")[1])
+            except:
+                duration = 0.0
+    return duration, width, height
+
+
+def extract_frames(video_path: str, frames_dir: Path):
+    frames_dir.mkdir(parents=True, exist_ok=True)
+    cmd = [
+        "ffmpeg", "-y", "-i", video_path,
+        "-vsync", "0",
+        str(frames_dir / "%06d.png")
+    ]
+    run_cmd(cmd)
+
+
+def reassemble_video(frames_dir: Path, audio_src: str, out_path: str):
+    tmp_video = str(frames_dir.parent / "tmp_video.mp4")
+    cmd_encode = [
+        "ffmpeg", "-y", "-framerate", "30",
+        "-i", str(frames_dir / "%06d.png"),
+        "-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p",
+        tmp_video
+    ]
+    run_cmd(cmd_encode)
+
+    # Check if source has audio
+    p = subprocess.run(
+        ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_type", "-of", "default=noprint_wrappers=1", audio_src],
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    )
+    has_audio = bool(p.stdout.decode().strip())
+
+    if has_audio:
+        cmd_mux = [
+            "ffmpeg", "-y",
+            "-i", tmp_video,
+            "-i", audio_src,
+            "-c:v", "copy", "-c:a", "aac",
+            "-map", "0:v:0", "-map", "1:a:0",
+            out_path
+        ]
+        run_cmd(cmd_mux)
+        os.remove(tmp_video)
+    else:
+        shutil.move(tmp_video, out_path)
+
+
+def simple_upscale_with_ffmpeg(frames_dir: Path, scale_factor: int = 2):
+    for p in sorted(frames_dir.glob("*.png")):
+        tmp = str(p) + ".tmp.png"
+        cmd = [
+            "ffmpeg", "-y", "-i", str(p),
+            "-vf", f"scale=iw*{scale_factor}:ih*{scale_factor}:flags=lanczos",
+            tmp
+        ]
+        run_cmd(cmd)
+        os.replace(tmp, p)
+
+
+def enhance_frames(frames_dir: Path):
+    # fallback to simple upscale for now (Real-ESRGAN requires GPU + weights)
+    simple_upscale_with_ffmpeg(frames_dir, scale_factor=2)
+
+
+def process_video(video_file) -> Tuple[str, str]:
+    """
+    Accepts uploaded video path from Gradio,
+    processes it and returns (message, path_to_result_video)
+    """
+    ts = int(time.time() * 1000)
+    base_dir = TEMP_DIR / f"job_{ts}"
+    base_dir.mkdir(parents=True, exist_ok=True)
+    in_path = base_dir / "input_video"
+
+    try:
+        shutil.copy(video_file, in_path)  # ✅ FIXED: treat input as path, not .read()
+    except Exception as e:
+        return f"Error saving uploaded file: {e}", ""
+
+    try:
+        duration, w, h = probe_video(str(in_path))
+    except Exception as e:
+        shutil.rmtree(base_dir, ignore_errors=True)
+        return f"Error probing video: {e}", ""
+
+    if duration > MAX_SECONDS:
+        shutil.rmtree(base_dir, ignore_errors=True)
+        return f"Video too long: {duration:.1f}s (limit {MAX_SECONDS}s).", ""
+
+    frames_dir = base_dir / "frames"
+    try:
+        extract_frames(str(in_path), frames_dir)
+    except Exception as e:
+        shutil.rmtree(base_dir, ignore_errors=True)
+        return f"Failed extracting frames: {e}", ""
+
+    try:
+        enhance_frames(frames_dir)
+    except Exception as e:
+        print(f"Enhancement failed: {e}")
+
+    out_video = base_dir / "enhanced_output.mp4"
+    try:
+        reassemble_video(frames_dir, str(in_path), str(out_video))
+    except Exception as e:
+        shutil.rmtree(base_dir, ignore_errors=True)
+        return f"Failed to reassemble video: {e}", ""
+
+    try:
+        shutil.rmtree(frames_dir)
+    except Exception:
+        pass
+
+    return "Processing complete. Download below.", str(out_video)
+
+
+# Gradio UI
+with gr.Blocks(title="AI Video Enhancer 4K") as demo:
+    gr.Markdown("# AI Video Enhancer 4K")
+    gr.Markdown("Upload a short video (<= 30s). It will be upscaled using AI/ffmpeg.")
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            video_in = gr.File(label="Upload video (mp4/avi/mov)", file_types=[".mp4", ".avi", ".mov"])
+            btn = gr.Button("Enhance Video")
+            status = gr.Textbox(label="Status", interactive=False)
+        with gr.Column(scale=1):
+            out_video = gr.Video(label="Enhanced video")
+
+    def on_click_process(file_obj):
+        if not file_obj:
+            return "Please upload a video file.", None
+        try:
+            msg, path = process_video(file_obj)
+            if path:
+                return msg, path
+            else:
+                return msg, None
+        except Exception as e:
+            return f"Unexpected error: {e}", None
+
+    btn.click(fn=on_click_process, inputs=[video_in], outputs=[status, out_video])
+
+if __name__ == "__main__":
+    demo.launch()