core/ui.py

import os
from datetime import datetime, timezone
from functools import partial
from pathlib import Path

try:
    import spaces  # pyright: ignore[reportMissingImports]
except ImportError:  # pragma: no cover - local fallback when not running on HF Spaces.
    class _SpacesShim:
        @staticmethod
        def GPU(*decorator_args, **decorator_kwargs):
            if decorator_args and callable(decorator_args[0]) and len(decorator_args) == 1 and not decorator_kwargs:
                return decorator_args[0]

            def _decorator(func):
                return func

            return _decorator

    spaces = _SpacesShim()

from benchmarks.harness import (
    DEFAULT_STAGE1_SMOKE_SCRIPT_IDS,
    evaluate_internal_benchmark,
    evaluate_stage1_integration_smoke,
)
from core.adapters import (
    AVAILABLE_EDGE_TTS_VOICES,
    DEFAULT_EDGE_TTS_VOICE,
    EdgeTTSAdapter,
    FasterWhisperAdapter,
    FunASRSenseVoiceAdapter,
    MockASRAdapter,
    MockTTSAdapter,
)
from core.pipeline import (
    AUTO_REPLY_DELAY_SECONDS,
    create_streaming_turn_state,
    maybe_auto_reply_streaming_turn,
    preview_streaming_turn,
    run_controlled_turn,
    run_streaming_turn,
    stop_streaming_turn_state,
)
from utils.config_loader import (
    DEFAULT_BENCHMARK_CSV_PATH,
    DEFAULT_EVALUATION_CONTRACT_PATH,
    DEFAULT_RULES_PATH,
    TABLE_HEADERS,
    load_evaluation_contract,
    load_rule_catalog,
    rules_from_editor_rows,
    rules_to_table_rows,
)
from utils.tracking import LocalJsonlTracker, RunMetadata, make_run_id, resolve_code_version

MINIMAL_UI_NOTICE = "当前为演示版，优先展示实时接话体验。首次加载模型可能稍慢；麦克风不可用时可改用下方兜底输入。"


def _is_zero_gpu_runtime() -> bool:
    return os.getenv("SPACES_ZERO_GPU", "").lower() in {"1", "t", "true"}


def _build_stream_preview_asr_adapters():
    # ZeroGPU forbids CUDA init from the main process, so preview callbacks must
    # stay on CPU when they are not wrapped by @spaces.GPU.
    if _is_zero_gpu_runtime():
        return {
            "funasr": FunASRSenseVoiceAdapter(device_order=("cpu",)),
            "faster-whisper": FasterWhisperAdapter(device_order=(("cpu", "int8"),)),
        }

    return {
        "funasr": FunASRSenseVoiceAdapter(),
        "faster-whisper": FasterWhisperAdapter(),
    }


