from pot_libs.es_util.es_utils import EsUtil
from pot_libs.logger import log
from unify_api import constants
from unify_api.constants import SDU_ALARM_LIST, SDU_ONE_TWO_GRADE_ALARM
from datetime import datetime
from pot_libs.mysql_util.mysql_util import MysqlUtil
from unify_api.utils.time_format import convert_es_str


async def sdu_alarm_statistics_dao(cid, start, end, is_sdu=True):
    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}}},
        {"term": {"cid": cid}},
    ]
    if is_sdu:
        # 新版sdu, 限制报警类型
        filter_list.append({"terms": {"type.keyword": SDU_ALARM_LIST}})
    query_body = {
        "size": 0,
        "query": {"bool": {"filter": filter_list}},
        "aggs": {
            "alarm_cnt": {
                "date_histogram": {
                    "field": "datetime",
                    "interval": "day",
                    "time_zone": "+08:00",
                    "format": "yyyy-MM-dd",
                    "min_doc_count": 0,
                    "extended_bounds": {"min": start_dt.strftime("%Y-%m-%d"),
                                        "max": end_dt.strftime("%Y-%m-%d")},
                },
                "aggs": {
                    "point_cnt": {
                        "terms": {"field": "point_id"}},
                    "appliance_cnt": {
                        "terms": {"field": "appliance.keyword"}},
                }
            },
        }
    }
    log.info("alarm_content_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)
    return es_result["aggregations"]["alarm_cnt"]["buckets"]


async def sdu_alarm_statistics_dao_new15(cid, start, end, is_sdu=True):
    mid_sql = ""
    if is_sdu:
        # 新版sdu, 限制报警类型
        mid_sql = f" and event_type in {tuple(SDU_ALARM_LIST)}"
    sql = f"SELECT pid,appliance,count(1) doc_count FROM `point_1min_event` " \
          f"where cid= {cid} and event_datetime BETWEEN '{start}' and '{end}' " \
          f"{mid_sql} GROUP BY pid, appliance ORDER BY pid"
    log.info(f"sql:{sql}")
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql)
    return datas


async def sdu_alarm_type_dao(start, end, points):
    """根据points分组, 再根据报警类型分组"""
    start_es = convert_es_str(start)
    end_es = convert_es_str(end)
    # 所有报警类型
    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "terms": {
                            "point_id": points
                        }
                    },
                    {
                        "range": {
                            "datetime": {
                                "gte": start_es,
                                "lte": end_es
                            }
                        }
                    }
                ]
            }
        },
        "aggs": {
            "points": {
                "terms": {
                    "field": "point_id",
                    "size": 1000
                },
                "aggs": {
                    "type": {
                        "terms": {
                            "field": "type.keyword",
                            "size": 20
                        }
                    }
                }
            }
        }
    }
    log.info("electric_behave query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    return es_result["aggregations"]["points"]["buckets"]


async def sdu_alarm_type_dao_new15(start, end, points):
    if points and len(points) == 1:
        mid_str = f"pid={points[0]}"
    else:
        mid_str = f"pid in {tuple(points)}"
    sql = f"SELECT pid,event_type,count(1) doc_count FROM point_1min_event " \
          f"where {mid_str} " \
          f"and event_datetime BETWEEN '{start}' and '{end}' " \
          f"GROUP BY pid, event_type ORDER BY pid"
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql)
    return datas


async def sdu_alarm_limit_type_dao(start, end, points):
    """根据points分组, 再根据报警类型分组, 限制type"""
    start_es = convert_es_str(start)
    end_es = convert_es_str(end)
    # 所有报警类型
    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "terms": {
                            "point_id": points
                        }
                    },
                    {
                        "range": {
                            "datetime": {
                                "gte": start_es,
                                "lte": end_es
                            }
                        }
                    },
                    # 新版sdu, 限制报警类型
                    {"terms": {"type.keyword": SDU_ALARM_LIST}}
                ]
            }
        },
        "aggs": {
            "points": {
                "terms": {
                    "field": "point_id",
                    "size": 1000
                },
                "aggs": {
                    "type": {
                        "terms": {
                            "field": "type.keyword",
                            "size": 20
                        }
                    }
                }
            }
        }
    }
    log.info("electric_behave query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    return es_result["aggregations"]["points"]["buckets"]


async def sdu_alarm_limit_type_dao_new15(start, end, points):
    sql = f"SELECT pid, event_type, count(1) doc_count " \
          f"FROM `point_1min_event` WHERE pid in %s and " \
          f"event_datetime BETWEEN '{start}' and '{end}' " \
          f"and event_type in {tuple(SDU_ALARM_LIST)} GROUP BY pid, event_type"
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql, args=(points, ))
    return datas


async def sdu_alarm_importance_dao(start, end, points, is_sdu=None):
    """根据points分组, 再根据报警等级分组"""
    start_es = convert_es_str(start)
    end_es = convert_es_str(end)

    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "terms": {
                            "point_id": points
                        }
                    },
                    {
                        "range": {
                            "datetime": {
                                "gte": start_es,
                                "lte": end_es
                            }
                        }
                    }
                ]
            }
        },
        "aggs": {
            "points": {
                "terms": {
                    "field": "point_id",
                    "size": 1000
                },
                "aggs": {
                    "importance": {
                        "terms": {
                            "field": "importance",
                            "size": 10
                        }
                    }
                }
            }
        }
    }
    if is_sdu:
        query_body["query"]["bool"]["must"].append(
            # 新版sdu, 限制报警类型
            {"terms": {"type.keyword": SDU_ALARM_LIST}}
        )
    log.info("electric_behave query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    return es_result["aggregations"]["points"]["buckets"]


async def sdu_alarm_importance_dao_new15(start, end, points, is_sdu=None):
    if points and len(points) == 1:
        mid_str = f"pid={points[0]}"
    else:
        mid_str = f"pid in {tuple(points)}"
    if is_sdu:
        mid_str += f" and event_type in {tuple(SDU_ALARM_LIST)}"
    sql = f"SELECT pid,importance,count(1) doc_count FROM point_1min_event " \
          f"where {mid_str} " \
          f"and event_datetime BETWEEN '{start}' and '{end}' " \
          f"GROUP BY pid, importance ORDER BY pid"
    log.info(f"sql:{sql}")
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql)
    return datas


