import time
import requests
import json

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

from aidso_geo.config.base_config import init_redis, init_redis8
from aidso_geo.utils import bh_utils,tos_utils

owner_map = {
    "XHSA": "崔士豪",
    "BDAI": "崔士豪",
    "KIMI": "崔士豪",
    "TXYBA": "崔士豪",
    "TYQWA": "崔士豪",
    "WXYY": "崔士豪",
    "DB": "李聪健",
    "DOUBA": "李聪健",
    "DP": "李聪健",
    "DPA": "李聪健",
    "DYAI": "李聪健",
    "TXYB": "李聪健",
    "TYQW": "李聪健",
}

platform_list = [
    "BDAI", "DB", "DOUBA", "DP", "DPA", "DYAI",
    "KIMI", "TXYB", "TXYBA", "TYQW", "TYQWA", "WXYY", "XHSA"
]


def feishu_tobot(data):
    url = "https://open.feishu.cn/open-apis/bot/v2/hook/151c5019-9f74-4a94-80b0-f0d643616c9f"

    payload = json.dumps({
        "msg_type": "text",
        "content": {
            "text": f"{data} 解析失败"
        }
    })
    headers = {
        'Content-Type': 'application/json'
    }

    try:
        response = requests.request("POST", url, headers=headers, data=payload)
        logger.info(f"feishu_tobot 发送完成，status_code={response.status_code}，response={response.text}")
    except Exception as e:
        logger.exception(f"feishu_tobot 发送失败: {e}")


def feishu_tobot_sse(data):
    url = "https://open.feishu.cn/open-apis/bot/v2/hook/151c5019-9f74-4a94-80b0-f0d643616c9f"

    payload = json.dumps({
        "msg_type": "text",
        "content": {
            "text": f"{data}"
        }
    })
    headers = {
        'Content-Type': 'application/json'
    }

    try:
        response = requests.request("POST", url, headers=headers, data=payload)
        logger.info(f"feishu_tobot 发送完成，status_code={response.status_code}，response={response.text}")
    except Exception as e:
        logger.exception(f"feishu_tobot_sse 发送失败: {e}")


def fail_task_send_feishu():
    logger.info("开始执行 fail_task_send_feishu")
    start_time = time.time()

    try:
        ts = int(time.time()) - 6 * 60 * 60

        data = bh_utils.query_data(f"""
                select
                    platform,
                    count() as total_cnt,
                    sum(case when type = 'success' then 1 else 0 end) as fail_cnt,
                    round(sum(case when type = 'success' then 1 else 0 end) / count() * 100, 2) as fail_rate
                from geo_commit_task
                where insertime > {ts}
                group by platform
                order by platform
            """)

        logger.info(f"fail_task_send_feishu 查询完成，返回平台数={len(data) if data else 0}")

        grouped = {}
        abnormal_list = []

        for item in data:
            platform = item.get("platform")
            owner = owner_map.get(platform, "未知负责人")
            total_cnt = item.get("total_cnt", 0)
            fail_cnt = item.get("fail_cnt", 0)
            fail_rate = item.get("fail_rate", 0)

            row = {
                "platform": platform,
                "total_cnt": total_cnt,
                "fail_cnt": fail_cnt,
                "fail_rate": fail_rate,
            }

            grouped.setdefault(owner, []).append(row)

            if fail_cnt > 0:
                abnormal_list.append({
                    "owner": owner,
                    "platform": platform,
                    "fail_cnt": fail_cnt,
                    "fail_rate": fail_rate,
                })

        for owner in grouped:
            grouped[owner].sort(key=lambda x: (-x["fail_cnt"], -x["fail_rate"], x["platform"]))

        abnormal_list.sort(key=lambda x: (-x["fail_cnt"], -x["fail_rate"], x["platform"]))

        result = "【过去六小时平台失败监控】\n\n"

        if abnormal_list:
            result += "❗异常摘要\n"
            for item in abnormal_list:
                result += (
                    f"- {item['owner']}：{item['platform']} "
                    f"失败 {item['fail_cnt']}，失败率 {item['fail_rate']}%\n"
                )
        else:
            result += "✅ 异常摘要：无\n"

        result += "\n"

        owner_order = ["李聪健", "崔士豪", "未知负责人"]

        for owner in owner_order:
            items = grouped.get(owner, [])
            if not items:
                continue

            result += f"👤 {owner}\n"
            for item in items:
                result += (
                    f"- {item['platform']}："
                    f"总任务 {item['total_cnt']}，"
                    f"失败 {item['fail_cnt']}，"
                    f"失败率 {item['fail_rate']}%\n"
                )
            result += "\n"

        feishu_tobot_sse(result)
        logger.info(f"fail_task_send_feishu 执行完成，异常平台数={len(abnormal_list)}，耗时={round(time.time() - start_time, 2)}s")

    except Exception as e:
        logger.exception(f"fail_task_send_feishu 执行失败: {e}")


