app.py

"""Private Hugging Face Space UI for the Mocovoice Audio Jev research demo."""

from __future__ import annotations

import importlib
import json
import os
from pathlib import Path
from typing import Any

import gradio as gr


SPACE_ROOT = Path(__file__).resolve().parent
CATALOG_PATH = SPACE_ROOT / "examples" / "catalog.json"

DEFAULT_QUESTIONS = {
    "本人確認": {
        "type": "noul",
        "instructions": "話者が本人確認に必要な情報を明確に述べているか確認してください。",
        "criteria": {
            "false": "必要な情報が不足している、または聞き取れません。",
            "true": "必要な情報を明確に述べています。",
        },
    },
    "聞き取りやすさ": {
        "type": "score",
        "instructions": "発話の明瞭さを評価してください。",
        "criteria": ["聞き取りにくい", "一部聞き取りにくい", "概ね明瞭", "明瞭"],
    },
}

PRESETS: dict[str, tuple[str, dict[str, Any]]] = {
    "本人確認": (
        "本人確認の会話です。氏名、生年月日など必要な情報が音声に含まれているか確認します。",
        DEFAULT_QUESTIONS,
    ),
    "応対品質": (
        "顧客応対の通話です。応対内容と話し方を評価します。",
        {
            "要件の確認": {
                "type": "noul",
                "instructions": "依頼・質問の要点を確認できているか評価してください。",
                "criteria": {"false": "要点を確認できていません。", "true": "要点を確認できています。"},
            },
            "応対評価": {
                "type": "score",
                "instructions": "丁寧で分かりやすい応対か評価してください。",
                "criteria": ["改善が必要", "やや不十分", "適切", "非常に良い"],
            },
        },
    ),
    "選択肢の分類": (
        "この音声を、定義した選択肢のいずれかに分類します。",
        {
            "お問い合わせ種別": {
                "type": "choice",
                "instructions": "主なお問い合わせ内容を一つ選んでください。",
                "criteria": {
                    "契約": "契約内容、変更、解約に関する問い合わせ",
                    "請求": "料金、請求、支払いに関する問い合わせ",
                    "技術": "利用方法や不具合に関する問い合わせ",
                },
            }
        },
    ),
}


def _pretty(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, indent=2)


def _user_error(message: str) -> None:
    raise gr.Error(message)


def _parse_json(value: str, label: str, *, allow_empty: bool = False) -> Any:
    if not value or not value.strip():
        if allow_empty:
            return None
        _user_error(f"{label}を入力してください。")
    try:
        return json.loads(value)
    except json.JSONDecodeError as exc:
        _user_error(f"{label}は有効なJSONにしてください（{exc.msg}）。")


def _validate_questions(raw: Any) -> dict[str, dict[str, Any]]:
    if not isinstance(raw, dict) or not raw:
        _user_error("質問JSONは、質問名をキーにした1件以上のオブジェクトにしてください。")
    validated: dict[str, dict[str, Any]] = {}
    for name, question in raw.items():
        if not isinstance(name, str) or not name.strip():
            _user_error("質問名は空でない文字列にしてください。")
        if not isinstance(question, dict):
            _user_error(f"「{name}」はオブジェクトにしてください。")
        kind = question.get("type")
        instructions = question.get("instructions")
        criteria = question.get("criteria")
        if kind not in {"noul", "choice", "score"}:
            _user_error(f"「{name}」のtypeは noul / choice / score のいずれかにしてください。")
        if not isinstance(instructions, str) or not instructions.strip():
            _user_error(f"「{name}」のinstructionsを入力してください。")
        if kind == "noul":
            if criteria is None:
                criteria = {"false": "該当しません。", "true": "該当します。"}
            if not isinstance(criteria, dict) or set(criteria) != {"false", "true"}:
                _user_error(f"「{name}」のnoul criteriaは false と true を持つオブジェクトにしてください。")
        elif kind == "choice":
            if not isinstance(criteria, dict) or not 2 <= len(criteria) <= 32:
                _user_error(f"「{name}」のchoice criteriaは2〜32件の選択肢オブジェクトにしてください。")
        else:
            if not isinstance(criteria, list) or not 2 <= len(criteria) <= 10:
                _user_error(f"「{name}」のscore criteriaは2〜10件の順序付きリストにしてください。")
        validated[name.strip()] = {"type": kind, "instructions": instructions.strip(), "criteria": criteria}
    return validated


