from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
from pathlib import Path
import uuid

import redis

from api.core.Gzip import gzip_zip
from api.utils.skill_detail import SKILL_DETAIL_NORMALIZE_VERSION, normalize_skill_detail
from api.utils.skill_tree import SKILL_TREE_WITH_DESC_SPECIAL_VERSION, load_skill_tree_with_desc_special
from core.basic.character import createCharacter
from core.basic.equipment import get_equipment
from core.character.adventure import adv, get_adv_list
from database.connect import BACKEND_DIR, RedisPool


PreheatLoader = Callable[[], object]
PreheatTask = tuple[str, PreheatLoader]
PREHEAT_LOCK_KEY = 'dcalc_sys:preheat:lock'
PREHEAT_LOCK_EXPIRE_SECONDS = 15 * 60
SKILL_DETAIL_PARTS = ('cn', 'skillDetail')


def _load_skills():
    with (BACKEND_DIR / 'openapi' / 'data' / 'skills.json').open(encoding='utf-8') as file:
        return json.load(file)


def _load_custom_cn():
    with (BACKEND_DIR / 'openapi' / 'custom_cn.json').open(encoding='utf-8') as file:
        return json.load(file)


def _iter_open_alters() -> list[str]:
    alters: list[str] = []
    for parent in adv:
        for child in parent['children']:
            alter = child.get('class')
            if child.get('open') and alter:
                alters.append(alter)
    return alters


def _build_character_preheat_tasks(equ_version: str = '0') -> list[PreheatTask]:
    tasks: list[PreheatTask] = []
    for alter in _iter_open_alters():
        tasks.append(
            (
                f'dcalc:character:{alter}:{equ_version}',
                lambda alter=alter: createCharacter(alter, equ_version).getInfo(),
            )
        )
    return tasks


def _iter_skill_detail_files() -> list[Path]:
    data_dir = BACKEND_DIR / 'openapi' / 'data'
    return sorted(data_dir.glob('*/' + '*/' + '/'.join(SKILL_DETAIL_PARTS) + '/*.json'))


def _skill_prefix_from_path(path: Path) -> str:
    relative = path.relative_to(BACKEND_DIR / 'openapi' / 'data').with_suffix('')
    return relative.as_posix()


def _openapi_skill_detail_key(prefix: str) -> str:
    parts = prefix.split('/')
    return f'openapi:{SKILL_DETAIL_NORMALIZE_VERSION}:{parts[0]}:{parts[1]}:{parts[-1]}'


def _load_skill_detail(path: Path):
    with path.open(encoding='utf-8') as file:
        return json.load(file)


def _option_value_count(skill_data: dict) -> int:
    rows = skill_data.get('levelInfo', {}).get('rows') or []
    if not rows:
        return 0
    return max(
        (
            int(key.replace('value', ''))
            for row in rows
            for key in (row.get('optionValue') or {})
            if key.startswith('value') and key.replace('value', '').isdigit()
        ),
        default=0,
    )


def _skill_data_value(skill_data: dict, index: int) -> list:
    key = f'value{index + 1}'
    rows = skill_data.get('levelInfo', {}).get('rows') or []
    data = [0] + [(row.get('optionValue') or {}).get(key, 0) for row in rows]
    return [0 if item is None or item == '-' else item for item in data]


def _custom_value(custom_cn: list[dict], prefix: str):
    parts = prefix.split('/')
    job = parts[0]
    job_grow = parts[1]
    skill_id = parts[-1]
    return next(
        (
            item.get('custom', 1.0)
            for item in custom_cn
            if item.get('job') == job and item.get('jobGrow') == job_grow and item.get('skillId') == skill_id
        ),
        1.0,
    )


def _build_skill_data_preheat_tasks(custom_cn: list[dict]) -> list[PreheatTask]:
    tasks: list[PreheatTask] = []
    for path in _iter_skill_detail_files():
        prefix = _skill_prefix_from_path(path)
        skill_data = _load_skill_detail(path)

        tasks.append(
            (
                _openapi_skill_detail_key(prefix),
                lambda skill_data=skill_data: normalize_skill_detail(skill_data),
            )
        )
        tasks.append((f'dcalc:skill:{prefix}:custom', lambda prefix=prefix: _custom_value(custom_cn, prefix)))

        if skill_data.get('evolution'):
            tasks.append(
                (
                    f'dcalc:skill:{prefix}:vps',
                    lambda skill_data=skill_data: [{'name': item['name'], 'desc': item['desc']} for item in skill_data.get('evolution', [])],
                )
            )

        for index in range(_option_value_count(skill_data)):
            tasks.append((f'dcalc:skill:{prefix}:{index}', lambda skill_data=skill_data, index=index: _skill_data_value(skill_data, index)))
    return tasks


def _build_skill_tree_preheat_tasks() -> list[PreheatTask]:
    tasks: list[PreheatTask] = []
    data_dir = BACKEND_DIR / 'openapi' / 'data'
    for path in sorted(data_dir.glob('*/' + '*/cn/skill_tree.json')):
        relative = path.relative_to(data_dir)
        job_id = relative.parts[0]
        job_grow_id = relative.parts[1]
        tasks.append(
            (
                f'openapi:{SKILL_TREE_WITH_DESC_SPECIAL_VERSION}:{job_id}:{job_grow_id}:skills',
                lambda job_id=job_id, job_grow_id=job_grow_id: load_skill_tree_with_desc_special(job_id, job_grow_id),
            )
        )
    return tasks


