"""Persist user/agent audio by device, scene mode, and time for later replay."""
|
|
from __future__ import annotations
|
|
import json
|
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 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.
|
|
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:
|
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)
|
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
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"音频归档失败: {exc}")
|
return None
|
|
|
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, text=text)
|