def _questions_from_text(questions_text: str) -> dict[str, dict[str, Any]]:
    return _validate_questions(_parse_json(questions_text, "質問JSON"))


def _criteria_hint(kind: str) -> str:
    if kind == "noul":
        return _pretty({"false": "該当しません", "true": "該当します"})
    if kind == "choice":
        return _pretty({"選択肢A": "説明", "選択肢B": "説明"})
    return _pretty(["低い", "高い"])


def _set_criteria_hint(kind: str) -> str:
    return _criteria_hint(kind)


def _edit_question(
    questions_text: str, name: str, kind: str, instructions: str, criteria_text: str, action: str
) -> tuple[str, str, str, str]:
    questions = _questions_from_text(questions_text)
    clean_name = name.strip()
    if not clean_name:
        _user_error("質問名を入力してください。")
    if action == "remove":
        if clean_name not in questions:
            _user_error(f"「{clean_name}」は質問JSONにありません。")
        del questions[clean_name]
        if not questions:
            _user_error("最後の質問は削除できません。代わりに内容を更新してください。")
        return _pretty(questions), "", "", _criteria_hint("noul")

    criteria = _parse_json(criteria_text, "基準") if criteria_text.strip() else None
    candidate = {clean_name: {"type": kind, "instructions": instructions, "criteria": criteria}}
    normalized = _validate_questions(candidate)[clean_name]
    questions[clean_name] = normalized
    return _pretty(questions), "", "", _criteria_hint(kind)


def _load_preset(name: str) -> tuple[str, str]:
    context, questions = PRESETS[name]
    return context, _pretty(questions)


def _catalog_examples() -> list[list[Any]]:
    if not CATALOG_PATH.is_file():
        return []
    try:
        records = json.loads(CATALOG_PATH.read_text(encoding="utf-8"))
        examples: list[list[Any]] = []
        for record in records:
            if not all(key in record for key in ("name", "audio", "context", "questions")):
                continue
            audio = SPACE_ROOT / str(record["audio"])
            if audio.is_file():
                examples.append([str(audio), str(record["context"]), _pretty(record["questions"])])
        return examples
    except (OSError, json.JSONDecodeError, TypeError):
        return []


def _backend() -> Any:
    try:
        return importlib.import_module("demo_backend")
    except Exception:
        _user_error("デモの実行準備がまだ完了していません。少し待ってから再度お試しください。")


def _summary(result: Any) -> str:
    if not isinstance(result, dict):
        return "結果を受け取りました。下のJSONで詳細を確認してください。"
    lines = ["### 読みやすい結果"]
    transcript = result.get("transcript") or result.get("text")
    if isinstance(transcript, str) and transcript.strip():
        lines.append(f"**文字起こし**  \\n{transcript.strip()}")
    answers = result.get("answers") or result.get("results") or result.get("decisions")
    if isinstance(answers, dict):
        for name, answer in answers.items():
            if isinstance(answer, dict):
                value = answer.get("value", answer.get("answer", answer.get("label", answer)))
                rationale = answer.get("reason") or answer.get("rationale")
                lines.append(f"- **{name}**: {value}" + (f" — {rationale}" if rationale else ""))
            else:
                lines.append(f"- **{name}**: {answer}")
    elif not transcript:
        lines.append("下のJSONで結果を確認してください。")
    return "\n".join(lines)


def _analyze(audio_path: str | None, context: str, questions_text: str) -> tuple[Any, str]:
    if not audio_path:
        _user_error("音声ファイルをアップロードするか、マイクで録音してください。")
    questions = _questions_from_text(questions_text)
    try:
        result = _backend().analyze(str(audio_path), context or "", _pretty(questions))
    except gr.Error:
        raise
    except Exception:
        _user_error("解析に失敗しました。音声と質問の内容を確認して、もう一度お試しください。")
    return result, _summary(result)


def _export_sample(
    audio_path: str | None, context: str, questions_text: str, expected_text: str, result: Any
) -> str:
    if not audio_path:
        _user_error("サンプルを保存するには音声を指定してください。")
    questions = _questions_from_text(questions_text)
    expected = _parse_json(expected_text, "期待する回答", allow_empty=True)
    result_json = result if isinstance(result, dict) else None
    try:
        return _backend().export_sample(str(audio_path), context or "", _pretty(questions), _pretty(expected) if expected is not None else "", result_json)
    except gr.Error:
        raise
    except Exception:
        _user_error("サンプルZIPを作成できませんでした。入力内容を確認して再度お試しください。")


