Blame for app.py

import gradio as gr
import subprocess
import tempfile
import shutil
import os
import struct

# ─────────────────────────────────────────────
# MP4 ATOM PATCHING (Python port of C# code)
# ─────────────────────────────────────────────

def index_of_sequence(buffer: bytes, pattern: bytes, start: int) -> int:
    if not buffer or not pattern or len(pattern) > len(buffer) - start:
        return -1
    for i in range(start, len(buffer) - len(pattern) + 1):
        if buffer[i:i+len(pattern)] == pattern:
            return i
    return -1


def patch_atom(atom_name: str, data: bytearray, scale_factor: float | None, log: list) -> int:
    patch_count = 0
    start_index = 0
    atom_bytes = atom_name.encode("utf-8")

    while True:
        found_index = index_of_sequence(bytes(data), atom_bytes, start_index)
        if found_index == -1:
            break

        header_offset = found_index - 4
        if header_offset < 0:
            start_index = found_index + 4
            continue

        if header_offset + 4 > len(data):
            start_index = found_index + 4
            continue

        box_size = struct.unpack_from(">I", data, header_offset)[0]
        if box_size < 8:
            start_index = found_index + 4
            continue

        if header_offset + 8 >= len(data):
            start_index = found_index + 4
            continue

        version = data[header_offset + 8]

        if version == 0:
            ts_off, dur_off = header_offset + 20, header_offset + 24
            if dur_off + 4 > len(data):
                start_index = found_index + 4
                continue
            old_ts  = struct.unpack_from(">I", data, ts_off)[0]
            old_dur = struct.unpack_from(">I", data, dur_off)[0]
            chosen  = scale_factor if scale_factor is not None else (30000.0 / old_ts)
            new_ts  = int(old_ts  * chosen)
            new_dur = int(old_dur * chosen)
            struct.pack_into(">I", data, ts_off,  new_ts)
            struct.pack_into(">I", data, dur_off, new_dur)
            log.append(f"  ✔ Patched '{atom_name}' (v0): timescale {old_ts}→{new_ts}, duration {old_dur}→{new_dur}")
            patch_count += 1

        elif version == 1:
            ts_off, dur_off = header_offset + 28, header_offset + 32
            if dur_off + 8 > len(data):
                start_index = found_index + 4
                continue
            old_ts  = struct.unpack_from(">I", data, ts_off)[0]
            old_dur = struct.unpack_from(">Q", data, dur_off)[0]
            chosen  = scale_factor if scale_factor is not None else (30000.0 / old_ts)
            new_ts  = int(old_ts  * chosen)
            new_dur = int(old_dur * chosen)
            struct.pack_into(">I", data, ts_off,  new_ts)
            struct.pack_into(">Q", data, dur_off, new_dur)
            log.append(f"  ✔ Patched '{atom_name}' (v1): timescale {old_ts}→{new_ts}, duration {old_dur}→{new_dur}")
            patch_count += 1
        else:
            log.append(f"  ⚠ '{atom_name}' (v{version}) skipped — unknown version")

        start_index = found_index + 4

    return patch_count


def patch_mp4(input_path: str, output_path: str, scale_factor: float | None) -> list[str]:
    log = []
    with open(input_path, "rb") as f:
        data = bytearray(f.read())
    log.append(f"📂 Read {len(data):,} bytes")

    n_mvhd = patch_atom("mvhd", data, scale_factor, log)
    n_mdhd = patch_atom("mdhd", data, scale_factor, log)
    total  = n_mvhd + n_mdhd

    if total == 0:
        log.append("⚠ No atoms patched — file may be unchanged")
    else:
        log.append(f"✅ Total atoms patched: {total}")

    with open(output_path, "wb") as f:
        f.write(data)
    log.append(f"💾 Saved to {output_path}")
    return log


# ─────────────────────────────────────────────
# FFMPEG HELPERS
# ─────────────────────────────────────────────

def ffmpeg_available() -> bool:
    return shutil.which("ffmpeg") is not None


def detect_fps(input_path: str) -> float:
    try:
        result = subprocess.run(
            ["ffprobe", "-v", "error", "-select_streams", "v:0",
             "-show_entries", "stream=r_frame_rate",
             "-of", "default=noprint_wrappers=1:nokey=1", input_path],
            capture_output=True, text=True, timeout=30
        )
        num, den = result.stdout.strip().split("/")
        return round(float(num) / float(den), 2)
    except Exception:
        return 0.0


