"""Local real-provider dialogue acceptance using isolated MySQL and actual /ws.

Requires an isolated MySQL on 127.0.0.1:13316 (root without password),
macOS afconvert, and existing local ASR/LLM/TTS test credentials.
Starts only the production WebSocket router and asset serving on loopback.
Never starts production MQTT or changes .env. Writes evidence under output/.
"""
import asyncio
import argparse
import hashlib
import json
import shutil
import struct
import sys
import time
import uuid
import wave
from pathlib import Path

import aiohttp
import pymysql
import uvicorn
import websockets
import yaml
from dotenv import dotenv_values
from fastapi import FastAPI

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from config import settings

OUT = ROOT / 'output/qwen-dialog-20261012'
OUT.mkdir(parents=True, exist_ok=True)
DEVICE_PREFIX = 'TalkingQ_E2E' + uuid.uuid4().hex[:6]
settings.db_host, settings.db_port = '127.0.0.1', 13316
settings.db_user, settings.db_password, settings.db_name = 'root', '', 'qwen_dialog_test'
settings.server_host, settings.server_port = '127.0.0.1', 18086
settings.assets_dir = str(OUT / 'assets')
settings.device_audio_cache_dir = str(OUT / 'device-audio')
settings.talkingq_mqtt_broker = ''
settings.selected_role_key = 'Banban'
tts_env = dotenv_values(ROOT / 'test/minimax.env')
settings.minimax_api_key = settings.minimax_api_key or tts_env.get('MINIMAX_API_KEY', '')
assert settings.minimax_api_key and settings.volcano_api_key and settings.aliyun_api_key
assert settings.aliyun_asr_model == 'qwen-audio-3.1-asr-flash-streaming'

from database.models import Base, Role, RoleLanguage, DeviceAuth, DeviceConfig, ConversationHistory, ConversationMessage
from database.connection import get_db_manager
from services.role_manager import role_manager
from services.audio_session import audio_session_manager
from services.task_manager import task_manager
from api.websocket import router
from api.assets import configure_static_assets
from sqlalchemy import select


def packet(device, sid, seq, kind, audio=b''):
    return device.encode() + b'\0' + sid.encode() + b'\0' + struct.pack('<IBI', seq, kind, len(audio)) + audio


async def setup(interrupt_only=False):
    conn = pymysql.connect(host='127.0.0.1', port=13316, user='root')
    with conn.cursor() as cur:
        cur.execute('CREATE DATABASE IF NOT EXISTS qwen_dialog_test CHARACTER SET utf8mb4')
    conn.close()
    manager = await get_db_manager()
    async with manager.async_engine.begin() as conn:
        await conn.run_sync(lambda sync: Base.metadata.create_all(sync, tables=[
            model.__table__ for model in (Role, RoleLanguage, DeviceAuth, DeviceConfig, ConversationHistory, ConversationMessage)
        ]))
    async with await manager.get_session() as db:
        for index, name in enumerate(['Banban', 'Baikeboshi']):
            d = yaml.safe_load((ROOT / f'assets/roles_definitions/{name}.yaml').read_text())
            existing = (await db.execute(select(Role).where(Role.role_key == name))).scalar_one_or_none()
            if not existing:
                role = Role(role_key=name, name=d['name'], content=d.get('content', ''), default_language='zh',
                            volcano_model_id=d['volcano_model_id'], minimax_voice_id=d.get('minimax_voice_id'), enabled=True)
                db.add(role)
                await db.flush()
                for language, config in d['multilingual'].items():
                    db.add(RoleLanguage(role_id=role.id, language_code=language, name=config.get('name'),
                                        content=config['content'], url=config.get('url'), minimax_voice_id=config.get('minimax_voice_id')))
                    source = ROOT / 'assets' / config['url'] / 'tts_error.mp3'
                    target = Path(settings.assets_dir) / config['url'] / 'tts_error.mp3'
                    if source.exists():
                        target.parent.mkdir(parents=True, exist_ok=True)
                        shutil.copyfile(source, target)
            device = f'{DEVICE_PREFIX}{index:02d}'
            if not (await db.execute(select(DeviceAuth).where(DeviceAuth.device_id == device))).scalar_one_or_none():
                db.add(DeviceAuth(device_id=device, serial_number=f'LOCAL-E2E-{index}', is_active=True))
                db.add(DeviceConfig(device_id=device, selected_role_key=name, preferred_language='zh'))
        await db.commit()
    await role_manager.initialize()
    return manager


