"""Dictation without a mic or a server: a fake PortAudio stream we push
frames into, and a fake /desk/stt socket that answers like Deepgram."""

import array
import json
import os
import queue
import sys
import threading
import time
from pathlib import Path

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pytest

from bolt_console import dictation


def frame(level: int, samples: int = 1280) -> bytes:
    return array.array("h", [level, -level] * (samples // 2)).tobytes()


def result(text, final):
    return json.dumps({"type": "Results", "is_final": final,
                       "channel": {"alternatives": [{"transcript": text}]}})


class FakeSocket:
    """Answers each audio chunk with the next scripted message; on CloseStream
    sends whatever is left, then closes."""

    def __init__(self, replies=()):
        self.replies = list(replies)
        self.audio = []
        self.closed_stream = False
        self._inbox = queue.Queue()

    def send_binary(self, data):
        self.audio.append(data)
        if self.replies and len(self.audio) % 2 == 0:
            self._inbox.put(self.replies.pop(0))

    def send(self, text):
        if json.loads(text).get("type") == "CloseStream":
            self.closed_stream = True
            for reply in self.replies:
                self._inbox.put(reply)
            self._inbox.put("")

    def recv(self):
        return self._inbox.get(timeout=5)

    def close(self):
        self._inbox.put("")


class FakeStream:
    def __init__(self):
        self.callback = None
        self.stopped = False

    def open(self, callback, _rate):
        self.callback = callback
        return self

    def push(self, *chunks):
        for chunk in chunks:
            self.callback(chunk)

    def stop(self):
        self.stopped = True

    def close(self):
        pass


class Recorder:
    def __init__(self):
        self.updates, self.levels, self.done = [], [], threading.Event()
        self.result = None

    def kwargs(self):
        return dict(on_update=lambda f, i: self.updates.append((f, i)),
                    on_level=self.levels.append,
                    on_done=lambda text, err: (setattr(self, "result", (text, err)), self.done.set()))


def make(sockets, stream, rec, **extra):
    sockets = list(sockets)

    def connect(_rate):
        item = sockets.pop(0)
        if isinstance(item, Exception):
            raise item
        return item

    return dictation.Dictation(**rec.kwargs(), connect_fn=connect, open_stream_fn=stream.open, **extra)


def test_live_words_arrive_while_talking_and_stop_returns_the_text():
    sock = FakeSocket([result("open the", False), result("open the readme", True),
                       result("please", True)])
    stream, rec = FakeStream(), Recorder()
    d = make([sock], stream, rec)
    d.start()
    assert d.live
    stream.push(*[frame(3000)] * 4)
    deadline = time.monotonic() + 3
    while len(rec.updates) < 2 and time.monotonic() < deadline:
        time.sleep(0.01)
    assert rec.updates[0] == ("", "open the")
    assert rec.updates[1] == ("open the readme", "")
    d.stop()
    assert rec.done.wait(5)
    assert rec.result == ("open the readme please", None)
    assert sock.closed_stream and stream.stopped
    assert max(rec.levels) > 0 and rec.levels[-1] == 0.0


def test_no_live_socket_means_the_whole_recording_is_sent_on_stop():
    late = FakeSocket([result("hello bolt", True)])
    stream, rec = FakeStream(), Recorder()
    d = make([ConnectionError("down"), late], stream, rec)
    d.start()
    assert not d.live
    stream.push(frame(3000), frame(3000), frame(0))
    time.sleep(0.3)
    d.stop()
    assert rec.done.wait(5)
    assert rec.result == ("hello bolt", None)
    assert b"".join(late.audio) == frame(3000) * 2 + frame(0)


def test_a_socket_dying_mid_recording_does_not_duplicate_words():
    class Dying(FakeSocket):
        def send_binary(self, data):
            if len(self.audio) >= 1:
                raise OSError("connection reset")
            super().send_binary(data)
            self._inbox.put(result("half", True))

    retry = FakeSocket([result("the whole thing", True)])
    stream, rec = FakeStream(), Recorder()
    d = make([Dying(), retry], stream, rec)
    d.start()
    stream.push(frame(3000), frame(3000), frame(3000))
    time.sleep(0.3)
    d.stop()
    assert rec.done.wait(5)
    assert rec.result == ("the whole thing", None)  # not "half the whole thing"


def test_pausing_after_speech_stops_on_its_own():
    sock = FakeSocket([result("done talking", True)])
    stream, rec = FakeStream(), Recorder()
    stopping = threading.Event()
    d = make([sock], stream, rec, silence_seconds=0.3, on_stopping=stopping.set)
    d.start()
    stream.push(frame(3000), frame(3000))
    for _ in range(8):
        stream.push(frame(10))
        time.sleep(0.05)
    assert rec.done.wait(5)  # nobody called stop()
    assert stopping.is_set()
    assert rec.result == ("done talking", None)


def test_silence_alone_never_auto_stops():
    stream, rec = FakeStream(), Recorder()
    d = make([FakeSocket()], stream, rec, silence_seconds=0.2)
    d.start()
    for _ in range(8):
        stream.push(frame(10))
        time.sleep(0.05)
    assert not rec.done.is_set()  # still waiting for you to start talking
    d.stop()
    assert rec.done.wait(5)
    assert rec.result[0] == "" and "Didn't catch anything" in rec.result[1]


def test_cancel_throws_the_result_away_without_a_fallback_upload():
    stream, rec = FakeStream(), Recorder()
    d = make([ConnectionError("down")], stream, rec)  # a fallback would need a 2nd socket
    d.start()
    stream.push(frame(3000))
    d.cancel()
    assert rec.done.wait(5)
    assert rec.result == ("", None)


def test_mic_failure_raises_from_start_and_closes_the_socket():
    sock = FakeSocket()
    rec = Recorder()

    def broken(_cb, _rate):
        raise OSError("no input device")

    d = dictation.Dictation(**rec.kwargs(), connect_fn=lambda _r: sock, open_stream_fn=broken)
    with pytest.raises(OSError):
        d.start()
    assert sock._inbox.get(timeout=1) == ""  # closed


def test_parse_result_is_tolerant():
    assert dictation.parse_result(result("hi", True)) == ("hi", True)
    assert dictation.parse_result(json.dumps({"type": "Metadata"})) == ("", False)
    assert dictation.parse_result(b"not json") == ("", False)
    assert dictation.parse_result(json.dumps({"type": "Results", "speech_final": True,
                                              "channel": {"alternatives": [{"transcript": "x"}]}})) == ("x", True)


def test_connect_url_carries_the_session_and_audio_format(monkeypatch):
    import types

    seen = {}
    fake_ws = types.SimpleNamespace(create_connection=lambda url, header, timeout: seen.update(url=url, header=header))
    monkeypatch.setitem(sys.modules, "websocket", fake_ws)
    monkeypatch.setattr(dictation.config, "SERVER_URL", "https://bolt.example")
    monkeypatch.setattr(dictation.config, "API_KEY", "k")
    monkeypatch.setattr(dictation.config, "SESSION_ID", "pet-console-x y")
    dictation.connect()
    assert seen["url"].startswith("wss://bolt.example/desk/stt?")
    assert "session_id=pet-console-x+y" in seen["url"] and "sample_rate=16000" in seen["url"]
    assert seen["header"] == ["X-Desk-Api-Key: k"]


# ── the window ──────────────────────────────────────────────────────────────

class ScriptedDictation:
    """Stands in for Dictation in the window: the test drives its callbacks."""

    last = None

    def __init__(self, on_update, on_level, on_done, silence_seconds=3.0, on_stopping=None):
        self.on_update, self.on_level, self.on_done = on_update, on_level, on_done
        self.stopped = self.cancelled = False
        self.live = True
        ScriptedDictation.last = self

    def start(self):
        pass

    def stop(self):
        self.stopped = True

    def cancel(self):
        self.cancelled = True


def test_window_mic_fills_the_composer_and_blocks_send_until_done(monkeypatch):
    from bolt_console import config
    from bolt_console.gui.window import MainWindow, make_app

    monkeypatch.setattr(config, "SERVER_URL", "http://127.0.0.1:9")
    monkeypatch.setattr(config, "API_KEY", "k")
    monkeypatch.setattr(dictation, "unavailable_reason", lambda: None)
    app = make_app(["test"])
    window = MainWindow(dictation_factory=ScriptedDictation)
    try:
        window.composer.edit.setPlainText("Please")
        window.toggle_dictation()
        rec = ScriptedDictation.last
        assert window.composer.mic.state == "recording"
        rec.on_update("open the readme", "and")
        app.processEvents()
        assert window.composer.edit.toPlainText() == "Please open the readme and"
        assert window.composer.edit.isReadOnly() and not window.composer.send.isEnabled()

        window.toggle_dictation()  # click again: stop
        assert rec.stopped and window.composer.mic.state == "finishing"
        rec.on_done("open the readme and summarise it", None)
        app.processEvents()
        assert window.composer.edit.toPlainText() == "Please open the readme and summarise it"
        assert not window.composer.edit.isReadOnly() and window.composer.send.isEnabled()
        assert window.composer.mic.state == "idle"
    finally:
        window.close()


def test_window_shows_why_the_mic_is_unavailable(monkeypatch):
    from bolt_console.gui.window import MainWindow, make_app

    monkeypatch.setattr(dictation, "unavailable_reason", lambda: "Dictation needs sounddevice")
    make_app(["test"])
    window = MainWindow(dictation_factory=ScriptedDictation)
    try:
        ScriptedDictation.last = None
        window.toggle_dictation()
        assert ScriptedDictation.last is None
        assert window.composer.mic.state == "unavailable"
        assert "sounddevice" in window.hint.text()
    finally:
        window.close()