@spaces.GPU
def _process_turn_for_space(
    audio_input,
    manual_text,
    asr_backend_key,
    tts_backend_key,
    tts_voice,
    rule_rows,
    *,
    tracker,
    manual_asr,
    asr_adapters,
    tts_adapters,
    code_version,
):
    session_rules = rules_from_editor_rows(rule_rows)
    audio_path = Path(audio_input) if audio_input else None
    hypothesis_ids = ["H-001", "H-002", "H-003"] if audio_path else ["H-002", "H-004"]
    route_id = "R-001" if asr_backend_key == "funasr" else "R-002" if asr_backend_key == "faster-whisper" else "R-001"
    run_id = make_run_id(f"stage1-turn-{asr_backend_key or 'manual'}")
    metadata = RunMetadata(
        run_id=run_id,
        experiment_name="stage1-controlled-turn",
        hypothesis_ids=hypothesis_ids,
        baseline_run_id=None,
        route_id=route_id,
        seed=42,
        code_version=code_version,
        data_version=str(audio_path) if audio_path else "ui-manual-text",
        started_at=datetime.now(timezone.utc).isoformat(),
        status="running",
    )

    try:
        outcome = run_controlled_turn(
            run_id=run_id,
            audio_path=audio_path,
            manual_text=manual_text,
            rules=session_rules,
            audio_asr=asr_adapters.get(asr_backend_key),
            manual_asr=manual_asr,
            tts_adapter=tts_adapters.get(tts_backend_key),
            tts_voice=tts_voice,
            tracker=tracker,
            metadata=metadata,
        )
    except Exception as exc:
        return (
            "",
            "",
            "未命中",
            "未命中",
            "",
            f"处理失败：{exc}",
            None,
            _format_run_summary(
                run_id=run_id,
                input_mode="audio" if audio_path else "manual_text",
                asr_backend=asr_backend_key,
                asr_latency_ms=None,
                rule_id=None,
                matched_keyword=None,
                tts_status="失败",
                runtime_note=None,
            ),
        )

    match = outcome.match_result
    tts_preview = outcome.tts_preview
    reply_text_value = match.reply or ("当前未命中规则。" if not match.matched else "")
    tts_status_text = _format_tts_status(tts_preview, match.matched, finalized=True)
    summary = _format_run_summary(
        run_id=outcome.run_id,
        input_mode=outcome.input_mode,
        asr_backend=outcome.asr_result.backend,
        asr_latency_ms=outcome.asr_result.latency_ms,
        rule_id=match.rule_id,
        matched_keyword=match.matched_keyword,
        tts_status=tts_status_text,
        runtime_note=outcome.asr_result.runtime_note,
    )
    return (
        outcome.asr_result.transcript,
        outcome.asr_result.partial_transcript,
        match.matched_keyword or "未命中",
        match.rule_id or "未命中",
        reply_text_value,
        tts_status_text,
        tts_preview.audio_path if tts_preview is not None else None,
        summary,
    )


def _preview_live_microphone_stream_for_space(
    audio_chunk,
    stream_state,
    asr_backend_key,
    rule_rows,
    *,
    asr_adapters,
):
    session_rules = rules_from_editor_rows(rule_rows)

    try:
        updated_state, asr_result, match = preview_streaming_turn(
            state=stream_state,
            audio_chunk=audio_chunk,
            rules=session_rules,
            audio_asr=asr_adapters[asr_backend_key],
        )
    except Exception as exc:
        safe_state = stream_state or create_streaming_turn_state(recording_active=True)
        return (
            safe_state,
            safe_state.transcript,
            safe_state.partial_transcript,
            safe_state.matched_keyword or "未命中",
            safe_state.matched_rule_id or "未命中",
            safe_state.matched_reply_text or "",
            f"实时识别失败：{exc}",
            safe_state.auto_reply_audio_path,
            _format_stream_preview_summary(state=safe_state, error=str(exc)),
        )

    reply_preview = match.reply or ("当前尚未命中规则。" if updated_state.transcript else "")
    return (
        updated_state,
        updated_state.transcript,
        updated_state.partial_transcript,
        match.matched_keyword or "未命中",
        match.rule_id or "未命中",
        reply_preview,
        _format_live_stream_status(updated_state, asr_result=asr_result, finalized=False),
        updated_state.auto_reply_audio_path,
        _format_stream_preview_summary(state=updated_state),
    )


def _auto_reply_live_microphone_stream_for_space(
    stream_state,
    tts_backend_key,
    tts_voice,
    *,
    tts_adapters,
):
    state = stream_state or create_streaming_turn_state()

    try:
        updated_state, tts_preview = maybe_auto_reply_streaming_turn(
            state=state,
            tts_adapter=tts_adapters.get(tts_backend_key),
            tts_voice=tts_voice,
        )
    except Exception as exc:
        return (
            state,
            f"自动播报失败：{exc}",
            state.auto_reply_audio_path,
            _format_stream_preview_summary(state=state, error=str(exc)),
        )

    return (
        updated_state,
        _format_live_stream_status(updated_state, finalized=False, auto_reply_triggered=tts_preview is not None),
        updated_state.auto_reply_audio_path,
        _format_stream_preview_summary(state=updated_state, auto_reply_triggered=tts_preview is not None),
    )