def _model_status() -> str:
    try:
        return str(_backend().model_status())
    except gr.Error:
        return "準備状況を確認できません。"
    except Exception:
        return "準備状況を確認できません。"


def build_app() -> gr.Blocks:
    css = """
    .gradio-container { max-width: 1120px !important; }
    #hero { padding: 1.2rem 0 .4rem; }
    #notice { border-left: 4px solid #e0a42b; padding: .7rem 1rem; background: #fff8e8; border-radius: 6px; }
    """
    with gr.Blocks(title="Mocovoice Audio Jev", css=css) as demo:
        gr.Markdown("# Mocovoice Audio Jev", elem_id="hero")
        gr.Markdown(
            "音声と文脈から、定義した質問への回答を試す研究プロトタイプです。"
            "新しい業務・質問への精度は検証されていません。結果は確認用であり、自動処理は行いません。",
            elem_id="notice",
        )
        with gr.Row():
            with gr.Column(scale=1):
                audio = gr.Audio(label="音声", sources=["upload", "microphone"], type="filepath")
                context = gr.Textbox(label="文脈・補足", lines=5, placeholder="例：この通話は契約変更についてのお問い合わせです。")
                preset = gr.Dropdown(label="プリセット", choices=list(PRESETS), value="本人確認")
                load_preset = gr.Button("プリセットを読み込む")
            with gr.Column(scale=1):
                questions = gr.Textbox(label="質問JSON", value=_pretty(DEFAULT_QUESTIONS), lines=18)
                with gr.Accordion("質問をフォームで編集", open=False):
                    gr.Markdown("質問名が同じ場合は更新します。削除は質問名を指定して行います。")
                    question_name = gr.Textbox(label="質問名")
                    question_type = gr.Radio(["noul", "choice", "score"], value="noul", label="回答タイプ")
                    question_instructions = gr.Textbox(label="指示", lines=3)
                    question_criteria = gr.Textbox(label="基準JSON", value=_criteria_hint("noul"), lines=5)
                    with gr.Row():
                        add_update = gr.Button("追加・更新", variant="secondary")
                        remove = gr.Button("削除", variant="stop")

        with gr.Row():
            analyze = gr.Button("解析する", variant="primary", scale=1)
            status_button = gr.Button("準備状況を確認", scale=0)
        status = gr.Textbox(label="モデルの準備状況", interactive=False, value="解析時にモデルを読み込みます。")

        with gr.Row():
            with gr.Column():
                summary = gr.Markdown("解析すると、ここに読みやすい結果を表示します。")
            with gr.Column():
                result = gr.JSON(label="結果JSON")

        with gr.Accordion("評価サンプルを保存", open=False):
            gr.Markdown("解析を待たずに、音声・文脈・質問をデータセット用ZIPとして保存できます。期待する回答は任意です。")
            expected = gr.Textbox(label="期待する回答JSON（任意）", lines=7, placeholder='例：{"本人確認": true}')
            export = gr.Button("サンプルZIPを作成")
            download = gr.File(label="サンプルZIP")

        examples = _catalog_examples()
        if examples:
            gr.Examples(examples=examples, inputs=[audio, context, questions], label="収録済みの例")

        load_preset.click(_load_preset, inputs=preset, outputs=[context, questions])
        question_type.change(_set_criteria_hint, inputs=question_type, outputs=question_criteria)
        add_update.click(
            lambda q, n, t, i, c: _edit_question(q, n, t, i, c, "upsert"),
            inputs=[questions, question_name, question_type, question_instructions, question_criteria],
            outputs=[questions, question_name, question_instructions, question_criteria],
        )
        remove.click(
            lambda q, n, t, i, c: _edit_question(q, n, t, i, c, "remove"),
            inputs=[questions, question_name, question_type, question_instructions, question_criteria],
            outputs=[questions, question_name, question_instructions, question_criteria],
        )
        analyze.click(_analyze, inputs=[audio, context, questions], outputs=[result, summary])
        status_button.click(_model_status, outputs=status)
        export.click(_export_sample, inputs=[audio, context, questions, expected, result], outputs=download)
    return demo.queue(default_concurrency_limit=1)


if __name__ == "__main__":
    build_app().launch(
        server_name="0.0.0.0",
        server_port=int(os.environ.get("PORT", "7860")),
        share=False,
    )