from concurrent.futures import ThreadPoolExecutor
from io import BytesIO
from pathlib import Path
import struct
import sqlite3
import sys
import threading
from types import SimpleNamespace
import uuid
import wave

from fastapi import HTTPException
from fastapi.testclient import TestClient
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import service


def wav():
    output = BytesIO()
    with wave.open(output, "wb") as audio:
        audio.setnchannels(1)
        audio.setsampwidth(2)
        audio.setframerate(16000)
        audio.writeframes(struct.pack("<h", 100) * 1600)
    return output.getvalue()


def upload(jobs, key=None, data=None, language="de"):
    return jobs.upload(BytesIO(wav() if data is None else data), "call.wav",
        key or "swyxweb:" + str(uuid.uuid4()), language, "meeting")


def test_job_and_audio_survive_restart_and_complete_without_summary(tmp_path):
    calls = []

    def transcribe(path, language):
        calls.append((path.read_bytes(), language))
        return "Bitte morgen zurückrufen."

    jobs = service.Jobs(tmp_path, transcribe)
    job = upload(jobs)
    assert job["status"] == "queued"
    # Simulate interruption after acknowledgement and during inference.
    with jobs.db() as db:
        db.execute("UPDATE jobs SET status='transcribing' WHERE id=?", (job["id"],))
    restarted = service.Jobs(tmp_path, transcribe)
    assert restarted.get(job["id"])["status"] == "queued"
    assert restarted.process_next()
    result = restarted.get(job["id"])
    assert result["status"] == "done"
    assert result["transcript"] == "Bitte morgen zurückrufen."
    assert result["summary"] == ""
    assert calls == [(wav(), "de")]
    assert not restarted.process_next()
    assert service.Jobs(tmp_path, transcribe).get(job["id"]) == result


def test_concurrent_duplicate_uploads_create_one_durable_job(tmp_path):
    jobs = service.Jobs(tmp_path, lambda *_: "Text")
    key = "swyxweb:" + str(uuid.uuid4())
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(lambda _: upload(jobs, key), range(8)))
    assert len({job["id"] for job in results}) == 1
    assert len(list(tmp_path.glob("*.wav"))) == 1
    with pytest.raises(HTTPException) as conflict:
        upload(jobs, key, language="en")
    assert conflict.value.status_code == 409
    with pytest.raises(HTTPException) as conflict:
        upload(jobs, key, data=wav() + b"changed")
    assert conflict.value.status_code == 409
    assert len(list(tmp_path.glob("*.wav"))) == 1


def test_failed_job_can_be_retried_without_reuploading(tmp_path):
    def failure(*_):
        raise ValueError("Unreadable audio")

    jobs = service.Jobs(tmp_path, failure)
    job = upload(jobs)
    assert jobs.process_next()
    assert jobs.get(job["id"])["status"] == "failed"
    assert jobs.get(job["id"])["error"]
    jobs.transcribe = lambda *_: "Erneut transkribiert."
    assert jobs.retry(job["id"])["status"] == "queued"
    with pytest.raises(HTTPException) as conflict:
        jobs.retry(job["id"])
    assert conflict.value.status_code == 409
    jobs.process_next()
    assert jobs.get(job["id"])["transcript"] == "Erneut transkribiert."
    assert jobs.get(job["id"])["error"] == ""


@pytest.mark.parametrize("data", [b"", b"not a WAV recording"])
def test_invalid_upload_does_not_leave_files_or_jobs(tmp_path, data):
    jobs = service.Jobs(tmp_path, lambda *_: "")
    with pytest.raises(HTTPException) as error:
        upload(jobs, data=data)
    assert error.value.status_code == 400
    assert not list(tmp_path.glob("*.wav"))
    assert not jobs.process_next()


def test_size_limit_cleans_up_partial_file(tmp_path, monkeypatch):
    monkeypatch.setattr(service, "MAX_BYTES", 100)
    jobs = service.Jobs(tmp_path, lambda *_: "")
    with pytest.raises(HTTPException) as error:
        upload(jobs)
    assert error.value.status_code == 413
    assert not list(tmp_path.glob("*.wav"))


def test_http_contract_remains_responsive_during_transcription(tmp_path):
    started, release = threading.Event(), threading.Event()

    def transcribe(*_):
        started.set()
        assert release.wait(5)
        return "Lokales Transkript"

    with TestClient(service.create_app(tmp_path, transcribe), base_url="http://127.0.0.1") as client:
        try:
            response = client.post("/api/jobs", files={"file": ("call.wav", wav(), "audio/wav")},
                data={"language": "de", "style": "meeting"}, headers={"Idempotency-Key": "swyxweb:" + str(uuid.uuid4())})
            assert response.status_code == 202
            job = response.json()
            assert set(job) == set(service.PUBLIC_FIELDS)
            assert started.wait(5)
            assert client.get("/health").status_code == 200
            assert client.get("/api/jobs/" + job["id"]).json()["status"] == "transcribing"
            assert client.post("/api/jobs/" + job["id"] + "/retry").status_code == 409
            assert client.get("/api/jobs/" + "0" * 32).status_code == 404
            assert client.get("/api/jobs/invalid").status_code == 400
            assert client.get("/health", headers={"Host": "untrusted.example"}).status_code == 400
        finally:
            release.set()


