import calendar
from datetime import datetime

import pendulum

from pot_libs.es_util.es_utils import EsUtil
from pot_libs.logger import log
from pot_libs.mysql_util.mysql_util import MysqlUtil
from unify_api import constants


def point_day2month(dt):
    if isinstance(dt, int) or isinstance(dt, float):
        dt = pendulum.from_timestamp(dt, tz="Asia/Shanghai")
        es_index = f"{constants.POINT_1MIN_INDEX}_{dt.year}_{dt.month}"

    elif isinstance(dt, datetime):
        es_index = f"{constants.POINT_1MIN_INDEX}_{dt.year}_{dt.month}"

    else:
        es_index = constants.POINT_1MIN_INDEX

    return es_index


async def today_alarm_cnt(cids):
    filters = [
        {"terms": {"cid": cids}},
        {"term": {"mode": "alarm"}},
    ]
    start_time = pendulum.today(tz="Asia/Shanghai")
    es_end_time = start_time.subtract(days=-1).format("YYYY-MM-DDTHH:mm:ss+08:00")
    es_start_time = start_time.format("YYYY-MM-DDTHH:mm:ss+08:00")
    filters.append({"range": {"datetime": {"gte": es_start_time, "lt": es_end_time,}}},)
    query_body = {
        "query": {"bool": {"filter": filters}},
        "size": 0,
        "aggs": {
            "cid_aggs": {
                "terms": {"field": "cid", "size": 10000},
                "aggs": {
                    "date_alarms": {
                        "date_histogram": {
                            "field": "datetime",
                            "order": {"_key": "desc"},
                            "min_doc_count": 0,
                            "interval": "day",
                            "format": "yyyy-MM-dd",
                            "time_zone": "+08:00",
                        }
                    }
                },
            }
        },
    }
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body, index=constants.POINT_1MIN_EVENT)

    cid_buckets = es_result.get("aggregations", {}).get("cid_aggs", {}).get("buckets", [])
    cid_bucket_map = {bucket["key"]: bucket for bucket in cid_buckets}

    now_time = datetime.now()
    cid_alarm_map = {cid: {"today_alarm_count": 0} for cid in cids}
    for cid in cids:
        bucket = {}
        if cid in cid_bucket_map:
            bucket = cid_bucket_map[cid]
        date_alarm_bucket = bucket.get("date_alarms", {}).get("buckets", [])
        for i in date_alarm_bucket:
            if i["key_as_string"] == str(now_time)[:10]:
                cid_alarm_map[cid]["today_alarm_count"] += i["doc_count"]
    return cid_alarm_map


async def proxy_safe_run_info(cids, start_time_str=None, end_time_str=None):
    """
    批量获取 各个工厂的安全运行天数以及今日报警数, 如果是获取月份的,那么计算这个月的安全运行天数
    :param cids:
    :return:
    """
    filters = [
        {"terms": {"cid": cids}},
        # {"term": {"mode": "alarm"}},
        {"term": {"importance": 1}},
    ]

    if start_time_str and end_time_str:
        start_dt = datetime.strptime(start_time_str, "%Y-%m-%d %H:%M:%S")
        end_dt = datetime.strptime(end_time_str, "%Y-%m-%d %H:%M:%S")
        now = datetime.now()
        if end_dt > now:
            end_dt = now
        es_start_str = datetime(
            year=start_dt.year, month=start_dt.month, day=start_dt.day
        ).strftime("%Y-%m-%dT%H:%M:%S+08:00")
        es_end_str = end_dt.strftime("%Y-%m-%dT%H:%M:%S+08:00")
        filters.append({"range": {"datetime": {"gte": es_start_str, "lt": es_end_str,}}},)

    query_body = {
        "query": {"bool": {"filter": filters}},
        "size": 0,
        "aggs": {
            "cid_aggs": {
                "terms": {"field": "cid", "size": 10000},
                "aggs": {
                    "date_alarms": {
                        "date_histogram": {
                            "field": "datetime",
                            "order": {"_key": "desc"},
                            "min_doc_count": 0,
                            "interval": "day",
                            "format": "yyyy-MM-dd",
                            "time_zone": "+08:00",
                        }
                    }
                },
            }
        },
    }

    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body, index=constants.POINT_1MIN_EVENT)

    now_time = datetime.now()
    # 获取到工厂安装时间create_time
    async with MysqlUtil() as conn:
        company_sql = "select cid, create_time from company where cid in %s"
        companys = await conn.fetchall(company_sql, (cids,))

    create_time_timestamp_map = {
        company["cid"]: datetime.fromtimestamp(company["create_time"]) for company in companys
    }
    cid_alarm_map = {cid: {"today_alarm_count": 0, "safe_run_days": 0} for cid in cids}

    cid_buckets = es_result.get("aggregations", {}).get("cid_aggs", {}).get("buckets", [])
    cid_bucket_map = {bucket["key"]: bucket for bucket in cid_buckets}
    for cid in cids:
        create_time = create_time_timestamp_map[cid]
        total_days = (now_time - create_time).days + 1
        if start_time_str and end_time_str:
            # 计算一段时间内安全运行天数,总天数的逻辑稍微不一样
            total_days = (end_dt.date() - start_dt.date()).days + 1
            if create_time > start_dt and create_time < end_dt:
                total_days = (end_dt.date() - create_time.date()).days + 1
            elif create_time > end_dt:
                total_days = 0

        has_alarm_days = 0
        bucket = {}
        if cid in cid_bucket_map:
            bucket = cid_bucket_map[cid]
        date_alarm_bucket = bucket.get("date_alarms", {}).get("buckets", [])
        for i in date_alarm_bucket:
            if i["doc_count"] != 0:
                # 没有报警，看做是安全运行了,统计累计安全运行的天数
                has_alarm_days += 1
        safe_run_days = total_days - has_alarm_days
        cid_alarm_map[cid]["safe_run_days"] = safe_run_days
        cid_alarm_map[cid]["total_days"] = total_days
    today_alarm_map = await today_alarm_cnt(cids)
    for cid in cid_alarm_map:
        cid_alarm_map[cid]["today_alarm_count"] = today_alarm_map[cid]["today_alarm_count"]
    return cid_alarm_map


