import logging import os import uuid import json import subprocess import tempfile _log = logging.getLogger(__name__) from converters._whisper import transcribe as _whisper_transcribe, to_srt, to_vtt SUPPORTED_FORMATS = { "3g2", "3gp", "avi", "flv", "mkv", "mov", "mp4", "mpg", "ogv", "webm", "wmv", "mp3" } TRANSCRIPTION_FORMATS = {"txt", "srt", "vtt"} 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 "" FORMAT_CODEC_MAP = { "3g2": ["-c:v", "mpeg4", "-c:a", "aac"], "3gp": ["-c:v", "mpeg4", "-c:a", "aac"], "avi": ["-c:v", "mpeg4", "-c:a", "libmp3lame"], "flv": ["-c:v", "flv1", "-c:a", "aac"], "mkv": ["-c:v", "libx264", "-c:a", "aac"], "mov": ["-c:v", "libx264", "-c:a", "aac"], "mp4": ["-c:v", "libx264", "-c:a", "aac"], "mpg": ["-c:v", "mpeg2video", "-c:a", "mp2"], "ogv": ["-c:v", "libtheora", "-c:a", "libvorbis"], "webm": ["-c:v", "libvpx-vp9", "-c:a", "libopus"], "wmv": ["-c:v", "wmv2", "-c:a", "wmav2"], } _SAFE_FALLBACK_CODEC_ARGS = ["-c:v", "libx264", "-c:a", "aac", "-strict", "experimental"] _ENCODER_ERROR_MARKERS = ( "encoder not found", "encoder selection failed", "no encoder", "unknown encoder", ) def _build_ffmpeg_cmd(input_path, output_path, options, extra_codec_args=None): cmd = ["ffmpeg", "-y", "-i", input_path] if options and isinstance(options, dict): trim = options.get("trim", {}) if trim and isinstance(trim, dict): start = trim.get("video_start", "") end = trim.get("video_end", "") if start and start.strip(): cmd.extend(["-ss", start.strip()]) if end and end.strip(): cmd.extend(["-to", end.strip()]) bitrate = options.get("bitrate") if bitrate and isinstance(bitrate, (int, float)) and bitrate > 0: cmd.extend(["-b:v", f"{int(bitrate)}k"]) framerate = options.get("framerate") if framerate and isinstance(framerate, (int, float)) and framerate > 0: cmd.extend(["-r", str(int(framerate))]) vf_filters = [] rotate = options.get("rotate", "") if rotate: rotate = str(rotate).strip() if rotate == "90": vf_filters.append("transpose=1") elif rotate == "180": vf_filters.append("transpose=1,transpose=1") elif rotate == "270": vf_filters.append("transpose=2") flip = options.get("flip", "") if flip: flip = str(flip).strip() if flip == "Horizontally": vf_filters.append("hflip") elif flip == "Vertically": vf_filters.append("vflip") if vf_filters: cmd.extend(["-vf", ",".join(vf_filters)]) if extra_codec_args: cmd.extend(extra_codec_args) cmd.append(output_path) return cmd def _transcribe_single(full_path, source_name, target_format_lower, upload_dir): """Extract audio from a video with ffmpeg then run Whisper on it.""" # Extract audio to a temp WAV file with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: tmp_wav = tmp.name try: cmd = ["ffmpeg", "-y", "-i", full_path, "-vn", "-ac", "1", "-ar", "16000", tmp_wav] subprocess.run(cmd, capture_output=True, timeout=600, check=False) if not os.path.exists(tmp_wav) or os.path.getsize(tmp_wav) == 0: raise RuntimeError("Could not extract audio from video with ffmpeg.") result = _whisper_transcribe(tmp_wav) finally: try: os.unlink(tmp_wav) except OSError: pass output_folder = uuid.uuid4().hex output_dir_path = os.path.join(upload_dir, output_folder) os.makedirs(output_dir_path, exist_ok=True) output_path = os.path.join(output_dir_path, f"{source_name}.{target_format_lower}") if target_format_lower == "txt": with open(output_path, "w", encoding="utf-8") as f: f.write(result["text"]) elif target_format_lower == "srt": with open(output_path, "w", encoding="utf-8") as f: f.write(to_srt(result["segments"])) elif target_format_lower == "vtt": with open(output_path, "w", encoding="utf-8") as f: f.write(to_vtt(result["segments"])) return output_path def convert(urls, target_format, options, config): try: upload_dir = config.get("UPLOAD_DIR", "static/uploads") target_format_lower = target_format.lower() # ── Whisper 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: 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] out = _transcribe_single(input_path, source_name, _tfmt, upload_dir) results.append(out) except Exception as e: errors.append(f"Transcription failed for {os.path.basename(file_path)}: {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 transcription."} return { "error": False, "results": results, "output_path": results[0] if results else "", "errors": errors if errors else None, } if target_format_lower not in SUPPORTED_FORMATS: return {"error": True, "message": f"Unsupported target format: {target_format}"} 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}.{target_format_lower}" 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) # Pre-check: verify the file has at least one video stream probe = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "v", "-show_entries", "stream=codec_type", "-of", "csv=p=0", input_path], capture_output=True, text=True, timeout=30 ) if not probe.stdout.strip(): errors.append( f"{os.path.basename(file_path)} has no video stream " f"and cannot be converted by the video converter." ) continue codec_args = FORMAT_CODEC_MAP.get(target_format_lower) cmd = _build_ffmpeg_cmd(input_path, output_path, options, extra_codec_args=codec_args) result = subprocess.run( cmd, capture_output=True, text=True, timeout=600 ) attempt = "primary" if result.returncode != 0 or not os.path.exists(output_path): stderr_lower = (result.stderr or "").lower() is_encoder_error = any(m in stderr_lower for m in _ENCODER_ERROR_MARKERS) if is_encoder_error: _log.warning( "FFmpeg encoder error on primary attempt for %s (format=%s); " "retrying with safe fallback (libx264/aac). stderr: %s", os.path.basename(file_path), target_format_lower, (result.stderr or "").strip()[-300:] ) if os.path.exists(output_path): try: os.remove(output_path) except Exception: pass cmd_fallback = _build_ffmpeg_cmd( input_path, output_path, options, extra_codec_args=_SAFE_FALLBACK_CODEC_ARGS ) result = subprocess.run( cmd_fallback, capture_output=True, text=True, timeout=600 ) attempt = "fallback" if result.returncode != 0 or not os.path.exists(output_path): stderr = (result.stderr or "").strip() lines = [l for l in stderr.splitlines() if l.strip()] tail = "\n".join(lines[-10:]) if lines else "Unknown error" _log.error( "FFmpeg failed on %s attempt for %s (format=%s): %s", attempt, os.path.basename(file_path), target_format_lower, tail ) errors.append(f"FFmpeg failed for {os.path.basename(file_path)}: {tail}") continue _log.info( "FFmpeg conversion succeeded on %s attempt: %s -> %s", attempt, os.path.basename(file_path), output_filename ) results.append(output_path) except subprocess.TimeoutExpired: fname = os.path.basename(file_path) if file_path else "unknown" errors.append(f"Conversion timed out for {fname}") 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)}"}