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