def _build_oath_preheat_tasks(equ_version: str = '0') -> list[PreheatTask]:
    tasks: list[PreheatTask] = []
    equ = get_equipment(equ_version)
    for oath in equ.oath_suits:
        for skill in oath.skills:
            key = f'dcalc:oath:{oath.suitId}:{oath.rarity}:{oath.level}:{skill.skillId}'
            tasks.append(
                (
                    key,
                    lambda oath=oath, skill=skill: {
                        **skill.__dict__,
                        'point': oath.points.split(',')[oath.level - 1],
                        'imageUrl': oath.imageUrl,
                    },
                )
            )
    return tasks


def _write_preheated_value(redis_conn: redis.Redis, key: str, value: object, expire: int = 0, without_gzip: bool = False):
    if value is None:
        return
    if without_gzip:
        redis_conn.set(key, json.dumps(value))
    else:
        redis_conn.set(key, gzip_zip(json.dumps(value)))
    if expire > 0:
        redis_conn.expire(key, expire)


def _run_preheat_task(key: str, loader: PreheatLoader) -> tuple[str, object]:
    return key, loader()


def delete_keys_with_prefix(redis_conn: redis.Redis, prefix: str):
    batch: list[bytes] = []
    for key in redis_conn.scan_iter(f'{prefix}*', count=500):
        batch.append(key)
        if len(batch) >= 500:
            redis_conn.delete(*batch)
            batch.clear()
    if batch:
        redis_conn.delete(*batch)


def _acquire_preheat_lock(redis_conn: redis.Redis) -> str | None:
    lock_token = str(uuid.uuid4())
    locked = redis_conn.set(PREHEAT_LOCK_KEY, lock_token, nx=True, ex=PREHEAT_LOCK_EXPIRE_SECONDS)
    if not locked:
        return None
    return lock_token


def _release_preheat_lock(redis_conn: redis.Redis, lock_token: str):
    redis_conn.eval(
        """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        end
        return 0
        """,
        1,
        PREHEAT_LOCK_KEY,
        lock_token,
    )


def preheat_redis_cache() -> list[str]:
    # if config.DEBUG_MODE:
    #     print('Skip redis preheat in debug mode.')
    #     return []
    warmed_keys: list[str] = []

    with redis.Redis(connection_pool=RedisPool) as redis_conn:
        try:
            lock_token = _acquire_preheat_lock(redis_conn)
        except redis.RedisError as exc:
            print(f'Skip redis preheat: redis unavailable ({exc}).')
            return []
        if not lock_token:
            print('Skip redis preheat: another instance is warming cache.')
            return []

        try:
            delete_keys_with_prefix(redis_conn, 'dcalc:')
            delete_keys_with_prefix(redis_conn, 'openapi:')

            custom_cn = _load_custom_cn()
            base_preheat_tasks: list[PreheatTask] = [
                ('dcalc:adventure', get_adv_list),
                ('dcalc:custom_cn', lambda: custom_cn),
                ('openapi:skills', _load_skills),
            ]
            shared_preheat_tasks = [
                *_build_skill_tree_preheat_tasks(),
                *_build_skill_data_preheat_tasks(custom_cn),
                *_build_oath_preheat_tasks(),
            ]
            character_preheat_tasks = _build_character_preheat_tasks()

            for key, loader in base_preheat_tasks:
                try:
                    _write_preheated_value(redis_conn, key, loader(), without_gzip=key == 'dcalc:custom_cn')
                    warmed_keys.append(key)
                except Exception as exc:
                    print(f'Redis preheat failed for {key}: {exc}')

            max_workers = min(8, len(shared_preheat_tasks))
            if max_workers > 0:
                with ThreadPoolExecutor(max_workers=max_workers) as executor:
                    future_map = {executor.submit(_run_preheat_task, key, loader): key for key, loader in shared_preheat_tasks}
                    for future in as_completed(future_map):
                        key = future_map[future]
                        try:
                            _, value = future.result()
                            _write_preheated_value(
                                redis_conn,
                                key,
                                value,
                                without_gzip=key.startswith('dcalc:skill:'),
                            )
                            warmed_keys.append(key)
                        except Exception as exc:
                            print(f'Redis preheat failed for {key}: {exc}')

            max_workers = min(8, len(character_preheat_tasks))
            if max_workers > 0:
                with ThreadPoolExecutor(max_workers=max_workers) as executor:
                    future_map = {executor.submit(_run_preheat_task, key, loader): key for key, loader in character_preheat_tasks}
                    for future in as_completed(future_map):
                        key = future_map[future]
                        try:
                            _, value = future.result()
                            _write_preheated_value(redis_conn, key, value)
                            warmed_keys.append(key)
                        except Exception as exc:
                            print(f'Redis preheat failed for {key}: {exc}')
        finally:
            _release_preheat_lock(redis_conn, lock_token)

    return warmed_keys