app.py

"""한국어 TTS 사람 청취평가 — 블라인드 받아쓰기 + 0~10점.

왜 필요한가
-----------
Open Ko-S2S 리더보드의 TTS 발음 지표는 합성음을 ASR 로 되받아쓴 CER 이다.
4090 실측에서 이 지표가 발음이 아니라 **숫자 표기**를 재고 있다는 것이 드러났다.

    (기존 20문장 세트, 표기 보정 없음)
    Qwen3-TTS 전체 CER 11.08%
      ├ 숫자 표기 문장(n=9)  23.67%   ← ASR 이 "삼십 분"을 "30분"으로 씀
      └ 숫자 없는 문장(n=11)  0.50%   ← 발음은 사실상 완벽

허용 표기(variants)를 도입해 이 편향을 걷어내자 숫자문장 CER 이 23.67% → 3.09%
로 떨어졌다(의원 접수대 100문장 기준). 남은 것이 진짜 발음 오차다.

    (의원 100문장, variants 적용)
    Qwen3-TTS  전체 1.50%  숫자 3.09%  숫자없음 1.02%
    MMS-TTS    전체 6.28%  숫자 7.74%  숫자없음 5.84%

다만 이 보정이 옳은지, 남은 오차가 정말 발음 문제인지는 자동 지표로 확정할 수
없다. 사람 귀가 유일한 기준선이라 이 앱으로 human CER 과 주관 점수를 모은다.

설계
----
- **블라인드**: 어떤 모델의 음성인지, 정답 문장이 무엇인지 제출 전에는 감춘다.
  정답을 먼저 보여주면 받아쓰기가 베끼기가 되어 명료도 측정이 무의미해진다.
- **표기 통제**: 사람도 "삼십 분"을 듣고 "30분"이라 적는다. 그래서 안내로
  한글 표기를 요청하고, 채점 시에도 숫자를 정규화해 표기 차이를 제거한다.
- **저장**: HF_TOKEN 시크릿이 있으면 데이터셋 repo 로 누적, 없으면 로컬 파일로
  받아 두고 화면에 경고한다(조용히 유실되지 않게).
"""

from __future__ import annotations

import json
import os
import random
import re
import unicodedata
import uuid
from datetime import datetime, timezone
from pathlib import Path

import gradio as gr

ROOT = Path(__file__).resolve().parent
AUDIO_DIR = ROOT / "audio"
MANIFEST = json.loads((AUDIO_DIR / "manifest.json").read_text(encoding="utf-8"))
ITEMS = MANIFEST["items"]
# 도메인: 의원(1차 의료기관) 접수대 100문장 × TTS 2종 = 200 샘플.
# 음성봇이 가장 먼저 투입되는 구간이라 이 도메인의 발음이 실사용을 좌우한다.

DATASET_REPO = os.environ.get("EVAL_DATASET_REPO", "baryonlabs/ko-tts-human-eval-votes")
HF_TOKEN = os.environ.get("HF_TOKEN")
LOCAL_LOG = Path(os.environ.get("EVAL_LOCAL_LOG", "/tmp/ko_tts_votes.jsonl"))

# ── 채점 ────────────────────────────────────────────────────────────────

# 한글 수사 → 숫자. 사람도 ASR 도 "삼십 분"을 "30분"으로 적을 수 있어서, 표기
# 차이가 명료도 점수를 오염시키지 않도록 양쪽을 같은 형태로 만든 뒤 비교한다.
_SINO = {"영": 0, "공": 0, "일": 1, "이": 2, "삼": 3, "사": 4, "오": 5,
         "육": 6, "칠": 7, "팔": 8, "구": 9}
_UNIT = {"십": 10, "백": 100, "천": 1000}
_NATIVE = {"한": 1, "두": 2, "세": 3, "네": 4, "다섯": 5, "여섯": 6, "일곱": 7,
           "여덟": 8, "아홉": 9, "열": 10, "스물": 20, "서른": 30, "마흔": 40,
           "쉰": 50, "예순": 60, "일흔": 70, "여든": 80, "아흔": 90}
