import calendar
import random
from collections import defaultdict
from datetime import datetime

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
from unify_api.constants import Importance, Product
from unify_api.modules.common.procedures.common_cps import (
    proxy_safe_run_info,
    alarm_time_distribution,
)
from unify_api.utils.time_format import get_start_end_by_tz_time_new, \
    proxy_power_slots, day_slots


async def alarm_count_info(company_ids, start, end, date_type):
    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")
    if date_type == "day":
        interval = "hour"
        _format = "yyyy-MM-dd HH:mm:ss"
        _min = start_dt.strftime("%Y-%m-%d %H:%M:%S")
        _max = end_dt.strftime("%Y-%m-%d %H:%M:%S")
    
    else:
        # date_type == "month"
        interval = "day"
        _format = "yyyy-MM-dd"
        _min = start_dt.strftime("%Y-%m-%d")
        _max = end_dt.strftime("%Y-%m-%d")
    filter_list = [
        {"range": {"datetime": {"gte": es_start_str, "lte": es_end_str, }}},
        {"terms": {"cid": company_ids}}]
    
    query_body = {
        "size": 0,
        "query": {"bool": {"filter": filter_list, }},
        "aggs": {
            "alarm_cnt": {
                "date_histogram": {
                    "field": "datetime",
                    "interval": interval,
                    "time_zone": "+08:00",
                    "format": _format,
                    "min_doc_count": 0,
                    "extended_bounds": {"min": _min, "max": _max, },
                },
                "aggs": {"type_cnt": {"terms": {"field": "importance"}}},
            },
            "cid_aggs": {"terms": {"field": "cid"}},
            "type_aggs": {"terms": {"field": "type.keyword"}},
        },
    }
    
    log.info("alarm_count_info query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    buckets = es_result["aggregations"]["alarm_cnt"]["buckets"]
    first_alarm = {"slots": [], "value": [0] * len(buckets)}
    second_alarm = {"slots": [], "value": [0] * len(buckets)}
    third_alarm = {"slots": [], "value": [0] * len(buckets)}
    
    cid_buckets = es_result["aggregations"]["cid_aggs"]["buckets"]
    cid_alarm_cnt_map = {i["key"]: i["doc_count"] for i in cid_buckets}
    
    type_buckets = es_result["aggregations"]["type_aggs"]["buckets"]
    type_alarm_cnt_map = {i["key"]: i["doc_count"] for i in type_buckets}
    
    for index, bucket in enumerate(buckets):
        if date_type == "day":
            time_str = bucket["key_as_string"][11:16]
        else:
            time_str = bucket["key_as_string"][5:10]
        first_alarm["slots"].append(time_str)
        second_alarm["slots"].append(time_str)
        third_alarm["slots"].append(time_str)
        
        if bucket["type_cnt"]["buckets"]:
            for item in bucket["type_cnt"]["buckets"]:
                if item["key"] == Importance.First.value:
                    first_alarm["value"][index] += item["doc_count"]
                elif item["key"] == Importance.Second.value:
                    second_alarm["value"][index] += item["doc_count"]
                elif item["key"] == Importance.Third.value:
                    third_alarm["value"][index] += item["doc_count"]
    
    log.info(f"first_alarm={first_alarm}")
    log.info(f"second_alarm={second_alarm}")
    log.info(f"third_alarm={third_alarm}")
    return {
        "first_alarm": first_alarm,
        "second_alarm": second_alarm,
        "third_alarm": third_alarm,
        "cid_alarm_cnt_map": cid_alarm_cnt_map,
        "type_alarm_cnt_map": type_alarm_cnt_map,
    }


async def alarm_count_info_new15(company_ids, start, end, date_type):
    if date_type == "day":
        date_fmt = "DATE_FORMAT(event_datetime,'%%H')"
        slots = day_slots('hours')
    else:
        # date_type == "month"
        date_fmt = "DATE_FORMAT(event_datetime,'%%m-%%d')"
        slots = proxy_power_slots(start, end, "MM-DD", True)
    alarm_sql = f"""
        select {date_fmt} date,importance,count(*) count from point_1min_event
        where cid in %s and event_datetime between %s and %s
        group by {date_fmt},importance
    """
    cid_sql = f"""
        select cid,count(*) count from point_1min_event
        where cid in %s and event_datetime between %s and %s
        group by cid
    """
    type_sql = f"""
        select event_type,count(*) count from point_1min_event
        where cid in %s and event_datetime between %s and %s
        group by event_type
    """
    async with MysqlUtil() as conn:
        args = (company_ids, start, end)
        alarm_result = await conn.fetchall(sql=alarm_sql, args=args)
        cid_result = await conn.fetchall(sql=cid_sql, args=args)
        type_result = await conn.fetchall(sql=type_sql, args=args)
    
    first_alarm = {"slots": [], "value": [0] * len(slots)}
    second_alarm = {"slots": [], "value": [0] * len(slots)}
    third_alarm = {"slots": [], "value": [0] * len(slots)}
    
    cid_alarm_cnt_map = {i["cid"]: i["count"] for i in cid_result}
    type_alarm_cnt_map = {i["event_type"]: i["count"] for i in type_result}
    for index, slot in enumerate(slots):
        show_slot = slot + ":00" if date_type == "day" else slot
        first_alarm["slots"].append(show_slot)
        second_alarm["slots"].append(show_slot)
        third_alarm["slots"].append(show_slot)
        for item in alarm_result:
            if item.get("date") == slot:
                if item["importance"] == Importance.First.value:
                    first_alarm["value"][index] += item["count"]
                elif item["importance"] == Importance.Second.value:
                    second_alarm["value"][index] += item["count"]
                elif item["importance"] == Importance.Third.value:
                    third_alarm["value"][index] += item["count"]
    
    log.info(f"first_alarm={first_alarm}")
    log.info(f"second_alarm={second_alarm}")
    log.info(f"third_alarm={third_alarm}")
    return {
        "first_alarm": first_alarm,
        "second_alarm": second_alarm,
        "third_alarm": third_alarm,
        "cid_alarm_cnt_map": cid_alarm_cnt_map,
        "type_alarm_cnt_map": type_alarm_cnt_map,
    }


async def alarm_content_time_distribution(company_ids, start, end, date_type):
    """
    电参数,温度,漏电流时间分布
    :param company_ids:
    :param start:
    :param end:
    :param date_type:
    :return:
    """
    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")
    if date_type == "day":
        interval = "hour"
        _format = "yyyy-MM-dd HH:mm:ss"
        _min = start_dt.strftime("%Y-%m-%d %H:%M:%S")
        _max = end_dt.strftime("%Y-%m-%d %H:%M:%S")
    
    else:
        # date_type == "month"
        interval = "day"
        _format = "yyyy-MM-dd"
        _min = start_dt.strftime("%Y-%m-%d")
        _max = end_dt.strftime("%Y-%m-%d")
    filter_list = [
        {"range": {"datetime": {"gte": es_start_str, "lte": es_end_str, }}},
        {"terms": {"cid": company_ids}}]
    
    query_body = {
        "size": 0,
        "query": {"bool": {"filter": filter_list, }},
        "aggs": {
            "alarm_cnt": {
                "date_histogram": {
                    "field": "datetime",
                    "interval": interval,
                    "time_zone": "+08:00",
                    "format": _format,
                    "min_doc_count": 0,
                    "extended_bounds": {"min": _min, "max": _max, },
                },
                "aggs": {"type_cnt": {
                    "terms": {"field": "type.keyword", "size": 10000}}},
            }
        },
    }
    
    log.info("alarm_count_info query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    buckets = es_result["aggregations"]["alarm_cnt"]["buckets"]
    temperature = {"slots": [], "value": [0] * len(buckets)}
    residual_current = {"slots": [], "value": [0] * len(buckets)}
    electric_param = {"slots": [], "value": [0] * len(buckets)}
    electric_param_detail = {
        "harmonic": 0,
        "voltage": 0,
        "current": 0,
        "power_factor": 0,
        "threephase_imbalance": 0,
        "load_rate": 0,
    }
    
    for index, bucket in enumerate(buckets):
        if date_type == "day":
            time_str = bucket["key_as_string"][11:16]
        else:
            time_str = bucket["key_as_string"][5:10]
        temperature["slots"].append(time_str)
        residual_current["slots"].append(time_str)
        electric_param["slots"].append(time_str)
        
        if bucket["type_cnt"]["buckets"]:
            for item in bucket["type_cnt"]["buckets"]:
                if item["key"] in [
                    "overTemp",
                    "overTempRange1min",
                    "overTempRange15min",
                    "overTempTrendDaily",
                    "overTempTrendQuarterly",
                ]:
                    temperature["value"][index] += item["doc_count"]
                elif item["key"] in [
                    "overResidualCurrent",
                ]:
                    residual_current["value"][index] += item["doc_count"]
                else:
                    electric_param["value"][index] += item["doc_count"]
                    if item["key"] in [
                        "overTHDI",  # 电流总谐波有效值越限
                        "overTHDU",  # 电压总谐波畸变率越限
                    ]:
                        electric_param_detail["harmonic"] += item["doc_count"]
                    elif item["key"] in [
                        "overU",  # 过压
                        "underU",  # 欠压
                    ]:
                        electric_param_detail["voltage"] += item["doc_count"]
                    elif item["key"] in [
                        "overI",
                    ]:
                        electric_param_detail["current"] += item["doc_count"]
                    elif item["key"] in [
                        "underPhasePF",  # 单相功率因数越下限
                        "underTotalPF",  # 总功率因数越下限
                    ]:
                        electric_param_detail["power_factor"] += item[
                            "doc_count"]
                    elif item["key"] in [
                        "unbalanceI",  # 三相电流不平衡度
                        "unbalanceU",  # 三相电压不平衡度
                    ]:
                        electric_param_detail["threephase_imbalance"] += item[
                            "doc_count"]
                    elif item["key"] in ["overPR"]:
                        electric_param_detail["load_rate"] += item["doc_count"]
    
    log.info(f"temperature={temperature}")
    log.info(f"residual_current={residual_current}")
    log.info(f"electric_param={electric_param}")
    return {
        "temperature": temperature,
        "residual_current": residual_current,
        "electric_param": electric_param,
        "electric_param_detail": electric_param_detail,
    }


async def alarm_summary(company_ids, start, end, date_type):
    """
    电参数,温度,漏电流时间分布
    :param company_ids:
    :param start:
    :param end:
    :param date_type:
    :return:
    """
    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")
    if date_type == "day":
        _format = "yyyy-MM-dd HH:mm:ss"
        _min = start_dt.strftime("%Y-%m-%d %H:%M:%S")
        _max = end_dt.strftime("%Y-%m-%d %H:%M:%S")
    
    else:
        # date_type == "month"
        _format = "yyyy-MM-dd"
        _min = start_dt.strftime("%Y-%m-%d")
        _max = end_dt.strftime("%Y-%m-%d")
    filter_list = [
        {"range": {"datetime": {"gte": es_start_str, "lte": es_end_str, }}},
        {"term": {"mode": "alarm"}},
    ]
    
    filter_list.append({"terms": {"cid": company_ids}})
    
    query_body = {
        "query": {"bool": {"filter": filter_list}},
        "size": 0,
        "aggs": {
            "cid_aggs": {
                "terms": {"field": "cid", "size": 10000},
                "aggs": {
                    "date_alarms": {
                        "date_histogram": {
                            "field": "datetime",
                            "order": {"_key": "desc"},
                            "min_doc_count": 1,
                            "interval": "day",
                            "format": "yyyy-MM-dd",
                            "time_zone": "+08:00",
                        }
                    }
                },
            }
        },
    }
    
    log.info("alarm_summary 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"es_result = {es_result}")
    buckets = es_result["aggregations"]["cid_aggs"]["buckets"] or []
    total_alarm_cnt, alarm_company_cnt = sum(
        [i["doc_count"] for i in buckets]), len(buckets)
    cid_alarmcnt_list = [i["doc_count"] for i in buckets]
    
    safe_run_map = await proxy_safe_run_info(company_ids, start_time_str=start,
                                             end_time_str=end)
    log.info(f"alarm_summary safe_run_map======{safe_run_map}")
    safe_run_days = sum(
        [safe_run_map[cid]["safe_run_days"] for cid in safe_run_map])
    
    # 时间分布： 白天， 黑夜, 凌晨
    time_distribution_map = await alarm_time_distribution(company_ids, start,
                                                          end)
    
    total_valid_company = 0
    for cid in safe_run_map:
        # 选择的这段时间，客户必须已经接入进来才算
        if safe_run_map[cid]["total_days"] != 0:
            total_valid_company += 1
    summary_map = {
        "total_alarm_cnt": total_alarm_cnt,
        "alarm_company_cnt": alarm_company_cnt,
        "avg_alarm_cnt": round(total_alarm_cnt / alarm_company_cnt,
                               1) if alarm_company_cnt else 0,
        "max_alarm_cnt": max(cid_alarmcnt_list) if cid_alarmcnt_list else 0,
        "safe_run_days": safe_run_days,
        "avg_safe_run_days": round(safe_run_days / total_valid_company,
                                   1) if total_valid_company else 0,
        "day_alarm_cnt": time_distribution_map["day_alarm_cnt"],
        "night_alarm_cnt": time_distribution_map["night_alarm_cnt"],
        "morning_alarm_cnt": time_distribution_map["morning_alarm_cnt"],
    }
    return summary_map