def get_video_info(input_path: str) -> dict:
    try:
        import json
        result = subprocess.run(
            ["ffprobe", "-v", "error", "-print_format", "json",
             "-show_streams", "-show_format", input_path],
            capture_output=True, text=True, timeout=30
        )
        data   = json.loads(result.stdout)
        fmt    = data.get("format", {})
        vstream = next((s for s in data.get("streams", []) if s.get("codec_type") == "video"), {})
        return {
            "duration": float(fmt.get("duration", 0)),
            "size_mb":  round(os.path.getsize(input_path) / 1e6, 2),
            "width":    vstream.get("width", "?"),
            "height":   vstream.get("height", "?"),
            "codec":    vstream.get("codec_name", "?"),
            "fps":      vstream.get("r_frame_rate", "?"),
        }
    except Exception:
        return {}


def reencode_video(input_path: str, output_path: str,
                   crf: int, fps_out: float,
                   use_nvenc: bool, log: list) -> bool:
    codec  = "h264_nvenc" if use_nvenc else "libx264"
    preset = "p4"         if use_nvenc else "slow"

    cmd = [
        "ffmpeg", "-y", "-i", input_path,
        "-c:v", codec,
        "-preset", preset,
        "-r", str(fps_out),
        "-pix_fmt", "yuv420p",
        "-movflags", "+faststart",
    ]
    if use_nvenc:
        cmd += ["-cq", str(crf)]
    else:
        cmd += ["-crf", str(crf)]

    cmd += ["-c:a", "aac", "-b:a", "192k", output_path]
    log.append(f"▶ FFmpeg: {' '.join(cmd)}")

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
        if result.returncode != 0:
            log.append(f"❌ FFmpeg error:\n{result.stderr[-800:]}")
            return False
        log.append("✅ FFmpeg encoding complete")
        return True
    except subprocess.TimeoutExpired:
        log.append("❌ FFmpeg timed out")
        return False


# ─────────────────────────────────────────────
# MAIN PIPELINE
# ─────────────────────────────────────────────

FPS_SCALE = {
    "120fps → 60fps": (120, 0.5),
    "120fps → 30fps": (120, 0.25),
    "90fps → 60fps":  (90,  0.5),
    "90fps → 30fps":  (90,  0.25),
    "60fps → 60fps":  (60,  1.0),
    "60fps → 30fps":  (60,  0.5),
    "30fps → 30fps":  (30,  1.0),
    "Auto (target 30k timescale)": (None, None),
}