@spaces.GPU
def _finalize_live_microphone_stream_for_space(
    stream_state,
    asr_backend_key,
    tts_backend_key,
    tts_voice,
    rule_rows,
    *,
    tracker,
    manual_asr,
    asr_adapters,
    tts_adapters,
    code_version,
):
    safe_state = stop_streaming_turn_state(stream_state or create_streaming_turn_state())
    session_rules = rules_from_editor_rows(rule_rows)
    run_id = make_run_id(f"stage1-live-stream-{asr_backend_key or 'manual'}")
    metadata = RunMetadata(
        run_id=run_id,
        experiment_name="stage1-live-stream-turn",
        hypothesis_ids=["H-001", "H-002", "H-003"],
        baseline_run_id=None,
        route_id="R-001" if asr_backend_key == "funasr" else "R-002",
        seed=42,
        code_version=code_version,
        data_version="ui-live-microphone-stream",
        started_at=datetime.now(timezone.utc).isoformat(),
        status="running",
    )

    reuse_auto_reply = bool(
        safe_state.current_match_key
        and safe_state.auto_reply_key == safe_state.current_match_key
        and safe_state.auto_reply_audio_path
    )

    try:
        outcome = run_streaming_turn(
            run_id=run_id,
            state=safe_state,
            rules=session_rules,
            audio_asr=asr_adapters[asr_backend_key],
            manual_asr=manual_asr,
            tts_adapter=None if reuse_auto_reply else tts_adapters.get(tts_backend_key),
            tts_voice=tts_voice,
            tracker=tracker,
            metadata=metadata,
        )
    except Exception as exc:
        return (
            create_streaming_turn_state(),
            safe_state.transcript,
            safe_state.partial_transcript,
            safe_state.matched_keyword or "未命中",
            safe_state.matched_rule_id or "未命中",
            safe_state.matched_reply_text or "",
            f"结束录音时处理失败：{exc}",
            safe_state.auto_reply_audio_path,
            _format_stream_preview_summary(state=safe_state, error=str(exc), finalized=True),
        )

    match = outcome.match_result
    tts_preview = outcome.tts_preview
    reply_text_value = safe_state.auto_reply_text or match.reply or ("当前未命中规则。" if not match.matched else "")
    tts_status_text = _format_live_stream_status(
        safe_state,
        asr_result=outcome.asr_result,
        finalized=True,
        auto_reply_triggered=reuse_auto_reply or tts_preview is not None,
    )
    summary = _format_run_summary(
        run_id=outcome.run_id,
        input_mode=outcome.input_mode,
        asr_backend=outcome.asr_result.backend,
        asr_latency_ms=outcome.asr_result.latency_ms,
        rule_id=match.rule_id,
        matched_keyword=match.matched_keyword,
        tts_status=tts_status_text,
        runtime_note=outcome.asr_result.runtime_note,
    )
    return (
        create_streaming_turn_state(),
        outcome.asr_result.transcript,
        outcome.asr_result.partial_transcript,
        match.matched_keyword or "未命中",
        match.rule_id or "未命中",
        reply_text_value,
        tts_status_text,
        safe_state.auto_reply_audio_path if reuse_auto_reply else (tts_preview.audio_path if tts_preview is not None else None),
        summary,
    )