_PCT = re.compile(r"퍼센트|프로")


def _sino_to_int(chunk: str):
    """'삼십사' → 34. 해석 불가하면 None."""
    total, current, seen = 0, 0, False
    for ch in chunk:
        if ch in _SINO:
            current = _SINO[ch]
            seen = True
        elif ch in _UNIT:
            total += (current or 1) * _UNIT[ch]
            current = 0
            seen = True
        else:
            return None
    return total + current if seen else None


def normalize_numbers(text: str) -> str:
    """숫자 표기를 통일한다: 아라비아 숫자·퍼센트 기호·한글 수사를 한 형태로."""
    t = _PCT.sub("%", text)
    t = re.sub(r"(\d),(?=\d{3})", r"\1", t)          # 4,500 → 4500
    t = t.replace("m", "미터").replace("M", "미터")

    for word, val in sorted(_NATIVE.items(), key=lambda kv: -len(kv[0])):
        t = t.replace(word, str(val))

    def repl(m):
        v = _sino_to_int(m.group(0))
        return str(v) if v is not None else m.group(0)

    t = re.sub(r"[영공일이삼사오육칠팔구십백천]{1,8}", repl, t)
    return t


def normalize(text: str, fold_numbers: bool = True) -> str:
    t = unicodedata.normalize("NFC", (text or "").strip())
    if fold_numbers:
        t = normalize_numbers(t)
    t = re.sub(r"[^\w가-힣%]", "", t)
    return t


def _edit_distance(a: str, b: str) -> int:
    if not a:
        return len(b)
    prev = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        cur = [i]
        for j, cb in enumerate(b, 1):
            cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
        prev = cur
    return prev[-1]


def char_error_rate(ref: str, hyp: str, fold_numbers: bool = True) -> float:
    r, h = normalize(ref, fold_numbers), normalize(hyp, fold_numbers)
    if not r:
        return 0.0
    return _edit_distance(r, h) / len(r)


# ── 저장 ────────────────────────────────────────────────────────────────

def _scheduler():
    """HF_TOKEN 이 있으면 데이터셋 repo 로 누적 저장하는 스케줄러를 만든다."""
    if not HF_TOKEN:
        return None
    try:
        from huggingface_hub import CommitScheduler

        LOCAL_LOG.parent.mkdir(parents=True, exist_ok=True)
        return CommitScheduler(
            repo_id=DATASET_REPO, repo_type="dataset", folder_path=LOCAL_LOG.parent,
            path_in_repo="data", every=5, token=HF_TOKEN, private=False,
        )
    except Exception as exc:  # 저장 실패를 조용히 삼키지 않는다
        print(f"[warn] CommitScheduler 초기화 실패: {exc}")
        return None


SCHEDULER = _scheduler()
STORAGE_NOTE = (
    f"✅ 응답이 `{DATASET_REPO}` 데이터셋에 누적됩니다."
    if SCHEDULER
    else "⚠️ **HF_TOKEN 시크릿이 없어 응답이 이 컨테이너에만 임시 저장됩니다.** "
         "Space 재시작 시 사라집니다. Settings → Secrets 에 write 권한 HF_TOKEN 을 넣어주세요."
)


def save_vote(record: dict) -> None:
    LOCAL_LOG.parent.mkdir(parents=True, exist_ok=True)
    line = json.dumps(record, ensure_ascii=False)
    if SCHEDULER is not None:
        with SCHEDULER.lock:
            with LOCAL_LOG.open("a", encoding="utf-8") as f:
                f.write(line + "\n")
    else:
        with LOCAL_LOG.open("a", encoding="utf-8") as f:
            f.write(line + "\n")


def load_votes() -> list[dict]:
    if not LOCAL_LOG.exists():
        return []
    out = []
    for line in LOCAL_LOG.read_text(encoding="utf-8").splitlines():
        if line.strip():
            try:
                out.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return out