def process_video(
    video_file,
    fps_mode: str,
    crf: int,
    use_nvenc: bool,
    enable_reencode: bool,
    enable_patch: bool,
):
    if video_file is None:
        return None, "⚠ Please upload a video first."

    input_path = video_file
    logs: list[str] = []
    logs.append("═" * 48)
    logs.append("🎬 TikTok HD Video Compressor — Starting...")
    logs.append("═" * 48)

    if not ffmpeg_available():
        logs.append("⚠ ffmpeg / ffprobe not found in PATH — skipping encode step")
        enable_reencode = False

    info = get_video_info(input_path)
    if info:
        logs.append(f"📊 Input  : {info.get('width')}×{info.get('height')}  "
                    f"{info.get('fps')} fps  {info.get('codec')}  "
                    f"{info.get('size_mb')} MB  {info.get('duration'):.1f}s")

    tmp_dir = tempfile.mkdtemp()
    try:
        step1 = os.path.join(tmp_dir, "step1_encoded.mp4")
        step2 = os.path.join(tmp_dir, "step2_patched.mp4")

        # Determine output FPS from fps_mode
        _, scale_factor = FPS_SCALE.get(fps_mode, (None, None))
        fps_map = {
            "120fps → 60fps": 60,
            "120fps → 30fps": 30,
            "90fps → 60fps":  60,
            "90fps → 30fps":  30,
            "60fps → 60fps":  60,
            "60fps → 30fps":  30,
            "30fps → 30fps":  30,
            "Auto (target 30k timescale)": 30,
        }
        fps_out = fps_map.get(fps_mode, 30)

        # ── Step 1: Re-encode with FFmpeg ──────────────────
        if enable_reencode:
            logs.append("\n── Step 1: FFmpeg Re-encode ──────────────────")
            ok = reencode_video(input_path, step1, crf, fps_out, use_nvenc, logs)
            if not ok:
                return None, "\n".join(logs)
            encoded_path = step1
        else:
            logs.append("\n── Step 1: Re-encode skipped ─────────────────")
            encoded_path = input_path

        # ── Step 2: Atom Patching ──────────────────────────
        if enable_patch:
            logs.append("\n── Step 2: MP4 Atom Patching ─────────────────")
            patch_logs = patch_mp4(encoded_path, step2, scale_factor)
            logs.extend(patch_logs)
            final_path = step2
        else:
            logs.append("\n── Step 2: Atom patch skipped ────────────────")
            final_path = encoded_path

        # ── Final output ───────────────────────────────────
        out_dir  = tempfile.mkdtemp()
        out_path = os.path.join(out_dir, "tiktok_hd_output.mp4")
        shutil.copy2(final_path, out_path)

        out_info = get_video_info(out_path)
        if out_info:
            logs.append(f"\n📊 Output : {out_info.get('width')}×{out_info.get('height')}  "
                        f"{out_info.get('fps')} fps  {out_info.get('codec')}  "
                        f"{out_info.get('size_mb')} MB")

        size_in  = info.get("size_mb", 0)
        size_out = out_info.get("size_mb", 0) if out_info else 0
        if size_in and size_out:
            pct = round((1 - size_out / size_in) * 100, 1)
            logs.append(f"📉 Size reduction: {size_in} MB → {size_out} MB  ({pct}% smaller)")

        logs.append("\n✅ Done! Download your TikTok-ready video below.")
        return out_path, "\n".join(logs)

    except Exception as e:
        logs.append(f"\n❌ Unexpected error: {e}")
        return None, "\n".join(logs)
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)


# ─────────────────────────────────────────────
# GRADIO UI
# ─────────────────────────────────────────────

CSS = """
@import url('https://fonts.googleapis.com/css2?family=Space+Mono:wght@400;700&family=DM+Sans:wght@300;400;600&display=swap');

:root {
    --bg: #0a0a0f;
    --surface: #12121a;
    --surface2: #1c1c28;
    --accent: #ff2d55;
    --accent2: #0af;
    --text: #f0f0f8;
    --muted: #888;
    --border: #2a2a3a;
    --radius: 12px;
    --mono: 'Space Mono', monospace;
    --sans: 'DM Sans', sans-serif;
}

body, .gradio-container {
    background: var(--bg) !important;
    font-family: var(--sans) !important;
    color: var(--text) !important;
}

.gradio-container { max-width: 860px !important; margin: 0 auto !important; }

/* Header */
.app-header {
    text-align: center;
    padding: 2.5rem 1rem 1.5rem;
    border-bottom: 1px solid var(--border);
    margin-bottom: 2rem;
}
.app-header h1 {
    font-family: var(--mono) !important;
    font-size: 2rem;
    letter-spacing: -0.02em;
    color: var(--text);
    margin: 0 0 .4rem;
}
.app-header h1 span { color: var(--accent); }
.app-header p { color: var(--muted); font-size: .9rem; margin: 0; }

/* Panels */
.panel {
    background: var(--surface) !important;
    border: 1px solid var(--border) !important;
    border-radius: var(--radius) !important;
    padding: 1.25rem !important;
    margin-bottom: 1rem !important;
}
.panel-title {
    font-family: var(--mono);
    font-size: .7rem;
    letter-spacing: .12em;
    text-transform: uppercase;
    color: var(--accent2);
    margin-bottom: .75rem;
}

/* Inputs */
label span, .gr-form label span {
    font-family: var(--sans) !important;
    font-size: .8rem !important;
    color: var(--muted) !important;
    font-weight: 600 !important;
    letter-spacing: .04em !important;
    text-transform: uppercase !important;
}
input, select, textarea, .gr-input, .gr-dropdown {
    background: var(--surface2) !important;
    border: 1px solid var(--border) !important;
    color: var(--text) !important;
    border-radius: 8px !important;
    font-family: var(--sans) !important;
}

/* Buttons */
button.primary, .gr-button-primary {
    background: var(--accent) !important;
    border: none !important;
    color: #fff !important;
    font-family: var(--mono) !important;
    font-size: .85rem !important;
    letter-spacing: .05em !important;
    border-radius: 8px !important;
    padding: .7rem 1.5rem !important;
    transition: opacity .2s !important;
}
button.primary:hover { opacity: .85 !important; }

/* Log box */
.log-box textarea {
    font-family: var(--mono) !important;
    font-size: .75rem !important;
    background: #08080e !important;
    color: #7dff8a !important;
    border: 1px solid var(--border) !important;
    border-radius: 8px !important;
    line-height: 1.7 !important;
}

/* Info badge */
.info-badge {
    display: inline-flex;
    gap: .4rem;
    align-items: center;
    background: var(--surface2);
    border: 1px solid var(--border);
    border-radius: 6px;
    padding: .3rem .7rem;
    font-family: var(--mono);
    font-size: .7rem;
    color: var(--muted);
    margin-right: .5rem;
    margin-bottom: .5rem;
}
"""

