"""Unified start_app_task / switch_app / mode helpers.""" from __future__ import annotations import asyncio import json import difflib import queue import random import re import uuid from typing import Any, Dict, List, Optional, TYPE_CHECKING from config.logger import setup_logging from plugins_func.register import Action, ActionResponse from .assessment import ( evaluate_turn, listen_window_ms, pcm_snapshot_to_wav, resolve_expected_text, resolve_listen_ratio, save_turn_wav, silent_pcm_snapshot, transcript_cache_dir, ) from .playback import ( clamp_play_speed, default_repeater_playlist, default_shadowing_playlist, expand_playlist_plays, lesson_day_index, load_last_lesson_params, mark_lesson_played_today, playlist_from_params, remember_last_lesson_params, resolve_play_speed, speeds_for_repeats, ) from .registry import ( default_app_id, get_app, get_registry, get_task, match_registry_voice, voice_menu_text, ) from .session import begin_session, current_session, end_session, parse_skill_event, schedule_close_parent_except if TYPE_CHECKING: from core.connection import ConnectionHandler TAG = __name__ logger = setup_logging() SOURCE_PRIORITY = { "mqtt_immediate": 3, "schedule": 2, "voice": 1, "default_chat": 0, } TRANSLATOR_PROMPT = ( "你是一台翻译机。只输出译文，不要解释、不要拼音课、不要追问、不要闲聊。" "未指定语言时：中文译成英文，英文译成中文；其它语言默认译成中文。" ) BILINGUAL_PROMPT = ( "你正在进行中英对照对话练习，不是翻译机，不要把用户的话原样译回去交差。" "严格执行：用户说中文时你只用英文完整回答；用户说英文时你只用中文完整回答。" "不要中英夹杂，不要同时给出两种语言，不要问要不要翻译。" "用户说退出中英对话、回到聊天时调用 switch_app，app_id=chat。" ) BILINGUAL_PAIR_PROMPTS = { "zh-en": BILINGUAL_PROMPT, "zh-fr": ( "你正在进行中法对照对话练习，不是翻译机，不要把用户的话原样译回去交差。" "用户说中文时你用法语完整回答；用户说法语时你用中文完整回答。" "不要两种语言同时输出。用户说退出对话、回到聊天时调用 switch_app，app_id=chat。" ), "zh-ja": ( "你正在进行中日对照对话练习，不是翻译机，不要把用户的话原样译回去交差。" "用户说中文时你用日语完整回答；用户说日语时你用中文完整回答。" "不要两种语言同时输出。用户说退出对话、回到聊天时调用 switch_app，app_id=chat。" ), } TRANSLATOR_PAIR_PROMPTS = { "zh-en": TRANSLATOR_PROMPT, "zh-fr": "你是一台翻译机。只输出译文。中文译成法语，法语译成中文；其它语言默认译成中文。", "zh-ja": "你是一台翻译机。只输出译文。中文译成日语，日语译成中文；其它语言默认译成中文。", } CHAT_PROACTIVE_HINT = ( "你是用户的语音伙伴。进入对话后可以先主动短句邀请聊天，例如「我们来聊天吧」。" "用户开口后按正常对话回答，不要反复催促。" ) CHAT_RETURN_TOPICS = ("今天过得怎么样", "你最近在忙什么", "有什么新鲜事") DEFAULT_SHADOWING_SENTENCES = ( "Hello, how are you today?", "I like to read English books.", "The weather is nice this morning.", "Can you say that again, please?", "Practice a little every day.", ) DEVICE_PLAY_TOOL = "self_mp3_play" DEVICE_STOP_TOOL = "self_mp3_stop" DEVICE_LIST_TOOL = "self_mp3_get_list" DEVICE_ENSURE_APP_TOOL = "self_sd_ensure_app" SD_LAYOUT_MODES = ("PLAY_AUDIO", "SHADOWING") SHADOWING_LISTEN_RATIO = 1.2 SHADOWING_LISTEN_MIN_MS = 800 REPEATER_END_PHRASE = "泛听结束" SHADOWING_END_PHRASE = "跟读结束" REPEATER_START_PHRASE = "开始泛听任务" SHADOWING_START_PHRASE = "开始跟读任务" CHAT_START_PHRASE = "开始对话任务" TRANSLATION_START_PHRASE = "开始翻译任务" BILINGUAL_START_PHRASE = "开始中英对话任务" def source_priority(source: str) -> int: return SOURCE_PRIORITY.get(source or "voice", 1) def _ensure_runtime_fields(conn: "ConnectionHandler") -> None: if not hasattr(conn, "active_app_id"): conn.active_app_id = default_app_id() if not hasattr(conn, "current_task_id"): conn.current_task_id = "conversation" if not hasattr(conn, "interaction_mode"): conn.interaction_mode = "CHAT" if not hasattr(conn, "current_source"): conn.current_source = "default_chat" if not hasattr(conn, "tool_policy"): conn.tool_policy = "inherit_agent" if not hasattr(conn, "tool_allowlist"): conn.tool_allowlist = [] if not hasattr(conn, "original_system_prompt"): conn.original_system_prompt = conn.prompt if not hasattr(conn, "shadowing_index"): conn.shadowing_index = 0 if not hasattr(conn, "shadowing_prompt"): conn.shadowing_prompt = "" if not hasattr(conn, "shadowing_clips"): conn.shadowing_clips = [] if not hasattr(conn, "shadowing_awaiting_user"): conn.shadowing_awaiting_user = False if not hasattr(conn, "shadowing_clip_gen"): conn.shadowing_clip_gen = 0 if not hasattr(conn, "shadowing_timeout_task"): conn.shadowing_timeout_task = None if not hasattr(conn, "last_asr_language"): conn.last_asr_language = None if not hasattr(conn, "task_session"): conn.task_session = None if not hasattr(conn, "last_asr_audio_snapshot"): conn.last_asr_audio_snapshot = None if not hasattr(conn, "shadowing_listen_pcm"): conn.shadowing_listen_pcm = [] if not hasattr(conn, "shadowing_audio_saved_index"): conn.shadowing_audio_saved_index = -1 if not hasattr(conn, "ensured_app_dirs"): conn.ensured_app_dirs = set() if not hasattr(conn, "repeater_speed_queue"): conn.repeater_speed_queue = [] if not hasattr(conn, "repeater_play_name"): conn.repeater_play_name = "" if not hasattr(conn, "repeater_power_off_after"): conn.repeater_power_off_after = False if not hasattr(conn, "repeater_playlist"): conn.repeater_playlist = [] if not hasattr(conn, "shadowing_last_speed"): conn.shadowing_last_speed = 1.0 if not hasattr(conn, "shadowing_assess_index"): conn.shadowing_assess_index = 0 if not hasattr(conn, "chat_greeted"): conn.chat_greeted = False if not hasattr(conn, "pending_play_after_tts"): conn.pending_play_after_tts = None def _task_needs_sd_layout(task: Optional[Dict[str, Any]]) -> bool: if not task: return False mode = task.get("interaction_mode") or "" if mode in SD_LAYOUT_MODES: return True for skill in task.get("skills") or []: name = (skill or {}).get("name") or "" if name in ("self.mp3.play", DEVICE_PLAY_TOOL): return True return False def app_sd_layout_paths(app_id: str, app: Optional[Dict[str, Any]] = None) -> list: """Return SD lesson dirs for an App: /{app_id}/{task_id}/lessons.""" if not app_id: return [] if app is None: app = get_app(app_id) if not app: return [] paths = [] for task_id, task in (app.get("tasks") or {}).items(): if _task_needs_sd_layout(task): paths.append(f"/{app_id}/{task_id}/lessons") return paths def _mcp_layout_ok(result) -> bool: if result is None: return False if result.action in (Action.ERROR, Action.NOTFOUND): return False blob = str(getattr(result, "response", None) or getattr(result, "result", None) or "") compact = blob.replace(" ", "").lower() if '"ok":false' in compact: return False return True async def ensure_app_sd_dirs(conn: "ConnectionHandler", app_id: str) -> bool: """Create /{app} and /{app}/{task}/lessons on device SD. Never raises; missing SD is ignored.""" _ensure_runtime_fields(conn) app = get_app(app_id) paths = app_sd_layout_paths(app_id, app) if not paths: return True already = getattr(conn, "ensured_app_dirs", None) if already is None: conn.ensured_app_dirs = set() already = conn.ensured_app_dirs if app_id in already: return True try: root = await _execute_named_tool(conn, DEVICE_ENSURE_APP_TOOL, {"app_id": app_id, "task_id": ""}) if not _mcp_layout_ok(root): logger.bind(tag=TAG).warning(f"创建设备 App 目录失败 {app_id}: {getattr(root, 'response', root)}") return False for task_id, task in (app.get("tasks") or {}).items(): if not _task_needs_sd_layout(task): continue result = await _execute_named_tool( conn, DEVICE_ENSURE_APP_TOOL, {"app_id": app_id, "task_id": task_id} ) if not _mcp_layout_ok(result): logger.bind(tag=TAG).warning( f"创建设备 Task 目录失败 {app_id}/{task_id}: {getattr(result, 'response', result)}" ) return False already.add(app_id) logger.bind(tag=TAG).info(f"已确保 SD 目录 {paths}") return True except Exception as e: logger.bind(tag=TAG).warning(f"创建设备 App 目录异常 {app_id}: {e}") return False async def ensure_all_subscribed_app_sd_dirs(conn: "ConnectionHandler") -> None: """After MCP tools/list, mkdir for every registry App that needs SD. Failures do not drop WS.""" try: apps = (get_registry().get("apps") or {}) for app_id, app in apps.items(): if app_sd_layout_paths(app_id, app): await ensure_app_sd_dirs(conn, app_id) except Exception as e: logger.bind(tag=TAG).warning(f"订阅 App 建目录失败（忽略）: {e}") def schedule_ensure_subscribed_app_sd_dirs(conn: "ConnectionHandler") -> None: """Fire-and-forget so tools/list handling is not blocked waiting for MCP mkdir.""" async def _run(): await ensure_all_subscribed_app_sd_dirs(conn) try: loop = getattr(conn, "loop", None) if loop: loop.create_task(_run()) else: asyncio.create_task(_run()) except Exception as e: logger.bind(tag=TAG).warning(f"无法调度 App 建目录: {e}") def filter_tools(conn: "ConnectionHandler", tools: Dict[str, Any]) -> Dict[str, Any]: """inherit_agent keeps the live agent toolset; allowlist keeps named tools only.""" _ensure_runtime_fields(conn) if getattr(conn, "tool_policy", "inherit_agent") != "allowlist": return tools allow = set(getattr(conn, "tool_allowlist", None) or []) if not allow: return {} filtered = {} for name, definition in tools.items(): if name in allow: filtered[name] = definition return filtered def _refresh_tools(conn: "ConnectionHandler") -> None: handler = getattr(conn, "func_handler", None) if handler and hasattr(handler, "upload_functions_desc"): handler.upload_functions_desc() def _normalize_language_pair(raw: Any) -> str: pair = str(raw or "zh-en").strip().lower().replace("_", "-") if pair in ("cn-en", "zh-cn-en", "zhcn-en"): return "zh-en" return pair or "zh-en" def _apply_prompt(conn: "ConnectionHandler", app: Dict[str, Any], params: Optional[Dict[str, Any]] = None) -> None: mode = app.get("system_prompt_mode") or "none" if conn.original_system_prompt is None: conn.original_system_prompt = conn.prompt pair = _normalize_language_pair((params or {}).get("language_pair")) conn.language_pair = pair conn.language_pair_fallback = pair not in BILINGUAL_PAIR_PROMPTS and pair not in TRANSLATOR_PAIR_PROMPTS prompt_pair = "zh-en" if conn.language_pair_fallback else pair if mode == "translator": pair_prompt = TRANSLATOR_PAIR_PROMPTS.get(prompt_pair, TRANSLATOR_PROMPT) conn.change_system_prompt( pair_prompt + "用户说话后应调用 translate；用户说退出翻译、回到聊天时调用 switch_app，app_id=chat。" "不要讲笑话、不要查天气、不要闲聊。" ) elif mode == "bilingual": conn.change_system_prompt(BILINGUAL_PAIR_PROMPTS.get(prompt_pair, BILINGUAL_PROMPT)) elif mode == "agent_proactive": base = conn.original_system_prompt or conn.prompt or "" if CHAT_PROACTIVE_HINT not in (base or ""): conn.change_system_prompt((base + "\n" + CHAT_PROACTIVE_HINT).strip()) else: conn.change_system_prompt(base) else: if conn.original_system_prompt: conn.change_system_prompt(conn.original_system_prompt) async def _execute_named_tool(conn: "ConnectionHandler", tool_name: str, arguments: Dict[str, Any]): handler = getattr(conn, "func_handler", None) if not handler or not hasattr(handler, "tool_manager"): return ActionResponse(action=Action.ERROR, response="工具处理器未就绪") try: return await handler.tool_manager.execute_tool(tool_name, arguments or {}) except Exception as e: logger.bind(tag=TAG).error(f"执行工具 {tool_name} 失败: {e}") return ActionResponse(action=Action.ERROR, response=str(e)) async def _stop_current_skills(conn: "ConnectionHandler") -> None: mode = getattr(conn, "interaction_mode", "CHAT") _cancel_shadowing_timeout(conn) if mode in ("PLAY_AUDIO", "SHADOWING"): await _execute_named_tool(conn, DEVICE_STOP_TOOL, {}) conn.awaiting_voice_app_choice = False conn.pending_play_after_tts = None conn.shadowing_index = 0 conn.shadowing_prompt = "" conn.shadowing_clips = [] conn.shadowing_awaiting_user = False conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1 conn.shadowing_assess_index = 0 conn.shadowing_listen_pcm = [] conn.shadowing_audio_saved_index = -1 conn.repeater_speed_queue = [] conn.repeater_play_name = "" conn.repeater_power_off_after = False conn.repeater_playlist = [] def tts_channels_ready(conn: "ConnectionHandler") -> bool: tts = getattr(conn, "tts", None) if tts is None: return False play = getattr(tts, "audio_play_priority_thread", None) text_th = getattr(tts, "tts_priority_thread", None) return bool( play is not None and play.is_alive() and text_th is not None and text_th.is_alive() ) async def wait_for_tts_channels(conn: "ConnectionHandler", timeout: float = 8.0) -> bool: loop = asyncio.get_running_loop() start = loop.time() while (loop.time() - start) < timeout: if tts_channels_ready(conn): return True await asyncio.sleep(0.1) return tts_channels_ready(conn) def _speak(conn: "ConnectionHandler", text: str) -> None: if not text: return tts = getattr(conn, "tts", None) if tts is None: logger.bind(tag=TAG).warning("TTS未初始化，跳过播报") return if not tts_channels_ready(conn): logger.bind(tag=TAG).warning("TTS通道未就绪，跳过播报") return from core.handle.intentHandler import speak_txt if not getattr(conn, "sentence_id", None): conn.sentence_id = uuid.uuid4().hex try: speak_txt(conn, text) except Exception as e: logger.bind(tag=TAG).warning(f"主动播报失败: {e}") async def _play_bind_prompt(conn: "ConnectionHandler") -> None: from core.handle.receiveAudioHandle import check_bind_device await check_bind_device(conn) async def prompt_bind_if_needed(conn: "ConnectionHandler") -> bool: """未绑定设备时播 6 位绑定码，并阻止欢迎词/进场词。已绑定则返回 False。""" event = getattr(conn, "bind_completed_event", None) if event is not None and not event.is_set(): try: await asyncio.wait_for(event.wait(), timeout=8) except asyncio.TimeoutError: pass if not getattr(conn, "need_bind", False): return False if not await wait_for_tts_channels(conn, timeout=8): logger.bind(tag=TAG).warning("TTS通道未就绪，仍尝试播放绑定码") await _play_bind_prompt(conn) return True def _greeting_text(conn: "ConnectionHandler") -> str: return (conn.config or {}).get("proactive_chat_greeting") or "我们来聊天吧。" def _display_name(app: Optional[Dict[str, Any]], task: Optional[Dict[str, Any]] = None) -> str: if task and task.get("display_name"): return str(task["display_name"]) if app and app.get("display_name"): return str(app["display_name"]) if app and app.get("name"): return str(app["name"]) return "当前模式" def _enter_prompt(app: Dict[str, Any], task: Dict[str, Any]) -> str: return (task.get("enter_prompt") or app.get("enter_prompt") or "").strip() def _switch_announcement( prev_app: Optional[Dict[str, Any]], prev_task: Optional[Dict[str, Any]], app: Dict[str, Any], task: Dict[str, Any], app_id: str, ) -> str: if app_id == "chat": prev_name = _display_name(prev_app, prev_task) topic = random.choice(CHAT_RETURN_TOPICS) return f"已经退出{prev_name}，现在是对话模式，我们来聊点{topic}吧。" return _enter_prompt(app, task) or f"现在已经切换至{_display_name(app, task)}。" def _shadowing_sentences(conn: "ConnectionHandler") -> list: cfg = (conn.config or {}).get("app_runtime") or {} sentences = cfg.get("shadowing_sentences") if not sentences: _, task = get_task("english_learning", "shadowing") sentences = (task or {}).get("sentences") if not sentences: sentences = DEFAULT_SHADOWING_SENTENCES return [str(s).strip() for s in sentences if str(s).strip()] def lesson_mp3_clips(names, app_id: str, task_id: str, lesson_id: str = "") -> list: """Pick sentence MP3s under /{app}/{task}/lessons/{lesson}/...""" prefix = f"/{app_id}/{task_id}/lessons/" clips = [] for raw in names or []: name = str(raw or "").replace("\\", "/") if not name.lower().endswith(".mp3"): continue if not name.startswith("/"): name = "/" + name.lstrip("/") if not name.startswith(prefix): continue rest = name[len(prefix) :] if not rest: continue base = rest.rsplit("/", 1)[-1].lower() if base == "full.mp3" or base.endswith("_full.mp3"): continue if lesson_id: if rest != lesson_id and not rest.startswith(str(lesson_id) + "/"): continue clips.append(name) clips.sort() if lesson_id or not clips: return clips first_lesson = clips[0][len(prefix) :].split("/", 1)[0] return [ name for name in clips if name[len(prefix) :] == first_lesson or name[len(prefix) :].startswith(first_lesson + "/") ] def repeater_lesson_mp3(names, app_id: str, task_id: str) -> str: """Pick the full-lesson MP3 under /{app}/{task}/lessons/, else the first file there.""" prefix = f"/{app_id}/{task_id}/lessons/" fulls = [] others = [] for raw in names or []: name = str(raw or "").replace("\\", "/") if not name.lower().endswith(".mp3"): continue if not name.startswith("/"): name = "/" + name.lstrip("/") if not name.startswith(prefix): continue base = name.rsplit("/", 1)[-1].lower() if base == "full.mp3" or base.endswith("_full.mp3"): fulls.append(name) else: others.append(name) fulls.sort() others.sort() if fulls: return fulls[0] return others[0] if others else "" def _is_clip_path(text: str) -> bool: t = (text or "").replace("\\", "/").lower() return t.endswith(".mp3") or t.startswith("/") def _shadowing_items(conn: "ConnectionHandler") -> list: clips = getattr(conn, "shadowing_clips", None) or [] if clips: return list(clips) return _shadowing_sentences(conn) def _begin_shadowing(conn: "ConnectionHandler", enter_text: str) -> str: sentences = _shadowing_sentences(conn) conn.shadowing_index = 0 if not sentences: conn.shadowing_prompt = "" return enter_text or "现在已经切换至跟读模式，但还没有跟读句子。" conn.shadowing_prompt = sentences[0] prefix = enter_text or "现在已经切换至跟读模式。" return f"{prefix}请跟我读：{sentences[0]}" def _cancel_shadowing_timeout(conn: "ConnectionHandler") -> None: task = getattr(conn, "shadowing_timeout_task", None) if task is not None: try: if not task.done(): task.cancel() except Exception: pass conn.shadowing_timeout_task = None def _task_for(conn: "ConnectionHandler") -> Optional[Dict[str, Any]]: session = current_session(conn) or getattr(conn, "task_session", None) app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", None) or "english_learning" task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", None) or "shadowing" _, task = get_task(app_id, task_id) return task def _expected_for_index(conn: "ConnectionHandler", index: int) -> Optional[str]: session = current_session(conn) or getattr(conn, "task_session", None) app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or "" task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or "" clips = list(getattr(conn, "shadowing_clips", None) or []) return resolve_expected_text( _task_for(conn), clips, index, transcript_cache_dir(app_id, task_id), ) def apply_deferred_assessment( conn: "ConnectionHandler", step_index: int, session: Any = None, clips: Optional[list] = None, app_id: str = "", task_id: str = "", audio_snapshot: Any = None, persist_audio: bool = True, ) -> None: """Write expected_text + metrics onto the captured session. Never raises into play.""" import time as _time session = session or getattr(conn, "task_session", None) if not session: return app_id = app_id or getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or "" task_id = task_id or getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or "" _, task = get_task(app_id or "english_learning", task_id or "shadowing") cfg = (task or {}).get("assessment") or {} clip_list = list(clips if clips is not None else (getattr(conn, "shadowing_clips", None) or [])) turn = session.find_turn(step_index) asr = _plain_user_text((turn or {}).get("asr_text") or "") expected = (turn or {}).get("expected_text") or resolve_expected_text( task, clip_list, step_index, transcript_cache_dir(app_id, task_id) ) snapshot = audio_snapshot if snapshot is None: snapshot = getattr(conn, "last_asr_audio_snapshot", None) wav_bytes = pcm_snapshot_to_wav(snapshot) existing_audio = (turn or {}).get("audio_id") audio_id = existing_audio if persist_audio and not existing_audio: audio_id = save_turn_wav( getattr(session, "task_session_id", "") or "", step_index, wav_bytes, conn=conn, text=asr, ) elif audio_id and asr: try: from core.audio_archive import write_sidecar_for_relative write_sidecar_for_relative(conn, audio_id, asr) except Exception: pass try: from core.handle.reportHandle import enqueue_asr_report enqueue_asr_report( conn, asr or "", snapshot, getattr(session, "task_session_id", None), int(step_index), ) except Exception: pass clip = "" if 0 <= int(step_index) < len(clip_list): clip = str(clip_list[step_index] or "") metrics = [] if not (cfg and cfg.get("enabled") is False): metrics = evaluate_turn( asr, expected or "", wav_bytes, cfg, extra={"clip": clip, "step_index": step_index}, ) session.upsert_turn( index=step_index, prompt_text=(turn or {}).get("prompt_text") or getattr(conn, "shadowing_prompt", "") or "", expected_text=expected, metrics=metrics or None, assessed_at=int(_time.time() * 1000) if metrics else None, audio_id=audio_id, ) session.save() def _schedule_deferred_assessment(conn: "ConnectionHandler", step_index: int) -> None: session = getattr(conn, "task_session", None) clips = list(getattr(conn, "shadowing_clips", None) or []) app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or "" task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or "" audio_snapshot = getattr(conn, "last_asr_audio_snapshot", None) async def _run(): try: apply_deferred_assessment( conn, step_index, session=session, clips=clips, app_id=app_id, task_id=task_id, audio_snapshot=audio_snapshot, persist_audio=False, ) except Exception as exc: logger.bind(tag=TAG).warning(f"跟读后评估失败: {exc}") loop = getattr(conn, "loop", None) try: loop = loop or asyncio.get_running_loop() loop.create_task(_run()) except RuntimeError: try: apply_deferred_assessment( conn, step_index, session=session, clips=clips, app_id=app_id, task_id=task_id, audio_snapshot=audio_snapshot, persist_audio=False, ) except Exception as exc: logger.bind(tag=TAG).warning(f"跟读后评估失败: {exc}") def _schedule_shadowing_listen_timeout(conn: "ConnectionHandler", duration_ms: int) -> None: _cancel_shadowing_timeout(conn) ratio = resolve_listen_ratio(conn, _task_for(conn)) wait_ms = listen_window_ms( duration_ms or 0, ratio, SHADOWING_LISTEN_MIN_MS, speed=getattr(conn, "shadowing_last_speed", None), wall_clock=True, ) conn.shadowing_awaiting_user = True conn.shadowing_assess_index = int(getattr(conn, "shadowing_index", 0) or 0) conn.shadowing_audio_saved_index = -1 conn.shadowing_listen_pcm = [] gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) logger.bind(tag=TAG).info( f"跟读听窗 {wait_ms}ms = 实际播放 {duration_ms}ms × {ratio}" f"（倍速已含在实际播放时长里） gen={gen}" ) async def _fire(): try: await asyncio.sleep(wait_ms / 1000.0) except asyncio.CancelledError: return if int(getattr(conn, "shadowing_clip_gen", 0) or 0) != gen: return if getattr(conn, "interaction_mode", "") != "SHADOWING": return if not getattr(conn, "shadowing_awaiting_user", False): return await _on_shadowing_listen_window_closed(conn) loop = getattr(conn, "loop", None) if loop: conn.shadowing_timeout_task = loop.create_task(_fire()) else: conn.shadowing_timeout_task = asyncio.create_task(_fire()) async def close_shadowing_listen_window(conn: "ConnectionHandler") -> None: """Device listen stop: archive this step once, cancel the clock, advance index.""" _cancel_shadowing_timeout(conn) await _on_shadowing_listen_window_closed(conn) def _drain_asr_queue_into_shadowing_pcm(conn: "ConnectionHandler") -> int: """Binary opus is queued; listen stop/timeout run on the WS loop first. Take pending PCM now.""" q = getattr(conn, "asr_audio_queue", None) if q is None or not getattr(conn, "shadowing_awaiting_user", False): return 0 drained = 0 while True: try: frame = q.get_nowait() except (queue.Empty, EOFError, OSError): break if frame: append_shadowing_listen_pcm(conn, frame) drained += 1 if drained: logger.bind(tag=TAG).info(f"跟读关窗前抽干 ASR 队列 frames={drained}") return drained def append_shadowing_listen_pcm(conn: "ConnectionHandler", pcm_frame: Any) -> None: if getattr(conn, "interaction_mode", "") != "SHADOWING": return if not getattr(conn, "shadowing_awaiting_user", False): return if not pcm_frame: return buf = getattr(conn, "shadowing_listen_pcm", None) if buf is None: conn.shadowing_listen_pcm = [] buf = conn.shadowing_listen_pcm buf.append(pcm_frame) def _take_shadowing_listen_pcm(conn: "ConnectionHandler") -> list: frames = list(getattr(conn, "shadowing_listen_pcm", None) or []) conn.shadowing_listen_pcm = [] return frames def save_shadowing_listen_audio(conn: "ConnectionHandler", step_index: Optional[int] = None) -> None: """Persist exactly one wav per follow-read step. Never merge neighboring windows.""" if getattr(conn, "interaction_mode", "") != "SHADOWING": return index = int( step_index if step_index is not None else (getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0) ) if int(getattr(conn, "shadowing_audio_saved_index", -1)) == index: conn.shadowing_listen_pcm = [] return snapshot = _take_shadowing_listen_pcm(conn) if not snapshot: logger.bind(tag=TAG).warning(f"跟读听窗无 PCM，写静音占位 step={index}") snapshot = silent_pcm_snapshot() conn.shadowing_audio_saved_index = index from core.app_runtime.session import stash_asr_audio stash_asr_audio(conn, snapshot) try: apply_deferred_assessment(conn, index, audio_snapshot=snapshot) except Exception as exc: logger.bind(tag=TAG).warning(f"跟读听窗录音落盘失败 step={index}: {exc}") async def _on_shadowing_listen_window_closed(conn: "ConnectionHandler") -> None: """Clock or listen stop: archive this listen. Device already sequences play → record → play.""" if not getattr(conn, "shadowing_awaiting_user", False): return _drain_asr_queue_into_shadowing_pcm(conn) conn.shadowing_awaiting_user = False listened_idx = int(getattr(conn, "shadowing_index", 0) or 0) listened_prompt = getattr(conn, "shadowing_prompt", "") or "" save_shadowing_listen_audio(conn, listened_idx) items = _shadowing_items(conn) nxt = listened_idx + 1 conn.shadowing_index = nxt if nxt < len(items): conn.shadowing_prompt = items[nxt] logger.bind(tag=TAG).info(f"跟读：听窗结束 step={listened_idx}") try: session = current_session(conn) or getattr(conn, "task_session", None) if session: if session.find_turn(listened_idx) is None: session.upsert_turn( index=listened_idx, prompt_text=listened_prompt, asr_text="", expected_text=_expected_for_index(conn, listened_idx), ) session.add_event("listen_window_closed") session.add_event("listen_timeout") session.save() except Exception as exc: logger.bind(tag=TAG).warning(f"跟读听窗归档失败: {exc}") def _parse_mp3_list_names(raw: Any) -> list: text = "" if raw is None: return [] if hasattr(raw, "response") or hasattr(raw, "result"): text = str(getattr(raw, "response", None) or getattr(raw, "result", None) or "") else: text = str(raw) text = text.strip() if not text: return [] try: import json data = json.loads(text) except Exception: return [] files = data.get("files") if isinstance(data, dict) else None names = [] if isinstance(files, list): for item in files: if isinstance(item, dict) and item.get("name"): names.append(str(item["name"])) elif isinstance(item, str): names.append(item) return names async def _list_sd_mp3_names(conn: "ConnectionHandler") -> list: result = await _execute_named_tool(conn, DEVICE_LIST_TOOL, {}) return _parse_mp3_list_names(result) async def _list_sd_mp3_names_safe(conn: "ConnectionHandler"): try: names = await _list_sd_mp3_names(conn) return names if isinstance(names, list) else [] except Exception as exc: logger.bind(tag=TAG).warning(f"列举 SD 课文失败: {exc}") return None def _play_lesson_names(play_params: Optional[Dict[str, Any]]) -> list: from .lesson_audio import unique_lesson_names if not isinstance(play_params, dict): return [] items = play_params.get("items") if isinstance(items, list) and items: return unique_lesson_names(items) return unique_lesson_names([play_params]) def _lesson_sync_timeout_sec(conn: "ConnectionHandler") -> float: runtime = {} cfg = getattr(conn, "config", None) if isinstance(cfg, dict) and isinstance(cfg.get("app_runtime"), dict): runtime = cfg.get("app_runtime") or {} raw = runtime.get("lesson_sync_timeout_sec", 90) try: return max(0.0, float(raw)) except (TypeError, ValueError): return 90.0 async def _request_lesson_sync( conn: "ConnectionHandler", missing: list, task_id: str, params: Optional[Dict[str, Any]], ) -> None: from config.manage_api_client import sync_lesson_audio from .lesson_audio import conn_org_id, course_id_for, lesson_audio_base_url, unique_lesson_names files = unique_lesson_names(missing) if not files: return params = params or {} lesson_id = str(params.get("lesson_id") or params.get("lesson") or "l1.001").strip() or "l1.001" org_id = conn_org_id(conn, params) payload = { "macAddress": getattr(conn, "device_id", None) or "", "courseId": course_id_for(task_id, lesson_id), "orgId": org_id, "baseUrl": lesson_audio_base_url(getattr(conn, "config", None), org_id), "files": files, } result = await sync_lesson_audio(payload) if result is None: logger.bind(tag=TAG).warning(f"课文 MQTT 同步失败 files={len(files)}") else: logger.bind(tag=TAG).info(f"课文 MQTT 同步已下发 org={org_id} files={len(files)}") async def _wait_lesson_files(conn: "ConnectionHandler", needed: list, timeout_sec: float) -> bool: from .lesson_audio import missing_lesson_names if not needed: return False deadline = asyncio.get_running_loop().time() + max(0.0, timeout_sec) while True: present = await _list_sd_mp3_names_safe(conn) if present is not None and not missing_lesson_names(needed, present): return True remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: return False await asyncio.sleep(min(2.0, remaining)) async def _ensure_lesson_mp3s( conn: "ConnectionHandler", names: list, task_id: str, params: Optional[Dict[str, Any]], ) -> bool: """READY if SD already has the package. Otherwise PRELOAD via MQTT then wait. Device paths have no org.""" from .lesson_audio import missing_lesson_names, unique_lesson_names needed = unique_lesson_names(names) if not needed: return False present = await _list_sd_mp3_names_safe(conn) if present is None: logger.bind(tag=TAG).warning("无法列举 SD 课文，跳过同步以免误报") return True missing = missing_lesson_names(needed, present) if not missing: return True params = params or {} conn.lesson_syncing = True _speak(conn, "正在同步音频，请稍等。") await _request_lesson_sync(conn, missing, task_id, params) timeout = _lesson_sync_timeout_sec(conn) ready = await _wait_lesson_files(conn, needed, timeout) conn.lesson_syncing = False return ready def _is_full_lesson_mp3(name: str) -> bool: base = str(name or "").replace("\\", "/").rsplit("/", 1)[-1].lower() return base == "full.mp3" or base.endswith("_full.mp3") def _compact_shadowing_playlist( conn: "ConnectionHandler", params: Optional[Dict[str, Any]], ) -> List[Dict[str, Any]]: """Keep {name, repeat, speed}. Device expands repeats locally with a listen window each pass.""" items = playlist_from_params(_task_for(conn), params, "shadowing", default_shadowing_playlist) compact: List[Dict[str, Any]] = [] for raw in items: if not isinstance(raw, dict): continue name = str(raw.get("name") or "") if not name or _is_full_lesson_mp3(name): continue try: repeat_n = int(raw.get("repeat") or 1) except (TypeError, ValueError): repeat_n = 1 if repeat_n < 1: repeat_n = 1 compact.append( { "name": name, "repeat": repeat_n, "speed": clamp_play_speed( raw["speed"] if raw.get("speed") is not None else 1.0 ), "power_off_after": False, "finish_task": False, } ) return compact def _begin_shadowing_playlist( conn: "ConnectionHandler", params: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """One shot: send the whole shadowing playlist. Device plays, records, plays.""" items = _compact_shadowing_playlist(conn, params) clips = expand_playlist_plays(items) conn.shadowing_clips = clips conn.shadowing_index = 0 conn.shadowing_prompt = clips[0] if clips else "" conn.shadowing_awaiting_user = False first = items[0] if items else dict(params or {}) play = dict(first) play["items"] = items play["finish_task"] = False play["listen_ratio"] = resolve_listen_ratio(conn, _task_for(conn)) if items: conn.shadowing_last_speed = items[0].get("speed") return play async def _play_shadowing_clip(conn: "ConnectionHandler", name: str) -> bool: if getattr(conn, "interaction_mode", "") != "SHADOWING": logger.bind(tag=TAG).info(f"跳过跟读播放，当前已离开跟读: {name}") return False conn.shadowing_awaiting_user = False conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1 speed = resolve_play_speed(_task_for(conn)) conn.shadowing_last_speed = speed notice = await _execute_named_tool_play( conn, { "name": name, "repeat": 1, "power_off_after": False, "speed": speed, "finish_task": False, "listen_ratio": resolve_listen_ratio(conn, _task_for(conn)), }, ) if notice and notice.action in (Action.ERROR, Action.NOTFOUND): return False return True async def _prepare_shadowing_sd(conn: "ConnectionHandler", params: Optional[Dict[str, Any]]) -> bool: params = params or {} items = _compact_shadowing_playlist(conn, params) clips = expand_playlist_plays(items) if not clips: lesson = str(params.get("lesson_id") or params.get("lesson") or "").strip() single = str(params.get("name") or params.get("path") or params.get("url") or "") if not lesson and single and not single.lower().endswith(".mp3"): lesson = single if lesson.lower().endswith(".mp3"): lesson = "" if not lesson: conn.shadowing_clips = [] conn.shadowing_prompt = "" return False names = await _list_sd_mp3_names(conn) clips = lesson_mp3_clips( names, getattr(conn, "active_app_id", "english_learning"), getattr(conn, "current_task_id", "shadowing"), lesson, ) clips = [name for name in clips if not _is_full_lesson_mp3(name)] if not clips: conn.shadowing_clips = [] conn.shadowing_prompt = "" return False conn.shadowing_clips = clips conn.shadowing_index = 0 conn.shadowing_prompt = clips[0] conn.shadowing_awaiting_user = False return True async def _start_shadowing_sd(conn: "ConnectionHandler", params: Optional[Dict[str, Any]]) -> bool: if not await _prepare_shadowing_sd(conn, params): return False play = _begin_shadowing_playlist(conn, params) names = _play_lesson_names(play) if not await _ensure_lesson_mp3s(conn, names, getattr(conn, "current_task_id", "shadowing") or "shadowing", params): return False conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1 notice = await _execute_named_tool_play(conn, play) if notice and notice.action in (Action.ERROR, Action.NOTFOUND): return False return True async def _fill_repeater_play_name( conn: "ConnectionHandler", params: Optional[Dict[str, Any]] ) -> Dict[str, Any]: play = dict(params or {}) if play.get("items"): return play if str(play.get("name") or "").strip(): return play stub = default_repeater_playlist(_task_for(conn)) if stub: first = stub[0] play["name"] = first.get("name") or "" if play.get("repeat") is None: play["repeat"] = first.get("repeat") or 1 if play.get("speed") is None and first.get("speed") is not None: play["speed"] = first.get("speed") conn.repeater_play_name = play["name"] return play async def maybe_flush_pending_play_after_tts(conn: "ConnectionHandler") -> None: """Start queued lesson audio only after enter-prompt TTS has finished on device.""" _ensure_runtime_fields(conn) pending = getattr(conn, "pending_play_after_tts", None) if not pending: return conn.pending_play_after_tts = None kind = pending.get("kind") if isinstance(pending, dict) else None try: if kind == "repeater": play_params = await _fill_repeater_play_name(conn, pending.get("params") or {}) await _execute_named_tool_play(conn, play_params) elif kind == "shadowing": play_params = pending.get("params") if isinstance(pending, dict) else None if not play_params: play_params = _begin_shadowing_playlist(conn, {}) conn.shadowing_awaiting_user = False conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1 await _execute_named_tool_play(conn, play_params) except Exception as exc: logger.bind(tag=TAG).warning(f"延后播放课文失败: {exc}") def _lesson_start_phrase(task_id: str) -> str: if task_id == "repeater": return REPEATER_START_PHRASE if task_id == "shadowing": return SHADOWING_START_PHRASE return "" def _schedule_talk_start_phrase(task_id: str) -> str: if task_id == "conversation": return CHAT_START_PHRASE if task_id == "live_translate": return TRANSLATION_START_PHRASE if task_id == "bilingual_chat": return BILINGUAL_START_PHRASE return "" def _speak_lesson_start(conn: "ConnectionHandler", task_id: str) -> None: phrase = _lesson_start_phrase(task_id) if not phrase: return conn.sentence_id = uuid.uuid4().hex _speak(conn, phrase) def _lesson_end_speech(conn: "ConnectionHandler", phrase: str) -> str: """Short lesson-end cue plus the same default chat greeting used after 翻译/中英对话.""" greeting = (_greeting_text(conn) or "").strip() phrase = (phrase or "").strip() if phrase and greeting: if phrase.endswith("。") or phrase.endswith("."): return f"{phrase}{greeting}" return f"{phrase}。{greeting}" return phrase or greeting async def _return_to_chat_after_lesson( conn: "ConnectionHandler", phrase: str, reason: str = "play_completed", ) -> None: """Playlist already finished on device: speak the end cue now, then chat. Do not MCP-stop.""" conn.repeater_speed_queue = [] conn.repeater_playlist = [] end_session(conn, reason) if getattr(conn, "tts", None) is not None: await wait_for_tts_channels(conn, timeout=3) conn.sentence_id = uuid.uuid4().hex _speak(conn, _lesson_end_speech(conn, phrase)) await start_app_task( conn, "chat", "conversation", {}, source="voice", execute_skills=False, speak_switch=False, stop_skills=False, ) async def _shadowing_play_next_or_finish(conn: "ConnectionHandler") -> None: """Legacy helper: device now sequences clips. Only used if a caller still asks to finish.""" conn.shadowing_index = 0 conn.shadowing_prompt = "" conn.shadowing_clips = [] await _return_to_chat_after_lesson(conn, SHADOWING_END_PHRASE, "finished") def _mark_repeater_day(conn: "ConnectionHandler") -> None: try: mark_lesson_played_today( str(getattr(conn, "device_id", "") or ""), getattr(conn, "active_app_id", "") or "", getattr(conn, "current_task_id", "") or "", getattr(conn, "repeater_play_name", "") or "", ) except Exception as exc: logger.bind(tag=TAG).warning(f"记录泛听天数失败: {exc}") def _apply_playback_speed( conn: "ConnectionHandler", task: Optional[Dict[str, Any]], params: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """Attach clamped speed. Multi-speed repeats are Server-driven (one pass per play).""" play = dict(params or {}) try: repeat_n = int(play.get("repeat") or 1) except (TypeError, ValueError): repeat_n = 1 if repeat_n < 1: repeat_n = 1 device_id = str(getattr(conn, "device_id", "") or "") app_id = getattr(conn, "active_app_id", "") or "" task_id = getattr(conn, "current_task_id", "") or "" lesson = str(play.get("name") or "") day_index = lesson_day_index(device_id, app_id, task_id, lesson) if play.get("speed") is not None: speeds = [clamp_play_speed(play.get("speed"))] * repeat_n else: speeds = speeds_for_repeats(task, repeat_n, day_index) conn.repeater_play_name = lesson conn.repeater_power_off_after = bool(play.get("power_off_after") or False) varied = len(speeds) > 1 and any(abs(item - speeds[0]) > 0.001 for item in speeds[1:]) if varied: conn.repeater_speed_queue = speeds[1:] play["speed"] = speeds[0] play["repeat"] = 1 play["power_off_after"] = False else: conn.repeater_speed_queue = [] play["speed"] = speeds[0] play["repeat"] = repeat_n # 泛听清单结束由 Server 收 audio.completed 后切回对话。设备不要 finish_task 写 NVS。 play["finish_task"] = False return play def _begin_repeater_playlist( conn: "ConnectionHandler", task: Optional[Dict[str, Any]], params: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """One shot: send the whole playlist. Repeat is just the same clip again, no mode switch.""" items = playlist_from_params(task, params, "repeater", default_repeater_playlist) normalized = [] for raw in items: if not isinstance(raw, dict): continue name = str(raw.get("name") or "") if not name: continue try: repeat_n = int(raw.get("repeat") or 1) except (TypeError, ValueError): repeat_n = 1 if repeat_n < 1: repeat_n = 1 item = { "name": name, "repeat": repeat_n, "speed": clamp_play_speed( raw["speed"] if raw.get("speed") is not None else 1.0 ), "power_off_after": bool(raw.get("power_off_after") or False), "finish_task": False, } normalized.append(item) conn.repeater_playlist = [] conn.repeater_speed_queue = [] first = normalized[0] if normalized else dict(params or {}) conn.repeater_play_name = str(first.get("name") or "") conn.repeater_power_off_after = bool(first.get("power_off_after") or False) play = dict(first) play["items"] = normalized play["finish_task"] = False return play async def _continue_repeater_if_needed(conn: "ConnectionHandler") -> bool: queue = list(getattr(conn, "repeater_speed_queue", None) or []) if queue: nxt = queue.pop(0) conn.repeater_speed_queue = queue last_speed = len(queue) == 0 and not list(getattr(conn, "repeater_playlist", None) or []) try: play_params = await _fill_repeater_play_name( conn, { "name": getattr(conn, "repeater_play_name", "") or "", "speed": nxt, "repeat": 1, "power_off_after": last_speed and bool(getattr(conn, "repeater_power_off_after", False)), "finish_task": False, }, ) await _execute_named_tool_play(conn, play_params) except Exception as exc: logger.bind(tag=TAG).warning(f"泛听下一遍失败: {exc}") conn.repeater_speed_queue = [] return False return True rest = list(getattr(conn, "repeater_playlist", None) or []) if not rest: return False nxt, conn.repeater_playlist = rest[0], rest[1:] play_params = _apply_playback_speed(conn, _task_for(conn), nxt) play_params["finish_task"] = False try: play_params = await _fill_repeater_play_name(conn, play_params) await _execute_named_tool_play(conn, play_params) except Exception as exc: logger.bind(tag=TAG).warning(f"泛听下一首失败: {exc}") conn.repeater_playlist = [] return False return True async def _execute_named_tool_play(conn, params): items = params.get("items") if isinstance(params.get("items"), list) else [] first = items[0] if items else params play_args = { "name": (first.get("name") if isinstance(first, dict) else None) or params.get("name") or "", "repeat": int((first.get("repeat") if isinstance(first, dict) else None) or params.get("repeat") or 1), "power_off_after": bool(params.get("power_off_after") or False), "speed": clamp_play_speed( (first.get("speed") if isinstance(first, dict) and first.get("speed") is not None else None) if items else (params["speed"] if params.get("speed") is not None else 1.0) ), "finish_task": bool(params.get("finish_task", True)), "app_id": str(getattr(conn, "active_app_id", "") or ""), "task_id": str(getattr(conn, "current_task_id", "") or ""), "force": True, } if items: play_args["items"] = json.dumps(items, ensure_ascii=False) play_args["speed"] = clamp_play_speed( first["speed"] if first.get("speed") is not None else 1.0 ) play_args["power_off_after"] = bool( items[-1].get("power_off_after") if items else False ) if params.get("listen_ratio") is not None: try: play_args["listen_ratio"] = float(params["listen_ratio"]) except (TypeError, ValueError): pass if play_args["task_id"] in ("repeater", "shadowing"): play_args["finish_task"] = False result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args) blob = "" if result: blob = str(result.response or result.result or "") compact = blob.replace(" ", "") if '"duplicate_ignored":true' in compact: logger.bind(tag=TAG).warning("设备忽略了同一课文的下一遍，强制再播一次") play_args["speed"] = clamp_play_speed(float(play_args["speed"]) + 0.02) play_args["force"] = True result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args) blob = "" if result: blob = str(result.response or result.result or "") compact = blob.replace(" ", "") rejected = (result and result.action in (Action.ERROR, Action.NOTFOUND)) or ( "no_mp3" in blob or '"accepted":false' in compact or '"duplicate_ignored":true' in compact ) if rejected: if not play_args["name"]: return ActionResponse( action=Action.RESPONSE, response="泛听已打开，但 SD 卡上还没有可播放的课文。", ) return ActionResponse( action=Action.RESPONSE, response=(result.response if result else None) or "这台设备没有本地泛听播放能力", ) return None def _apply_last_lesson_params( conn: "ConnectionHandler", task_id: str, params: Optional[Dict[str, Any]], source: str, ) -> Dict[str, Any]: params = dict(params or {}) if task_id not in ("repeater", "shadowing"): return params device_id = str(getattr(conn, "device_id", "") or "") has_lesson = str(params.get("lesson_id") or params.get("lesson") or "").strip() if has_lesson: remember_last_lesson_params(device_id, task_id, params) return params saved = load_last_lesson_params(device_id, task_id) if saved: merged = dict(saved) for key, value in params.items(): if value not in (None, "", []): merged[key] = value return merged return params def _lesson_missing_speech(task_id: str, params: Optional[Dict[str, Any]], fallback: str) -> str: kind = "泛听" if task_id == "repeater" else "跟读" lesson = str((params or {}).get("lesson_id") or (params or {}).get("lesson") or "").strip() if not lesson: return f"还没有下发过{kind}课文，请先在家长端选一课。" return fallback async def start_app_task( conn: "ConnectionHandler", app_id: str, task_id: str = "", params: Optional[Dict[str, Any]] = None, source: str = "voice", execute_skills: bool = True, speak_switch: bool = True, stop_skills: bool = True, ) -> ActionResponse: """Single entry for MQTT / schedule / voice / default chat.""" _ensure_runtime_fields(conn) if await prompt_bind_if_needed(conn): return ActionResponse(action=Action.NONE, response="设备未绑定") conn.client_abort = False tts = getattr(conn, "tts", None) if tts is not None: tts.tts_stop_request = False params = params or {} if params.get("announce") is False: speak_switch = False app, task = get_task(app_id, task_id) if not app: return ActionResponse(action=Action.ERROR, response=f"未知的 App: {app_id}") if not task: default_task = app.get("default_task") or "" app, task = get_task(app_id, default_task) if not task: return ActionResponse(action=Action.ERROR, response=f"未知的 Task: {app_id}/{task_id}") resolved_task_id = task_id or app.get("default_task") or "conversation" params = _apply_last_lesson_params(conn, resolved_task_id, params, source) new_p = source_priority(source) cur_p = source_priority(getattr(conn, "current_source", "default_chat")) same_task = ( getattr(conn, "active_app_id", "") == app_id and getattr(conn, "current_task_id", "") == resolved_task_id ) # 口令回对话要能退出家长即时/计划任务。其它低优先级切换仍然拦住。 voice_exit_to_chat = ( source == "voice" and app_id == "chat" and resolved_task_id == "conversation" ) if new_p < cur_p and not same_task and not voice_exit_to_chat: logger.bind(tag=TAG).info( f"拒绝低优先级任务 {source}({new_p}) < {conn.current_source}({cur_p})" ) return ActionResponse( action=Action.NONE, response="当前有更高优先级的任务，先完成后再切换", ) if source in ("voice", "mqtt_immediate", "schedule"): schedule_close_parent_except(conn, source, app_id, resolved_task_id) prev_app_id = getattr(conn, "active_app_id", "chat") prev_task_id = getattr(conn, "current_task_id", "conversation") prev_app, prev_task = get_task(prev_app_id, prev_task_id) switching = (prev_app_id != app_id) or (prev_task_id != resolved_task_id) if not same_task: if stop_skills: await _stop_current_skills(conn) if app_id != "chat": conn.chat_greeted = False conn.active_app_id = app_id conn.current_task_id = resolved_task_id conn.interaction_mode = task.get("interaction_mode") or app.get("mode") or "CHAT" conn.current_source = source conn.tool_policy = app.get("tool_policy") or "inherit_agent" conn.tool_allowlist = list(app.get("tools") or []) _apply_prompt(conn, app, params) _refresh_tools(conn) logger.bind(tag=TAG).info( f"start_app_task app={app_id} task={resolved_task_id} mode={conn.interaction_mode} source={source}" ) device_ready = bool(params.get("sd_ready")) if device_ready: speak_switch = False logger.bind(tag=TAG).info("课文已在 SD，先播开始语，再由设备本地开播") _speak_lesson_start(conn, resolved_task_id) if app_id == "chat": end_session(conn, "user_exit") # 家长下发的对话要记一条执行记录；开机欢迎词 / 口令回对话不要另开 session。 if source in ("mqtt_immediate", "schedule"): begin_session(conn, app_id, resolved_task_id, source, conn.interaction_mode) elif switching: begin_session(conn, app_id, resolved_task_id, source, conn.interaction_mode) pending_play = None lesson_missing = False if execute_skills and not device_ready: if app_sd_layout_paths(app_id, app): await ensure_app_sd_dirs(conn, app_id) for skill in task.get("skills") or []: skill_name = (skill or {}).get("name") or "" if skill_name in ("self.mp3.play", DEVICE_PLAY_TOOL): play_params = _begin_repeater_playlist(conn, task, params) items = play_params.get("items") if isinstance(play_params, dict) else None if not items and not (isinstance(play_params, dict) and play_params.get("name")): lesson_missing = True else: names = _play_lesson_names(play_params) if not await _ensure_lesson_mp3s(conn, names, resolved_task_id, params): lesson_missing = True else: pending_play = {"kind": "repeater", "params": play_params} elif skill_name == "self.power_off": await _execute_named_tool(conn, "self_power_off", {}) if source == "schedule": talk_phrase = _schedule_talk_start_phrase(resolved_task_id) if talk_phrase: conn.sentence_id = uuid.uuid4().hex _speak(conn, talk_phrase) speak_switch = False if resolved_task_id == "conversation": conn.chat_greeted = True announce = "" boot_chat = app_id == "chat" and source == "default_chat" if boot_chat: # 开机进 chat：必须播欢迎词。不要被 same_task / just_woken_up 跳过。 enable = (conn.config or {}).get("enable_proactive_greeting", True) if enable and not getattr(conn, "chat_greeted", False): if not await wait_for_tts_channels(conn, timeout=10): logger.bind(tag=TAG).error("开机欢迎词跳过：TTS通道未就绪") else: conn.chat_greeted = True conn.sentence_id = uuid.uuid4().hex _speak(conn, _greeting_text(conn)) elif switching: announce = _switch_announcement(prev_app, prev_task, app, task, app_id) if lesson_missing: prefix = announce or "现在已经切换至泛听模式。" announce = _lesson_missing_speech( resolved_task_id, params, f"{prefix}这课还没有放到SD卡。", ) if getattr(conn, "language_pair_fallback", False) and conn.interaction_mode in ( "BILINGUAL", "TRANSLATION", ): announce = f"{announce or ''}这套语言先用中英来练。" if conn.interaction_mode == "SHADOWING" and not device_ready: play_params = _begin_shadowing_playlist(conn, params) names = _play_lesson_names(play_params) if not names: if await _prepare_shadowing_sd(conn, params): play_params = _begin_shadowing_playlist(conn, params) names = _play_lesson_names(play_params) if not names: prefix = announce or "现在已经切换至跟读模式。" announce = _lesson_missing_speech( resolved_task_id, params, f"{prefix}跟读课文还没有放到SD卡。", ) elif not await _ensure_lesson_mp3s(conn, names, resolved_task_id, params): prefix = announce or "现在已经切换至跟读模式。" announce = _lesson_missing_speech( resolved_task_id, params, f"{prefix}跟读课文还没有放到SD卡。", ) else: announce = _lesson_start_phrase(resolved_task_id) or announce or "开始跟读任务" pending_play = { "kind": "shadowing", "params": play_params, } start_phrase = _lesson_start_phrase(resolved_task_id) if start_phrase and pending_play and not lesson_missing: announce = start_phrase if speak_switch and announce: conn.sentence_id = uuid.uuid4().hex conn.pending_play_after_tts = pending_play _speak(conn, announce) elif pending_play: conn.pending_play_after_tts = pending_play await maybe_flush_pending_play_after_tts(conn) if app_id == "chat": conn.chat_greeted = True elif conn.interaction_mode == "CHAT": enable = (conn.config or {}).get("enable_proactive_greeting", True) woken = bool(getattr(conn, "just_woken_up", False)) if enable and not woken and not getattr(conn, "chat_greeted", False): conn.chat_greeted = True conn.sentence_id = uuid.uuid4().hex _speak(conn, _greeting_text(conn)) elif conn.interaction_mode == "SHADOWING" and not device_ready and not getattr(conn, "shadowing_clips", None): started = await _start_shadowing_sd(conn, params) if not started and speak_switch: conn.sentence_id = uuid.uuid4().hex _speak(conn, _lesson_missing_speech(resolved_task_id, params, "跟读课文还没有放到SD卡。")) elif pending_play: conn.pending_play_after_tts = pending_play await maybe_flush_pending_play_after_tts(conn) app_name = _display_name(app, task) return ActionResponse( action=Action.NONE, result=f"{app_id}/{resolved_task_id}", response=announce or f"已打开{app_name}", ) async def switch_app( conn: "ConnectionHandler", app_id: str, source: str = "voice", task_id: str = "", ) -> ActionResponse: app = get_app(app_id) if not app: return ActionResponse(action=Action.ERROR, response=f"不支持的 App: {app_id}") resolved = task_id or app.get("default_task") or "" return await start_app_task(conn, app_id, resolved, {}, source) async def on_user_wakeup(conn: "ConnectionHandler") -> None: """User wake word returns to chat; do not treat [app_task] as wakeup.""" _ensure_runtime_fields(conn) if getattr(conn, "interaction_mode", "CHAT") == "CHAT" and getattr(conn, "active_app_id", "chat") == "chat": return await start_app_task(conn, "chat", "conversation", {}, source="voice") def match_voice_switch(text: str) -> Optional[tuple]: """Map a short phrase to (app_id, task_id) using apps.json voice_enter / voice_exit.""" if not text: return None _, filtered = _strip(text) return match_registry_voice(filtered) def _strip(text: str) -> tuple: from core.utils.util import remove_punctuation_and_length return remove_punctuation_and_length(text) def is_exit_to_chat(text: str) -> bool: mapped = match_voice_switch(text) return bool(mapped and mapped[0] == "chat") def should_skip_listen_stop_asr(conn: Any) -> bool: """True when listen stop is closing the mic for lesson playback, not a follow-read.""" mode = getattr(conn, "interaction_mode", "CHAT") if mode == "PLAY_AUDIO": return True if mode == "SHADOWING" and not getattr(conn, "shadowing_awaiting_user", False): return True return False def is_exit_translation(text: str) -> bool: return is_exit_to_chat(text) def _plain_user_text(text: str) -> str: """ASR 可能把原文包在 JSON 里，翻译只要 content。""" if not text: return "" raw = str(text).strip() if raw.startswith("{") and raw.endswith("}"): try: import json data = json.loads(raw) if isinstance(data, dict) and data.get("content"): return str(data["content"]).strip() except Exception: pass return raw def _norm_for_echo(text: str) -> str: _, filtered = _strip(text or "") return re.sub(r"\s+", "", (filtered or "").lower()) def is_echo_of_prompt(user_text: str, prompt: str) -> bool: a = _norm_for_echo(user_text) b = _norm_for_echo(prompt) if not a or not b: return False if a == b or a in b or b in a: return True return difflib.SequenceMatcher(None, a, b).ratio() >= 0.82 def detect_user_language(text: str, hint: Optional[str] = None) -> str: if hint: h = str(hint).lower() if h.startswith("zh") or h in ("cn", "yue", "chinese"): return "zh" if h.startswith("en") or h in ("english",): return "en" body = _plain_user_text(text) cjk = len(re.findall(r"[\u4e00-\u9fff]", body)) latin = len(re.findall(r"[A-Za-z]", body)) return "zh" if cjk >= max(latin, 1) else "en" def wrap_bilingual_user_text(text: str, user_lang: str) -> str: if user_lang == "zh": return "Respond in English only. Do not use Chinese characters.\n\n" + text return "请只用中文回答，不要出现英文单词或句子。\n\n" + text def _looks_like_tool_error(text: str) -> bool: if not text: return True s = str(text) if "Error executing tool" in s or "isError=True" in s or "isError=true" in s: return True if "未配置" in s and "api_key" in s: return True return False def _record_shadowing_turn( conn: "ConnectionHandler", text: str, echo_rejected: bool = False, index: Optional[int] = None, ) -> None: session = current_session(conn) if not session: return if index is None: index = int(getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0) session.upsert_turn( index=index, prompt_text=getattr(conn, "shadowing_prompt", "") or "", asr_text=_plain_user_text(text), echo_rejected=echo_rejected, expected_text=_expected_for_index(conn, index), ) session.save() def apply_shadowing_user_text(conn: "ConnectionHandler", text: str) -> str: """Apply one ASR utterance. Returns exit | ignore | echo | recorded. ASR never advances the clip index; the listen-window clock does. Late ASR after the window still scores the sentence that was just listened to. Device plays the next clip itself; Server does not MCP the next sentence. """ if is_exit_to_chat(text): end_session(conn, "user_exit") return "exit" if getattr(conn, "client_is_speaking", False) and not getattr( conn, "shadowing_awaiting_user", False ): logger.bind(tag=TAG).info("跟读：播放中忽略拾音，避免喇叭回声触发对话") return "ignore" prompt = getattr(conn, "shadowing_prompt", "") or "" if ( getattr(conn, "shadowing_awaiting_user", False) and prompt and not _is_clip_path(prompt) and is_echo_of_prompt(text, prompt) ): logger.bind(tag=TAG).info("跟读：忽略与范读过于相似的识别结果") session = current_session(conn) if session: session.add_event("echo_rejected") session.save() return "echo" _record_shadowing_turn(conn, text, echo_rejected=False) return "recorded" async def handle_shadowing_turn(conn: "ConnectionHandler", text: str) -> None: """Score this utterance off the chat path. Next clip is sequenced on device.""" outcome = apply_shadowing_user_text(conn, text) if outcome == "exit": _cancel_shadowing_timeout(conn) await start_app_task(conn, "chat", "conversation", {}, source="voice") return if outcome in ("ignore", "echo"): return step_index = int( getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0 ) logger.bind(tag=TAG).info(f"跟读 ASR 仅评估不上对话 step={step_index}") _schedule_deferred_assessment(conn, step_index) async def handle_skill_event(conn: "ConnectionHandler", text: str) -> bool: """Handle `[skill_event]audio.completed|app|task|duration_ms`. Returns True if consumed.""" parsed = parse_skill_event(text) if parsed is None: return False _ensure_runtime_fields(conn) name = parsed.get("name") or "" app_id = parsed.get("app_id") or "" task_id = parsed.get("task_id") or "" try: duration_ms = int(parsed.get("duration_ms") or 0) except (TypeError, ValueError): duration_ms = 0 logger.bind(tag=TAG).info(f"skill_event {name} {app_id}/{task_id} duration_ms={duration_ms}") mode = getattr(conn, "interaction_mode", "CHAT") # 听窗时钟必须先于 session.save()：磁盘/HTTP 不能挡住 1.2×duration 的热路径。 shadowing_playlist_done = name in ("audio.playlist_completed", "playlist.completed") and ( mode == "SHADOWING" or task_id == "shadowing" ) if shadowing_playlist_done: _cancel_shadowing_timeout(conn) if getattr(conn, "shadowing_awaiting_user", False): await _on_shadowing_listen_window_closed(conn) else: save_shadowing_listen_audio(conn) session = getattr(conn, "task_session", None) if session: session.add_event(name, app_id, task_id) await _return_to_chat_after_lesson(conn, SHADOWING_END_PHRASE, "finished") return True if mode == "SHADOWING" and name == "audio.completed": if getattr(conn, "shadowing_awaiting_user", False): await _on_shadowing_listen_window_closed(conn) conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1 _schedule_shadowing_listen_timeout(conn, duration_ms) session = getattr(conn, "task_session", None) if session: session.add_event(name, app_id, task_id) return True if mode == "SHADOWING" and name == "audio.stopped": session = getattr(conn, "task_session", None) if session: session.add_event(name, app_id, task_id) end_session(conn, "play_stopped") await start_app_task( conn, "chat", "conversation", {}, source="voice", execute_skills=False, speak_switch=False, ) return True repeater_done = name == "audio.completed" and ( mode == "PLAY_AUDIO" or task_id == "repeater" ) if repeater_done: continued = await _continue_repeater_if_needed(conn) session = getattr(conn, "task_session", None) if session: session.add_event(name, app_id, task_id) if continued: if session: session.save() return True _mark_repeater_day(conn) await _return_to_chat_after_lesson(conn, REPEATER_END_PHRASE, "play_completed") return True if mode == "PLAY_AUDIO" and name == "audio.stopped": conn.repeater_speed_queue = [] session = getattr(conn, "task_session", None) if session: session.add_event(name, app_id, task_id) session.save() if name == "audio.completed": end_session(conn, "play_completed") await start_app_task( conn, "chat", "conversation", {}, source="voice", execute_skills=False, speak_switch=True, ) elif name == "audio.stopped": end_session(conn, "play_stopped") if getattr(conn, "active_app_id", "chat") != "chat": await start_app_task( conn, "chat", "conversation", {}, source="voice", execute_skills=False, speak_switch=False, ) else: logger.bind(tag=TAG).warning(f"未知 skill_event: {name}") return True async def run_translation(conn: "ConnectionHandler", text: str) -> str: """Translate with the connection LLM first (智控台已配的模型). MCP is fallback only.""" text = _plain_user_text(text) if not text: return "" from plugins_func.register import all_function_registry item = all_function_registry.get("translate") if item: try: maybe = item.func(conn, text=text) if hasattr(maybe, "__await__"): maybe = await maybe if isinstance(maybe, ActionResponse) and maybe.action not in ( Action.ERROR, Action.NOTFOUND, ): out = (maybe.response or maybe.result or "").strip() if out and not _looks_like_tool_error(out): return out except Exception as e: logger.bind(tag=TAG).warning(f"翻译插件失败: {e}") handler = getattr(conn, "func_handler", None) if handler and handler.has_tool("translate"): result = await _execute_named_tool(conn, "translate", {"text": text}) if result and result.action not in (Action.ERROR, Action.NOTFOUND): out = _extract_translate_text(result.response or result.result or "") if out and not _looks_like_tool_error(out): return out llm = getattr(conn, "llm", None) if llm and hasattr(llm, "response_no_stream"): try: return ( llm.response_no_stream(TRANSLATOR_PROMPT, text, temperature=0.1) or "" ).strip() except Exception as e: logger.bind(tag=TAG).error(f"翻译 LLM 失败: {e}") return "" def _extract_translate_text(raw: Any) -> str: if raw is None: return "" if isinstance(raw, str): text = raw.strip() if text.startswith("{") and "text" in text: try: import json data = json.loads(text) if isinstance(data, dict): return str(data.get("text") or data.get("content") or text) except Exception: pass return text if hasattr(raw, "content"): content = raw.content if isinstance(content, list) and content: first = content[0] return getattr(first, "text", None) or str(first) return str(content) return str(raw).strip()