@spaces.GPU
def _run_stage1_smoke_for_space(
    asr_backend_key,
    tts_backend_key,
    tts_voice,
    rule_rows,
    script_ids_text,
    *,
    benchmark_csv_path,
    tracker,
    asr_adapters,
    tts_adapters,
    code_version,
):
    session_rules = rules_from_editor_rows(rule_rows)
    selected_script_ids = _parse_script_ids(script_ids_text)
    run_id = make_run_id(f"stage1-smoke-{asr_backend_key}")
    route_id = "R-001" if asr_backend_key == "funasr" else "R-002"
    metadata = RunMetadata(
        run_id=run_id,
        experiment_name="stage1-synthetic-audio-smoke",
        hypothesis_ids=["H-001", "H-002", "H-003", "H-005"],
        baseline_run_id=None,
        route_id=route_id,
        seed=42,
        code_version=code_version,
        data_version=f"{benchmark_csv_path}#synthetic-audio-smoke",
        started_at=datetime.now(timezone.utc).isoformat(),
        status="running",
    )
    summary = evaluate_stage1_integration_smoke(
        benchmark_csv_path,
        session_rules,
        asr_adapters[asr_backend_key],
        tts_adapters[tts_backend_key],
        tts_adapters[tts_backend_key],
        script_ids=selected_script_ids,
        tracker=tracker,
        metadata=metadata,
        input_voice=tts_voice,
        reply_voice=tts_voice,
    )
    return summary.to_markdown()


