""" Voice2Text — Chuyển video/audio thành văn bản tiếng Việt. Nguồn: upload file HOẶC dán link (file trực tiếp / YouTube / Facebook...). Engine: Gemini hoặc OpenAI Whisper. Có chống AI "bịa" trên file câm. """ from __future__ import annotations import re import shutil import subprocess import time import tempfile from collections import Counter from pathlib import Path import gradio as gr VIDEO_EXT = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v", ".mpg", ".mpeg", ".3gp"} AUDIO_EXT = {".mp3", ".wav", ".m4a", ".aac", ".ogg", ".opus", ".flac", ".wma", ".amr"} MEDIA_EXT = VIDEO_EXT | AUDIO_EXT HAS_FFMPEG = shutil.which("ffmpeg") is not None GEMINI_MODELS = ["gemini-2.5-flash", "gemini-3.1-flash-lite", "gemini-2.5-pro"] WHISPER_MODELS = ["whisper-1", "gpt-4o-mini-transcribe", "gpt-4o-transcribe"] WHISPER_MAX_BYTES = 25 * 1024 * 1024 # OpenAI giới hạn 25MB/file NO_SPEECH = "(không có nội dung thoại)" TRANSCRIBE_PROMPT = """Bạn là công cụ gỡ băng (transcribe) audio. Hãy chuyển toàn bộ lời nói trong file thành văn bản tiếng Việt VERBATIM (đúng từng từ). YÊU CẦU BẮT BUỘC: 1. Ngôn ngữ chính là tiếng Việt; nếu người nói xen tiếng Anh thì GIỮ NGUYÊN, KHÔNG dịch. 2. KHÔNG tóm tắt, KHÔNG bình luận, KHÔNG thêm timestamp. 3. TUYỆT ĐỐI KHÔNG thêm mô tả âm thanh dạng [im lặng], [tiếng nhạc], [tiếng cười]... 4. Mỗi câu hoàn chỉnh hoặc mỗi lượt nói nằm trên một dòng riêng. 5. Nếu HOÀN TOÀN không có lời nói, chỉ trả về đúng một dòng: (không có nội dung thoại) 6. Chỉ trả về phần text transcript thuần, không tiêu đề/ghi chú/markdown. """.strip() # ---------- Tiện ích ---------- def extract_audio(src: str, bitrate: str = "32k") -> str | None: """Tách audio từ video/audio -> mp3 nhẹ để upload nhanh. None nếu lỗi.""" if not HAS_FFMPEG: return None out = str(Path(tempfile.gettempdir()) / f"a_{int(time.time()*1000)}.mp3") cmd = ["ffmpeg", "-y", "-i", src, "-vn", "-ac", "1", "-ar", "16000", "-c:a", "libmp3lame", "-b:a", bitrate, out] try: r = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=900) if r.returncode == 0 and Path(out).exists() and Path(out).stat().st_size > 0: return out except Exception: pass return None def prepare_audio(path: str) -> tuple[str, bool]: """Trả về (đường_dẫn_ASCII_an_toàn, is_temp). Tách/nén ra mp3 tên ASCII để tránh lỗi 'ascii codec' với tên file tiếng Việt trên server locale ASCII.""" a = extract_audio(path) if a: return a, True # Fallback (không có ffmpeg): copy sang tên ASCII, giữ định dạng gốc dst = str(Path(tempfile.gettempdir()) / f"in_{int(time.time()*1000)}{Path(path).suffix.lower()}") shutil.copy(path, dst) return dst, True def looks_like_loop(text: str) -> bool: """Phát hiện AI bịa (lặp 'dạ vâng' vô tận) trên file câm/nhiễu.""" lines = [ln.strip() for ln in text.splitlines() if ln.strip()] if len(lines) < 40: return False fillers = {"dạ", "vâng", "ạ", "ừm", "à", "ờ", "alo", "đúng rồi", "dạ vâng", "vâng ạ", "dạ.", "vâng."} def norm(s: str) -> str: s = re.sub(r"[.,!?;:…\"'\-_/()]+", " ", s.lower()) return re.sub(r"\s+", " ", s).strip() normed = [norm(x) for x in lines] filler_ratio = sum(1 for n in normed if n in fillers or all(w in fillers for w in n.split())) / len(normed) counter = Counter(normed) top3 = sum(c for _, c in counter.most_common(3)) / len(normed) distinct = len(counter) / len(normed) return filler_ratio >= 0.55 or top3 >= 0.6 or distinct <= 0.2 def download_url(url: str) -> tuple[str | None, str]: """Tải video/audio từ link (file trực tiếp / YouTube / FB...) bằng yt-dlp. Trả (đường_dẫn_audio_mp3, tên_hiển_thị). Đã tách sẵn audio.""" import yt_dlp ts = int(time.time() * 1000) out_tmpl = str(Path(tempfile.gettempdir()) / f"dl_{ts}.%(ext)s") opts = { "format": "bestaudio/best", "outtmpl": out_tmpl, "quiet": True, "no_warnings": True, "noplaylist": True, # Thử các client ít bị YouTube chặn bot hơn (android/ios) trên server cloud "extractor_args": {"youtube": {"player_client": ["android", "ios", "web"]}}, "postprocessors": [{ "key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "64", }], } with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url, download=True) title = info.get("title") or info.get("id") or url # File sau postprocess có đuôi .mp3 mp3 = Path(tempfile.gettempdir()) / f"dl_{ts}.mp3" if mp3.exists(): return str(mp3), str(title) # fallback: tìm file dl_ts.* for p in Path(tempfile.gettempdir()).glob(f"dl_{ts}.*"): return str(p), str(title) return None, str(title) # ---------- Engine: Gemini ---------- def transcribe_gemini(api_key: str, path: str, model: str) -> str: from google import genai from google.genai import types client = genai.Client(api_key=api_key) # Luôn chuyển sang audio mp3 tên ASCII (tránh lỗi tên tiếng Việt + nhẹ + rẻ) upload_path, audio_tmp = prepare_audio(path) uploaded = client.files.upload(file=upload_path) try: # đợi ACTIVE deadline = time.time() + 300 while uploaded.state.name == "PROCESSING": if time.time() > deadline: raise TimeoutError("File xử lý quá lâu.") time.sleep(2) uploaded = client.files.get(name=uploaded.name) if uploaded.state.name != "ACTIVE": raise RuntimeError(f"File lỗi: {uploaded.state.name}") cfg = types.GenerateContentConfig( system_instruction=TRANSCRIBE_PROMPT, temperature=0.0, thinking_config=types.ThinkingConfig(thinking_budget=0), ) for wait in (0, 5, 10, 20): if wait: time.sleep(wait) try: resp = client.models.generate_content( model=model, contents=[uploaded, "Gỡ băng file trên theo đúng quy tắc."], config=cfg, ) text = (resp.text or "").strip() return NO_SPEECH if looks_like_loop(text) else text except Exception as e: if "503" in str(e) or "UNAVAILABLE" in str(e): continue raise return NO_SPEECH finally: try: client.files.delete(name=uploaded.name) except Exception: pass if audio_tmp: Path(upload_path).unlink(missing_ok=True) # ---------- Engine: OpenAI Whisper ---------- def transcribe_whisper(api_key: str, path: str, model: str) -> str: from openai import OpenAI client = OpenAI(api_key=api_key) # Whisper giới hạn 25MB -> luôn tách audio nén nhẹ (tên ASCII) audio, _ = prepare_audio(path) try: if Path(audio).stat().st_size > WHISPER_MAX_BYTES: # nén mạnh hơn nếu vẫn lớn smaller = extract_audio(path, bitrate="16k") if smaller and Path(smaller).stat().st_size <= WHISPER_MAX_BYTES: if audio not in (path,): Path(audio).unlink(missing_ok=True) audio = smaller else: raise RuntimeError("File quá dài cho Whisper (>25MB audio). " "Hãy dùng engine Gemini cho file dài.") with open(audio, "rb") as f: resp = client.audio.transcriptions.create( model=model, file=f, language="vi", prompt="Gỡ băng tiếng Việt, giữ nguyên tiếng Anh xen kẽ.", ) text = (resp.text or "").strip() return NO_SPEECH if looks_like_loop(text) else text finally: if audio != path: Path(audio).unlink(missing_ok=True) # ---------- Chạy ---------- def run(engine, api_key, model, files, urls_text, progress=gr.Progress()): if not api_key or not api_key.strip(): raise gr.Error("Chưa nhập API key.") api_key = api_key.strip() # Gom nguồn: file/folder upload + link jobs = [] # (path, ten_hien_thi, is_temp) for f in (files or []): p = f.name if hasattr(f, "name") else f # Khi upload cả folder, chỉ lấy file media (bỏ file lạ) if Path(p).suffix.lower() in MEDIA_EXT: jobs.append((p, Path(p).name, False)) urls = [u.strip() for u in (urls_text or "").splitlines() if u.strip()] if urls: for i, u in enumerate(urls): progress(0, desc=f"Đang tải link {i+1}/{len(urls)}...") try: path, title = download_url(u) if path: jobs.append((path, title, True)) except Exception as e: jobs.append((None, f"[Lỗi tải link: {u[:50]}] {str(e)[:80]}", False)) if not jobs: raise gr.Error("Chưa có file, folder hoặc link nào.") transcribe = transcribe_gemini if engine == "Gemini" else transcribe_whisper sections, errors = [], [] total = len(jobs) for i, (path, name, is_temp) in enumerate(jobs): progress(i / total, desc=f"[{i+1}/{total}] {name[:50]}") if path is None: errors.append(name) continue try: text = transcribe(api_key, path, model) sections.append(f"## {name}\n\n{text}\n") except Exception as e: msg = str(e) # Match CHẶT dấu hiệu hết quota thật (tránh nhận nhầm như số chứa '429') is_quota = ("RESOURCE_EXHAUSTED" in msg or "insufficient_quota" in msg.lower() or "exceeded your current quota" in msg.lower()) if is_quota: errors.append(f"{name}: hết quota/giới hạn API key. → {msg[:120]}") break errors.append(f"{name}: {msg[:200]}") finally: if is_temp and path: Path(path).unlink(missing_ok=True) progress(1.0, desc="Xong") combined = "\n---\n\n".join(sections) if sections else "(không có kết quả)" if errors: combined += "\n\n---\n\n### ⚠️ Ghi chú lỗi\n" + "\n".join(f"- {e}" for e in errors) out = Path(tempfile.gettempdir()) / "transcripts.txt" out.write_text(combined, encoding="utf-8") status = f"✅ Xong {len(sections)}/{total}." + (f" ({len(errors)} lỗi)" if errors else "") return combined, str(out), status # ---------- Giao diện ---------- INTER = gr.themes.GoogleFont("Inter") theme = gr.themes.Soft(font=[INTER, "system-ui", "sans-serif"]) with gr.Blocks(title="Voice2Text", theme=theme) as demo: gr.HTML( """
VINAMALL