async def sdu_alarm_behavior_dao(start, end, points):
    """用电行为统计, 目前只有违规电器
       如果还有其他统计, 则可先根据type分组, 再根据appliance分组
    """
    start_es = convert_es_str(start)
    end_es = convert_es_str(end)

    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "filter": [
                    {
                        "terms": {
                            "point_id": points
                        }
                    },
                    {
                        "range": {
                            "datetime": {
                                "gte": start_es,
                                "lte": end_es
                            }
                        }
                    },
                    {
                        "terms": {
                            "type.keyword": [
                                "illegal_ele_app"
                            ]
                        }
                    }
                ]
            }
        },
        "aggs": {
            "appliance": {
                "terms": {
                    "field": "appliance.keyword",
                    "size": 100
                }
            }
        }
    }
    log.info("alarm_behavior query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    return es_result["aggregations"]["appliance"]["buckets"]


async def sdu_alarm_behavior_dao_new15(start, end, points):
    sql = f"SELECT appliance, count(1) doc_count " \
          f"FROM `point_1min_event` WHERE pid in %s " \
          f"and event_type = 'illegal_ele_app' and " \
          f"event_datetime BETWEEN '{start}' and '{end}'GROUP BY appliance"
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql, args=(points,))
    return datas


async def sdu_alarm_aggs_date(cid):
    """sdu求安全运行
       根据每日聚合,再根据points聚合
    """
    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "term": {
                            "cid": cid
                        }
                    },
                    {"terms": {
                        "type.keyword": SDU_ONE_TWO_GRADE_ALARM}}
                ]
            }
        },
        "aggs": {
            "date_day": {
                "date_histogram": {
                    "field": "datetime",
                    "interval": "day",
                    "time_zone": "+08:00",
                    "format": "yyyy-MM-dd",
                    "min_doc_count": 0
                }
            }
        }
    }
    log.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)
    return es_result["aggregations"]["date_day"]["buckets"]


async def zdu_alarm_aggs_date_impotent(cid, start, end):
    """zdu求安全运行
       根据每日聚合,再根据等级聚合
    """
    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")
    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "term": {
                            "cid": cid
                        }
                    },
                    {
                        "range": {
                            "datetime": {
                                "gte": es_start_str,
                                "lte": es_end_str
                            }
                        }
                    }
                ]
            }
        },
        "aggs": {
            "date_day": {
                "date_histogram": {
                    "field": "datetime",
                    "interval": "day",
                    "time_zone": "+08:00",
                    "format": "yyyy-MM-dd",
                    "min_doc_count": 0,
                    "extended_bounds": {"min": start_dt.strftime("%Y-%m-%d"),
                                        "max": end_dt.strftime(
                                            "%Y-%m-%d")},
                },
                "aggs": {"important_cnt": {"terms": {"field": "importance"}}}
            }
        }
    }
    log.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)
        return es_result["aggregations"]["date_day"]["buckets"]


async def sdu_alarm_aggs_date_importance(cid):
    """按日期,再按等级聚合"""
    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "term": {
                            "cid": cid
                        }
                    }
                ]
            }
        },
        "aggs": {
            "date_day": {
                "date_histogram": {
                    "field": "datetime",
                    "interval": "day",
                    "time_zone": "+08:00",
                    "format": "yyyy-MM-dd",
                    "min_doc_count": 0
                },
                "aggs": {
                    "importance": {
                        "terms": {
                            "field": "importance",
                            "size": 10
                        }
                    }
                }
            }
        }
    }
    log.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)
    return es_result["aggregations"]["date_day"]["buckets"]


async def sdu_alarm_aggs_type(cid, start, end):
    sql = f"""
        SELECT 
            COUNT(*) doc_count, 
            event_type
        FROM
            point_1min_event pevent
        WHERE
            cid = {cid}
        AND pevent.event_datetime >= '{start}'
        AND pevent.event_datetime <= '{end}'
        GROUP BY
            pevent.event_type;
    """
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql)
    return datas if datas else []


async def sdu_alarm_aggs_type_old(cid, 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")
    """根据类型聚合"""
    query_body = {
        "size": 0,
        "query": {
            "bool": {
                "must": [
                    {
                        "term": {
                            "cid": cid
                        }
                    },
                    {
                        "range": {
                            "datetime": {
                                "gte": es_start_str,
                                "lte": es_end_str
                            }
                        }
                    }
                ]
            }
        },
        "aggs": {
            "type": {
                "terms": {
                    "field": "type.keyword",
                    "size": 40
                }
            }
        }
    }
    log.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)
    return es_result["aggregations"]["type"]["buckets"]


async def alarm_aggs_importance(cid, start, end):
    """按报警等级聚合"""
    sql = f"""
        SELECT
            COUNT(*) alarm_count,
            importance
        FROM
            point_1min_event pevent
        WHERE
            cid = {54}
        AND pevent.event_datetime >= '{start}'
        AND pevent.event_datetime <= '{end}'
        GROUP BY
            pevent.importance
    """
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql)
    return datas if datas else []
