"""
FunASR Paraformer 语音识别 FastAPI 服务

ASR 模型: speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch
VAD 模型: speech_fsmn_vad_zh-cn-16k-common-pytorch
PUNC 模型: punc_ct-transformer_zh-cn-common-vocab272727-pytorch
端口: 59805
"""

import os
import re
import json
import asyncio
import tempfile
import logging
import subprocess
import wave
from typing import Optional

import requests
import uvicorn
try:
    import websockets
except ImportError:
    websockets = None
from fastapi import FastAPI, UploadFile, File, Form, Query, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse, PlainTextResponse
from funasr import AutoModel


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


_CN_NUM = {
    "零": 0,
    "一": 1,
    "幺": 1,
    "二": 2,
    "两": 2,
    "三": 3,
    "四": 4,
    "五": 5,
    "六": 6,
    "七": 7,
    "八": 8,
    "九": 9,
}

_CN_UNIT = {
    "十": 10,
    "百": 100,
    "千": 1000,
    "万": 10000,
    "亿": 100000000,
}

_CN_PATTERN = re.compile(r"[零一幺二两三四五六七八九十百千万亿]+")


def _cn2num(s: str) -> int:
    total, section, num = 0, 0, 0

    for ch in s:
        if ch in _CN_NUM:
            num = _CN_NUM[ch]
        elif ch in _CN_UNIT:
            unit = _CN_UNIT[ch]
            if unit >= 10000:
                section = (section + (num or 1)) * unit
                total += section
                section = 0
            else:
                section += (num or 1) * unit
            num = 0

    return total + section + num


def cn_to_arabic(text: str) -> str:
    return _CN_PATTERN.sub(lambda m: str(_cn2num(m.group())), text)


def read_txt_file(file_path: str) -> str:
    try:
        with open(file_path, "r", encoding="utf-8") as file:
            return file.read()
    except FileNotFoundError:
        logger.warning(f"找不到热词文件: {file_path}")
        return ""
    except Exception as e:
        logger.warning(f"读取热词文件失败: {e}")
        return ""


app = FastAPI(
    title="Paraformer ASR API",
    description="基于 FunASR Paraformer 的中文语音识别服务",
    version="1.0.0",
)


model = None
VOICE_DETECTED_URL = os.environ.get("VOICE_DETECTED_URL", "http://127.0.0.1:18000")
VOICEPRINT_STORE = os.environ.get("VOICEPRINT_STORE", "/tmp/asr_voiceprints.json")
VOICEPRINT_MATCH_THRESHOLD = float(os.environ.get("VOICEPRINT_MATCH_THRESHOLD", "0.45"))
REALTIME_PARTIAL_INTERVAL_SECONDS = float(os.environ.get("REALTIME_PARTIAL_INTERVAL_SECONDS", "2.0"))
REALTIME_PARTIAL_MIN_SECONDS = float(os.environ.get("REALTIME_PARTIAL_MIN_SECONDS", "1.5"))
OFFLINE_REALTIME_WS_URL = os.environ.get(
    "OFFLINE_REALTIME_WS_URL", "ws://fastapi-wss:10095"
)
OFFLINE_REALTIME_CHUNK_MS = int(
    os.environ.get("OFFLINE_REALTIME_CHUNK_MS", "60")
)
OFFLINE_REALTIME_FINAL_TIMEOUT = float(
    os.environ.get("OFFLINE_REALTIME_FINAL_TIMEOUT", "15")
)


def _voiceprint_key(database_name: str, collection_name: str) -> str:
    return f"{database_name or 'NB'}::{collection_name or 'voice'}"