def task_queue_backlog():
    logger.info("开始执行 task_queue_backlog")
    start_time = time.time()

    try:
        r = init_redis()

        grouped = {}
        abnormal_list = []

        for platform in platform_list:
            owner = owner_map.get(platform, "未知负责人")

            stream_batch_key = f"{platform}:geo:stream_batch:list"
            batch_key = f"{platform}:geo:batch:list"

            stream_batch_cnt = r.llen(stream_batch_key)
            batch_cnt = r.llen(batch_key)

            row = {
                "platform": platform,
                "stream_batch_cnt": stream_batch_cnt,
                "batch_cnt": batch_cnt,
            }

            grouped.setdefault(owner, []).append(row)

            if stream_batch_cnt > 0:
                abnormal_list.append({
                    "owner": owner,
                    "platform": platform,
                    "task_type": "stream_batch",
                    "cnt": stream_batch_cnt,
                })

            if batch_cnt > 0:
                abnormal_list.append({
                    "owner": owner,
                    "platform": platform,
                    "task_type": "batch",
                    "cnt": batch_cnt,
                })

        for owner in grouped:
            grouped[owner].sort(
                key=lambda x: (-(x["stream_batch_cnt"] + x["batch_cnt"]), x["platform"])
            )

        abnormal_list.sort(key=lambda x: (-x["cnt"], x["owner"], x["platform"], x["task_type"]))

        result = "【平台队列积压监控】\n\n"

        if abnormal_list:
            result += "❗异常摘要\n"
            for item in abnormal_list:
                result += (
                    f"- {item['owner']}：{item['platform']} "
                    f"{item['task_type']} 积压 {item['cnt']}\n"
                )
            result += "\n"

            owner_order = ["李聪健", "崔士豪", "未知负责人"]

            for owner in owner_order:
                items = grouped.get(owner, [])
                if not items:
                    continue

                result += f"👤 {owner}\n"
                for item in items:
                    result += (
                        f"- {item['platform']}："
                        f"stream_batch 积压 {item['stream_batch_cnt']}，"
                        f"batch 积压 {item['batch_cnt']}\n"
                    )
                result += "\n"

            feishu_tobot_sse(result)
            logger.info(
                f"task_queue_backlog 执行完成，异常项数={len(abnormal_list)}，耗时={round(time.time() - start_time, 2)}s")
        else:
            logger.info(f"task_queue_backlog 无异常，不发送飞书，耗时={round(time.time() - start_time, 2)}s")

    except Exception as e:
        logger.exception(f"task_queue_backlog 执行失败: {e}")

from datetime import datetime

redis_client = init_redis8()


def safe_int(value, default=0):
    try:
        if value is None:
            return default
        return int(value)
    except Exception:
        return default


def get_active_channels():
    """
    查询所有启用中的第三方 channel
    """
    sql = """
        select channel, daily_limit, total_limit
        from geo_third_token
        where status = 1
    """
    return bh_utils.query_data(sql) or []