# ── 출제 ────────────────────────────────────────────────────────────────

def pick_item(seen_ids: list[str]) -> dict:
    """아직 안 들은 것 우선, 그 안에서 응답 수가 적은 것 우선."""
    counts = {}
    for v in load_votes():
        counts[v["item_id"]] = counts.get(v["item_id"], 0) + 1
    pool = [it for it in ITEMS if it["id"] not in seen_ids] or ITEMS
    fewest = min(counts.get(it["id"], 0) for it in pool)
    return random.choice([it for it in pool if counts.get(it["id"], 0) == fewest])


def start(seen_ids):
    item = pick_item(seen_ids or [])
    return (
        str(AUDIO_DIR / item["audio"]),
        item,
        gr.update(value="", interactive=True),
        gr.update(value=5, interactive=True),
        gr.update(visible=False, value=""),
        gr.update(visible=True),
        gr.update(visible=False),
    )


def _best_cer(item, heard, fold_numbers):
    """참조에 허용 표기(variants)가 있으면 오류가 가장 적은 것을 택한다.

    `삼십 분` / `30분` 처럼 발음이 같고 표기만 다른 형태를 하나만 정답으로 쓰면,
    그 표기를 적은 평가자만 유리해진다. KsponSpeech 가 이중전사를 제공하는 것과
    같은 이유다.
    """
    refs = item.get("variants") or [item["ref"]]
    return min(char_error_rate(r, heard, fold_numbers=fold_numbers) for r in refs)


def submit(item, heard, score, seen_ids, rater_id):
    if not item:
        return (gr.update(), gr.update(visible=True, value="먼저 '다음 문장'을 눌러주세요."),
                gr.update(), gr.update(), seen_ids)
    if not (heard or "").strip():
        return (gr.update(), gr.update(visible=True, value="⚠️ 들린 내용을 입력해주세요."),
                gr.update(visible=True), gr.update(visible=False), seen_ids)

    cer_folded = _best_cer(item, heard, True)
    cer_raw = _best_cer(item, heard, False)
    record = {
        "vote_id": str(uuid.uuid4()),
        "rater_id": rater_id,
        "ts": datetime.now(timezone.utc).isoformat(),
        "item_id": item["id"],
        "model_key": item["model_key"],
        "model_name": item["model_name"],
        "sentence_id": item.get("sentence_id", item.get("id")),
        "category": item.get("category"),
        "has_number": item.get("has_number"),
        "ref": item["ref"],
        "heard": heard.strip(),
        "score": int(score),
        "human_cer": round(cer_folded, 4),
        "human_cer_raw": round(cer_raw, 4),
    }
    save_vote(record)

    seen = (seen_ids or []) + [item["id"]]
    reveal = (
        f"### 제출 완료\n"
        f"**정답** : {item['ref']}\n\n"
        f"**입력** : {heard.strip()}\n\n"
        f"| 지표 | 값 |\n| --- | --- |\n"
        f"| 사람 CER (숫자 정규화) | **{cer_folded * 100:.1f}%** |\n"
        f"| 사람 CER (원문 그대로) | {cer_raw * 100:.1f}% |\n"
        f"| 내 점수 | {int(score)} / 10 |\n\n"
        f"두 CER 이 다르면 그 차이가 곧 **표기 때문에 생긴 오차**입니다.\n\n"
        f"누적 응답 {len(load_votes())}개 · 계속하려면 **다음 문장**"
    )
    return (gr.update(visible=True, value=reveal), gr.update(visible=False, value=""),
            gr.update(visible=False), gr.update(visible=True), seen)


