"""Persist user/agent audio by device, scene mode, and time for later replay."""
|
|
from __future__ import annotations
|
|
import os
|
import time
|
from typing import Any, Optional
|
|
from config.config_loader import get_project_dir
|
from config.logger import setup_logging
|
|
TAG = __name__
|
logger = setup_logging()
|
|
MODE_DIR = {
|
"CHAT": "chat",
|
"SHADOWING": "shadowing",
|
"PLAY_AUDIO": "repeater",
|
"TRANSLATION": "translation",
|
"BILINGUAL": "bilingual",
|
}
|
|
|
def _safe_segment(value: Any, fallback: str = "unknown") -> str:
|
raw = str(value or "").strip() or fallback
|
out = []
|
for ch in raw:
|
if ch.isalnum() or ch in "-_":
|
out.append(ch)
|
elif ch in (":", " ", "."):
|
out.append("-")
|
else:
|
out.append("_")
|
text = "".join(out).strip("-_") or fallback
|
return text[:80]
|
|
|
def _mode_dir(conn: Any) -> str:
|
mode = str(getattr(conn, "interaction_mode", "") or "").strip().upper()
|
return MODE_DIR.get(mode, _safe_segment(mode.lower() or "unknown"))
|
|
|
def _archive_cfg(conn: Any) -> dict:
|
return dict(((getattr(conn, "config", None) or {}).get("audio_archive") or {}))
|
|
|
def archive_enabled(conn: Any = None) -> bool:
|
cfg = _archive_cfg(conn)
|
if cfg.get("enabled") is False:
|
return False
|
return True
|
|
|
def archive_root(conn: Any = None) -> str:
|
cfg = _archive_cfg(conn)
|
rel = str(cfg.get("dir") or "data/recordings")
|
if os.path.isabs(rel):
|
return rel
|
return os.path.join(get_project_dir(), rel)
|
|
|
def archive_wav(
|
conn: Any,
|
wav_bytes: bytes,
|
*,
|
kind: str = "asr",
|
extra: str = "",
|
) -> Optional[str]:
|
"""Write WAV under recordings/{device}/{mode}/{date}/{time}_{kind}_{extra}.wav."""
|
if not wav_bytes or not archive_enabled(conn):
|
return None
|
try:
|
now = time.time()
|
local = time.localtime(now)
|
day = time.strftime("%Y-%m-%d", local)
|
stamp = time.strftime("%H%M%S", local) + f"{int((now % 1) * 1000):03d}"
|
device = _safe_segment(getattr(conn, "device_id", "") or "", "unknown")
|
mode = _mode_dir(conn)
|
parts = [stamp, _safe_segment(kind, "asr")]
|
extra_part = _safe_segment(extra, "") if extra else ""
|
if extra_part:
|
parts.append(extra_part)
|
filename = "_".join(parts) + ".wav"
|
directory = os.path.join(archive_root(conn), device, mode, day)
|
os.makedirs(directory, exist_ok=True)
|
path = os.path.join(directory, filename)
|
with open(path, "wb") as handle:
|
handle.write(wav_bytes)
|
relative = os.path.join(device, mode, day, filename).replace("\\", "/")
|
logger.bind(tag=TAG).info(f"音频归档 {relative} bytes={len(wav_bytes)}")
|
return relative
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"音频归档失败: {exc}")
|
return None
|
|
|
def archive_pcm_snapshot(conn: Any, snapshot: Any, *, extra: str = "") -> 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)
|
