"""L'aiguillage des événements ADK, sans réseau : on fabrique les événements que run_live produirait."""
import asyncio
import json

from google.adk.events import Event
from google.genai import types

from copilote.appel import Appel


def ev(**kw):
    return Event(author="accueil_cabinet", **kw)


def audio(n=4800):
    return ev(content=types.Content(role="model", parts=[types.Part(inline_data=types.Blob(data=b"\x00" * n, mime_type="audio/pcm;rate=24000"))]))


def test_un_tour_complet(cfg, agenda):
    recus = []

    async def emettre(m):
        recus.append(m)

    async def scenario():
        a = Appel(cfg, agenda, emettre, appel_id="t1")
        await a.traiter(ev(input_transcription=types.Transcription(text="Oui, mon numéro est le 06 12 34 56 78"), partial=False))
        assert a.outils.derniere_phrase_appelant.startswith("Oui")
        a.fin_de_parole(a.t0)
        await a.traiter(ev(turn_complete=True))  # fin de tour sans audio (appel d'outil) : on attend encore
        assert not a.tour_fini.is_set()
        await a.traiter(audio())
        await a.traiter(ev(output_transcription=types.Transcription(text="C'est noté."), partial=False))
        await a.traiter(ev(turn_complete=True))
        assert a.tour_fini.is_set() and len(a.latences) == 1
        return a.fermer()

    bilan = asyncio.run(scenario())
    assert [t["role"] for t in bilan["transcription"]] == ["appelant", "assistant"]
    assert any(isinstance(m, bytes) and m[:1] == b"\x01" for m in recus)
    journal = (cfg.sorties / "appels" / "t1" / "journal.jsonl").read_text()
    assert "56 78" not in journal and "06 •• •• •• 78" in journal  # numéro masqué dans le journal
    assert json.loads(journal.splitlines()[-1])["type"] == "fin"


def test_appel_d_outil_journalise_et_masque(cfg, agenda):
    async def scenario():
        a = Appel(cfg, agenda, appel_id="t2")
        fc = types.FunctionCall(name="reserver_creneau", args={"creneau_id": 3, "telephone": "0612345678"})
        await a.traiter(ev(content=types.Content(role="model", parts=[types.Part(function_call=fc)])))
        a.fermer()
        return a

    a = asyncio.run(scenario())
    assert a.outils_appeles[0]["nom"] == "reserver_creneau"
    assert "0612345678" not in (cfg.sorties / "appels" / "t2" / "journal.jsonl").read_text()


def test_transfert_termine_l_appel(cfg, agenda):
    async def scenario():
        a = Appel(cfg, agenda, appel_id="t3")
        a.outils.transferer_humain("facture", "double paiement")
        await a.traiter(audio())
        await a.traiter(ev(turn_complete=True))
        return a.termine.is_set()

    assert asyncio.run(scenario())


def test_cout_somme_tous_les_releves():
    from copilote.e2e import cout_usd
    u = {"total": 0, "entree_detail": {"TEXT": 1_000_000, "AUDIO": 0}, "sortie_detail": {"AUDIO": 1_000_000}}
    assert cout_usd([u, u]) == 2 * (0.75 + 12.00)


def test_compteurs_de_jetons_absents(cfg, agenda):
    """Vertex AI renvoie parfois token_count=None : le calcul de coût ne doit pas planter."""
    from copilote.e2e import cout_usd
    um = types.GenerateContentResponseUsageMetadata(
        prompt_token_count=None, total_token_count=10,
        prompt_tokens_details=[types.ModalityTokenCount(modality=types.MediaModality.AUDIO, token_count=None)])

    async def scenario():
        a = Appel(cfg, agenda, appel_id="t5")
        await a.traiter(ev(usage_metadata=um))
        a.fermer()
        return a

    a = asyncio.run(scenario())
    assert a.usage[0]["entree"] == 0 and cout_usd(a.usage) == 0


def test_refus_d_enregistrement_rien_n_est_ecrit(cfg, agenda):
    from dataclasses import replace

    async def scenario():
        a = Appel(replace(cfg, enregistrer=True), agenda, appel_id="t6")
        a.envoyer_audio(b"\x01\x00" * 1600)
        await a.traiter(audio())
        a.outils.refuser_enregistrement()
        a.fermer()

    asyncio.run(scenario())
    dossier = cfg.sorties / "appels" / "t6"
    assert (dossier / "journal.jsonl").exists() and not (dossier / "enregistrement.wav").exists()


def test_vad_reglable(cfg, agenda):
    from dataclasses import replace

    async def scenario():
        defaut, long = Appel(cfg, agenda, appel_id="t7"), Appel(replace(cfg, silence_ms=800), agenda, appel_id="t8")
        r = defaut._vad(), long._vad()
        defaut.fermer(), long.fermer()
        return r

    defaut, long = asyncio.run(scenario())
    assert defaut is None and long.automatic_activity_detection.silence_duration_ms == 800