def _load_voiceprints() -> dict:
    if not os.path.exists(VOICEPRINT_STORE):
        return {}
    try:
        with open(VOICEPRINT_STORE, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except Exception as e:
        logger.warning(f"读取声纹库失败，将使用空库: {e}")
        return {}


def _save_voiceprints(data: dict):
    os.makedirs(os.path.dirname(VOICEPRINT_STORE) or ".", exist_ok=True)
    tmp_path = f"{VOICEPRINT_STORE}.tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    os.replace(tmp_path, VOICEPRINT_STORE)


def _extract_embedding(audio_path: str, filename: str) -> list:
    with open(audio_path, "rb") as f:
        response = requests.post(
            f"{VOICE_DETECTED_URL.rstrip('/')}/extract_embedding",
            files={"wav_file": (filename or "voice.wav", f, "audio/wav")},
            timeout=120,
        )
    response.raise_for_status()
    payload = response.json()
    embedding = payload.get("embedding")
    if not isinstance(embedding, list):
        raise ValueError("voice-dected 返回中没有有效 embedding")
    return embedding


def _convert_to_embedding_wav(audio_path: str) -> str:
    output = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
    output_path = output.name
    output.close()

    try:
        subprocess.run(
            [
                "ffmpeg",
                "-y",
                "-i",
                audio_path,
                "-acodec",
                "pcm_s16le",
                "-ar",
                "16000",
                "-ac",
                "1",
                output_path,
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=60,
        )
        return output_path
    except Exception:
        if os.path.exists(output_path):
            os.unlink(output_path)
        raise


def _cosine_similarity(a: list, b: list) -> float:
    if not a or not b or len(a) != len(b):
        return -1.0
    dot = sum(float(x) * float(y) for x, y in zip(a, b))
    norm_a = sum(float(x) * float(x) for x in a) ** 0.5
    norm_b = sum(float(y) * float(y) for y in b) ** 0.5
    if norm_a == 0 or norm_b == 0:
        return -1.0
    return dot / (norm_a * norm_b)


def _match_voiceprint(audio_path: str, filename: str, database_name: str = "NB", collection_name: str = "voice"):
    store = _load_voiceprints()
    records = store.get(_voiceprint_key(database_name, collection_name), {})
    if not records:
        return None, None

    embedding_path = _convert_to_embedding_wav(audio_path)
    try:
        embedding = _extract_embedding(embedding_path, "match.wav")
    finally:
        if os.path.exists(embedding_path):
            os.unlink(embedding_path)

    best_name = None
    best_score = -1.0

    for name, record in records.items():
        score = _cosine_similarity(embedding, record.get("embedding", []))
        if score > best_score:
            best_name = name
            best_score = score

    logger.info(f"声纹匹配结果: name={best_name}, score={best_score:.4f}, threshold={VOICEPRINT_MATCH_THRESHOLD}")
    if best_name and best_score >= VOICEPRINT_MATCH_THRESHOLD:
        return best_name, best_score
    return None, best_score


def load_model():
    global model

    if model is None:
        logger.info("正在加载 Paraformer 模型...")
        model = AutoModel(
            model="/app/models/seacomodel",
            vad_model="/app/models/vadmodel",
            punc_model="/app/models/ctpuncmodel",
        )
        logger.info("模型加载完成")

    return model


@app.on_event("startup")
async def startup_event():
    load_model()


@app.get("/")
async def root():
    return {
        "status": "ok",
        "message": "Paraformer ASR 服务运行中",
    }


@app.get("/health")
async def health():
    return {"status": "healthy"}


def _write_pcm_wav(audio_bytes: bytes) -> str:
    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
        tmp_path = tmp.name

    with wave.open(tmp_path, "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(16000)
        wav_file.writeframes(audio_bytes)

    return tmp_path


def _text_delta(previous: str, current: str) -> str:
    previous = previous or ""
    current = current or ""
    if not current or current == previous:
        return ""
    if current.startswith(previous):
        return current[len(previous):]

    max_overlap = min(len(previous), len(current), 80)
    for size in range(max_overlap, 0, -1):
        if previous[-size:] == current[:size]:
            return current[size:]
    return current


async def run_asr_websocket(websocket: WebSocket):
    await websocket.accept(subprotocol="binary")
    audio_chunks = []
    partial_chunks = []
    last_partial_at = 0.0

    try:
        while True:
            message = await websocket.receive()

            if "bytes" in message and message["bytes"]:
                audio_chunks.append(message["bytes"])
                partial_chunks.append(message["bytes"])
                audio_seconds = sum(len(chunk) for chunk in audio_chunks) / 32000.0
                partial_seconds = sum(len(chunk) for chunk in partial_chunks) / 32000.0
                now = asyncio.get_running_loop().time()

                if (
                    partial_seconds >= REALTIME_PARTIAL_MIN_SECONDS
                    and now - last_partial_at >= REALTIME_PARTIAL_INTERVAL_SECONDS
                ):
                    last_partial_at = now
                    partial_audio = b"".join(partial_chunks)
                    partial_chunks = []
                    partial_path = None
                    try:
                        partial_path = _write_pcm_wav(partial_audio)
                        result = await run_asr(partial_path, language="zh", response_format="json")
                        partial_text = result.get("text", "") if isinstance(result, dict) else ""
                        if partial_text:
                            await websocket.send_json({
                                "mode": "2pass-online",
                                "text": partial_text,
                                "timestamp": str(int(audio_seconds * 1000)),
                                "is_final": False,
                            })
                    except Exception as e:
                        logger.warning(f"实时部分识别失败: {e}")
                    finally:
                        if partial_path and os.path.exists(partial_path):
                            os.unlink(partial_path)
                continue

            if "text" in message and message["text"]:
                try:
                    payload = json.loads(message["text"])
                except Exception:
                    payload = {}

                if payload.get("type") == "stop" or payload.get("is_speaking") is False:
                    break
                continue

            if message.get("type") == "websocket.disconnect":
                break
    except WebSocketDisconnect:
        pass

    text = ""
    tmp_path = None
    try:
        if audio_chunks:
            tmp_path = _write_pcm_wav(b"".join(audio_chunks))
            result = await run_asr(tmp_path, language="zh", response_format="json")
            if isinstance(result, dict):
                text = result.get("text", "")

        try:
            await websocket.send_json({
                "mode": "2pass-offline",
                "text": text,
                "is_final": True,
            })
        except Exception as e:
            logger.warning(f"Skip final transcript because websocket is closed: {e}")
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)
        try:
            await websocket.close()
        except Exception:
            pass


@app.websocket("/ws/asr")
async def websocket_asr(websocket: WebSocket):
    await run_asr_websocket(websocket)


@app.websocket("/")
async def websocket_asr_root(websocket: WebSocket):
    await run_asr_websocket(websocket)


def make_speakr_response(text: str, language: str = "zh", speaker: str = "SPEAKER_00"):
    text = (text or "").strip()
    return {
        "text": text,
        "language": language,
        "segments": [
            {
                "start": None,
                "end": None,
                "text": text,
                "speaker": speaker or "SPEAKER_00",
            }
        ] if text else [],
    }


def _request_diarization(audio_path: str, speaker_num: int, threshold: float) -> list:
    wav_path = _convert_to_embedding_wav(audio_path)
    try:
        with open(wav_path, "rb") as f:
            response = requests.post(
                f"{VOICE_DETECTED_URL.rstrip('/')}/diarize",
                files={"wav_file": ("audio.wav", f, "audio/wav")},
                data={"speaker_num": speaker_num, "threshold": threshold},
                timeout=600,
            )
        response.raise_for_status()
        segments = response.json().get("segments", [])
        return segments if isinstance(segments, list) else []
    finally:
        if os.path.exists(wav_path):
            os.unlink(wav_path)


def _resolve_diarized_speakers(
    segments: list,
    database_name: str,
    collection_name: str,
) -> list:
    """Bind each cluster to one enrolled name, or keep one anonymous label."""
    records = _load_voiceprints().get(
        _voiceprint_key(database_name, collection_name), {}
    )
    cluster_embeddings = {}
    for segment in segments:
        speaker = segment.get("speaker") or "SPEAKER_00"
        embedding = segment.get("embedding")
        if isinstance(embedding, list) and embedding:
            cluster_embeddings.setdefault(speaker, []).append(embedding)

    speaker_names = {}
    for speaker, embeddings in cluster_embeddings.items():
        size = len(embeddings[0])
        valid = [item for item in embeddings if len(item) == size]
        if not valid:
            continue
        mean_embedding = [
            sum(float(item[index]) for item in valid) / len(valid)
            for index in range(size)
        ]
        best_name = None
        best_score = -1.0
        for name, record in records.items():
            score = _cosine_similarity(
                mean_embedding, record.get("embedding", [])
            )
            if score > best_score:
                best_name = name
                best_score = score
        if best_name and best_score >= VOICEPRINT_MATCH_THRESHOLD:
            speaker_names[speaker] = best_name
            logger.info(
                f"离线聚类 {speaker} 统一匹配为 {best_name}, score={best_score:.4f}"
            )
        else:
            speaker_names[speaker] = speaker

    resolved = []
    for segment in segments:
        item = dict(segment)
        anonymous_name = item.get("speaker") or "SPEAKER_00"
        item["speaker"] = speaker_names.get(anonymous_name, anonymous_name)
        item.pop("embedding", None)
        resolved.append(item)
    return resolved


def _merge_neighboring_segments(segments: list, max_gap: float = 0.6) -> list:
    """Merge nearby utterances of the same resolved speaker before ASR."""
    merged = []
    for segment in sorted(segments, key=lambda item: float(item["start"])):
        item = dict(segment)
        if (
            merged
            and merged[-1]["speaker"] == item["speaker"]
            and float(item["start"]) - float(merged[-1]["end"]) <= max_gap
        ):
            merged[-1]["end"] = max(
                float(merged[-1]["end"]), float(item["end"])
            )
        else:
            merged.append(item)
    return merged


def _extract_audio_segment(audio_path: str, start: float, end: float) -> str:
    output = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
    output_path = output.name
    output.close()
    try:
        subprocess.run(
            [
                "ffmpeg", "-y", "-i", audio_path,
                "-ss", str(max(0.0, start)),
                "-to", str(max(start, end)),
                "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
                output_path,
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=120,
        )
        return output_path
    except Exception:
        if os.path.exists(output_path):
            os.unlink(output_path)
        raise


def _convert_to_pcm(audio_path: str) -> str:
    output = tempfile.NamedTemporaryFile(delete=False, suffix=".pcm")
    output_path = output.name
    output.close()
    try:
        subprocess.run(
            [
                "ffmpeg", "-y", "-i", audio_path,
                "-f", "s16le", "-acodec", "pcm_s16le",
                "-ar", "16000", "-ac", "1", output_path,
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=120,
        )
        return output_path
    except Exception:
        if os.path.exists(output_path):
            os.unlink(output_path)
        raise


async def _transcribe_via_realtime(audio_path: str, language: str) -> dict:
    """Replay a file through the same WebSocket and voiceprint path as live audio."""
    if websockets is None:
        raise RuntimeError("当前 ASR 镜像未安装 websockets")
    pcm_path = _convert_to_pcm(audio_path)
    raw_results = {}
    speaker_results = {}
    arrival_order = []
    chunk_bytes = max(320, int(16000 * 2 * OFFLINE_REALTIME_CHUNK_MS / 1000))

    async def receive_results(websocket):
        while True:
            try:
                message = await asyncio.wait_for(
                    websocket.recv(),
                    timeout=OFFLINE_REALTIME_FINAL_TIMEOUT,
                )
            except asyncio.TimeoutError:
                break
            except websockets.exceptions.ConnectionClosed:
                break
            if not isinstance(message, str):
                continue
            try:
                data = json.loads(message)
            except json.JSONDecodeError:
                continue
            if not isinstance(data, dict):
                continue

            speaker_ref = data.get("speaker_ref")
            if data.get("mode") == "2pass-offline":
                text = (data.get("text") or "").strip()
                if text and speaker_ref:
                    raw_results[speaker_ref] = text
                    if speaker_ref not in arrival_order:
                        arrival_order.append(speaker_ref)
            if (
                data.get("type") == "asr_with_speaker"
                and data.get("source") in {"voiceprint", "acoustic"}
                and speaker_ref
            ):
                text = (data.get("text") or "").strip()
                speaker = (data.get("speaker") or "").strip()
                if text and speaker:
                    # Voiceprint is the same role source used by live recording
                    # and takes precedence over an acoustic anonymous label.
                    previous = speaker_results.get(speaker_ref)
                    if previous is None or data.get("source") == "voiceprint":
                        speaker_results[speaker_ref] = {
                            "text": text,
                            "speaker": speaker,
                            "source": data.get("source"),
                        }

    try:
        logger.info(
            f"离线文件开始复用实时转录链路: {OFFLINE_REALTIME_WS_URL}"
        )
        async with websockets.connect(
            OFFLINE_REALTIME_WS_URL,
            ping_interval=None,
            max_size=None,
            open_timeout=15,
        ) as websocket:
            receiver = asyncio.create_task(receive_results(websocket))
            with open(pcm_path, "rb") as pcm_file:
                while True:
                    chunk = pcm_file.read(chunk_bytes)
                    if not chunk:
                        break
                    await websocket.send(chunk)
                    # The live voiceprint matcher uses a rolling PCM buffer.
                    # Preserve media time so each ASR result sees its own audio.
                    await asyncio.sleep(len(chunk) / (16000 * 2))

            await websocket.send(json.dumps({
                "type": "stop",
                "is_speaking": False,
            }))
            await receiver
    finally:
        if os.path.exists(pcm_path):
            os.unlink(pcm_path)

    segments = []
    for speaker_ref in arrival_order:
        role_result = speaker_results.get(speaker_ref)
        if role_result:
            segments.append({
                "start": None,
                "end": None,
                "text": role_result["text"],
                "speaker": role_result["speaker"],
            })
        else:
            text = raw_results.get(speaker_ref, "")
            if text:
                segments.append({
                    "start": None,
                    "end": None,
                    "text": text,
                    "speaker": "UNKNOWN_SPEAKER",
                })

    if not segments:
        raise RuntimeError("实时链路未返回任何最终转录片段")
    logger.info(
        f"离线文件实时复用完成: segments={len(segments)}, "
        f"speakers={sorted(set(item['speaker'] for item in segments))}"
    )
    return {
        "text": "".join(item["text"] for item in segments),
        "language": language,
        "segments": segments,
    }


async def run_asr(
    audio_path: str,
    language: str = "zh",
    response_format: str = "json",
):
    try:
        hotword = read_txt_file("/app/hotword.txt")
        result = model.generate(input=audio_path, hotword=hotword)

        if result and len(result) > 0:
            text = result[0].get("text", "")
        else:
            text = ""

        text = cn_to_arabic(text)
        logger.info(f"识别结果: {text}")

        if response_format == "text":
            return PlainTextResponse(content=text)

        return {
            "text": text,
            "language": language,
        }

    except Exception as e:
        logger.error(f"识别出错: {e}")
        raise HTTPException(status_code=500, detail=f"识别失败: {str(e)}")


@app.post("/api/v1/audio/transcriptions")
async def transcribe(
    request: Request,
    file: Optional[UploadFile] = File(None),
    language: Optional[str] = Form("zh"),
    response_format: Optional[str] = Form("json"),
):
    """
    语音转文字接口，兼容两种方式：

    1. multipart/form-data 上传文件：
       - file: 音频文件
       - language: 默认 zh
       - response_format: json/text

    2. application/json 传服务器本地文件路径：
       {
         "file_path": "/app/audio/test.wav",
         "language": "zh",
         "response_format": "json"
       }
    """

    tmp_path = None
    should_delete = False

    try:
        content_type = request.headers.get("content-type", "").lower()

        if "application/json" in content_type:
            body = await request.json()

            file_path = body.get("file_path")
            language = body.get("language", language)
            response_format = body.get("response_format", response_format)

            if not file_path:
                raise HTTPException(status_code=400, detail="缺少 file_path")

            if not os.path.exists(file_path):
                raise HTTPException(status_code=400, detail=f"文件不存在: {file_path}")

            if not os.path.isfile(file_path):
                raise HTTPException(status_code=400, detail=f"不是有效文件: {file_path}")

            audio_path = file_path
            logger.info(f"处理服务器本地文件: {audio_path}")

        else:
            if file is None:
                raise HTTPException(status_code=400, detail="缺少上传文件 file")

            suffix = os.path.splitext(file.filename or "")[1] or ".wav"

            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
                content = await file.read()

                if not content:
                    raise HTTPException(status_code=400, detail="上传文件为空")

                tmp.write(content)
                tmp_path = tmp.name
                should_delete = True

            audio_path = tmp_path
            logger.info(
                f"处理上传文件: {file.filename}, 大小: {len(content)} bytes, 临时路径: {audio_path}"
            )

        return await run_asr(
            audio_path=audio_path,
            language=language or "zh",
            response_format=response_format or "json",
        )

    finally:
        if should_delete and tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)


@app.post("/v1/audio/transcriptions")
async def transcribe_v1(
    request: Request,
    file: Optional[UploadFile] = File(None),
    language: Optional[str] = Form("zh"),
    response_format: Optional[str] = Form("json"),
):
    return await transcribe(
        request=request,
        file=file,
        language=language,
        response_format=response_format,
    )


@app.post("/asr")
async def transcribe_speakr_compatible(
    audio_file: UploadFile = File(...),
    language: Optional[str] = Form("zh"),
    database_name: Optional[str] = Form("NB"),
    collection_name: Optional[str] = Form("voice"),
    spk_diarization: bool = Query(False),
    spk_num: int = Query(0, ge=0),
    spk_threshold: float = Query(0.6, ge=0.0, le=1.0),
):
    """
    Speakr 兼容接口：
    - 上传字段：audio_file
    - 返回字段：segments，其中每个 segment 含 text/speaker/start/end
    """

    tmp_path = None

    try:
        suffix = os.path.splitext(audio_file.filename or "")[1] or ".wav"

        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            content = await audio_file.read()

            if not content:
                raise HTTPException(status_code=400, detail="上传文件为空")

            tmp.write(content)
            tmp_path = tmp.name

        logger.info(
            f"处理 Speakr /asr 上传文件: {audio_file.filename}, 大小: {len(content)} bytes, 临时路径: {tmp_path}"
        )

        if spk_diarization:
            try:
                return await _transcribe_via_realtime(
                    tmp_path, language or "zh"
                )
            except Exception as e:
                logger.exception(
                    f"实时链路复用失败，回退到离线VAD/聚类方案: {e}"
                )

            try:
                diarized_segments = _request_diarization(
                    tmp_path, spk_num, spk_threshold
                )
                diarized_segments = _resolve_diarized_speakers(
                    diarized_segments,
                    database_name or "NB",
                    collection_name or "voice",
                )
                diarized_segments = _merge_neighboring_segments(
                    diarized_segments
                )
                output_segments = []
                for item in diarized_segments:
                    start = float(item["start"])
                    end = float(item["end"])
                    if end - start < 0.3:
                        continue

                    segment_path = _extract_audio_segment(tmp_path, start, end)
                    try:
                        segment_result = await run_asr(
                            audio_path=segment_path,
                            language=language or "zh",
                            response_format="json",
                        )
                        text = segment_result.get("text", "").strip()
                        if not text:
                            continue

                        output_segments.append({
                            "start": start,
                            "end": end,
                            "text": text,
                            "speaker": item.get("speaker") or "SPEAKER_00",
                        })
                    finally:
                        if os.path.exists(segment_path):
                            os.unlink(segment_path)

                if output_segments:
                    return {
                        "text": "".join(
                            segment["text"] for segment in output_segments
                        ),
                        "language": language or "zh",
                        "segments": output_segments,
                    }
                logger.warning("说话人分离未产生有效文本，回退到整段识别")
            except Exception as e:
                logger.exception(f"说话人分离失败，回退到整段识别: {e}")

        result = await run_asr(
            audio_path=tmp_path,
            language=language or "zh",
            response_format="json",
        )
        speaker_name = "SPEAKER_00"
        try:
            matched_name, score = _match_voiceprint(
                tmp_path,
                audio_file.filename or "audio.wav",
                database_name or "NB",
                collection_name or "voice",
            )
            if matched_name:
                speaker_name = matched_name
        except Exception as e:
            logger.warning(f"声纹匹配失败，使用默认说话人: {e}")

        return make_speakr_response(
            text=result.get("text", "") if isinstance(result, dict) else "",
            language=result.get("language", language or "zh") if isinstance(result, dict) else language or "zh",
            speaker=speaker_name,
        )

    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)


@app.get("/get_voice_name")
async def get_voice_name(
    database_name: str = "NB",
    collection_name: str = "voice",
):
    store = _load_voiceprints()
    records = store.get(_voiceprint_key(database_name, collection_name), {})
    return {
        "code": 0,
        "success": True,
        "message": "ok",
        "data": [
            {"name": name, "voice_name": name}
            for name in sorted(records.keys())
        ],
    }


@app.post("/voice_insert")
async def voice_insert(
    voice_name: str = Form(...),
    database_name: str = Form("NB"),
    collection_name: str = Form("voice"),
    voice_file: UploadFile = File(...),
):
    tmp_path = None

    try:
        suffix = os.path.splitext(voice_file.filename or "")[1] or ".wav"
        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            content = await voice_file.read()
            if not content:
                raise HTTPException(status_code=400, detail="上传声纹文件为空")
            tmp.write(content)
            tmp_path = tmp.name

        embedding = _extract_embedding(tmp_path, voice_file.filename or f"{voice_name}.wav")
        store = _load_voiceprints()
        key = _voiceprint_key(database_name, collection_name)
        records = store.setdefault(key, {})
        records[voice_name] = {
            "name": voice_name,
            "database_name": database_name,
            "collection_name": collection_name,
            "embedding": embedding,
        }
        _save_voiceprints(store)

        return {
            "code": 0,
            "success": True,
            "message": "voice inserted",
            "data": {"name": voice_name, "embedding_size": len(embedding)},
        }
    except requests.RequestException as e:
        logger.error(f"调用 voice-dected 失败: {e}")
        raise HTTPException(status_code=502, detail=f"voice-dected 调用失败: {e}")
    except Exception as e:
        logger.error(f"写入声纹失败: {e}")
        raise HTTPException(status_code=500, detail=f"写入声纹失败: {e}")
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)


@app.delete("/delete_voice")
async def delete_voice(
    name: str,
    database_name: str = "NB",
    collection_name: str = "voice",
):
    store = _load_voiceprints()
    key = _voiceprint_key(database_name, collection_name)
    records = store.setdefault(key, {})
    existed = name in records
    records.pop(name, None)
    _save_voiceprints(store)

    return {
        "code": 0,
        "success": True,
        "message": "voice deleted" if existed else "voice not found",
        "data": {"name": name, "deleted": existed},
    }


if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=59805,
        reload=False,
        workers=1,
        log_level="info",
    )