"""Audiobook ingest: raw-cache transcoding must terminate and clean up."""
from __future__ import annotations

import asyncio
import sys
import tempfile
import threading
import unittest
from collections import namedtuple
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, PropertyMock, patch

import numpy as np

SRC_SR = 44100
SRC_SECONDS = 25  # two full 10 s blocks + one partial block
TARGET_SR = 22050

_DiskUsage = namedtuple("_DiskUsage", "total used free")


class TranscodeToRawTests(unittest.TestCase):
    def setUp(self) -> None:
        try:
            import soundfile as sf
        except ImportError:
            self.skipTest("soundfile not installed")
        from javis.voices import audiobook_ingest
        self.sf = sf
        self.ingest = audiobook_ingest
        self._tmp = tempfile.TemporaryDirectory()
        self.root = Path(self._tmp.name)
        self.source = self.root / "book.wav"
        t = np.arange(SRC_SR * SRC_SECONDS) / SRC_SR
        tone = 0.3 * np.sin(2 * np.pi * 220 * t)
        sf.write(str(self.source), np.stack([tone, tone], axis=1), SRC_SR, subtype="PCM_16")
        self.target = self.root / "cache" / "ingest_audio_cache.i16"

    def tearDown(self) -> None:
        self._tmp.cleanup()

    def _bogus_frame_count(self):
        """Mimic an MP3 with a malformed ID3 tag: INT64_MAX frames, unseekable."""
        frames = patch.object(self.sf.SoundFile, "frames", new_callable=PropertyMock,
                              return_value=sys.maxsize)
        seekable = patch.object(self.sf.SoundFile, "seekable", return_value=False)
        return frames, seekable

    def _run_with_timeout(self, timeout: float = 60.0, **kw) -> BaseException | None:
        errors: list[BaseException] = []

        def run():
            try:
                self.ingest._transcode_to_raw(self.source, self.target, TARGET_SR, **kw)
            except BaseException as e:  # noqa: BLE001
                errors.append(e)

        th = threading.Thread(target=run, daemon=True)
        th.start()
        th.join(timeout)
        self.assertFalse(th.is_alive(), "transcode did not stop at EOF (runaway)")
        return errors[0] if errors else None

    def test_unknown_frame_count_stops_at_eof(self) -> None:
        frames, seekable = self._bogus_frame_count()
        with frames, seekable:
            self.assertGreaterEqual(self.sf.info(str(self.source)).frames, sys.maxsize // 2)
            err = self._run_with_timeout()
        self.assertIsNone(err)
        self.assertEqual(self.target.stat().st_size, SRC_SECONDS * TARGET_SR * 2)
        self.assertFalse(self.target.with_suffix(".part").exists())

    def test_stop_request_aborts_and_removes_partial_file(self) -> None:
        err = self._run_with_timeout(should_stop=lambda: True)
        self.assertIsInstance(err, self.ingest.TranscodeCancelled)
        self.assertFalse(self.target.exists())
        self.assertFalse(self.target.with_suffix(".part").exists())

    def test_disk_running_full_aborts_and_removes_partial_file(self) -> None:
        frames, seekable = self._bogus_frame_count()
        full = patch.object(self.ingest.shutil, "disk_usage",
                            return_value=_DiskUsage(10**12, 10**12, 0))
        every_block = patch.object(self.ingest, "_DISK_CHECK_EVERY_BLOCKS", 1)
        with frames, seekable, full, every_block:
            err = self._run_with_timeout()
        self.assertIsInstance(err, RuntimeError)
        self.assertIn("free space", str(err))
        self.assertFalse(self.target.exists())
        self.assertFalse(self.target.with_suffix(".part").exists())

    def test_insufficient_space_for_known_length_fails_fast(self) -> None:
        with patch.object(self.ingest.shutil, "disk_usage",
                          return_value=_DiskUsage(10**12, 10**12, 0)):
            err = self._run_with_timeout()
        self.assertIsInstance(err, RuntimeError)
        self.assertIn("Not enough disk space", str(err))
        self.assertFalse(self.target.with_suffix(".part").exists())

    def test_stop_button_during_transcode_ends_ingest_as_stopped(self) -> None:
        events: list[dict] = []

        class Bus:
            async def publish(self, topic, payload):
                events.append(payload)

        mgr = MagicMock()
        mgr.root = self.root
        mgr.create.return_value = SimpleNamespace(dir=self.root / "voice")
        stop = asyncio.Event()
        stop.set()
        with patch("javis.voices.manager.VoiceManager", return_value=mgr):
            stats = asyncio.run(self.ingest.ingest_audiobook(
                None, self.source, "voice", bus=Bus(), stop_event=stop, resume=False))
        self.assertEqual(stats.total_segments, 0)
        self.assertEqual(events[-1]["phase"], "stopped")
        self.assertEqual(events[-1]["during"], "transcoding")
        self.assertFalse((self.root / "voice" / "ingest_audio_cache.i16").exists())
        self.assertFalse((self.root / "voice" / "ingest_audio_cache.part").exists())


if __name__ == "__main__":
    unittest.main()