| import gradio as gr |
| import subprocess |
| import os |
| import random |
| import shutil |
| import tempfile |
| import json |
| import time |
| import threading |
| import uuid |
| import re |
| from pathlib import Path |
| from datetime import datetime |
| from queue import Queue |
|
|
| |
| jobs = {} |
|
|
| class UltimateVideoEvader: |
| def __init__(self): |
| self.ffmpeg_path = self._find_ffmpeg() |
| self.ffprobe_path = self.ffmpeg_path.replace("ffmpeg", "ffprobe") |
| self.output_base_dir = tempfile.mkdtemp(prefix="vfe_jobs_") |
| print(f"[*] Job output directory: {self.output_base_dir}") |
| |
| def _find_ffmpeg(self): |
| for cmd in ["ffmpeg", "/usr/bin/ffmpeg"]: |
| if shutil.which(cmd): |
| return cmd |
| return "ffmpeg" |
| |
| def _get_resolution(self, input_file): |
| try: |
| cmd = [ |
| self.ffprobe_path, |
| "-v", "error", |
| "-select_streams", "v:0", |
| "-show_entries", "stream=width,height,duration", |
| "-of", "json", |
| input_file |
| ] |
| result = subprocess.run(cmd, capture_output=True, text=True) |
| if result.returncode == 0: |
| data = json.loads(result.stdout) |
| if "streams" in data and data["streams"]: |
| w = int(data["streams"][0]["width"]) |
| h = int(data["streams"][0]["height"]) |
| duration = float(data["streams"][0].get("duration", 0)) |
| return (w, h, duration) |
| except: |
| pass |
| return (1920, 1080, 0) |
| |
| def _random_float(self, a, b): |
| return round(random.uniform(a, b), 4) |
| |
| def build_brutal_command(self, input_file, output_file): |
| cmd = [self.ffmpeg_path, "-i", input_file] |
| cmd.extend(["-map_metadata", "-1"]) |
| |
| scale = self._random_float(0.6, 1.4) |
| w, h, duration = self._get_resolution(input_file) |
| new_w = int(w * scale) |
| new_h = int(h * scale) |
| new_w = new_w if new_w % 2 == 0 else new_w + 1 |
| new_h = new_h if new_h % 2 == 0 else new_h + 1 |
| |
| fps = self._random_float(15, 60) |
| speed = self._random_float(0.8, 1.3) |
| tempo = self._random_float(0.8, 1.3) * speed |
| tempo = min(tempo, 2.0) |
| pitch = self._random_float(-1.0, 1.0) |
| noise_db = self._random_float(1, 12) |
| bright = self._random_float(-0.1, 0.1) |
| contrast = self._random_float(0.85, 1.15) |
| saturation = self._random_float(0.7, 1.3) |
| hue = self._random_float(0, 360) |
| |
| crop_percent = self._random_float(0.90, 0.98) |
| crop_w = int(new_w * crop_percent) |
| crop_h = int(new_h * crop_percent) |
| crop_w = crop_w if crop_w % 2 == 0 else crop_w + 1 |
| crop_h = crop_h if crop_h % 2 == 0 else crop_h + 1 |
| pad_left = (new_w - crop_w) // 2 |
| pad_top = (new_h - crop_h) // 2 |
| |
| video_filters = [] |
| video_filters.append(f"scale={new_w}:{new_h}") |
| video_filters.append(f"setpts={1/speed}*PTS") |
| video_filters.append(f"noise=alls={noise_db}:allf=t+u") |
| video_filters.append(f"eq=brightness={bright}:contrast={contrast}:saturation={saturation}") |
| video_filters.append(f"hue=H={hue}") |
| video_filters.append(f"crop={crop_w}:{crop_h}:{pad_left}:{pad_top},pad={new_w}:{new_h}:{pad_left}:{pad_top}") |
| |
| offset = random.randint(0, 120) |
| if offset > 0: |
| video_filters.append(f"trim=start_frame={offset}") |
| |
| if random.choice([True, False]): |
| video_filters.append("reverse") |
| |
| if random.choice([True, False]): |
| blur = self._random_float(0.5, 2.0) |
| video_filters.append(f"gblur=sigma={blur}") |
| |
| filter_chain = ",".join(video_filters) |
| cmd.extend(["-vf", filter_chain]) |
| cmd.extend(["-r", str(round(fps, 2))]) |
| |
| audio_filters = [] |
| if 0.5 <= tempo <= 2.0: |
| audio_filters.append(f"atempo={tempo}") |
| pitch_factor = 2 ** (pitch / 12) |
| audio_filters.append(f"rubberband=pitch={pitch_factor}") |
| if audio_filters: |
| cmd.extend(["-af", ",".join(audio_filters)]) |
| |
| codec = random.choice(["libx264", "libx265", "libvpx-vp9"]) |
| cmd.extend(["-c:v", codec]) |
| preset = random.choice(["ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow"]) |
| cmd.extend(["-preset", preset]) |
| crf = random.randint(18, 28) |
| cmd.extend(["-crf", str(crf)]) |
| audio_codec = random.choice(["aac", "libmp3lame"]) |
| cmd.extend(["-c:a", audio_codec, "-b:a", "128k"]) |
| container = random.choice(["mp4", "mkv", "mov"]) |
| base = os.path.splitext(output_file)[0] |
| output_file = f"{base}.{container}" |
| cmd.extend(["-y", output_file]) |
| return cmd, output_file |
| |
| def process_video_background(self, job_id, input_path): |
| log_queue = jobs[job_id]["log_queue"] |
| |
| def log(msg): |
| log_queue.put(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}") |
| |
| try: |
| log("🚀 Job started") |
| jobs[job_id]["status"] = "processing" |
| jobs[job_id]["progress"] = 10 |
| |
| job_dir = os.path.join(self.output_base_dir, job_id) |
| os.makedirs(job_dir, exist_ok=True) |
| output_base = os.path.join(job_dir, "evaded_video") |
| |
| log("📦 Building FFmpeg command...") |
| cmd, final_output = self.build_brutal_command(input_path, output_base) |
| jobs[job_id]["progress"] = 20 |
| |
| input_copy = os.path.join(job_dir, "input_" + os.path.basename(input_path)) |
| shutil.copy2(input_path, input_copy) |
| log(f"📁 Input file: {os.path.basename(input_path)}") |
| |
| _, _, duration = self._get_resolution(input_path) |
| if duration > 0: |
| log(f"⏱️ Input duration: {duration:.2f} seconds") |
| |
| jobs[job_id]["progress"] = 25 |
| log("⚙️ Starting FFmpeg processing...") |
| |
| process = subprocess.Popen( |
| cmd, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| text=True, |
| bufsize=1, |
| universal_newlines=True |
| ) |
| |
| speed_pattern = re.compile(r'speed=\s*([\d.]+)x') |
| fps_pattern = re.compile(r'fps=\s*([\d.]+)') |
| time_pattern = re.compile(r'time=(\d{2}):(\d{2}):(\d{2})\.(\d{2})') |
| |
| start_time = time.time() |
| last_log_time = 0 |
| |
| while True: |
| line = process.stderr.readline() |
| if not line: |
| break |
| line = line.strip() |
| |
| if 'frame=' in line or 'time=' in line: |
| time_match = time_pattern.search(line) |
| if time_match: |
| h, m, s, cs = time_match.groups() |
| current_time = int(h) * 3600 + int(m) * 60 + int(s) + int(cs) / 100 |
| |
| speed_match = speed_pattern.search(line) |
| speed_val = speed_match.group(1) if speed_match else "?" |
| fps_match = fps_pattern.search(line) |
| fps_val = fps_match.group(1) if fps_match else "?" |
| |
| if duration > 0 and time_match and speed_match: |
| if float(speed_val) > 0: |
| eta = (duration - current_time) / float(speed_val) |
| eta_min = int(eta // 60) |
| eta_sec = int(eta % 60) |
| eta_str = f"{eta_min}m {eta_sec}s" |
| else: |
| eta_str = "calculating..." |
| |
| percent = min(100, int((current_time / duration) * 100)) if duration > 0 else 0 |
| jobs[job_id]["progress"] = min(25 + int(percent * 0.7), 95) |
| |
| if time.time() - last_log_time > 3: |
| log(f"⏳ Progress: {percent}% | Time: {current_time:.1f}s/{duration:.1f}s | Speed: {speed_val}x | FPS: {fps_val} | ETA: {eta_str}") |
| last_log_time = time.time() |
| |
| process.wait() |
| |
| if process.returncode != 0: |
| jobs[job_id]["status"] = "failed" |
| jobs[job_id]["error"] = "FFmpeg error" |
| jobs[job_id]["progress"] = 0 |
| log("❌ FFmpeg failed with error") |
| return |
| |
| if os.path.exists(final_output) and os.path.getsize(final_output) > 0: |
| jobs[job_id]["status"] = "completed" |
| jobs[job_id]["output_file"] = final_output |
| jobs[job_id]["progress"] = 100 |
| jobs[job_id]["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
| file_size = round(os.path.getsize(final_output) / (1024 * 1024), 2) |
| log(f"✅ Processing complete! Output size: {file_size} MB") |
| log(f"📥 Download ready: {os.path.basename(final_output)}") |
| else: |
| jobs[job_id]["status"] = "failed" |
| jobs[job_id]["error"] = "Output file not created" |
| jobs[job_id]["progress"] = 0 |
| log("❌ Output file not created") |
| |
| except Exception as e: |
| jobs[job_id]["status"] = "failed" |
| jobs[job_id]["error"] = str(e) |
| jobs[job_id]["progress"] = 0 |
| log(f"❌ Error: {str(e)}") |
| |
| log("🏁 Job finished") |
| log_queue.put("__END__") |
| |
| def start_job(self, input_path): |
| if not input_path or not os.path.exists(input_path): |
| return "❌ कोई फ़ाइल नहीं", None |
| |
| job_id = str(uuid.uuid4())[:8] |
| log_queue = Queue() |
| jobs[job_id] = { |
| "status": "queued", |
| "input_file": input_path, |
| "output_file": None, |
| "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| "progress": 0, |
| "error": None, |
| "job_id": job_id, |
| "log_queue": log_queue |
| } |
| |
| thread = threading.Thread( |
| target=self.process_video_background, |
| args=(job_id, input_path) |
| ) |
| thread.daemon = True |
| thread.start() |
| |
| return f"✅ Job started! ID: {job_id}", job_id |
| |
| def get_job_status(self, job_id): |
| if job_id not in jobs: |
| return "❌ Job not found", None, None |
| job = jobs[job_id] |
| status = job["status"] |
| progress = job["progress"] |
| error = job.get("error", "") |
| |
| status_text = f"Status: {status}\nProgress: {progress}%" |
| if error: |
| status_text += f"\nError: {error}" |
| |
| output_file = job.get("output_file") |
| return status_text, output_file, status |
| |
| def get_live_logs(self, job_id): |
| if job_id not in jobs: |
| return "No job found" |
| queue = jobs[job_id].get("log_queue") |
| if not queue: |
| return "Log queue not initialized" |
| |
| logs = [] |
| while not queue.empty(): |
| msg = queue.get() |
| if msg == "__END__": |
| break |
| logs.append(msg) |
| return "\n".join(logs) if logs else "Waiting for logs..." |
| |
| def get_all_jobs(self): |
| job_list = [] |
| for job_id, job in jobs.items(): |
| job_list.append({ |
| "job_id": job_id, |
| "status": job["status"], |
| "timestamp": job.get("timestamp", "Unknown"), |
| "output_file": job.get("output_file", ""), |
| "input_file": os.path.basename(job.get("input_file", "unknown")), |
| "progress": job.get("progress", 0), |
| "has_output": job.get("output_file") is not None and os.path.exists(job.get("output_file", "")) |
| }) |
| job_list.sort(key=lambda x: x["timestamp"], reverse=True) |
| return job_list |
| |
| def get_download_link(self, job_id): |
| if job_id not in jobs: |
| return None, "❌ Job not found" |
| job = jobs[job_id] |
| if job.get("status") != "completed": |
| return None, f"⏳ Job is {job.get('status')}" |
| if not job.get("output_file") or not os.path.exists(job.get("output_file")): |
| return None, "❌ File missing" |
| file_path = job["output_file"] |
| file_name = os.path.basename(file_path) |
| |
| return file_path, f"✅ Ready: {file_name}" |
|
|
|
|
| |
| def create_interface(): |
| evader = UltimateVideoEvader() |
| |
| with gr.Blocks(title="Video Fingerprint Evader") as demo: |
| gr.Markdown(""" |
| # 🚀 Video Fingerprint Evader – Live Logs + Background Processing |
| """) |
| |
| job_id_state = gr.State("") |
| |
| with gr.Tab("📤 Upload & Process"): |
| with gr.Row(): |
| with gr.Column(scale=1): |
| input_video = gr.File( |
| label="📤 वीडियो अपलोड करें", |
| file_types=[".mp4", ".mkv", ".avi", ".mov", ".webm", ".flv"], |
| type="filepath" |
| ) |
| process_btn = gr.Button("🔥 Process in Background", variant="primary") |
| job_status = gr.Textbox(label="📊 Job Status", lines=3, interactive=False) |
| |
| gr.Markdown("---") |
| gr.Markdown("### 📥 Download Current Job") |
| download_btn = gr.Button("📥 Download Video", variant="secondary") |
| download_file = gr.File(label="📥 Download", visible=False) |
| download_link_html = gr.HTML(label="", visible=False) |
| download_status = gr.Textbox(label="Status", lines=2, interactive=False) |
| |
| with gr.Column(scale=2): |
| log_display = gr.Textbox( |
| label="📋 Live Logs", |
| lines=25, |
| interactive=False, |
| autoscroll=True |
| ) |
| refresh_logs_btn = gr.Button("🔄 Refresh Logs", variant="secondary") |
| status_display = gr.Textbox(label="📊 Quick Status", lines=2, interactive=False) |
| |
| with gr.Tab("📜 Job History"): |
| gr.Markdown("### सभी Jobs की History") |
| |
| with gr.Row(): |
| with gr.Column(scale=2): |
| history_refresh_btn = gr.Button("🔄 Refresh History") |
| job_history = gr.Dataframe( |
| headers=["Job ID", "Input File", "Status", "Timestamp", "Progress"], |
| datatype=["str", "str", "str", "str", "str"], |
| label="Job History", |
| interactive=False |
| ) |
| with gr.Column(scale=1): |
| gr.Markdown("### 📥 Download by Job ID") |
| download_job_id = gr.Textbox(label="Enter Job ID", placeholder="e.g., 5b7c3970") |
| download_by_id_btn = gr.Button("📥 Download Video", variant="primary") |
| download_by_id_file = gr.File(label="Download", visible=False) |
| download_by_id_link = gr.HTML(label="", visible=False) |
| download_by_id_status = gr.Textbox(label="Status", lines=2, interactive=False) |
| |
| |
| |
| def start_process(file_path): |
| if not file_path: |
| return "❌ कृपया पहले वीडियो अपलोड करें।", "", "No job", "", "" |
| status, job_id = evader.start_job(file_path) |
| return status, job_id, f"Job {job_id} running...", "", job_id |
| |
| def refresh_logs(job_id): |
| if not job_id or job_id not in jobs: |
| return "❌ No job running or invalid ID", "No job" |
| logs = evader.get_live_logs(job_id) |
| status = jobs.get(job_id, {}).get("status", "unknown") |
| progress = jobs.get(job_id, {}).get("progress", 0) |
| return logs, f"Status: {status} | Progress: {progress}%" |
| |
| def download_current(job_id): |
| if not job_id or job_id not in jobs: |
| return None, "", False, "❌ No job" |
| file_path, status = evader.get_download_link(job_id) |
| if file_path and os.path.exists(file_path): |
| file_name = os.path.basename(file_path) |
| |
| html_link = f'<a href="/file={file_path}" download="{file_name}" target="_blank">📥 Click here to download: {file_name}</a>' |
| return file_path, html_link, True, status |
| return None, "", False, status |
| |
| def download_by_id(job_id): |
| if not job_id: |
| return None, "", False, "❌ Please enter Job ID" |
| file_path, status = evader.get_download_link(job_id) |
| if file_path and os.path.exists(file_path): |
| file_name = os.path.basename(file_path) |
| html_link = f'<a href="/file={file_path}" download="{file_name}" target="_blank">📥 Click here to download: {file_name}</a>' |
| return file_path, html_link, True, status |
| return None, "", False, status |
| |
| def refresh_history(): |
| job_list = evader.get_all_jobs() |
| if not job_list: |
| return gr.update(value=[], visible=True) |
| |
| rows = [] |
| for j in job_list: |
| status_display = j["status"] |
| if j["status"] == "completed": |
| status_display = "✅ completed" |
| elif j["status"] == "processing": |
| status_display = "⏳ processing" |
| elif j["status"] == "failed": |
| status_display = "❌ failed" |
| rows.append([ |
| j["job_id"], |
| j["input_file"], |
| status_display, |
| j["timestamp"], |
| f"{j['progress']}%" |
| ]) |
| return gr.update(value=rows, visible=True) |
| |
| |
| |
| process_btn.click( |
| start_process, |
| inputs=[input_video], |
| outputs=[job_status, job_id_state, status_display, log_display, job_id_state] |
| ).then( |
| refresh_logs, |
| inputs=[job_id_state], |
| outputs=[log_display, status_display] |
| ) |
| |
| refresh_logs_btn.click( |
| refresh_logs, |
| inputs=[job_id_state], |
| outputs=[log_display, status_display] |
| ) |
| |
| download_btn.click( |
| download_current, |
| inputs=[job_id_state], |
| outputs=[download_file, download_link_html, download_file, download_status] |
| ) |
| |
| history_refresh_btn.click( |
| refresh_history, |
| inputs=[], |
| outputs=[job_history] |
| ) |
| |
| download_by_id_btn.click( |
| download_by_id, |
| inputs=[download_job_id], |
| outputs=[download_by_id_file, download_by_id_link, download_by_id_file, download_by_id_status] |
| ) |
| |
| |
| timer = gr.Timer(value=5, active=False) |
| |
| def toggle_timer(job_id): |
| if job_id and job_id in jobs: |
| return gr.update(active=True) |
| return gr.update(active=False) |
| |
| process_btn.click( |
| toggle_timer, |
| inputs=[job_id_state], |
| outputs=[timer] |
| ) |
| |
| timer.tick( |
| refresh_logs, |
| inputs=[job_id_state], |
| outputs=[log_display, status_display] |
| ) |
| |
| return demo |
|
|
|
|
| if __name__ == "__main__": |
| demo = create_interface() |
| demo.launch(debug=True) |