def build_app(
    benchmark_csv_path: Path = DEFAULT_BENCHMARK_CSV_PATH,
    rules_path: Path = DEFAULT_RULES_PATH,
    evaluation_contract_path: Path = DEFAULT_EVALUATION_CONTRACT_PATH,
):
    gr = _require_gradio()
    seed_rules = load_rule_catalog(rules_path)
    evaluation_contract = load_evaluation_contract(evaluation_contract_path)
    tracker = LocalJsonlTracker(Path(__file__).resolve().parents[1] / "results" / "tracking")
    manual_asr = MockASRAdapter()
    asr_adapters = {
        "funasr": FunASRSenseVoiceAdapter(),
        "faster-whisper": FasterWhisperAdapter(),
    }
    # Keep preview ASR instances separate so a ZeroGPU live-preview CPU fallback
    # does not pin the finalize/upload paths to CPU as well.
    stream_preview_asr_adapters = _build_stream_preview_asr_adapters()
    tts_adapters = {
        "edge-tts": EdgeTTSAdapter(),
        "mock": MockTTSAdapter(),
    }
    code_version = resolve_code_version(Path(__file__).resolve().parents[1])

    def apply_rules(rule_rows):
        session_rules = rules_from_editor_rows(rule_rows)
        return rules_to_table_rows(session_rules), _format_rule_status(session_rules)

    def start_live_microphone_stream():
        return (
            gr.Timer(value=0.25, active=True),
            create_streaming_turn_state(recording_active=True),
            "",
            "",
            "未命中",
            "未命中",
            "",
            f"实时监听中，命中规则后稳定 {int(AUTO_REPLY_DELAY_SECONDS)} 秒将自动播放回复语音。",
            None,
            _initial_live_stream_placeholder(),
        )

    process_turn = partial(
        _process_turn_for_space,
        tracker=tracker,
        manual_asr=manual_asr,
        asr_adapters=asr_adapters,
        tts_adapters=tts_adapters,
        code_version=code_version,
    )
    preview_live_stream = partial(
        _preview_live_microphone_stream_for_space,
        asr_adapters=stream_preview_asr_adapters,
    )
    auto_reply_live_stream = partial(
        _auto_reply_live_microphone_stream_for_space,
        tts_adapters=tts_adapters,
    )
    finalize_live_stream = partial(
        _finalize_live_microphone_stream_for_space,
        tracker=tracker,
        manual_asr=manual_asr,
        asr_adapters=asr_adapters,
        tts_adapters=tts_adapters,
        code_version=code_version,
    )

    def stop_live_microphone_stream(stream_state, asr_backend_key, tts_backend_key, tts_voice, rule_rows):
        result = finalize_live_stream(stream_state, asr_backend_key, tts_backend_key, tts_voice, rule_rows)
        return (gr.Timer(value=0.25, active=False), *result)

    def preview_benchmark(rule_rows):
        session_rules = rules_from_editor_rows(rule_rows)
        run_id = make_run_id("stage1-rule-preview")
        metadata = RunMetadata(
            run_id=run_id,
            experiment_name="stage1-rule-only-benchmark-preview",
            hypothesis_ids=["H-002"],
            baseline_run_id=None,
            route_id="R-001",
            seed=42,
            code_version=code_version,
            data_version=str(benchmark_csv_path),
            started_at=datetime.now(timezone.utc).isoformat(),
            status="running",
        )
        summary = evaluate_internal_benchmark(
            benchmark_csv_path,
            session_rules,
            tracker=tracker,
            metadata=metadata,
        )
        return summary.to_markdown()

    run_stage1_smoke = partial(
        _run_stage1_smoke_for_space,
        benchmark_csv_path=benchmark_csv_path,
        tracker=tracker,
        asr_adapters=asr_adapters,
        tts_adapters=tts_adapters,
        code_version=code_version,
    )

    with gr.Blocks(title="VoiceDirector 语音场控接话演示") as app:
        live_stream_state = gr.State(value=create_streaming_turn_state())
        auto_reply_timer = gr.Timer(value=0.25, active=False)

        gr.Markdown("# VoiceDirector 语音场控接话演示")
        gr.Markdown("**实时语音接话演示**：连续麦克风转写、自动规则匹配、自动回复语音。")
        gr.Markdown(f"提示：{MINIMAL_UI_NOTICE}")

        with gr.Row():
            with gr.Column(scale=5):
                gr.Markdown(
                    f"### 主路径：麦克风连续流式识别\n点击麦克风开始说话，识别文本会持续刷新；命中规则稳定 {int(AUTO_REPLY_DELAY_SECONDS)} 秒后会自动播放回复语音。"
                )
                live_audio_input = gr.Audio(
                    label="实时麦克风输入（连续流式）",
                    sources=["microphone"],
                    type="numpy",
                    streaming=True,
                )
                asr_backend = gr.Dropdown(
                    choices=[
                        ("FunASR / SenseVoice（主路径）", "funasr"),
                        ("faster-whisper（兜底）", "faster-whisper"),
                    ],
                    value="funasr",
                    label="识别后端",
                )
                tts_backend = gr.Dropdown(
                    choices=[("edge-tts（真实语音）", "edge-tts"), ("mock（调试预览）", "mock")],
                    value="edge-tts",
                    label="回复语音后端",
                )
                tts_voice = gr.Dropdown(
                    choices=[(voice, voice) for voice in AVAILABLE_EDGE_TTS_VOICES],
                    value=DEFAULT_EDGE_TTS_VOICE,
                    label="回复音色",
                )
                with gr.Accordion("上传音频 / 手工输入兜底", open=False):
                    upload_audio_input = gr.Audio(
                        label="上传音频兜底",
                        sources=["upload"],
                        type="filepath",
                        format="wav",
                    )
                    manual_text = gr.Textbox(
                        label="手工文本兜底",
                        placeholder="没有音频时，可直接输入中文文本做规则演示。",
                        lines=3,
                    )
                    run_turn_button = gr.Button("执行上传/手工兜底", variant="secondary")
                recognized_text = gr.Textbox(label="识别文本")
                partial_text = gr.Textbox(label="实时转写预览")
                matched_keyword = gr.Textbox(label="命中关键词")
                matched_rule = gr.Textbox(label="命中规则")
                reply_text = gr.Textbox(label="回复文本", lines=3)
                tts_status = gr.Textbox(label="当前状态")
                reply_audio = gr.Audio(label="回复语音", interactive=False, type="filepath", autoplay=True)
                run_summary = gr.Markdown(value=_initial_live_stream_placeholder())
            with gr.Column(scale=7):
                rule_table = gr.Dataframe(
                    headers=TABLE_HEADERS,
                    datatype=["str", "str", "str", "str"],
                    row_count=(len(seed_rules), "fixed"),
                    column_count=(len(TABLE_HEADERS), "fixed"),
                    value=rules_to_table_rows(seed_rules),
                    label="规则编辑表（当前会话）",
                    interactive=True,
                )
                apply_button = gr.Button("应用规则编辑")
                rule_status = gr.Textbox(label="规则编辑状态", value=_format_rule_status(seed_rules))
                gr.Markdown(_format_contract(evaluation_contract))

        with gr.Accordion("基准与烟测面板", open=False):
            gr.Markdown(
                "下面的规则预览保留了内部基准可见性。端到端烟测会使用 edge-tts 生成输入音频来打通真实 ASR/TTS 代码路径。"
            )
            benchmark_button = gr.Button("预览规则基准")
            benchmark_output = gr.Markdown(value=_initial_benchmark_placeholder(benchmark_csv_path))
            script_ids = gr.Textbox(
                label="烟测脚本 ID",
                value=", ".join(DEFAULT_STAGE1_SMOKE_SCRIPT_IDS),
                lines=2,
            )
            smoke_asr_backend = gr.Dropdown(
                choices=[
                    ("FunASR / SenseVoice（主路径）", "funasr"),
                    ("faster-whisper（兜底）", "faster-whisper"),
                ],
                value="funasr",
                label="烟测识别后端",
            )
            smoke_tts_backend = gr.Dropdown(
                choices=[("edge-tts（真实语音）", "edge-tts"), ("mock（调试预览）", "mock")],
                value="edge-tts",
                label="烟测语音后端",
            )
            smoke_tts_voice = gr.Dropdown(
                choices=[(voice, voice) for voice in AVAILABLE_EDGE_TTS_VOICES],
                value=DEFAULT_EDGE_TTS_VOICE,
                label="烟测音色",
            )
            stage1_smoke_button = gr.Button("执行端到端烟测")
            stage1_smoke_output = gr.Markdown(value=_initial_stage1_smoke_placeholder(benchmark_csv_path))

        run_turn_button.click(
            process_turn,
            inputs=[upload_audio_input, manual_text, asr_backend, tts_backend, tts_voice, rule_table],
            outputs=[recognized_text, partial_text, matched_keyword, matched_rule, reply_text, tts_status, reply_audio, run_summary],
        )
        live_audio_input.start_recording(
            start_live_microphone_stream,
            outputs=[auto_reply_timer, live_stream_state, recognized_text, partial_text, matched_keyword, matched_rule, reply_text, tts_status, reply_audio, run_summary],
            queue=False,
            show_progress="hidden",
        )
        stream_event = live_audio_input.stream(
            preview_live_stream,
            inputs=[live_audio_input, live_stream_state, asr_backend, rule_table],
            outputs=[live_stream_state, recognized_text, partial_text, matched_keyword, matched_rule, reply_text, tts_status, reply_audio, run_summary],
            show_progress="hidden",
            trigger_mode="always_last",
            concurrency_limit=1,
            concurrency_id="live-stream-session",
            stream_every=0.75,
        )
        auto_reply_event = auto_reply_timer.tick(
            auto_reply_live_stream,
            inputs=[live_stream_state, tts_backend, tts_voice],
            outputs=[live_stream_state, tts_status, reply_audio, run_summary],
            show_progress="hidden",
            trigger_mode="always_last",
            concurrency_limit=1,
            concurrency_id="live-stream-session",
        )
        live_audio_input.stop_recording(
            stop_live_microphone_stream,
            inputs=[live_stream_state, asr_backend, tts_backend, tts_voice, rule_table],
            outputs=[auto_reply_timer, live_stream_state, recognized_text, partial_text, matched_keyword, matched_rule, reply_text, tts_status, reply_audio, run_summary],
            show_progress="minimal",
            cancels=[stream_event, auto_reply_event],
        )
        live_audio_input.pause_recording(
            stop_live_microphone_stream,
            inputs=[live_stream_state, asr_backend, tts_backend, tts_voice, rule_table],
            outputs=[auto_reply_timer, live_stream_state, recognized_text, partial_text, matched_keyword, matched_rule, reply_text, tts_status, reply_audio, run_summary],
            show_progress="minimal",
            cancels=[stream_event, auto_reply_event],
        )
        apply_button.click(apply_rules, inputs=[rule_table], outputs=[rule_table, rule_status])
        benchmark_button.click(preview_benchmark, inputs=[rule_table], outputs=[benchmark_output])
        stage1_smoke_button.click(
            run_stage1_smoke,
            inputs=[smoke_asr_backend, smoke_tts_backend, smoke_tts_voice, rule_table, script_ids],
            outputs=[stage1_smoke_output],
        )

    return app