HEADER_HTML = """
<div class="app-header">
  <h1>🎬 TikTok<span>HD</span> Compressor</h1>
  <p>MP4 Atom Patching · FFmpeg H.264 · Smooth FPS for TikTok upload</p>
</div>
"""

INFO_HTML = """
<div style="display:flex;flex-wrap:wrap;gap:.5rem;margin-bottom:1rem">
  <span class="info-badge">⚡ MP4 Atom Patch</span>
  <span class="info-badge">🎞 FFmpeg H.264</span>
  <span class="info-badge">🎯 TikTok-optimised</span>
  <span class="info-badge">🖥 CPU + NVIDIA GPU</span>
</div>
"""

FPS_CHOICES = list(FPS_SCALE.keys())

with gr.Blocks(css=CSS, title="TikTok HD Compressor") as demo:

    gr.HTML(HEADER_HTML)
    gr.HTML(INFO_HTML)

    with gr.Row():
        with gr.Column(scale=1):

            gr.HTML('<div class="panel-title">📁 Input Video</div>')
            video_input = gr.Video(label="Upload MP4 / MOV / AVI", sources=["upload"])

            gr.HTML('<div class="panel-title" style="margin-top:1.25rem">⚙ Encoding Settings</div>')

            fps_mode = gr.Dropdown(
                label="FPS Mode (Atom Patch Scale)",
                choices=FPS_CHOICES,
                value="60fps → 30fps",
            )

            crf_slider = gr.Slider(
                label="CRF Quality (lower = better, larger file)",
                minimum=15, maximum=35, value=23, step=1,
            )

            with gr.Row():
                enable_reencode = gr.Checkbox(label="Re-encode with FFmpeg", value=True)
                enable_patch    = gr.Checkbox(label="MP4 Atom Patch",        value=True)
                use_nvenc       = gr.Checkbox(label="NVIDIA GPU (h264_nvenc)", value=False)

            run_btn = gr.Button("▶ Compress for TikTok", variant="primary", size="lg")

        with gr.Column(scale=1):
            gr.HTML('<div class="panel-title">📤 Output</div>')
            video_output = gr.Video(label="Processed Video")

            gr.HTML('<div class="panel-title" style="margin-top:1.25rem">📋 Log</div>')
            log_output = gr.Textbox(
                label="",
                lines=18,
                interactive=False,
                elem_classes=["log-box"],
                placeholder="Logs will appear here after processing...",
            )

    gr.HTML("""
    <div style="margin-top:1.5rem;border-top:1px solid #2a2a3a;padding-top:1rem;
                text-align:center;font-size:.72rem;color:#555;font-family:'Space Mono',monospace;">
      MP4 Atom Patching based on Editing News / Evilfy (GPLv3) ·
      FFmpeg licensed under LGPL/GPL ·
      Built for TikTok HD upload optimisation
    </div>
    """)

    run_btn.click(
        fn=process_video,
        inputs=[video_input, fps_mode, crf_slider, use_nvenc, enable_reencode, enable_patch],
        outputs=[video_output, log_output],
    )

if __name__ == "__main__":
    demo.launch()