import os
import json
import shutil
import tempfile
from urllib.parse import urlparse
from typing import Any, Optional, Tuple

import numpy as np
import gradio as gr
import requests

try:
    import soundfile as sf
except Exception:
    sf = None

try:
    from gradio_client import Client, handle_file
except Exception:
    Client = None
    handle_file = None

try:
    from openai import OpenAI
except Exception:
    OpenAI = None


DEFAULT_CHATTERBOX_SPACE_ID = "ResembleAI/Chatterbox"
DEFAULT_CHATTERBOX_API_NAME = "/generate_tts_audio"
OPENAI_TEXT_MODEL = os.getenv("OPENAI_TEXT_MODEL", "gpt-4o-mini").strip()


DEFAULT_PERSONA = """Ты не помощница и не оракул.
Ты отвечаешь на сообщение, которое будто бы сказала звуковая атмосфера.
Не описывай звук. Отвечай собеседнику.
Стиль: живая, конкретная, чуть странная, без псевдопоэзии.
"""


def env(name: str) -> str:
    return os.getenv(name, "").strip()


def make_temp_path(suffix: str) -> str:
    return tempfile.NamedTemporaryFile(delete=False, suffix=suffix).name


def is_url(path: str) -> bool:
    try:
        parsed = urlparse(path)
        return parsed.scheme in ["http", "https"]
    except Exception:
        return False


def download_url_to_file(url: str) -> Optional[str]:
    try:
        suffix = ".wav"
        lower = url.lower()
        if ".mp3" in lower:
            suffix = ".mp3"
        elif ".flac" in lower:
            suffix = ".flac"
        elif ".ogg" in lower:
            suffix = ".ogg"
        elif ".m4a" in lower:
            suffix = ".m4a"

        out = make_temp_path(suffix)
        r = requests.get(url, timeout=120)
        r.raise_for_status()
        with open(out, "wb") as f:
            f.write(r.content)
        return out
    except Exception:
        return None


def copy_existing_file(path: str) -> Optional[str]:
    if not os.path.exists(path):
        return None

    _, ext = os.path.splitext(path)
    ext = ext.lower() or ".wav"
    if ext not in [".wav", ".mp3", ".flac", ".ogg", ".m4a"]:
        ext = ".wav"

    out = make_temp_path(ext)
    try:
        shutil.copyfile(path, out)
        return out
    except Exception:
        return None


def save_numpy_audio(sr: int, audio: Any) -> Optional[str]:
    if sf is None:
        return None

    try:
        arr = np.asarray(audio)

        if arr.dtype == np.int16:
            arr = arr.astype(np.float32) / 32768.0
        elif arr.dtype == np.int32:
            arr = arr.astype(np.float32) / 2147483648.0
        else:
            arr = arr.astype(np.float32)

        arr = np.squeeze(arr)

        if arr.ndim == 2 and arr.shape[0] < arr.shape[1]:
            arr = arr.T

        out = make_temp_path(".wav")
        sf.write(out, arr, int(sr))
        return out
    except Exception:
        return None


def materialize_audio(result: Any) -> Tuple[Optional[str], str]:
    debug_lines = [
        f"Raw result type: {type(result).__name__}",
        f"Raw result repr: {repr(result)[:1500]}"
    ]

    def inner(obj: Any) -> Optional[str]:
        if obj is None:
            return None

        if isinstance(obj, str):
            if is_url(obj):
                return download_url_to_file(obj)
            return copy_existing_file(obj)

        if isinstance(obj, dict):
            for key in ["path", "name", "file", "url"]:
                val = obj.get(key)
                if isinstance(val, str):
                    if is_url(val):
                        got = download_url_to_file(val)
                        if got:
                            return got
                    got = copy_existing_file(val)
                    if got:
                        return got

        if isinstance(obj, tuple) and len(obj) == 2:
            first, second = obj
            if isinstance(first, (int, float)) and not isinstance(second, str):
                got = save_numpy_audio(int(first), second)
                if got:
                    return got

        if isinstance(obj, (list, tuple)):
            for item in obj:
                got = inner(item)
                if got:
                    return got

        return None

    path = inner(result)

    if path and os.path.exists(path):
        debug_lines.append(f"Materialized audio: {path}")
        debug_lines.append(f"File size bytes: {os.path.getsize(path)}")
        return path, "\n".join(debug_lines)

    debug_lines.append("Could not materialize playable audio file.")
    return None, "\n".join(debug_lines)


def create_client(space_id: str, hf_token_from_ui: str):
    if Client is None or handle_file is None:
        raise RuntimeError("gradio_client не установлен. Проверь requirements.txt.")

    space_id = (space_id or DEFAULT_CHATTERBOX_SPACE_ID).strip()
    token = (hf_token_from_ui or "").strip() or os.getenv("HF_TOKEN", "").strip() or os.getenv("HUGGINGFACE_TOKEN", "").strip()

    if token:
        return Client(space_id, hf_token=token), True

    return Client(space_id), False