🎙️ Voice2Text

Chuyển video / audio thành văn bản tiếng Việt bằng AI. Upload file hoặc dán link (file trực tiếp / YouTube / Facebook...).

""" ) with gr.Row(): with gr.Column(scale=1): engine = gr.Radio( ["Gemini", "OpenAI Whisper"], value="Gemini", label="⚙️ Engine AI", info="Gemini: rẻ, tiếng Việt tốt. Whisper: của OpenAI (file ≤ ~100 phút).", ) api_key = gr.Textbox( label="🔑 API key", type="password", placeholder="Dán Gemini key (AIza...) — aistudio.google.com/apikey", info="Key = tiền công ty. KHÔNG chia sẻ.", ) model = gr.Dropdown(GEMINI_MODELS, value=GEMINI_MODELS[0], label="Model") files = gr.File( label="📁 Upload file / cả folder", file_count="directory", ) urls = gr.Textbox( label="🔗 Hoặc dán link video (mỗi dòng 1 link)", placeholder="https://youtube.com/...\nhttps://.../video.mp4", lines=2, ) btn = gr.Button("▶️ Bắt đầu", variant="primary") status = gr.Textbox(label="Trạng thái", interactive=False) with gr.Column(scale=2): out_text = gr.Textbox(label="📝 Kết quả transcript", lines=27, show_copy_button=True) out_file = gr.File(label="⬇️ Tải file .txt") # Đổi model + placeholder key theo engine def on_engine(e): if e == "Gemini": return (gr.update(choices=GEMINI_MODELS, value=GEMINI_MODELS[0]), gr.update(placeholder="Dán Gemini key (AIza...) — aistudio.google.com/apikey")) return (gr.update(choices=WHISPER_MODELS, value=WHISPER_MODELS[0]), gr.update(placeholder="Dán OpenAI key (sk-...) — platform.openai.com/api-keys")) engine.change(on_engine, inputs=engine, outputs=[model, api_key]) btn.click(run, inputs=[engine, api_key, model, files, urls], outputs=[out_text, out_file, status]) gr.Markdown( "---\n" "**Mẹo:** file câm trả `(không có nội dung thoại)`. " "Link YouTube/FB phải là video công khai. " "Whisper giới hạn ~100 phút/file — file dài hơn dùng Gemini." ) if __name__ == "__main__": demo.queue(max_size=20).launch(show_api=False, ssr_mode=False)