def sync_third_usage_once():
    """
    每小时同步第三方 token 用量快照。

    读取：
    1. Redis hash third_geo_daily：field = channel + YYYYMMDD
    2. Redis hash third_geo_total：field = channel

    写入：
    geo_third_usage_snapshot
    """
    now = datetime.now()
    today_pt = now.strftime("%Y%m%d")
    sync_time = now.strftime("%Y-%m-%d %H:%M:%S")
    insertime = int(time.time())

    try:
        channels = get_active_channels()
        if not channels:
            logger.info(f"[{sync_time}] sync_third_usage_once 无启用中的 channel")
            return

        # 一次性拉 Redis hash，避免用户多时频繁 hget
        daily_map = redis_client.hgetall("third_geo_daily") or {}
        total_map = redis_client.hgetall("third_geo_total") or {}

        insert_rows = []

        for item in channels:
            channel = str(item.get("channel", "")).strip()
            if not channel:
                continue

            daily_limit = safe_int(item.get("daily_limit"), 0)
            total_limit = safe_int(item.get("total_limit"), 0)

            daily_key = f"{channel}{today_pt}"

            today_used = safe_int(daily_map.get(daily_key), 0)
            total_used = safe_int(total_map.get(channel), 0)

            insert_rows.append({
                "channel": channel,
                "pt": today_pt,
                "daily_limit": daily_limit,
                "total_limit": total_limit,
                "today_used": today_used,
                "total_used": total_used,
                "today_remain": max(daily_limit - today_used, 0),
                "total_remain": max(total_limit - total_used, 0),
                "sync_time": sync_time,
                "insertime": insertime
            })

        if not insert_rows:
            logger.info(f"[{sync_time}] sync_third_usage_once 无可写入数据")
            return

        ok = bh_utils.insert_data("geo_third_usage_snapshot", insert_rows)
        if ok:
            logger.info(f"[{sync_time}] sync_third_usage_once 同步成功 rows={len(insert_rows)}")
        else:
            logger.error(f"[{sync_time}] sync_third_usage_once 同步失败 rows={len(insert_rows)}")

    except Exception as e:
        logger.exception(f"[{sync_time}] sync_third_usage_once 执行异常: {e}")


def save_image():
    try:
        result = bh_utils.query_data("""
            select distinct t1.eco_id, t1.eco_pic
        from geo_eco_data t1
                 left join geo_eco_image_data t2 on t1.eco_id = t2.eco_id
        where t2.eco_id is null limit 10000
            """)
        bh_result = []
        for r in result:
            eco_pic = r.get("eco_pic")
            eco_id = r.get("eco_id")
            logger.info(f"{eco_id}处理成功")
            tt = tos_utils.save_image_to_tos(r.get("eco_pic"), f'geo_image/{r.get("eco_id")}')
            bh_result.append({"eco_id": eco_id, "eco_pic": eco_pic, "eco_pic_tos": tt})
        bh_utils.insert_data("geo_eco_image_data", bh_result)
    except Exception as e:
        logger.exception(f"task_queue_backlog 执行失败: {e}")


if __name__ == '__main__':
    logger.info("监控调度器启动")
    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
    scheduler.add_job(
        fail_task_send_feishu,
        trigger='cron',
        minute=10,
        id='fail_task_send_feishu',
        max_instances=1,
        coalesce=True,
        replace_existing=True
    )


    scheduler.add_job(
        task_queue_backlog,
        trigger='cron',
        minute=0,
        id='task_queue_backlog',
        max_instances=1,
        coalesce=True,
        replace_existing=True
    )


    scheduler.add_job(
        save_image,
        trigger='cron',
        minute=30,
        id='save_image',
        max_instances=1,
        coalesce=True,
        replace_existing=True
    )

    scheduler.add_job(
        sync_third_usage_once,
        trigger='cron',
        minute=5,
        id='sync_third_usage_once',
        max_instances=1,
        coalesce=True,
        replace_existing=True
    )

    logger.info(
        "定时任务注册完成："
        "fail_task_send_feishu(每6小时), "
        "task_queue_backlog(每小时)"
        "save_image(每小时)"
        "sync_third_usage_once(每小时)"
    )
    scheduler.start()
