Spaces:
Sleeping
Sleeping
| """ | |
| 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( | |
| """ | |
| <div style="display:flex;align-items:center;gap:12px;margin-bottom:4px"> | |
| <span style="background:#e11d48;color:#fff;font-weight:800; | |
| padding:4px 12px;border-radius:8px;font-size:14px; | |
| letter-spacing:1px">VINAMALL</span> | |
| <h1 style="margin:0;font-size:28px">🎙️ Voice2Text</h1> | |
| </div> | |
| <p style="margin:4px 0 0;color:var(--body-text-color-subdued)"> | |
| <b>Chuyển video / audio thành văn bản tiếng Việt bằng AI.</b> | |
| Upload file hoặc dán link (file trực tiếp / YouTube / Facebook...). | |
| </p> | |
| """ | |
| ) | |
| 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) | |