"""SD MP3 playback speed: clamp [0.5, 2.0], default 1.0. Gradients live on Server."""
|
|
from __future__ import annotations
|
|
import json
|
import os
|
import re
|
import time
|
from typing import Any, Dict, List, Optional
|
|
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()
|
|
PLAY_SPEED_MIN = 0.5
|
PLAY_SPEED_MAX = 2.0
|
PLAY_SPEED_DEFAULT = 1.0
|
|
# 语音口令未带 lesson 时，不再使用测试清单；必须先有家长下发的课文。
|
TEMP_REPEATER_PLAYLIST = (
|
{
|
"name": "/english_learning/repeater/lessons/l1/001/cn/01.mp3",
|
"repeat": 1,
|
"speed": 1.0,
|
},
|
{
|
"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": 1.0,
|
},
|
)
|
|
TEMP_SHADOWING_PLAYLIST = tuple(
|
{
|
"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:
|
try:
|
value = float(raw)
|
except (TypeError, ValueError):
|
return PLAY_SPEED_DEFAULT
|
if value != value: # NaN
|
return PLAY_SPEED_DEFAULT
|
if value < PLAY_SPEED_MIN:
|
return PLAY_SPEED_MIN
|
if value > PLAY_SPEED_MAX:
|
return PLAY_SPEED_MAX
|
return value
|
|
|
def playback_config(task: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
cfg = (task or {}).get("playback")
|
return cfg if isinstance(cfg, dict) else {}
|
|
|
def _progress_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", "playback_progress", f"{safe}.json")
|
|
|
def _load_progress(device_id: str) -> Dict[str, Any]:
|
path = _progress_path(device_id)
|
if not os.path.isfile(path):
|
return {}
|
try:
|
with open(path, "r", encoding="utf-8") as handle:
|
data = json.load(handle)
|
return data if isinstance(data, dict) else {}
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"读取播放进度失败: {exc}")
|
return {}
|
|
|
def _save_progress(device_id: str, data: Dict[str, Any]) -> None:
|
path = _progress_path(device_id)
|
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}")
|
|
|
_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:
|
return f"{app_id or ''}/{task_id or ''}"
|
|
|
def lesson_day_index(
|
device_id: str,
|
app_id: str,
|
task_id: str,
|
lesson: str = "",
|
) -> int:
|
"""0-based count of distinct calendar days this lesson has already been played."""
|
data = _load_progress(device_id)
|
bucket = data.get(_lesson_key(app_id, task_id)) or {}
|
lessons = bucket.get("lessons") if isinstance(bucket, dict) else {}
|
key = lesson or "_default"
|
days = lessons.get(key) if isinstance(lessons, dict) else None
|
if not isinstance(days, list):
|
return 0
|
return len([d for d in days if d])
|
|
|
def mark_lesson_played_today(
|
device_id: str,
|
app_id: str,
|
task_id: str,
|
lesson: str = "",
|
) -> None:
|
today = time.strftime("%Y-%m-%d")
|
data = _load_progress(device_id)
|
key = _lesson_key(app_id, task_id)
|
bucket = data.get(key)
|
if not isinstance(bucket, dict):
|
bucket = {}
|
lessons = bucket.get("lessons")
|
if not isinstance(lessons, dict):
|
lessons = {}
|
lesson_key = lesson or "_default"
|
days = lessons.get(lesson_key)
|
if not isinstance(days, list):
|
days = []
|
if today not in days:
|
days.append(today)
|
lessons[lesson_key] = days
|
bucket["lessons"] = lessons
|
data[key] = bucket
|
_save_progress(device_id, data)
|
|
|
def daily_base_speed(task: Optional[Dict[str, Any]], day_index: int = 0) -> float:
|
cfg = playback_config(task)
|
daily = cfg.get("daily_speeds")
|
if isinstance(daily, list) and daily:
|
idx = max(0, int(day_index or 0))
|
if idx >= len(daily):
|
idx = len(daily) - 1
|
return clamp_play_speed(daily[idx])
|
if cfg.get("speed") is not None:
|
return clamp_play_speed(cfg.get("speed"))
|
return PLAY_SPEED_DEFAULT
|
|
|
def speeds_for_repeats(
|
task: Optional[Dict[str, Any]],
|
repeat_n: int = 1,
|
day_index: int = 0,
|
) -> List[float]:
|
"""One speed per pass. repeat=1 uses daily curve; repeat>1 samples repeat_speeds if set."""
|
n = 1
|
try:
|
n = int(repeat_n or 1)
|
except (TypeError, ValueError):
|
n = 1
|
if n < 1:
|
n = 1
|
base = daily_base_speed(task, day_index)
|
curve = playback_config(task).get("repeat_speeds")
|
if n == 1 or not isinstance(curve, list) or not curve:
|
return [base] * n
|
values = [clamp_play_speed(item) for item in curve]
|
if n == len(values):
|
return values
|
if n == 1:
|
return [values[0]]
|
out: List[float] = []
|
last = len(values) - 1
|
for i in range(n):
|
pos = (i / (n - 1)) * last
|
lo = int(pos)
|
hi = min(lo + 1, last)
|
frac = pos - lo
|
out.append(clamp_play_speed(values[lo] * (1.0 - frac) + values[hi] * frac))
|
return out
|
|
|
def resolve_play_speed(
|
task: Optional[Dict[str, Any]],
|
day_index: int = 0,
|
explicit: Any = None,
|
) -> float:
|
if explicit is not None:
|
return clamp_play_speed(explicit)
|
return daily_base_speed(task, day_index)
|
|
|
def normalize_media_name(raw: Any) -> str:
|
"""SD files[].name, MQTT path, or HTTP url → /app/task/lessons/... form."""
|
name = str(raw or "").strip().replace("\\", "/")
|
if not name:
|
return ""
|
for marker in ("/english_learning/", "english_learning/"):
|
idx = name.find(marker)
|
if idx >= 0:
|
rest = name[idx:]
|
return rest if rest.startswith("/") else "/" + rest
|
if "://" in name:
|
path = name.split("://", 1)[1]
|
_, _, rest = path.partition("/")
|
if rest:
|
return "/" + rest.lstrip("/")
|
return ""
|
if not name.startswith("/"):
|
name = "/" + name.lstrip("/")
|
return name
|
|
|
def _item_repeat(raw: Dict[str, Any]) -> int:
|
for key in ("repeat", "count", "play_count"):
|
if raw.get(key) is None:
|
continue
|
try:
|
value = int(raw.get(key))
|
except (TypeError, ValueError):
|
continue
|
return max(1, value)
|
return 1
|
|
|
def playlist_item_from_value(raw: Any) -> Optional[Dict[str, Any]]:
|
if isinstance(raw, str):
|
name = normalize_media_name(raw)
|
if not name:
|
return None
|
return {"name": name, "repeat": 1}
|
if not isinstance(raw, dict):
|
return None
|
name = normalize_media_name(
|
raw.get("name")
|
or raw.get("path")
|
or raw.get("target_path")
|
or raw.get("url")
|
or ""
|
)
|
if not name:
|
return None
|
item: Dict[str, Any] = {"name": name, "repeat": _item_repeat(raw)}
|
if raw.get("speed") is not None:
|
item["speed"] = clamp_play_speed(raw.get("speed"))
|
if raw.get("power_off_after") is not None:
|
item["power_off_after"] = bool(raw.get("power_off_after"))
|
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 {}
|
raw_items = params.get("items") or params.get("playlist") or params.get("clips")
|
if isinstance(raw_items, list) and raw_items:
|
items = []
|
for raw in raw_items:
|
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(
|
{
|
"name": params.get("name") or params.get("path") or params.get("url") or "",
|
"repeat": params.get("repeat") or params.get("count") or 1,
|
"speed": params.get("speed"),
|
"power_off_after": params.get("power_off_after"),
|
}
|
)
|
if item and item["name"]:
|
return [item]
|
return []
|
|
|
def expand_playlist_plays(items: Optional[List[Dict[str, Any]]]) -> List[str]:
|
"""Flatten {name, repeat} into one path per play. Shadowing uses this so each pass has its own listen window."""
|
clips: List[str] = []
|
for item in items or []:
|
if not isinstance(item, dict):
|
continue
|
name = str(item.get("name") or "")
|
if not name:
|
continue
|
clips.extend([name] * _item_repeat(item))
|
return clips
|
|
|
def default_repeater_playlist(task: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
|
"""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 []
|
|
|
def default_shadowing_playlist(task: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
|
"""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 []
|
