From 220e4fab27b7f419b3e89e2f2d28ae789aac09d7 Mon Sep 17 00:00:00 2001
From: wangzhibo <wangzhibo@shsening.com>
Date: 星期四, 08 十月 2026 16:29:55 +0800
Subject: [PATCH] 5种场景模式家长端ok
---
xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/playback.py | 288 +++++++++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 268 insertions(+), 20 deletions(-)
diff --git a/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/playback.py b/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/playback.py
index e40dca9..88b0d51 100644
--- a/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/playback.py
+++ b/xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/playback.py
@@ -10,6 +10,15 @@
from config.config_loader import get_project_dir
from config.logger import setup_logging
+from core.app_runtime.idiom_catalog import (
+ default_en_passes,
+ extensive_paths,
+ find_story,
+ normalize_pass,
+ sentence_count,
+ shadowing_paths,
+)
+from core.app_runtime.lesson_audio import list_lesson_clips
TAG = __name__
logger = setup_logging()
@@ -18,32 +27,187 @@
PLAY_SPEED_MAX = 2.0
PLAY_SPEED_DEFAULT = 1.0
-# 灏忕▼搴�/瀹氭椂浠诲姟涓嬪彂娓呭崟涔嬪墠鐨勪复鏃惰鏂囥�傝矾寰勭浉瀵� SD 鍗℃牴锛�/english_learning/...锛夈��
+# 璇煶鍙ｄ护鏈甫 lesson 鏃讹紝涓嶅啀浣跨敤娴嬭瘯娓呭崟锛涘繀椤诲厛鏈夊闀夸笅鍙戠殑璇炬枃銆�
TEMP_REPEATER_PLAYLIST = (
{
- "name": "/english_learning/shadowing/lessons/unit01/01.mp3",
- "repeat": 10,
- "speed": PLAY_SPEED_DEFAULT,
+ "name": "/english_learning/repeater/lessons/l1/001/cn/01.mp3",
+ "repeat": 1,
+ "speed": 1.0,
},
{
- "name": "/english_learning/shadowing/lessons/unit01/02.mp3",
+ "name": "/english_learning/repeater/lessons/l1/001/en/01.mp3",
+ "repeat": 1,
+ "speed": 0.8,
+ },
+ {
+ "name": "/english_learning/repeater/lessons/l1/001/en/01.mp3",
+ "repeat": 2,
+ "speed": 0.9,
+ },
+ {
+ "name": "/english_learning/repeater/lessons/l1/001/en/01.mp3",
"repeat": 5,
- "speed": PLAY_SPEED_DEFAULT,
+ "speed": 1.0,
},
)
-TEMP_SHADOWING_PLAYLIST = (
+TEMP_SHADOWING_PLAYLIST = tuple(
{
- "name": "/english_learning/shadowing/lessons/unit01/01.mp3",
- "repeat": 5,
- "speed": PLAY_SPEED_DEFAULT,
- },
- {
- "name": "/english_learning/shadowing/lessons/unit01/02.mp3",
- "repeat": 10,
- "speed": PLAY_SPEED_DEFAULT,
- },
+ "name": f"/english_learning/shadowing/lessons/l1/001/en/{index:02d}.mp3",
+ "repeat": 1,
+ "speed": 0.8,
+ }
+ for index in range(1, 59)
)
+
+
+def expand_lesson_playlist(
+ task_id: str,
+ lesson_id: str,
+ repeat: Any = None,
+ speed: Any = None,
+ language: Any = None,
+ en_passes: Any = None,
+ cn_repeat: Any = None,
+ cn_speed: Any = None,
+ sentence_count_override: Any = None,
+) -> List[Dict[str, Any]]:
+ """lesson_id + optional repeat/speed 鈫� compact playlist. Unknown lesson returns []."""
+ lesson = str(lesson_id or "").strip()
+ tid = str(task_id or "").strip()
+ if not lesson:
+ return []
+ story = find_story(lesson)
+ if tid == "repeater":
+ items = _expand_repeater(lesson, story, en_passes, cn_repeat, cn_speed)
+ elif tid == "shadowing":
+ items = _expand_shadowing(lesson, story, speed, language, sentence_count_override)
+ else:
+ return []
+ return _apply_repeat_speed(items, repeat, speed if tid == "repeater" else None)
+
+
+def _lesson_lang_paths(kind: str, lesson: str, lang: str, catalog_paths: List[str]) -> List[str]:
+ """Prefer MP3s that actually exist under the lesson folder. Catalog is only a fallback."""
+ listed = list_lesson_clips(None, None, kind, lesson, lang)
+ if listed is not None:
+ return listed
+ return [path for path in catalog_paths if path]
+
+
+def _expand_repeater(
+ lesson: str,
+ story: Optional[Dict[str, Any]],
+ en_passes: Any,
+ cn_repeat: Any,
+ cn_speed: Any,
+) -> List[Dict[str, Any]]:
+ if story is None:
+ return []
+ catalog_cn, catalog_en = extensive_paths(story, lesson)
+ cn_paths = _lesson_lang_paths("repeater", lesson, "cn", [catalog_cn])
+ en_paths = _lesson_lang_paths("repeater", lesson, "en", [catalog_en])
+ if not cn_paths or not en_paths:
+ return []
+ try:
+ cn_times = max(1, int(cn_repeat)) if cn_repeat is not None else 1
+ except (TypeError, ValueError):
+ cn_times = 1
+ cn_rate = clamp_play_speed(cn_speed) if cn_speed is not None else 1.0
+ items: List[Dict[str, Any]] = [
+ {"name": path, "repeat": min(10, cn_times), "speed": cn_rate}
+ for path in cn_paths
+ ]
+ passes = []
+ if isinstance(en_passes, list):
+ for raw in en_passes:
+ row = normalize_pass(raw)
+ if row:
+ passes.append(row)
+ if not passes:
+ passes = default_en_passes()
+ for row in passes[:8]:
+ for path in en_paths:
+ items.append(
+ {"name": path, "repeat": row["repeat"], "speed": clamp_play_speed(row["speed"])}
+ )
+ return items
+
+
+def _expand_shadowing(
+ lesson: str,
+ story: Optional[Dict[str, Any]],
+ speed: Any,
+ language: Any,
+ sentence_count_override: Any,
+) -> List[Dict[str, Any]]:
+ if story is None:
+ return []
+ lang = str(language or "en").strip().lower()
+ if lang not in ("cn", "en"):
+ lang = "en"
+ listed = list_lesson_clips(None, None, "shadowing", lesson, lang)
+ if listed is not None:
+ paths = listed
+ else:
+ count = None
+ if sentence_count_override is not None:
+ try:
+ count = int(sentence_count_override)
+ except (TypeError, ValueError):
+ count = None
+ if count is None:
+ count = sentence_count(story)
+ paths = shadowing_paths(story, lang, count, lesson)
+ if not paths:
+ return []
+ rate = clamp_play_speed(speed) if speed is not None else 0.8
+ return [
+ {"name": name, "repeat": 1, "speed": rate}
+ for name in paths
+ ]
+
+
+def _apply_repeat_speed(items: List[Dict[str, Any]], repeat: Any, speed: Any) -> List[Dict[str, Any]]:
+ if repeat is not None:
+ try:
+ count = max(1, int(repeat))
+ except (TypeError, ValueError):
+ count = 1
+ for item in items:
+ item["repeat"] = count
+ if speed is not None:
+ value = clamp_play_speed(speed)
+ for item in items:
+ item["speed"] = value
+ return items
+
+
+def playlist_from_params(
+ task: Optional[Dict[str, Any]],
+ params: Optional[Dict[str, Any]],
+ task_id: str,
+ default_factory=None,
+) -> List[Dict[str, Any]]:
+ """Explicit items win; lesson_id expands catalog. No last lesson 鈫� empty (not the test playlist)."""
+ params = params or {}
+ explicit = parse_playlist(params)
+ if explicit:
+ return explicit
+ lesson_id = str(params.get("lesson_id") or params.get("lesson") or "").strip()
+ if lesson_id:
+ return expand_lesson_playlist(
+ task_id,
+ lesson_id,
+ params.get("repeat"),
+ params.get("speed"),
+ language=params.get("language") or params.get("lang"),
+ en_passes=params.get("en_passes"),
+ cn_repeat=params.get("cn_repeat"),
+ cn_speed=params.get("cn_speed"),
+ sentence_count_override=params.get("sentence_count"),
+ )
+ return []
def clamp_play_speed(raw: Any) -> float:
@@ -91,6 +255,82 @@
json.dump(data, handle, ensure_ascii=False, indent=2)
except Exception as exc:
logger.bind(tag=TAG).warning(f"淇濆瓨鎾斁杩涘害澶辫触: {exc}")
+
+
+_LAST_LESSON_KEYS = (
+ "lesson_id",
+ "lesson",
+ "en_passes",
+ "cn_repeat",
+ "cn_speed",
+ "language",
+ "lang",
+ "speed",
+ "on_complete",
+ "listen_ratio",
+ "sentence_count",
+ "org_id",
+ "orgId",
+)
+
+
+def _last_lesson_path(device_id: str) -> str:
+ safe = re.sub(r"[^A-Za-z0-9_-]+", "_", device_id or "unknown") or "unknown"
+ return os.path.join(get_project_dir(), "data", "last_lesson", f"{safe}.json")
+
+
+def _compact_last_lesson(params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+ src = params or {}
+ out: Dict[str, Any] = {}
+ for key in _LAST_LESSON_KEYS:
+ if key in src and src[key] not in (None, ""):
+ out[key] = src[key]
+ lesson = str(out.get("lesson_id") or out.get("lesson") or "").strip()
+ if lesson:
+ out["lesson_id"] = lesson
+ return out
+
+
+def remember_last_lesson_params(device_id: str, task_id: str, params: Optional[Dict[str, Any]]) -> None:
+ kind = str(task_id or "").strip()
+ if kind not in ("repeater", "shadowing"):
+ return
+ compact = _compact_last_lesson(params)
+ if not str(compact.get("lesson_id") or "").strip():
+ return
+ path = _last_lesson_path(device_id)
+ data: Dict[str, Any] = {}
+ if os.path.isfile(path):
+ try:
+ with open(path, "r", encoding="utf-8") as handle:
+ loaded = json.load(handle)
+ if isinstance(loaded, dict):
+ data = loaded
+ except Exception:
+ data = {}
+ data[kind] = compact
+ try:
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "w", encoding="utf-8") as handle:
+ json.dump(data, handle, ensure_ascii=False, indent=2)
+ except Exception as exc:
+ logger.bind(tag=TAG).warning(f"淇濆瓨涓婃璇炬枃澶辫触: {exc}")
+
+
+def load_last_lesson_params(device_id: str, task_id: str) -> Dict[str, Any]:
+ kind = str(task_id or "").strip()
+ path = _last_lesson_path(device_id)
+ if kind not in ("repeater", "shadowing") or not os.path.isfile(path):
+ return {}
+ try:
+ with open(path, "r", encoding="utf-8") as handle:
+ data = json.load(handle)
+ except Exception:
+ return {}
+ row = data.get(kind) if isinstance(data, dict) else None
+ if not isinstance(row, dict):
+ return {}
+ return _compact_last_lesson(row)
def _lesson_key(app_id: str, task_id: str) -> str:
@@ -255,6 +495,9 @@
return item
+PLAYLIST_ITEM_MAX = 256
+
+
def parse_playlist(params: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Repeater/shadowing playlist. Each item is {name, repeat, optional speed}."""
params = params or {}
@@ -265,6 +508,11 @@
item = playlist_item_from_value(raw)
if item:
items.append(item)
+ if len(items) > PLAYLIST_ITEM_MAX:
+ logger.bind(tag=TAG).warning(
+ f"playlist truncated incoming={len(items)} max={PLAYLIST_ITEM_MAX}"
+ )
+ items = items[:PLAYLIST_ITEM_MAX]
return items
item = playlist_item_from_value(
{
@@ -293,20 +541,20 @@
def default_repeater_playlist(task: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
- """No downlink yet: task.default_playlist, else the unit01 01脳10 / 02脳5 stub."""
+ """Only used when a caller still has an explicit default_playlist on the task."""
raw = (task or {}).get("default_playlist")
if isinstance(raw, list) and raw:
items = parse_playlist({"items": raw})
if items:
return items
- return [dict(item) for item in TEMP_REPEATER_PLAYLIST]
+ return []
def default_shadowing_playlist(task: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
- """No downlink yet: task.default_playlist, else the unit01 01脳5 / 02脳10 stub."""
+ """Only used when a caller still has an explicit default_playlist on the task."""
raw = (task or {}).get("default_playlist")
if isinstance(raw, list) and raw:
items = parse_playlist({"items": raw})
if items:
return items
- return [dict(item) for item in TEMP_SHADOWING_PLAYLIST]
+ return []
--
Gitblit v1.9.1