import json
import time
import threading
import signal
from loguru import logger
from concurrent.futures import ThreadPoolExecutor, as_completed
from aidso_geo.config.base_config import init_redis8
from aidso_geo.models.process import main_process, process_call_back
from aidso_geo.utils import bh_utils

redis_client8 = init_redis8()

RUNNING = True

def stop_handler(signum, frame):
    global RUNNING
    logger.success("开始退出")
    RUNNING = False

def handle_item(data):
    if data.get('comWordsMap'):
        data['comWordsMap'] = json.loads(data.get('comWordsMap'))
    if data.get('brandWords'):
        data['brandWords'] = json.loads(data.get('brandWords'))
    if data.get('comWords'):
        data['comWords'] = json.loads(data.get('comWords'))
    if data.get('keywords'):
        data['keywords'] = json.loads(data.get('keywords'))
    if data.get('productWordsMap'):
        data['productWordsMap'] = json.loads(data.get('productWordsMap'))
    return main_process(data)

def handle_one_task(task):
    try:
        task_data = task.get("task_data")
        result = task.get("result")
        process_call_back(task_data, result)
    except Exception as e:
        print(f"回调任务处理失败: {e}, task={task}")


def run_stream_task():
    global RUNNING
    while RUNNING:
        status = 'ING'
        stream_task = bh_utils.query_data(

            f"select * from geo_commit_task where type = 'stream' and status = '{status}' limit 20"
        )

        if not stream_task:
            time.sleep(5)
            continue
        if stream_task:
            with ThreadPoolExecutor(max_workers=20) as executor:
                futures = [executor.submit(handle_item, item) for item in stream_task]

                for future in as_completed(futures):
                    try:
                        future.result()
                    except Exception as e:
                        print(f"任务执行失败: {e}")


def run_callback_task():
    global RUNNING
    REDIS_KEY = "geo:call_back:list"

    while RUNNING:
        batch = []

        # 一次从 redis 取 20 个
        for _ in range(20):
            result = redis_client8.rpop(REDIS_KEY)
            if not result:
                break
            batch.append(json.loads(result))

        # 没数据就休眠，避免死循环空转
        if not batch:
            time.sleep(1)
            continue

        # 20 个并发处理
        with ThreadPoolExecutor(max_workers=20) as executor:
            futures = [executor.submit(handle_one_task, task) for task in batch]

            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    print(f"线程执行异常: {e}")


if __name__ == '__main__':
    # signal.signal(signal.SIGINT, stop_handler)
    # signal.signal(signal.SIGTERM, stop_handler)

    t1 = threading.Thread(target=run_stream_task, name="stream-thread")
    # t2 = threading.Thread(target=run_callback_task, name="callback-thread")
    t1.start()
    # t2.start()
    # logger.success("stream-thread 线程已启动")
    # logger.success("callback-thread 线程已启动")
    t1.join()
    # t2.join()
    # logger.success("所有线程已退出")






