import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from loguru import logger
from apscheduler.schedulers.blocking import BlockingScheduler

from aidso_geo.config.base_config import init_redis, init_redis8
from aidso_geo.models.process import main_process, process_call_back

# redis_client = init_redis()
redis_client = init_redis8()

QUEUE_KEY = "geo:call_back:list"

# 每轮最多拉取任务数量
BATCH_SIZE = 50

# 每 30 秒执行一次
INTERVAL_SECONDS = 20

# 每轮内部并发数
CONCURRENT_WORKERS = 50


def parse_task(raw):
    try:
        if isinstance(raw, bytes):
            raw = raw.decode("utf-8")

        return json.loads(raw)

    except Exception as e:
        return None




def safe_json_loads(value, default=None):
    if value is None:
        return default

    # 已经是 list/dict 了，直接返回
    if isinstance(value, (list, dict)):
        return value

    # Redis 读出来有时是 bytes
    if isinstance(value, bytes):
        value = value.decode("utf-8")

    # 字符串才尝试 json.loads
    if isinstance(value, str):
        value = value.strip()
        if value == "":
            return default

        try:
            return json.loads(value)
        except Exception as e:
            return default

    return default


def pull_tasks():
    """
    每次最多从 Redis 队列拉取 BATCH_SIZE 条任务
    Redis 连接异常时，只停止本轮拉取，不影响下次定时任务继续跑
    """
    task_list = []

    for _ in range(BATCH_SIZE):
        try:
            raw = redis_client.rpop(QUEUE_KEY)

            if not raw:
                break

            data = parse_task(raw)

            if data:
                task_list.append(data)

        except (ConnectionError, TimeoutError) as e:
            logger.warning(f"Redis连接异常，本轮停止拉取: {e}")

            try:
                redis_client.connection_pool.disconnect()
            except Exception:
                pass

            break

        except Exception as e:
            logger.exception(f"拉取队列任务异常: {e}")
            break

    return task_list


def handle_one_task(task):
    """
    单条任务处理。
    成功：正常结束
    失败：重新放回 Redis 队列
    """
    task_data = task.get("task_data")
    result = task.get("result")
    req_id = task_data.get('reqId')
    platform = task_data.get('platform')

    task_data["comWordsMap"] = safe_json_loads(task_data.get("comWordsMap"), {})
    task_data["brandWords"] = safe_json_loads(task_data.get("brandWords"), [])
    task_data["comWords"] = safe_json_loads(task_data.get("comWords"), [])
    task_data["keywords"] = safe_json_loads(task_data.get("keywords"), [])
    task_data["productWordsMap"] = safe_json_loads(task_data.get("productWordsMap"), {})

    try:
        process_call_back(task_data,result)
        return True

    except Exception as e:
        logger.exception(f"{req_id}--{platform}--处理失败，重新放回队列: {e}")

        redis_client.lpush(
            QUEUE_KEY,
            json.dumps(task, ensure_ascii=False)
        )

        return False


def consume_call_back():
    """
    定时任务：
    1. 拉取 Redis 队列
    2. 使用线程池并发执行 main_process
    """
    start_time = time.time()

    try:
        tasks = pull_tasks()

        if not tasks:
            logger.info("本轮没有任务")
            return

        with ThreadPoolExecutor(max_workers=CONCURRENT_WORKERS) as executor:
            futures = [
                executor.submit(handle_one_task, task)
                for task in tasks
            ]

            for future in as_completed(futures):
                try:
                    ok = future.result()

                except Exception as e:
                    logger.exception(f"线程任务异常: {e}")

        cost = round(time.time() - start_time, 2)

        logger.success(
            f"本轮处理完成，总数: {len(tasks)}, "
            f"耗时: {cost}s"
        )

    except Exception as e:
        logger.exception(f"定时消费任务异常: {e}")


if __name__ == "__main__":
    scheduler = BlockingScheduler(timezone="Asia/Shanghai")

    scheduler.add_job(
        consume_call_back,
        trigger="interval",
        seconds=INTERVAL_SECONDS,
        id="consume_call_back",
        max_instances=2,       # 允许最多 5 个调度批次同时跑
        coalesce=True,
        misfire_grace_time=300
    )

    logger.success(
        f"geo task commit consumer 启动，每 {INTERVAL_SECONDS}s 执行一次，"
        f"QUEUE_KEY={QUEUE_KEY}, "
        f"BATCH_SIZE={BATCH_SIZE}, "
        f"CONCURRENT_WORKERS={CONCURRENT_WORKERS}, "
        f"max_instances=2"
    )


    scheduler.start()