def _format_contract(contract: dict) -> str:
    h003 = contract["h003_mos"]
    notes = contract["evaluation_constraints"]
    return "\n".join(
        [
            "## 固定评测约束",
            f"- H-003 MOS 阈值：{h003['threshold']}/{h003['scale_max']}",
            f"- 最少评审人数：{h003['minimum_raters']}",
            f"- H-004 正式验证阶段：{notes['h004_formal_validation_phase']}",
            f"- H-002 适用范围：{notes['h002_precision_scope']}",
        ]
    )


def _format_rule_status(session_rules) -> str:
    return f"当前会话已加载 {len(session_rules)} 条规则，编辑结果会立即在本次会话生效。"


def _initial_benchmark_placeholder(benchmark_csv_path: Path) -> str:
    return "\n".join(
        [
            "### 规则基准已就绪",
            f"- 数据集：{benchmark_csv_path}",
            "- 范围：仅内部基准",
            "- 用途：查看规则与文本基准，不代表真实语音识别表现。",
        ]
    )


def _initial_stage1_smoke_placeholder(benchmark_csv_path: Path) -> str:
    return "\n".join(
        [
            "### 端到端烟测已就绪",
            f"- 数据集：{benchmark_csv_path}",
            "- 输入音频来源：edge-tts 根据脚本文本合成",
            "- 用途：检查真实 ASR/TTS 链路是否跑通。",
        ]
    )


