# -*- coding: utf-8 -*-
"""edge-tts synthesis with content-hash caching (edit a line -> only that
line re-synthesises)."""
import asyncio
import hashlib
from pathlib import Path

from .ffbin import probe_duration

VOICES = [
    "zh-TW-YunJheNeural",
    "zh-TW-HsiaoChenNeural",
    "zh-TW-HsiaoYuNeural",
]


MIN_VALID_BYTES = 800   # a real line is >13 KB; a failed stream leaves 0

# Words edge-tts reads with the wrong homograph. Applied ONLY to the text
# handed to the synthesiser — the subtitle keeps the correct spelling, so
# viewers read 心率 while hearing the right 「ㄌㄩˋ」 rather than 「ㄕㄨㄞˋ」.
# 率 carries both readings and the engine picks shuài inside 心率.
SAY_AS = {
    "心率": "心律",
}


def for_speech(text: str) -> str:
    for wrong, right in SAY_AS.items():
        text = text.replace(wrong, right)
    return text


def _path(tts_dir: Path, i: int, spoken: str, voice: str, rate: str) -> Path:
    # Key on what is actually SPOKEN, not what is written: otherwise adding a
    # SAY_AS entry leaves every previously cached clip in place, still saying
    # the wrong thing, because the written text never changed.
    h = hashlib.md5(
        f"{for_speech(spoken)}|{voice}|{rate}".encode()).hexdigest()[:10]
    return tts_dir / f"n{i:02d}_{h}.mp3"


def _is_cached(p: Path) -> bool:
    """Presence alone is not proof of a usable file: an interrupted or
    service-failed synthesis can leave a 0-byte mp3 behind, which the cache
    then happily skips forever — the render only blows up much later, when
    ffprobe refuses to read it. Size-check the cache entry instead."""
    try:
        return p.is_file() and p.stat().st_size >= MIN_VALID_BYTES
    except OSError:
        return False


RETRIES = 4


async def _synth_one(edge_tts, text, voice, rate, dst, log):
    """edge-tts streams from a Microsoft cloud endpoint, which intermittently
    returns an empty stream (NoAudioReceived) even for input that succeeds on
    the very next attempt — verified by running an identical call repeatedly.
    A whole render shouldn't die because one request out of thirty blipped."""
    for attempt in range(1, RETRIES + 1):
        try:
            await edge_tts.Communicate(text, voice, rate=rate,
                                       pitch="+3Hz").save(str(dst))
            if _is_cached(dst):
                return
            raise RuntimeError("TTS 產生了空檔案或過小的檔案")
        except Exception as e:  # noqa: BLE001
            dst.unlink(missing_ok=True)
            if attempt == RETRIES:
                raise RuntimeError(
                    f"語音合成連續失敗 {RETRIES} 次（{type(e).__name__}）："
                    f"「{text[:20]}」。通常是網路或微軟語音服務暫時異常，"
                    f"稍後再試一次即可。") from e
            wait = 1.5 * attempt
            log(f"    ↻ 第 {attempt} 次失敗（{type(e).__name__}），"
                f"{wait:.1f} 秒後重試")
            await asyncio.sleep(wait)


async def _synth_all(lines, tts_dir: Path, voice: str, rate: str, log,
                     on_progress=None, should_cancel=None):
    import edge_tts
    from .ffbin import JobCancelled
    todo = [(i, ln) for i, ln in enumerate(lines)
            if not _is_cached(_path(tts_dir, i, ln["spoken"], voice, rate))]
    for n, (i, ln) in enumerate(todo, start=1):
        if should_cancel and should_cancel():
            raise JobCancelled("使用者中止語音合成")
        dst = _path(tts_dir, i, ln["spoken"], voice, rate)
        await _synth_one(edge_tts, for_speech(ln["spoken"]), voice, rate, dst, log)
        log(f"  tts {i:02d} @ {ln['start']:.1f}s  {ln['spoken'][:24]}")
        if on_progress:
            on_progress(n, len(todo))


MIN_GAP = 0.2   # seconds of silence enforced between two narration clips


LEAD = 0.5      # seconds every narration line starts EARLY (see synthesize)