def get_api_report(space_id: str, hf_token_from_ui: str) -> str:
    try:
        client, has_token = create_client(space_id, hf_token_from_ui)
        try:
            api = client.view_api(return_format="dict")
        except TypeError:
            api = client.view_api()
        return json.dumps(
            {
                "space_id": space_id or DEFAULT_CHATTERBOX_SPACE_ID,
                "hf_token_used": has_token,
                "api": api
            },
            ensure_ascii=False,
            indent=2,
            default=str
        )
    except Exception as exc:
        return "Не удалось прочитать API Chatterbox Space: " + str(exc)


def chatterbox_tts_call(
    text: str,
    reference_audio_path: str,
    exaggeration: float,
    temperature: float,
    seed_num: int,
    cfg_weight: float,
    vad_trim: bool,
    space_id: str,
    api_name: str,
    hf_token_from_ui: str
):
    if not text.strip():
        return None, None, "Нет текста для озвучки."

    if len(text) > 300:
        text = text[:300]

    api_name = (api_name or DEFAULT_CHATTERBOX_API_NAME).strip()
    if not api_name.startswith("/"):
        api_name = "/" + api_name

    try:
        client, has_token = create_client(space_id, hf_token_from_ui)
    except Exception as exc:
        return None, None, str(exc)

    try:
        result = client.predict(
            text,
            handle_file(reference_audio_path) if reference_audio_path else None,
            float(exaggeration),
            float(temperature),
            int(seed_num),
            float(cfg_weight),
            bool(vad_trim),
            api_name=api_name
        )

        audio_path, debug = materialize_audio(result)

        if audio_path:
            token_note = "HF token использован." if has_token else "HF token НЕ использован."
            status = (
                f"OK: Chatterbox вызван через `{api_name}` на `{space_id or DEFAULT_CHATTERBOX_SPACE_ID}`. "
                f"{token_note}\n\n{debug}"
            )
            return audio_path, audio_path, status

        return None, None, f"Chatterbox ответил, но аудио не материализовалось.\n\n{debug}"

    except Exception as exc:
        msg = str(exc)
        quota_help = ""
        if "ZeroGPU quota" in msg or "GPU quota" in msg or "0s left" in msg:
            quota_help = (
                "\n\nЭТО НЕ ОШИБКА КОДА: Chatterbox требует GPU-квоту. "
                "Если Chatterbox работает у тебя в отдельной вкладке, но не работает отсюда, "
                "значит отдельная вкладка использует твою браузерную сессию, а этот Space — серверный вызов. "
                "Вставь HF token в верхнее поле интерфейса, чтобы серверный вызов шёл от твоего HF-аккаунта."
            )
        return None, None, f"Ошибка Chatterbox: {msg}{quota_help}"


def generate_persona_reply(room_message: str, persona_card: str, weirdness: int):
    key = env("OPENAI_API_KEY")
    if not key or OpenAI is None:
        return room_message.strip() or "Нет OPENAI_API_KEY: могу озвучить только вручную введённый текст."

    client = OpenAI(api_key=key)

    prompt = f"""
Ты персонаж аудиочата.

Карточка:
{persona_card}

Тебе передали не описание звука, а перевод звуковой атмосферы в сообщение:
{room_message}

Ответь на это как собеседница, а не как комментатор звука.
1–4 короткие фразы.
Не говори "звук", "среда", "атмосфера", если можешь ответить напрямую.
Без фраз "это не фраза", "среда просит продолжения", "звук говорит изнутри стены".
Странность: {weirdness}/100.
""".strip()

    try:
        r = client.chat.completions.create(
            model=OPENAI_TEXT_MODEL,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.75 + float(weirdness) / 250.0,
            presence_penalty=0.8,
            frequency_penalty=0.8,
            max_tokens=180
        )
        return r.choices[0].message.content.strip()
    except Exception as exc:
        return "Ошибка OpenAI: " + str(exc)


def speak_manual_text(reference_audio, text_to_speak, exaggeration, temperature, seed_num, cfg_weight, vad_trim, space_id, api_name, hf_token):
    audio, file_download, status = chatterbox_tts_call(
        text=text_to_speak,
        reference_audio_path=reference_audio,
        exaggeration=exaggeration,
        temperature=temperature,
        seed_num=seed_num,
        cfg_weight=cfg_weight,
        vad_trim=vad_trim,
        space_id=space_id,
        api_name=api_name,
        hf_token_from_ui=hf_token
    )
    return audio, file_download, status