def _initial_live_stream_placeholder() -> str:
    return "\n".join(
        [
            "### 实时监听已就绪",
            "- 点击麦克风开始说话，识别文本会持续刷新。",
            f"- 命中规则稳定 {int(AUTO_REPLY_DELAY_SECONDS)} 秒后，会自动播放回复语音。",
            "- 上传音频 / 手工输入兜底仍可用。",
            f"- 说明：{MINIMAL_UI_NOTICE}",
        ]
    )


def _parse_script_ids(raw_value: str | None) -> list[str]:
    if not raw_value:
        return list(DEFAULT_STAGE1_SMOKE_SCRIPT_IDS)
    return [item.strip() for item in raw_value.replace("\n", ",").split(",") if item.strip()]


def _format_tts_status(tts_preview, matched: bool, *, finalized: bool) -> str:
    if tts_preview is None:
        return "录音结束，当前未生成回复语音。" if finalized and not matched else "当前未生成回复语音。"

    status = f"已生成回复语音（{tts_preview.latency_ms} ms）"
    if tts_preview.runtime_note:
        status = f"{status}；{tts_preview.runtime_note}"
    return status


def _format_run_summary(
    *,
    run_id: str,
    input_mode: str,
    asr_backend: str | None,
    asr_latency_ms: int | None,
    rule_id: str | None,
    matched_keyword: str | None,
    tts_status: str,
    runtime_note: str | None,
) -> str:
    latency_text = f"{asr_latency_ms} ms" if asr_latency_ms is not None else "未知"
    input_mode_text = "音频输入" if input_mode == "audio" else "手工文本"
    lines = [
        "### 本轮处理结果",
        f"- 运行 ID：{run_id}",
        f"- 输入方式：{input_mode_text}",
        f"- 识别后端：{asr_backend or '手工文本'}",
        f"- 最新识别延迟：{latency_text}",
        f"- 命中规则：{rule_id or '未命中'}",
        f"- 命中关键词：{matched_keyword or '未命中'}",
        f"- 回复语音状态：{tts_status}",
        f"- 说明：{MINIMAL_UI_NOTICE}",
    ]
    if runtime_note:
        lines.append(f"- 运行提示：{runtime_note}")
    return "\n".join(lines)