def test_whisper_runs_cpu_offline_and_consumes_all_segments(monkeypatch):
    consumed = []

    class Model:
        def __init__(self, path, **options):
            assert path == "/models/test"
            assert options == {"device": "cpu", "compute_type": "int8", "local_files_only": True,
                               "cpu_threads": 2, "num_workers": 1}

        def transcribe(self, path, **options):
            assert path == "call.wav"
            assert options["language"] is None
            assert options["task"] == "transcribe"

            def segments():
                for text in [" Guten Tag.", " Bitte zurückrufen."]:
                    consumed.append(text)
                    yield SimpleNamespace(text=text)

            return segments(), SimpleNamespace(language="de")

    monkeypatch.setenv("WHISPER_MODEL_PATH", "/models/test")
    monkeypatch.setenv("WHISPER_THREADS", "2")
    monkeypatch.setitem(sys.modules, "faster_whisper", SimpleNamespace(WhisperModel=Model))
    assert service.load_whisper()(Path("call.wav"), "auto") == "Guten Tag. Bitte zurückrufen."
    assert len(consumed) == 2


def test_real_audio_decoder_is_compatible_with_installed_pyav():
    audio = pytest.importorskip("faster_whisper.audio")
    decoded = audio.decode_audio(BytesIO(wav()), sampling_rate=16000)
    assert decoded.shape == (1600,)
    assert decoded[0] == pytest.approx(100 / 32768)


def test_summary_failure_preserves_transcript_and_retry_skips_whisper(tmp_path):
    calls = []

    def transcribe(*_):
        calls.append("whisper")
        return "Rückruf am Freitag."

    def fail_summary(text, style):
        assert text == "Rückruf am Freitag."
        assert style == "meeting"
        raise service.SummaryError("LLM vorübergehend nicht erreichbar.")

    jobs = service.Jobs(tmp_path, transcribe, fail_summary)
    job = upload(jobs)
    jobs.process_next()
    result = jobs.get(job["id"])
    assert result["status"] == "failed"
    assert result["stage"] == "Zusammenfassung fehlgeschlagen"
    assert result["transcript"] == "Rückruf am Freitag."
    assert "LLM" in result["error"]
    jobs.summarize = lambda text, style: "Am Freitag zurückrufen."
    jobs.retry(job["id"])
    jobs.process_next()
    result = jobs.get(job["id"])
    assert result["status"] == "done"
    assert result["summary"] == "Am Freitag zurückrufen."
    assert result["error"] == ""
    assert calls == ["whisper"]


def test_transcript_is_available_while_summary_is_still_running(tmp_path):
    started, release = threading.Event(), threading.Event()

    def summarize(*_):
        started.set()
        assert release.wait(5)
        return "Zusammenfassung"

    with TestClient(service.create_app(tmp_path, lambda *_: "Gesprächstext", summarize), base_url="http://127.0.0.1") as client:
        try:
            response = client.post("/api/jobs", files={"file": ("call.wav", wav(), "audio/wav")},
                headers={"Idempotency-Key": "swyxweb:" + str(uuid.uuid4())})
            assert response.status_code == 202
            assert started.wait(5)
            job = client.get("/api/jobs/" + response.json()["id"]).json()
            assert job["status"] == "summarizing"
            assert job["transcript"] == "Gesprächstext"
            assert job["summary"] == ""
            assert client.get("/health").status_code == 200
        finally:
            release.set()


def test_restart_resumes_summary_without_retranscribing(tmp_path):
    jobs = service.Jobs(tmp_path, lambda *_: "Gesprächstext")
    job = upload(jobs)
    jobs.process_next()
    with jobs.db() as db:
        db.execute("UPDATE jobs SET status='summarizing' WHERE id=?", (job["id"],))

    def unexpected_transcription(*_):
        pytest.fail("An existing transcript must be reused")

    restarted = service.Jobs(tmp_path, unexpected_transcription, lambda text, _: "Ergebnis: " + text)
    assert restarted.get(job["id"])["status"] == "queued"
    restarted.process_next()
    assert restarted.get(job["id"])["summary"] == "Ergebnis: Gesprächstext"


def test_silent_recording_does_not_call_llm(tmp_path):
    def unexpected_summary(*_):
        pytest.fail("Silence must not be sent to the LLM")

    jobs = service.Jobs(tmp_path, lambda *_: "", unexpected_summary)
    job = upload(jobs)
    jobs.process_next()
    assert jobs.get(job["id"])["status"] == "done"
    assert jobs.get(job["id"])["stage"] == "Keine Sprache erkannt"


def test_existing_database_is_migrated_without_losing_results(tmp_path):
    jobs = service.Jobs(tmp_path, lambda *_: "Vorhandenes Transkript")
    job = upload(jobs)
    jobs.process_next()
    with sqlite3.connect(tmp_path / "jobs.sqlite3") as db:
        db.execute("ALTER TABLE jobs DROP COLUMN transcribed")
    migrated = service.Jobs(tmp_path, lambda *_: "")
    assert migrated.get(job["id"])["transcript"] == "Vorhandenes Transkript"
    assert migrated.get(job["id"])["status"] == "done"
    with migrated.db() as db:
        assert db.execute("SELECT transcribed FROM jobs WHERE id=?", (job["id"],)).fetchone()[0] == 1