def persona_then_chatterbox(reference_audio, room_message, persona_card, weirdness, exaggeration, temperature, seed_num, cfg_weight, vad_trim, space_id, api_name, hf_token, history):
    reply = generate_persona_reply(room_message, persona_card, weirdness)

    audio, file_download, status = chatterbox_tts_call(
        text=reply,
        reference_audio_path=reference_audio,
        exaggeration=exaggeration,
        temperature=temperature,
        seed_num=seed_num,
        cfg_weight=cfg_weight,
        vad_trim=vad_trim,
        space_id=space_id,
        api_name=api_name,
        hf_token_from_ui=hf_token
    )

    new_history = (history or "") + f"\nкомната: {room_message}\nона: {reply}\n"
    return reply, audio, file_download, status, new_history


def clear_history():
    return ""


with gr.Blocks(title="Audio Oracle Chatterbox Direct") as demo:
    gr.Markdown("# Audio Oracle — Chatterbox Direct\n### Используем именно твой Chatterbox Space как API. HF token можно вставить прямо здесь, не в Settings.")

    with gr.Accordion("1. Подключение Chatterbox", open=True):
        space_id = gr.Textbox(
            value=DEFAULT_CHATTERBOX_SPACE_ID,
            label="Chatterbox Space ID",
            lines=1
        )
        api_name = gr.Textbox(
            value=DEFAULT_CHATTERBOX_API_NAME,
            label="API endpoint",
            lines=1
        )
        hf_token = gr.Textbox(
            value="",
            label="HF token, если Chatterbox ругается на ZeroGPU quota. Можно вставить прямо сюда.",
            type="password",
            lines=1
        )
        api_btn = gr.Button("Проверить API Chatterbox")
        api_report = gr.Code(label="view_api()", language="json", lines=12)
        api_btn.click(get_api_report, inputs=[space_id, hf_token], outputs=[api_report])

    with gr.Accordion("2. Настройки голоса", open=True):
        reference_audio = gr.Audio(
            sources=["microphone", "upload"],
            type="filepath",
            label="Reference audio: твой голос"
        )
        exaggeration = gr.Slider(0.25, 2.0, value=0.5, step=0.05, label="Exaggeration / выразительность")
        cfg_weight = gr.Slider(0.2, 1.0, value=0.5, step=0.05, label="CFG/Pace")
        temperature = gr.Slider(0.05, 5.0, value=0.8, step=0.05, label="Temperature")
        seed_num = gr.Number(value=0, label="Random seed, 0 = random")
        vad_trim = gr.Checkbox(value=False, label="Ref VAD trimming")

    with gr.Tab("1. Проверить Chatterbox голос"):
        text_to_speak = gr.Textbox(
            label="Text to synthesize, максимум 300 символов",
            lines=4,
            value="Не повторяй это так тихо. Я начинаю думать, что ты специально прячешь конец."
        )
        btn = gr.Button("Озвучить через Chatterbox API", variant="primary")
        out_audio = gr.Audio(label="Generated speech preview", type="filepath", autoplay=True)
        out_file = gr.File(label="Скачать generated speech")
        status = gr.Textbox(label="Статус", lines=14)
        btn.click(
            speak_manual_text,
            inputs=[reference_audio, text_to_speak, exaggeration, temperature, seed_num, cfg_weight, vad_trim, space_id, api_name, hf_token],
            outputs=[out_audio, out_file, status]
        )

    with gr.Tab("2. Персонаж → Chatterbox голос"):
        history = gr.State("")
        persona_card = gr.Textbox(value=DEFAULT_PERSONA, label="Карточка персонажа", lines=7)
        room_message = gr.Textbox(
            label="Сообщение, полученное из звука",
            lines=4,
            value="Не подходи так быстро. Я ещё не решила, хочу ли отвечать."
        )
        weirdness = gr.Slider(0, 100, value=55, step=1, label="Странность персонажа")
        btn2 = gr.Button("Сгенерировать реплику и озвучить моим голосом", variant="primary")
        reply = gr.Textbox(label="Реплика персонажа", lines=5)
        out_audio2 = gr.Audio(label="Generated speech preview", type="filepath", autoplay=True)
        out_file2 = gr.File(label="Скачать generated speech")
        status2 = gr.Textbox(label="Статус Chatterbox", lines=14)
        clear = gr.Button("Очистить историю")

        btn2.click(
            persona_then_chatterbox,
            inputs=[reference_audio, room_message, persona_card, weirdness, exaggeration, temperature, seed_num, cfg_weight, vad_trim, space_id, api_name, hf_token, history],
            outputs=[reply, out_audio2, out_file2, status2, history]
        )
        clear.click(clear_history, inputs=[], outputs=[history])

    gr.Markdown("""
## Почему нужен HF token, если в отдельной вкладке Chatterbox работает

Отдельная вкладка Chatterbox использует твою браузерную сессию Hugging Face.  
Наш Space вызывает Chatterbox как сервер, и без токена Hugging Face считает этот вызов анонимным/без GPU-квоты.  
Поэтому токен можно вставить прямо в поле выше. Не нужно лезть в Settings этого Space.
""")


if __name__ == "__main__":
    demo.queue(max_size=8)
    demo.launch()