"""Platform lesson MP3s live under xiaozhi-server, keyed by org. Device paths have no org."""
|
|
from __future__ import annotations
|
|
import os
|
import re
|
from typing import Any, Dict, Iterable, List, Optional
|
|
DEFAULT_ORG_ID = "default_org"
|
LESSON_AUDIO_DIRNAME = "lesson_audio"
|
ALLOWED_PREFIX = "english_learning/"
|
_ORG_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
|
_REL_RE = re.compile(
|
r"^english_learning/(repeater|shadowing)/lessons/[A-Za-z0-9][A-Za-z0-9_./-]*\.mp3$"
|
)
|
|
|
def resolve_org_id(org_id: Optional[str]) -> str:
|
value = str(org_id or "").strip()
|
if not value:
|
return DEFAULT_ORG_ID
|
if not _ORG_RE.fullmatch(value):
|
return DEFAULT_ORG_ID
|
return value
|
|
|
def normalize_sd_name(name: str) -> str:
|
text = str(name or "").replace("\\", "/").strip()
|
if not text:
|
return ""
|
if not text.startswith("/"):
|
text = "/" + text
|
return text
|
|
|
def unique_lesson_names(items: Iterable[Any]) -> List[str]:
|
seen = set()
|
names: List[str] = []
|
for raw in items or []:
|
if isinstance(raw, dict):
|
name = normalize_sd_name(str(raw.get("name") or raw.get("path") or raw.get("url") or ""))
|
else:
|
name = normalize_sd_name(str(raw or ""))
|
if not name or name in seen:
|
continue
|
seen.add(name)
|
names.append(name)
|
return names
|
|
|
def missing_lesson_names(needed: Iterable[str], present: Iterable[str]) -> List[str]:
|
have = {normalize_sd_name(name) for name in present or []}
|
have.discard("")
|
return [name for name in unique_lesson_names(needed) if name not in have]
|
|
|
def lesson_audio_root(config: Optional[Dict[str, Any]] = None) -> str:
|
from config.config_loader import get_project_dir
|
|
data_dir = "data"
|
if isinstance(config, dict):
|
log_cfg = config.get("log") if isinstance(config.get("log"), dict) else {}
|
data_dir = str(log_cfg.get("data_dir") or data_dir)
|
root = data_dir if os.path.isabs(data_dir) else os.path.join(get_project_dir(), data_dir)
|
return os.path.join(root, LESSON_AUDIO_DIRNAME)
|
|
|
def org_audio_dir(config: Optional[Dict[str, Any]], org_id: Optional[str]) -> str:
|
return os.path.join(lesson_audio_root(config), resolve_org_id(org_id))
|
|
|
def sanitize_relative_path(rel: str) -> Optional[str]:
|
text = str(rel or "").replace("\\", "/").lstrip("/")
|
if not text or ".." in text.split("/"):
|
return None
|
if not _REL_RE.fullmatch(text):
|
return None
|
return text
|
|
|
def resolve_disk_file(
|
config: Optional[Dict[str, Any]], org_id: Optional[str], rel: str
|
) -> Optional[str]:
|
relative = sanitize_relative_path(rel)
|
if not relative:
|
return None
|
org = resolve_org_id(org_id)
|
primary = os.path.normpath(os.path.join(org_audio_dir(config, org), relative))
|
root = os.path.normpath(lesson_audio_root(config))
|
if not primary.startswith(root + os.sep) and primary != root:
|
return None
|
if os.path.isfile(primary):
|
return primary
|
if org != DEFAULT_ORG_ID:
|
fallback = os.path.normpath(os.path.join(org_audio_dir(config, DEFAULT_ORG_ID), relative))
|
if fallback.startswith(root + os.sep) and os.path.isfile(fallback):
|
return fallback
|
return None
|
|
|
def lesson_audio_public_root(config: Optional[Dict[str, Any]]) -> str:
|
server = (config or {}).get("server") if isinstance(config, dict) else {}
|
if not isinstance(server, dict):
|
server = {}
|
raw = str(server.get("lesson_audio_public_url") or "").strip()
|
port = int(server.get("http_port", 8003) or 8003)
|
if not raw or "你" in raw:
|
raw = f"http://{_local_ip()}:{port}/lesson-audio"
|
return raw.rstrip("/")
|
|
|
def _local_ip() -> str:
|
import socket
|
|
try:
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
sock.connect(("8.8.8.8", 80))
|
ip = sock.getsockname()[0]
|
sock.close()
|
return ip
|
except Exception:
|
return "127.0.0.1"
|
|
|
def lesson_audio_base_url(config: Optional[Dict[str, Any]], org_id: Optional[str]) -> str:
|
return f"{lesson_audio_public_root(config)}/{resolve_org_id(org_id)}"
|
|
|
_CLIP_RE = re.compile(r"^(\d+)\.mp3$", re.I)
|
|
|
def list_lesson_clips(
|
config: Optional[Dict[str, Any]],
|
org_id: Optional[str],
|
kind: str,
|
lesson_id: str,
|
lang: str,
|
) -> Optional[List[str]]:
|
"""SD paths for one lesson folder.
|
|
None means the folder is not on disk yet (caller may use the catalog).
|
A list, including empty, means the folder exists and those MP3s are the lesson.
|
"""
|
level, story = _split_lesson_id(lesson_id)
|
kind_name = str(kind or "").strip().lower()
|
language = str(lang or "").strip().lower()
|
if (
|
not level
|
or not story
|
or kind_name not in ("repeater", "shadowing")
|
or language not in ("cn", "en")
|
):
|
return None
|
rel_dir = f"english_learning/{kind_name}/lessons/{level}/{story}/{language}"
|
folder = _existing_lesson_dir(config, org_id, rel_dir)
|
if folder is None:
|
return None
|
numbered = []
|
try:
|
for entry in os.listdir(folder):
|
match = _CLIP_RE.match(entry)
|
if match:
|
numbered.append((int(match.group(1)), entry))
|
except OSError:
|
return None
|
numbered.sort()
|
return [f"/{rel_dir}/{name}" for _, name in numbered]
|
|
|
def _split_lesson_id(lesson_id: str) -> tuple:
|
from core.app_runtime.idiom_catalog import parse_lesson_id
|
|
return parse_lesson_id(lesson_id)
|
|
|
def _existing_lesson_dir(
|
config: Optional[Dict[str, Any]], org_id: Optional[str], rel_dir: str
|
) -> Optional[str]:
|
org = resolve_org_id(org_id)
|
primary = os.path.join(org_audio_dir(config, org), *rel_dir.split("/"))
|
if os.path.isdir(primary):
|
return primary
|
if org != DEFAULT_ORG_ID:
|
fallback = os.path.join(org_audio_dir(config, DEFAULT_ORG_ID), *rel_dir.split("/"))
|
if os.path.isdir(fallback):
|
return fallback
|
return None
|
|
|
def course_id_for(task_id: str, lesson_id: str) -> str:
|
lesson = re.sub(r"[^A-Za-z0-9_-]+", "-", str(lesson_id or "l1.001").strip()) or "l1-001"
|
kind = "rpt" if str(task_id or "").strip() == "repeater" else "shd"
|
course = f"{lesson}-{kind}"
|
return course[:31]
|
|
|
def conn_org_id(conn: Any, params: Optional[Dict[str, Any]] = None) -> str:
|
params = params or {}
|
raw = params.get("org_id") if params.get("org_id") is not None else params.get("orgId")
|
if raw:
|
return resolve_org_id(str(raw))
|
cfg = getattr(conn, "config", None) if conn is not None else None
|
runtime = cfg.get("app_runtime") if isinstance(cfg, dict) else None
|
if isinstance(runtime, dict) and runtime.get("lesson_org_id") is not None:
|
return resolve_org_id(str(runtime.get("lesson_org_id")))
|
return DEFAULT_ORG_ID
|
|
|
def ensure_org_dirs(config: Optional[Dict[str, Any]] = None) -> None:
|
os.makedirs(org_audio_dir(config, DEFAULT_ORG_ID), exist_ok=True)
|
