app.py

"""
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)