5种场景模式家长端ok

wangzhibo

10 天以前 220e4fab27b7f419b3e89e2f2d28ae789aac09d7

 xiaozhi-esp32-server/main/xiaozhi-server/core/audio_archive.py

@@ -2,6 +2,7 @@

from __future__ import annotations

import json
import os
import time
from typing import Any, Optional
@@ -59,14 +60,63 @@
    return os.path.join(get_project_dir(), rel)


def plain_transcript(text: Any) -> str:
    """ASR/TTS 归档文本。FunASR 可能是 {content, speaker, ...} JSON。"""
    if text is None:
        return ""
    raw = str(text)
    stripped = raw.strip()
    if stripped.startswith("{") and stripped.endswith("}"):
        try:
            data = json.loads(stripped)
            if isinstance(data, dict) and "content" in data:
                return str(data.get("content") or "")
        except Exception:
            pass
    return raw


def write_sidecar_text(wav_path: str, text: Optional[str]) -> Optional[str]:
    """Write UTF-8 .txt next to a wav. Same stem, suffix .txt. None skips."""
    if text is None or not wav_path:
        return None
    sidecar = os.path.splitext(wav_path)[0] + ".txt"
    try:
        body = plain_transcript(text)
        with open(sidecar, "w", encoding="utf-8") as handle:
            handle.write(body)
            if body and not body.endswith("\n"):
                handle.write("\n")
        return sidecar
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"文本归档失败: {exc}")
        return None


def write_sidecar_for_relative(conn: Any, relative: str, text: Optional[str]) -> Optional[str]:
    """Update the .txt next to an already archived wav (relative audio_id)."""
    if not relative or text is None:
        return None
    rel = str(relative).replace("\\", "/").lstrip("/")
    if "/" in rel:
        path = os.path.join(archive_root(conn), *rel.split("/"))
    else:
        path = os.path.join(get_project_dir(), "data", "task_sessions", "audio", rel)
    return write_sidecar_text(path, text)


def archive_wav(
    conn: Any,
    wav_bytes: bytes,
    *,
    kind: str = "asr",
    extra: str = "",
    text: Optional[str] = None,
) -> Optional[str]:
    """Write WAV under recordings/{device}/{mode}/{date}/{time}_{kind}_{extra}.wav."""
    """Write WAV under recordings/{device}/{mode}/{date}/{time}_{kind}_{extra}.wav.

    If text is not None, also write the same stem with .txt for later analysis.
    """
    if not wav_bytes or not archive_enabled(conn):
        return None
    try:
@@ -86,6 +136,8 @@
        path = os.path.join(directory, filename)
        with open(path, "wb") as handle:
            handle.write(wav_bytes)
        if text is not None:
            write_sidecar_text(path, text)
        relative = os.path.join(device, mode, day, filename).replace("\\", "/")
        logger.bind(tag=TAG).info(f"音频归档 {relative} bytes={len(wav_bytes)}")
        return relative
@@ -94,10 +146,16 @@
        return None


def archive_pcm_snapshot(conn: Any, snapshot: Any, *, extra: str = "") -> Optional[str]:
def archive_pcm_snapshot(
    conn: Any,
    snapshot: Any,
    *,
    extra: str = "",
    text: Optional[str] = None,
) -> Optional[str]:
    from core.app_runtime.assessment import pcm_snapshot_to_wav

    wav_bytes = pcm_snapshot_to_wav(snapshot)
    if not wav_bytes:
        return None
    return archive_wav(conn, wav_bytes, kind="asr", extra=extra)
    return archive_wav(conn, wav_bytes, kind="asr", extra=extra, text=text)