def _format_live_stream_status(state, *, asr_result=None, finalized: bool, auto_reply_triggered: bool = False) -> str:
    if finalized:
        status = "录音已结束。"
        if state.current_match_key and (state.auto_reply_key == state.current_match_key or auto_reply_triggered):
            status = "录音已结束，已保留当前自动回复语音。"
        elif state.current_match_key:
            status = "录音已结束，已完成本轮规则匹配。"
        else:
            status = "录音已结束，当前未命中规则。"
    elif state.auto_reply_key == state.current_match_key and state.auto_reply_audio_path:
        status = f"已自动播放规则 {state.matched_rule_id or '未命中'} 的回复语音，继续说话可触发新的规则。"
    elif state.current_match_key:
        status = f"已命中规则 {state.matched_rule_id or '未命中'}，稳定 {int(AUTO_REPLY_DELAY_SECONDS)} 秒后将自动播放回复语音。"
    elif state.transcript:
        status = "实时转写中，当前尚未命中规则。"
    else:
        status = "实时监听中，请开始说话。"

    effective_asr_result = asr_result
    if effective_asr_result is None and state.asr_latency_ms is not None:
        latency_text = f"最新识别延迟 {state.asr_latency_ms} ms。"
        status = f"{status} {latency_text}"
    elif effective_asr_result is not None:
        status = f"{status} 最新识别延迟 {effective_asr_result.latency_ms} ms。"

    runtime_note = effective_asr_result.runtime_note if effective_asr_result is not None else state.runtime_note
    if runtime_note:
        status = f"{status} {runtime_note}"
    return status


def _format_stream_preview_summary(*, state, error: str | None = None, finalized: bool = False, auto_reply_triggered: bool = False) -> str:
    if error:
        return "\n".join(
            [
                "### 实时状态",
                f"- 已接收音频片段：{state.chunk_count}",
                f"- 错误：{error}",
                f"- 说明：{MINIMAL_UI_NOTICE}",
            ]
        )

    if state.auto_reply_key == state.current_match_key and state.auto_reply_audio_path:
        auto_reply_status = "已自动播放"
    elif state.current_match_key:
        auto_reply_status = f"等待稳定 {int(AUTO_REPLY_DELAY_SECONDS)} 秒"
    else:
        auto_reply_status = "当前未触发"

    if auto_reply_triggered:
        auto_reply_status = "刚刚完成自动播放"
    if finalized and state.current_match_key:
        auto_reply_status = "录音结束，已完成本轮处理"

    latency_text = f"{state.asr_latency_ms} ms" if state.asr_latency_ms is not None else "未知"
    lines = [
        "### 实时状态",
        f"- 已接收音频片段：{state.chunk_count}",
        f"- 当前采样率：{state.sample_rate or '未知'} Hz",
        f"- 识别后端：{state.asr_backend or '尚未开始'}",
        f"- 最新识别延迟：{latency_text}",
        f"- 当前命中规则：{state.matched_rule_id or '未命中'}",
        f"- 当前命中关键词：{state.matched_keyword or '未命中'}",
        f"- 自动播报状态：{auto_reply_status}",
        f"- 说明：{MINIMAL_UI_NOTICE}",
    ]
    if state.runtime_note:
        lines.append(f"- 运行提示：{state.runtime_note}")
    return "\n".join(lines)


def _require_gradio():
    try:
        import gradio as gr
    except ImportError as exc:
        raise RuntimeError("未安装 Gradio，请先执行 `pip install -r requirements.txt`。") from exc
    return gr