import os import uuid import json import re 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): 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 RESOLUTION_MAP = { "360p": "640x360", "480p": "854x480", "2k": "2560x1440", "4K (UHD)": "3840x2160", } 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 _parse_size_string(size_str): if not size_str: return None, None, None quality = "MQ" if "HQ" in size_str: quality = "HQ" elif "MQ" in size_str: quality = "MQ" resolution = None res_match = re.search(r'(\d{3,5})x(\d{3,5})', size_str) if res_match: resolution = f"{res_match.group(1)}x{res_match.group(2)}" else: for label, res in RESOLUTION_MAP.items(): if label in size_str: resolution = res break fps = None fps_match = re.search(r'(\d+(?:\.\d+)?)fps', size_str) if fps_match: fps = fps_match.group(1) return resolution, fps, quality 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} results = [] errors = [] size_str = "" disable_audio = False trim = {} if options and isinstance(options, dict): size_str = options.get("size", "") disable_audio = options.get("disableAudio", False) trim = options.get("trim", {}) resolution, fps, quality = _parse_size_string(size_str) if quality == "HQ": crf = "18" else: crf = "23" 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"] if trim and isinstance(trim, dict): video_start = trim.get("video_start", "") video_end = trim.get("video_end", "") if video_start and video_start != "00:00:00": cmd.extend(["-ss", video_start]) if video_end and video_end != "00:00:00": cmd.extend(["-to", video_end]) cmd.extend(["-i", input_path]) cmd.extend(["-c:v", "libx264"]) cmd.extend(["-crf", crf]) cmd.extend(["-preset", "medium"]) if resolution: keep_dims = False if size_str and "keep dimensions" in size_str.lower(): keep_dims = True if not keep_dims: cmd.extend(["-s", resolution]) elif size_str and "keep dimensions" in size_str.lower(): pass if fps: cmd.extend(["-r", fps]) if disable_audio: cmd.append("-an") else: cmd.extend(["-c:a", "aac", "-b:a", "128k"]) cmd.extend(["-movflags", "+faststart"]) cmd.extend(["-pix_fmt", "yuv420p"]) 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)}"}