"""Provide functionality for TTS."""
import asyncio
import functools as ft
import hashlib
import io
import logging
import mimetypes
import os
import re
from typing import Dict, Optional

from aiohttp import web
import mutagen
import voluptuous as vol

from homeassistant.components.http import HomeAssistantView
from homeassistant.components.media_player.const import (
    ATTR_MEDIA_CONTENT_ID,
    ATTR_MEDIA_CONTENT_TYPE,
    DOMAIN as DOMAIN_MP,
    MEDIA_TYPE_MUSIC,
    SERVICE_PLAY_MEDIA,
)
from homeassistant.const import (
    ATTR_ENTITY_ID,
    CONF_PLATFORM,
    HTTP_BAD_REQUEST,
    HTTP_NOT_FOUND,
    HTTP_OK,
)
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_per_platform, discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.network import get_url
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_prepare_setup_platform

# mypy: allow-untyped-defs, no-check-untyped-defs

_LOGGER = logging.getLogger(__name__)

ATTR_CACHE = "cache"
ATTR_LANGUAGE = "language"
ATTR_MESSAGE = "message"
ATTR_OPTIONS = "options"
ATTR_PLATFORM = "platform"

BASE_URL_KEY = "tts_base_url"

CONF_BASE_URL = "base_url"
CONF_CACHE = "cache"
CONF_CACHE_DIR = "cache_dir"
CONF_LANG = "language"
CONF_SERVICE_NAME = "service_name"
CONF_TIME_MEMORY = "time_memory"

DEFAULT_CACHE = True
DEFAULT_CACHE_DIR = "tts"
DEFAULT_TIME_MEMORY = 300
DOMAIN = "tts"

MEM_CACHE_FILENAME = "filename"
MEM_CACHE_VOICE = "voice"

SERVICE_CLEAR_CACHE = "clear_cache"
SERVICE_SAY = "say"

_RE_VOICE_FILE = re.compile(r"([a-f0-9]{40})_([^_]+)_([^_]+)_([a-z_]+)\.[a-z0-9]{3,4}")
KEY_PATTERN = "{0}_{1}_{2}_{3}"


def _deprecated_platform(value):
    """Validate if platform is deprecated."""
    if value == "google":
        raise vol.Invalid(
            "google tts service has been renamed to google_translate,"
            " please update your configuration."
        )
    return value


PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_PLATFORM): vol.All(cv.string, _deprecated_platform),
        vol.Optional(CONF_CACHE, default=DEFAULT_CACHE): cv.boolean,
        vol.Optional(CONF_CACHE_DIR, default=DEFAULT_CACHE_DIR): cv.string,
        vol.Optional(CONF_TIME_MEMORY, default=DEFAULT_TIME_MEMORY): vol.All(
            vol.Coerce(int), vol.Range(min=60, max=57600)
        ),
        vol.Optional(CONF_BASE_URL): cv.string,
        vol.Optional(CONF_SERVICE_NAME): cv.string,
    }
)
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE.extend(PLATFORM_SCHEMA.schema)

SCHEMA_SERVICE_SAY = vol.Schema(
    {
        vol.Required(ATTR_MESSAGE): cv.string,
        vol.Optional(ATTR_CACHE): cv.boolean,
        vol.Required(ATTR_ENTITY_ID): cv.comp_entity_ids,
        vol.Optional(ATTR_LANGUAGE): cv.string,
        vol.Optional(ATTR_OPTIONS): dict,
    }
)

SCHEMA_SERVICE_CLEAR_CACHE = vol.Schema({})


async def async_setup(hass, config):
    """Set up TTS."""
    tts = SpeechManager(hass)

    try:
        conf = config[DOMAIN][0] if config.get(DOMAIN, []) else {}
        use_cache = conf.get(CONF_CACHE, DEFAULT_CACHE)
        cache_dir = conf.get(CONF_CACHE_DIR, DEFAULT_CACHE_DIR)
        time_memory = conf.get(CONF_TIME_MEMORY, DEFAULT_TIME_MEMORY)
        base_url = conf.get(CONF_BASE_URL) or get_url(hass)
        hass.data[BASE_URL_KEY] = base_url

        await tts.async_init_cache(use_cache, cache_dir, time_memory, base_url)
    except (HomeAssistantError, KeyError):
        _LOGGER.exception("Error on cache init")
        return False

    hass.http.register_view(TextToSpeechView(tts))
    hass.http.register_view(TextToSpeechUrlView(tts))

    async def async_setup_platform(p_type, p_config=None, discovery_info=None):
        """Set up a TTS platform."""
        if p_config is None:
            p_config = {}

        platform = await async_prepare_setup_platform(hass, config, DOMAIN, p_type)
        if platform is None:
            return

        try:
            if hasattr(platform, "async_get_engine"):
                provider = await platform.async_get_engine(
                    hass, p_config, discovery_info
                )
            else:
                provider = await hass.async_add_executor_job(
                    platform.get_engine, hass, p_config, discovery_info
                )

            if provider is None:
                _LOGGER.error("Error setting up platform %s", p_type)
                return

            tts.async_register_engine(p_type, provider, p_config)
        except Exception:  # pylint: disable=broad-except
            _LOGGER.exception("Error setting up platform: %s", p_type)
            return

        async def async_say_handle(service):
            """Service handle for say."""
            entity_ids = service.data[ATTR_ENTITY_ID]
            message = service.data.get(ATTR_MESSAGE)
            cache = service.data.get(ATTR_CACHE)
            language = service.data.get(ATTR_LANGUAGE)
            options = service.data.get(ATTR_OPTIONS)

            try:
                url = await tts.async_get_url(
                    p_type, message, cache=cache, language=language, options=options
                )
            except HomeAssistantError as err:
                _LOGGER.error("Error on init TTS: %s", err)
                return

            data = {
                ATTR_MEDIA_CONTENT_ID: url,
                ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
                ATTR_ENTITY_ID: entity_ids,
            }

            await hass.services.async_call(
                DOMAIN_MP,
                SERVICE_PLAY_MEDIA,
                data,
                blocking=True,
                context=service.context,
            )

        service_name = p_config.get(CONF_SERVICE_NAME, f"{p_type}_{SERVICE_SAY}")
        hass.services.async_register(
            DOMAIN, service_name, async_say_handle, schema=SCHEMA_SERVICE_SAY
        )

    setup_tasks = [
        async_setup_platform(p_type, p_config)
        for p_type, p_config in config_per_platform(config, DOMAIN)
    ]

    if setup_tasks:
        await asyncio.wait(setup_tasks)

    async def async_platform_discovered(platform, info):
        """Handle for discovered platform."""
        await async_setup_platform(platform, discovery_info=info)

    discovery.async_listen_platform(hass, DOMAIN, async_platform_discovered)

    async def async_clear_cache_handle(service):
        """Handle clear cache service call."""
        await tts.async_clear_cache()

    hass.services.async_register(
        DOMAIN,
        SERVICE_CLEAR_CACHE,
        async_clear_cache_handle,
        schema=SCHEMA_SERVICE_CLEAR_CACHE,
    )

    return True


def _hash_options(options: Dict) -> str:
    """Hashes an options dictionary."""
    opts_hash = hashlib.blake2s(digest_size=5)
    for key, value in sorted(options.items()):
        opts_hash.update(str(key).encode())
        opts_hash.update(str(value).encode())

    return opts_hash.hexdigest()


class SpeechManager:
    """Representation of a speech store."""

    def __init__(self, hass):