
from loguru import logger
import os
import sys
import json
from apscheduler.schedulers.blocking import BlockingScheduler
import redis


BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(BASE_DIR)
from aidso_geo.utils import bh_utils


_redis8_client = None
_redis8_pool = None
def init_redis8():
    global _redis8_client, _redis8_pool

    if _redis8_client is not None:
        return _redis8_client

    try:
        _redis8_pool = redis.ConnectionPool(
            host='redis-cnlfmu7rl14awitrz.redis.ivolces.com',
            port=6379,
            db=8,
            password='aiyingli@@123',
            socket_timeout=120,
            socket_connect_timeout=5,
            decode_responses=True,
            retry_on_timeout=True,
            health_check_interval=30,
            socket_keepalive=True,
            max_connections=100,
        )

        _redis8_client = redis.Redis(connection_pool=_redis8_pool)
        _redis8_client.ping()
        return _redis8_client
    except Exception as e:
        logger.error(f"Redis 初始化失败: {e}")
        _redis8_client = None
        _redis8_pool = None
        return None




KEEP_FIELDS = (
    "prompt",
    "taskId",
    "reqId",
    "platform",
    "type",
    "thinkingEnabled",
    "channel",
    "is_share"
)


def task_commit_api():
    query_queue = bh_utils.query_data(
        """
select t1.reqId,
t1.prompt,
t1.taskId,
t1.platform,
t1.type,
t1.insertime,
t1.status,
t1.thinkingEnabled,
t1.channel,
ifnull(t1.is_share,0) as is_share
from (SELECT *
      FROM geo_third_task_data
      where status = 'ING') t1
         left join (select * from geo_commit_task where pt > date_format(date_sub(now(), 10), '%Y%m%d')) t2
                   on t1.reqId = t2.reqId
where t2.reqId is null 
        """
    )

    if not query_queue:
        logger.info("task_commit_api 没有待提交任务")
        return

    # 普通任务
    normal_values = []

    channel_one_values = []
    for task_data in query_queue:
        payload = {
            key: task_data.get(key)
            for key in KEEP_FIELDS
        }

        value = json.dumps(payload, ensure_ascii=False)

        if task_data.get("platform") =="DB":
            channel_one_values.append(value)
        else:
            normal_values.append(value)

        # 修改数据库记录状态
        task_data["status"] = "PROCESSING"

    redis_client8 = init_redis8()
    pipeline = redis_client8.pipeline(transaction=True)

    if normal_values:
        pipeline.lpush(
            "geo:task_commit:list",
            *normal_values
        )

    if channel_one_values:
        pipeline.lpush(
            "geo:task_commit:JIKE:list",
            *channel_one_values
        )
        logger.info(
            f"令狐豆包 提交任务数量：{len(channel_one_values)}")

    pipeline.execute()
    bh_utils.insert_data(
        "geo_third_task_data",
        query_queue,
    )

    logger.info(
        f"task_commit_api 提交任务数量：{len(normal_values)}")

def get_result_api():
    try:
        query_result = bh_utils.query_data("""
            select
        t1.reqId,
        t1.prompt,
        t1.taskId,
        t1.platform,
        t1.type,
        t1.insertime,
        t2.status,
        t1.thinkingEnabled,
        t1.channel,
        t1.is_share
         from (
        SELECT *
                FROM geo_third_task_data
                WHERE status = 'PROCESSING'
                ) t1   left join (select * from geo_commit_task where pt > date_format(date_sub(now(), 10), '%Y%m%d')) t2 on t1.reqId = t2.reqId where t2.status = 'SUCCESS'
            """)

        if not query_result:
            logger.info("get_result_api 没有完成任务")
            return

        bh_utils.insert_data(
            "geo_third_task_data",
            query_result,
        )
        logger.info(
            f"get_result_api 完成任务数量：{len(query_result)}")
    except Exception as e:
        logger.error(e)


def task_commit_api1():
    query_queue = bh_utils.query_data(
        """
select t1.reqId,
       t1.prompt,
       t1.taskId,
       t1.platform,
       t1.type,
       t1.insertime,
       t1.status,
       t1.thinkingEnabled,
       t1.channel
from (SELECT *
      FROM geo_third_task_data
      where platform = 'DB' ) t1
         left join (select * from geo_commit_task where pt > date_format(date_sub(now(), 10), '%Y%m%d')) t2
                   on t1.reqId = t2.reqId  where t2.status ='ING'
        """
    )

    if not query_queue:
        logger.info("task_commit_api 没有待提交任务")
        return

    # 普通任务
    normal_values = []

    channel_one_values = []
    for task_data in query_queue:
        payload = {
            key: task_data.get(key)
            for key in KEEP_FIELDS
        }

        value = json.dumps(payload, ensure_ascii=False)

        if task_data.get("platform") =="DB":
            channel_one_values.append(value)
        else:
            normal_values.append(value)

        # 修改数据库记录状态
        task_data["status"] = "PROCESSING"

    redis_client8 = init_redis8()
    pipeline = redis_client8.pipeline(transaction=True)

    if normal_values:
        pipeline.lpush(
            "geo:task_commit:list",
            *normal_values
        )

    if channel_one_values:
        pipeline.lpush(
            "geo:task_commit:JIKE:list",
            *channel_one_values
        )
        logger.info(
            f"令狐豆包 提交任务数量：{len(channel_one_values)}")

    pipeline.execute()
    bh_utils.insert_data(
        "geo_third_task_data",
        query_queue,
    )

    logger.info(
        f"task_commit_api 提交任务数量：{len(normal_values)}")

if __name__ == "__main__":
    logger.info("调度 启动")
    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
    scheduler.add_job(
        task_commit_api,
        trigger="interval",
        seconds=30,
        id='task_commit_api',
        max_instances=1,
        coalesce=True,
        replace_existing=True
    )

    scheduler.add_job(
        get_result_api,
        trigger="interval",
        seconds=30,
        id='get_result_api',
        max_instances=1,
        coalesce=True,
        replace_existing=True
    )
    scheduler.start()
    logger.info("third_task_commit 启动")
    logger.info("third_task_check 启动")