def leaderboard():
    votes = load_votes()
    if not votes:
        return "아직 응답이 없습니다. 몇 개 평가하면 여기에 모델별 집계가 나옵니다."
    agg = {}
    for v in votes:
        a = agg.setdefault(v["model_name"], {"n": 0, "score": 0.0, "cer": 0.0, "cer_raw": 0.0,
                                             "n_num": 0, "cer_num": 0.0})
        a["n"] += 1
        a["score"] += v["score"]
        a["cer"] += v["human_cer"]
        a["cer_raw"] += v["human_cer_raw"]
        if v.get("has_number"):
            a["n_num"] += 1
            a["cer_num"] += v["human_cer"]
    rows = sorted(agg.items(), key=lambda kv: kv[1]["cer"] / kv[1]["n"])
    out = ["| 모델 | 응답수 | 평균 점수 (0~10) | 사람 CER | 사람 CER(원문) | 숫자문장 CER |",
           "| --- | ---: | ---: | ---: | ---: | ---: |"]
    for name, a in rows:
        n = a["n"]
        num = f"{a['cer_num'] / a['n_num'] * 100:.2f}%" if a["n_num"] else "—"
        out.append(f"| {name} | {n} | {a['score'] / n:.2f} | "
                   f"{a['cer'] / n * 100:.2f}% | {a['cer_raw'] / n * 100:.2f}% | {num} |")
    out.append(f"\n총 {len(votes)}개 응답 · 평가자 {len({v['rater_id'] for v in votes})}명")
    out.append("\n※ '사람 CER'은 숫자 표기를 정규화한 값, '원문'은 적힌 그대로입니다. "
               "둘의 차이가 표기 관습이 만드는 오차의 크기입니다.")
    return "\n".join(out)


INTRO = f"""# 🎧 한국어 TTS 사람 청취평가

합성음을 듣고 **들리는 대로** 받아쓴 뒤 0~10점을 매겨주세요.
어떤 모델인지, 정답 문장이 무엇인지는 제출 후에 공개됩니다.

**받아쓰기 요령**
- 들린 그대로 적어주세요. 못 알아들은 부분은 비워두거나 `?` 로 두면 됩니다.
- **숫자는 들린 대로 한글로** 적어주세요. (`30분`이 아니라 `삼십 분`)
  자동 채점이 숫자 표기 때문에 발음 오류를 과대평가하는 문제를 확인하려는 것이 이 평가의 목적입니다.

**0~10점 기준** — 0: 전혀 못 알아들음 · 5: 알아듣지만 어색함 · 10: 사람 발화와 구분 안 됨

{STORAGE_NOTE}
"""

with gr.Blocks(title="한국어 TTS 사람 청취평가") as demo:
    seen_state = gr.State([])
    item_state = gr.State(None)
    rater_state = gr.State(lambda: str(uuid.uuid4())[:8])

    gr.Markdown(INTRO)

    with gr.Tab("평가하기"):
        audio = gr.Audio(label="합성음 (모델명 비공개)", interactive=False, autoplay=False)
        heard_box = gr.Textbox(label="들린 대로 받아쓰기", lines=2,
                               placeholder="예) 오늘 회의는 오후 세 시 삼십 분에 시작합니다")
        score_slider = gr.Slider(0, 10, value=5, step=1, label="자연스러움·명료도 (0~10)")
        warn = gr.Markdown(visible=False)
        with gr.Row():
            submit_btn = gr.Button("제출", variant="primary", visible=False)
            next_btn = gr.Button("다음 문장", variant="secondary")
        reveal_box = gr.Markdown(visible=False)

    with gr.Tab("집계"):
        board = gr.Markdown()
        gr.Button("새로고침").click(leaderboard, outputs=board)
        demo.load(leaderboard, outputs=board)

    next_btn.click(start, inputs=[seen_state],
                   outputs=[audio, item_state, heard_box, score_slider,
                            reveal_box, submit_btn, next_btn])
    submit_btn.click(submit,
                     inputs=[item_state, heard_box, score_slider, seen_state, rater_state],
                     outputs=[reveal_box, warn, submit_btn, next_btn, seen_state])

if __name__ == "__main__":
    demo.launch()