import os import uuid import json import subprocess from converters._whisper import transcribe as _whisper_transcribe, to_srt, to_vtt FFMPEG_PATH = "ffmpeg" 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 "" def _transcribe_single(full_path, source_name, target_format_lower, upload_dir): """Run Whisper on an audio file and write TXT / SRT / VTT output.""" result = _whisper_transcribe(full_path) 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 target format OR the option ── # When the user explicitly selects txt/srt/vtt as the target format we # always run Whisper; the options checkbox is a secondary trigger that # lets users override the output format via options.transcribe_format. if target_format_lower in TRANSCRIPTION_FORMATS or ( 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: full_path = os.path.join(upload_dir, file_path) if not os.path.exists(full_path): full_path = file_path if not os.path.exists(full_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(full_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, } results = [] errors = [] for url_entry in urls: file_path = _parse_url_entry(url_entry) if not file_path or file_path == "empty": continue try: full_path = os.path.join(upload_dir, file_path) if not os.path.exists(full_path): full_path = file_path if not os.path.exists(full_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) cmd = [FFMPEG_PATH, "-y", "-i", full_path, "-vn"] if options and isinstance(options, dict): bitrate = options.get("bitrate", "") if bitrate and isinstance(bitrate, str): bitrate_val = bitrate.replace(" kbps", "").strip() if bitrate_val: cmd.extend(["-b:a", f"{bitrate_val}k"]) frequency = options.get("frequency", "") if frequency and isinstance(frequency, str): freq_val = frequency.replace(" Hz", "").strip() if freq_val: cmd.extend(["-ar", freq_val]) channels = options.get("channels", "") if channels and isinstance(channels, str): if channels.lower() == "mono": cmd.extend(["-ac", "1"]) elif channels.lower() == "stereo": cmd.extend(["-ac", "2"]) trim = options.get("trim", {}) if trim and isinstance(trim, dict): audio_start = trim.get("audio_start", "") audio_end = trim.get("audio_end", "") if audio_start and audio_start.strip(): cmd.extend(["-ss", audio_start.strip()]) if audio_end and audio_end.strip(): cmd.extend(["-to", audio_end.strip()]) cmd.append(output_path) result = subprocess.run( cmd, capture_output=True, text=True, timeout=600 ) if result.returncode != 0: stderr_tail = result.stderr[-500:].strip() if result.stderr else "" if "No audio stream" in (result.stderr or "") or "no audio" in (result.stderr or "").lower(): errors.append(f"{os.path.basename(file_path)} has no audio track and cannot be converted to an audio format.") else: errors.append(f"FFmpeg error for {os.path.basename(file_path)}: {stderr_tail or 'Unknown error'}") continue if os.path.exists(output_path) and os.path.getsize(output_path) > 0: results.append(output_path) else: errors.append(f"Output file not created for {os.path.basename(file_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)}"}