async def alarm_time_distribution(company_ids, start, end):
    sql = f"""
        SELECT
            HOUR (pevent.event_datetime) event_hour,
            COUNT(*) event_count
        FROM
            point_1min_event pevent
        WHERE
            cid IN %s
        AND pevent.event_datetime >= '{start}'
        AND pevent.event_datetime <= '{end}'
        GROUP BY
            HOUR (pevent.event_datetime)
    """
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql, args=(company_ids, ))

    time_distribution_map = {"day_alarm_cnt": 0, "night_alarm_cnt": 0, "morning_alarm_cnt": 0}
    for data in datas:
        hour = int(data["event_hour"])
        if hour >= 6 and hour < 18:
            time_distribution_map["day_alarm_cnt"] += data["event_count"]
        elif hour >= 18 and hour <= 23:
            time_distribution_map["night_alarm_cnt"] += data["event_count"]
        else:
            time_distribution_map["morning_alarm_cnt"] += data["event_count"]
    return time_distribution_map


async def alarm_time_distribution_old(company_ids, start, end):
    start_dt = datetime.strptime(start, "%Y-%m-%d %H:%M:%S")
    end_dt = datetime.strptime(end, "%Y-%m-%d %H:%M:%S")
    es_start_str = datetime(year=start_dt.year, month=start_dt.month, day=start_dt.day).strftime(
        "%Y-%m-%dT%H:%M:%S+08:00"
    )
    es_end_str = end_dt.strftime("%Y-%m-%dT%H:%M:%S+08:00")
    filter_list = [
        {"range": {"datetime": {"gte": es_start_str, "lte": es_end_str, }}},
        {"terms": {"cid": company_ids}}]

    query_body = {
        "query": {"bool": {"filter": filter_list}},
        "size": 0,
        "aggs": {
            "cid_aggs": {
                "terms": {"field": "cid", "size": 10000},
                "aggs": {
                    "time_alarms": {
                        "date_histogram": {
                            "field": "datetime",
                            "order": {"_key": "desc"},
                            "min_doc_count": 1,
                            "interval": "hour",
                            "format": "HH",
                            "time_zone": "+08:00",
                        }
                    }
                },
            }
        },
    }

    log.info("alarm time distribute query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body, index=constants.POINT_1MIN_EVENT)
        print(f"alarm time distribute es_result = {es_result}")
    buckets = es_result["aggregations"]["cid_aggs"]["buckets"] or []
    time_distribution_map = {"day_alarm_cnt": 0, "night_alarm_cnt": 0, "morning_alarm_cnt": 0}
    for i in buckets:
        cid_buckets = i.get("time_alarms", {}).get("buckets", [])
        for item in cid_buckets:
            hour = int(item["key_as_string"])
            if hour >= 6 and hour < 18:
                time_distribution_map["day_alarm_cnt"] += item["doc_count"]
            elif hour >= 18 and hour <= 23:
                time_distribution_map["night_alarm_cnt"] += item["doc_count"]
            else:
                time_distribution_map["morning_alarm_cnt"] += item["doc_count"]
    return time_distribution_map
