import os import uuid import json import subprocess import tempfile from converters._whisper import transcribe as _whisper_transcribe, to_srt, to_vtt TRANSCRIPTION_FORMATS = {"txt", "srt", "vtt"} def _transcribe_single(full_path, source_name, fmt, upload_dir): """Extract audio from video then run Whisper.""" with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: tmp_wav = tmp.name try: subprocess.run( ["ffmpeg", "-y", "-i", full_path, "-vn", "-ac", "1", "-ar", "16000", tmp_wav], capture_output=True, timeout=600, check=False ) result = _whisper_transcribe(tmp_wav) finally: try: os.unlink(tmp_wav) except OSError: pass output_folder = uuid.uuid4().hex output_dir = os.path.join(upload_dir, output_folder) os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, f"{source_name}.{fmt}") if fmt == "srt": with open(output_path, "w", encoding="utf-8") as f: f.write(to_srt(result["segments"])) elif fmt == "vtt": with open(output_path, "w", encoding="utf-8") as f: f.write(to_vtt(result["segments"])) else: with open(output_path, "w", encoding="utf-8") as f: f.write(result["text"]) return output_path DEVICE_PRESETS = { "android": { "args": ["-c:v", "libx264", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart"], }, "blackberry": { "args": ["-c:v", "libx264", "-s", "480x360", "-c:a", "aac", "-b:a", "96k"], }, "ipad": { "args": ["-c:v", "libx264", "-c:a", "aac", "-b:a", "160k", "-movflags", "+faststart"], }, "iphone": { "args": ["-c:v", "libx264", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart"], }, "ipod": { "args": ["-c:v", "libx264", "-s", "320x240", "-c:a", "aac", "-b:a", "96k"], }, "playstation": { "args": ["-c:v", "libx264", "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart"], }, "psp": { "args": ["-c:v", "libx264", "-s", "480x272", "-c:a", "aac", "-b:a", "96k"], }, "wii": { "args": ["-c:v", "libx264", "-s", "640x480", "-c:a", "aac", "-b:a", "128k"], }, "xbox": { "args": ["-c:v", "libx264", "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart"], }, } ALLOWED_INPUTS = { "3g2", "3gp", "avi", "flv", "mkv", "mov", "mp4", "mpg", "ogv", "webm", "wmv" } def _parse_url_entry(url_entry): if isinstance(url_entry, str): try: parsed = json.loads(url_entry) if isinstance(parsed, dict): return parsed.get("path", "") except (json.JSONDecodeError, TypeError): return url_entry elif isinstance(url_entry, dict): return url_entry.get("path", "") return "" def convert(urls, target_format, options, config): try: upload_dir = config.get("UPLOAD_DIR", "static/uploads") # AI Transcription — triggered by the "transcribe" option if options and isinstance(options, dict) and options.get("transcribe"): _tfmt = (options.get("transcribe_format") or "txt").lower().strip() if _tfmt not in TRANSCRIPTION_FORMATS: _tfmt = "txt" results, errors = [], [] for url_entry in urls: fp = _parse_url_entry(url_entry) if not fp or fp == "empty": continue try: full = os.path.join(upload_dir, fp) if not os.path.exists(full): full = fp if not os.path.exists(full): errors.append(f"File not found: {os.path.basename(fp)}") continue src = os.path.splitext(os.path.basename(fp))[0] results.append(_transcribe_single(full, src, _tfmt, upload_dir)) except Exception as e: errors.append(f"Transcription failed for {os.path.basename(fp)}: {e}") if not results and errors: return {"error": True, "message": "; ".join(errors)} if not results: return {"error": True, "message": "No files provided for transcription."} return {"error": False, "results": results, "output_path": results[0], "errors": errors or None} device_key = target_format.lower() if device_key not in DEVICE_PRESETS: return {"error": True, "message": f"Unsupported device: {target_format}"} preset = DEVICE_PRESETS[device_key] results = [] errors = [] for url_entry in urls: file_path = _parse_url_entry(url_entry) if not file_path or file_path == "empty": continue try: input_path = os.path.join(upload_dir, file_path) if not os.path.exists(input_path): input_path = file_path if not os.path.exists(input_path): errors.append(f"File not found: {os.path.basename(file_path)}") continue source_name = os.path.splitext(os.path.basename(file_path))[0] output_filename = f"{source_name}.mp4" output_folder = uuid.uuid4().hex output_dir = os.path.join(upload_dir, output_folder) os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, output_filename) cmd = ["ffmpeg", "-y", "-i", input_path] if options and isinstance(options, dict): trim = options.get("trim", {}) if trim and isinstance(trim, dict): video_start = trim.get("video_start", "") video_end = trim.get("video_end", "") if video_start and video_start.strip(): cmd.extend(["-ss", video_start.strip()]) if video_end and video_end.strip(): cmd.extend(["-to", video_end.strip()]) preset_args = list(preset["args"]) if options and isinstance(options, dict): size = options.get("size", "") if size and isinstance(size, str) and "x" in size: if "-s" in preset_args: s_idx = preset_args.index("-s") preset_args[s_idx + 1] = size.strip() else: codec_idx = preset_args.index("-c:v") + 2 if "-c:v" in preset_args else 0 preset_args.insert(codec_idx, "-s") preset_args.insert(codec_idx + 1, size.strip()) disable_audio = options.get("disableAudio", False) if disable_audio: filtered_args = [] skip_next = False for i, arg in enumerate(preset_args): if skip_next: skip_next = False continue if arg in ("-c:a", "-b:a"): skip_next = True continue filtered_args.append(arg) preset_args = filtered_args preset_args.append("-an") cmd.extend(preset_args) cmd.append(output_path) result = subprocess.run( cmd, capture_output=True, text=True, timeout=600 ) if result.returncode != 0: errors.append(f"FFmpeg error for {os.path.basename(file_path)}: {result.stderr[:200]}") continue if not os.path.exists(output_path): errors.append(f"Output file not created for {os.path.basename(file_path)}") continue results.append(output_path) except subprocess.TimeoutExpired: errors.append(f"Conversion timed out for {os.path.basename(file_path)}") except Exception as e: fname = os.path.basename(file_path) if file_path else "unknown" errors.append(f"Failed to convert {fname}: {str(e)}") if not results and errors: return {"error": True, "message": "; ".join(errors)} if not results and not errors: return {"error": True, "message": "No files were provided for conversion."} return { "error": False, "results": results, "output_path": results[0] if results else "", "errors": errors if errors else None } except Exception as e: return {"error": True, "message": f"Conversion failed: {str(e)}"}