def synthesize(lines: list, tts_dir: str, voice: str = VOICES[0],
               rate: str = "+18%", log=print, on_progress=None,
               should_cancel=None, lead: float = LEAD) -> list:
    """Returns lines enriched with file + dur.

    Two timing corrections happen here, in order.

    1. `lead` — start every line EARLY by this many seconds. A line timed at
       t=X describes what is on screen at t=X, but speech is not
       instantaneous: start speaking at exactly X and the words identifying
       the moment ("他從左邊超過去！") land a beat or two later, by which
       point the rider is gone and the narration reads as describing the
       wrong shot. Beginning slightly before the moment puts the meaningful
       part of the sentence over the moment itself. This is the same reason
       broadcast subtitling uses a lead-in.

    2. Overlap compaction — a line's spoken length isn't known until after
       synthesis, so the "start" times scriptgen picked are estimates. When
       a clip still plays at the next line's intended start, both get mixed
       together (audible overlap) and the narration keeps describing the
       previous scene. Nudging the next start forward just enough fixes it.

    Both act on "start", and render.build_ass reads that same field for the
    subtitle, so audio and subtitle always shift together and never desync.
    """
    d = Path(tts_dir)
    d.mkdir(parents=True, exist_ok=True)
    asyncio.run(_synth_all(lines, d, voice, rate, log, on_progress,
                           should_cancel))
    out = []
    for i, ln in enumerate(lines):
        p = _path(d, i, ln["spoken"], voice, rate)
        out.append({**ln, "file": str(p), "dur": probe_duration(str(p))})

    if lead:
        # clamp at 0 — nothing can start before the video does
        for o in out:
            o["start"] = max(0.0, round(o["start"] - lead, 2))
        log(f"  旁白整體提前 {lead:.2f}s（讓描述落在該畫面上）")

    cursor = -1.0
    for i, o in enumerate(out):
        if o["start"] < cursor:
            drift = cursor - o["start"]
            log(f"  ! 第{i}句語音較長，往後延 {drift:.1f}s 避免跟上一句重疊"
                f"（{o['start']:.1f}s → {cursor:.1f}s）")
            o["start"] = round(cursor, 2)
        cursor = o["start"] + o["dur"] + MIN_GAP
    return out


def synth_line(text: str, tts_dir: str, voice: str = VOICES[0],
               rate: str = "+18%") -> str:
    """Synthesise a single line for auditioning and return its path.

    Slot 99 is reserved for previews so they never collide with the batch's
    per-line n00..n98 files. That does mean a previewed line is synthesised
    again at render time (the batch keys on its own index) — the cost is one
    short clip, and keeping the two caches apart avoids a preview of edited
    text masquerading as the rendered line.
    """
    d = Path(tts_dir)
    d.mkdir(parents=True, exist_ok=True)
    dst = _path(d, 99, text, voice, rate)
    if _is_cached(dst):
        return str(dst)
    import edge_tts
    asyncio.run(_synth_one(edge_tts, for_speech(text), voice, rate, dst,
                           log=lambda *a: None))
    return str(dst)


# Hand-written notes for the voices worth describing; anything not listed
# falls back to Microsoft's own FriendlyName.
VOICE_NOTES = {
    "zh-TW-YunJheNeural":       "台灣・男聲・沉穩有力，適合熱血轉播",
    "zh-TW-HsiaoChenNeural":    "台灣・女聲・清亮標準，像新聞主播",
    "zh-TW-HsiaoYuNeural":      "台灣・女聲・溫柔親切",
    "zh-CN-YunjianNeural":      "大陸・男聲・體育賽事風格，語速快有力",
    "zh-CN-YunxiNeural":        "大陸・男聲・年輕活潑",
    "zh-CN-YunyangNeural":      "大陸・男聲・專業播報",
    "zh-CN-YunxiaNeural":       "大陸・男聲・偏少年音",
    "zh-CN-XiaoxiaoNeural":     "大陸・女聲・自然口語",
    "zh-CN-XiaoyiNeural":       "大陸・女聲・活潑",
    "zh-CN-liaoning-XiaobeiNeural": "大陸・女聲・東北腔",
    "zh-CN-shaanxi-XiaoniNeural":   "大陸・女聲・陝西腔",
    "zh-HK-WanLungNeural":      "香港・男聲・粵語",
    "zh-HK-HiuMaanNeural":      "香港・女聲・粵語",
    "zh-HK-HiuGaaiNeural":      "香港・女聲・粵語",
}


def list_chinese_voices() -> list:
    """All zh-* voices the installed edge-tts can reach, TW first."""
    import edge_tts
    voices = asyncio.run(edge_tts.list_voices())
    out = []
    for v in voices:
        if not v["Locale"].startswith("zh"):
            continue
        out.append({"short": v["ShortName"], "locale": v["Locale"],
                    "gender": v["Gender"],
                    "label": VOICE_NOTES.get(v["ShortName"],
                                             v.get("FriendlyName", v["ShortName"]))})
    order = {"zh-TW": 0, "zh-CN": 1}
    out.sort(key=lambda v: (order.get(v["locale"], 2), v["short"]))
    return out