async def download(http, url, sid, index):
    assert url.startswith('http://127.0.0.1:18086/assets/'), url
    async with http.get(url) as response:
        body = await response.read()
        assert response.status == 200 and body, f'empty/error audio: {response.status}'
    local = OUT / 'downloads' / f'{sid}-{index}.mp3'
    local.parent.mkdir(exist_ok=True)
    local.write_bytes(body)
    wav = local.with_suffix('.wav')
    process = await asyncio.create_subprocess_exec('/usr/bin/afconvert', '-f', 'WAVE', '-d', 'LEI16', str(local), str(wav),
                                                 stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    _, err = await process.communicate()
    assert process.returncode == 0, err.decode()
    with wave.open(str(wav)) as decoded:
        seconds = decoded.getnframes() / decoded.getframerate()
        pcm = decoded.readframes(decoded.getnframes())
        assert seconds > 0 and any(pcm), 'empty/silent decoded reply'
    async with http.get(url, headers={'Range': 'bytes=0-1023'}) as response:
        assert response.status == 206 and await response.read() == body[:1024]
    return {'url': url, 'file': str(local.relative_to(ROOT)), 'bytes': len(body), 'sha256': hashlib.sha256(body).hexdigest(), 'decoded_seconds': round(seconds, 3), 'http': 200, 'range': 206}


async def run_turn(ws, http, manager, device, role, case, interrupt=False):
    sid = uuid.uuid4().hex
    result = {'role': role, 'case': case['name'], 'session_id': sid, 'model': settings.aliyun_asr_model,
              'messages': [], 'audio': [], 'passed': False, 'interrupted': interrupt}
    started = time.perf_counter()
    try:
        await ws.send(packet(device, sid, 0, 1))
        assert await asyncio.wait_for(ws.recv(), 15) == 'START_ACK'
        result['messages'].append('START_ACK')
        # Device START_ACK precedes ASR readiness. Real handler queues subsequent
        # WebSocket frames while establishing the upstream connection.
        audio = (ROOT / case['audio']).read_bytes()
        for seq, offset in enumerate(range(0, len(audio), 640), 1):
            await ws.send(packet(device, sid, seq, 0, audio[offset:offset + 640]))
            await asyncio.sleep(len(audio[offset:offset + 640]) / 2000)
            if interrupt and offset >= 1280:
                break
        ending = time.perf_counter()
        await ws.send(packet(device, sid, seq + 1, 3 if interrupt else 2))
        while True:
            message = await asyncio.wait_for(ws.recv(), 90)
            result['messages'].append(message)
            if message.startswith(('TTS_URL:', 'PROMPT_SOUND_URL:')):
                if 'first_audio_seconds' not in result:
                    result['first_audio_seconds'] = round(time.perf_counter() - ending, 3)
                result['audio'].append(await download(http, message.split(':', 1)[1], sid, len(result['audio'])))
                if message.startswith('PROMPT_SOUND_URL:'):
                    break
            if message == 'TTS_END' or (interrupt and message == 'INTERRUPT_ACK'):
                break
            if message == 'TTS_ERROR':
                raise AssertionError('TTS_ERROR received')
        # END may be delivered before the history transaction completes.
        for _ in range(100):
            session = await audio_session_manager.get_session((device, sid))
            if session and session.asr_service._client is None and (not session.llm_service.connected):
                break
            await asyncio.sleep(.05)
        if session:
            result['transcript'] = session.transcript
            result['asr_error'] = session.asr_error
            result['resource_state'] = {
                'asr_closed': session.asr_service._client is None and session.asr_service._receiver is None,
                'llm_closed': not session.llm_service.connected and (session.llm_service.session is None or session.llm_service.session.closed),
                'tts_closed': session.tts_service._client_session is None or session.tts_service._client_session.closed,
            }
            result['resources_released'] = all(result['resource_state'].values())
        async with await manager.get_session() as db:
            query = select(ConversationMessage).join(ConversationHistory).where(ConversationHistory.device_id == device,
                    ConversationHistory.role_key == role).order_by(ConversationMessage.timestamp)
            messages = (await db.execute(query)).scalars().all()
            result['history_messages'] = [{'is_user': m.is_user, 'text': m.content} for m in messages]
        if interrupt:
            assert 'INTERRUPT_ACK' in result['messages'] and not result['audio']
        elif not case['expected_keywords']:
            assert not result.get('transcript') and any(m.startswith('PROMPT_SOUND_URL:') for m in result['messages'])
        else:
            assert not result.get('asr_error')
            assert all(word.lower() in result.get('transcript', '').lower() for word in case['expected_keywords'])
            assert 'TTS_START' in result['messages'] and 'TTS_END' in result['messages'] and result['audio']
            assert messages and not messages[-1].is_user and messages[-1].content.strip(), 'assistant history missing'
            assert len(messages) >= 2 and messages[-2].content == result['transcript'], 'latest history does not match current transcript'
            result['reply'] = messages[-1].content
        assert result.get('resources_released'), 'provider resources not released'
        result['passed'] = True
    except Exception as exc:
        result['error'] = f'{type(exc).__name__}: {exc}'
    result['total_seconds'] = round(time.perf_counter() - started, 3)
    return result


async def main(interrupt_only=False):
    manager = await setup(interrupt_only)
    app = FastAPI()
    configure_static_assets(app)
    app.include_router(router)
    server = uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=18086, log_level='warning', lifespan='off'))
    server_task = asyncio.create_task(server.serve())
    results = []
    try:
        for _ in range(100):
            if server.started: break
            await asyncio.sleep(.05)
        assert server.started
        cases = json.loads((ROOT / 'tests/fixtures/asr/manifest.json').read_text())
        async with aiohttp.ClientSession() as http:
            for index, role in enumerate(['Banban', 'Baikeboshi']):
                device = f'{DEVICE_PREFIX}{index:02d}'
                async with websockets.connect('ws://127.0.0.1:18086/ws', ping_interval=None) as ws:
                    await ws.send(json.dumps({'device_id': device, 'serial_number': f'LOCAL-E2E-{index}'}))
                    assert json.loads(await ws.recv())['status'] == 'authenticated'
                    plan = [(case, False) for case in cases]
                    if interrupt_only:
                        plan = [(cases[1], True), (cases[0], False)]
                    for case, interrupted in plan:
                        result = await asyncio.wait_for(run_turn(ws, http, manager, device, role, case, interrupt=interrupted), 150)
                        results.append(result)
                        (OUT / ('results-interrupt.json' if interrupt_only else 'results.json')).write_text(json.dumps(results, ensure_ascii=False, indent=2))
                        print('E2E_RESULT', json.dumps(result, ensure_ascii=False), flush=True)
                        if not result['passed'] and not interrupt_only:
                            return False
        return all(r['passed'] for r in results)
    finally:
        server.should_exit = True
        await server_task
        await task_manager.cleanup_all_tasks()
        await manager.close()


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--interrupt-only', action='store_true')
    args = parser.parse_args()
    sys.exit(0 if asyncio.run(main(args.interrupt_only)) else 1)