5种场景模式家长端ok

wangzhibo

10 天以前 220e4fab27b7f419b3e89e2f2d28ae789aac09d7

 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 []