import os
import sqlite3
from typing import List

import jinja2
import numpy as np
import pandas as pd


def render_html(params: dict) -> str:
    searchpath = os.path.dirname(os.path.realpath(__file__))
    env = jinja2.Environment(loader=jinja2.FileSystemLoader(searchpath=searchpath))
    template = env.get_template('template.html')
    return template.render(params)


def render_simple_html(params: dict) -> str:
    searchpath = os.path.dirname(os.path.realpath(__file__))
    env = jinja2.Environment(loader=jinja2.FileSystemLoader(searchpath=searchpath))
    template = env.get_template('simple-template.html')
    return template.render(params)


def fix_data_type(df: pd.DataFrame) -> pd.DataFrame:
    df['Time'] = pd.to_datetime(df['Time'].astype(int), unit='s')
    df['Query_time'] = df.Query_time.apply(lambda x: float(x) if len(x) > 0 else 0)
    df['Mem_max'] = df.Mem_max.apply(lambda x: int(x) if x != '' else 0)
    df['Is_internal'] = df.Is_internal.apply(lambda x: x == 'true')
    return df


def explode_table(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df['table'] = df.tables.str.split(';').apply(lambda xs: list(set(xs)))
    return df[df.tables != ''].explode('table')


class SlowlogAnalyzer:
    def __init__(self, filepathlist: list):
        dfs = []
        for filepath in filepathlist:
            conn = sqlite3.connect(filepath)
            cur = conn.cursor()
            # get table names
            cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
            tables = cur.fetchall()
            for t in tables:
                table_name = t[0]
                if table_name == 'slowlog':
                    tempdf = pd.read_sql_query("select * from slowlog", conn)
                    dfs.append(tempdf)
            conn.close()
        frames = pd.concat(dfs)
        self.dataframe = fix_data_type(frames).drop_duplicates()
        self.internal_dataframe = self.dataframe[self.dataframe.Is_internal]
        self.dataframe = self.dataframe[~self.dataframe.Is_internal]

    @property
    def total_query_time(self) -> float:
        return self.dataframe.Query_time.sum()

    @property
    def new_digest_current(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        digest = self.dataframe[self.dataframe.fresh_sql_digest != ''].fresh_sql_digest
        df = self.dataframe[self.dataframe.Digest.isin(digest)].groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def digest_statistics(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return {
            'order_by_time': df.sort_values('query_time', ascending=False).to_dict('records'),
            'order_by_count': df.sort_values('count', ascending=False).to_dict('records'),
        }

    @property
    def new_database_current(self) -> List[dict]:
        dbs = self.dataframe[self.dataframe.fresh_db != ''].fresh_db
        df = self.dataframe[self.dataframe.DB.isin(dbs)]
        df = df.groupby('DB').Query_time.sum().to_frame('query_time').reset_index()
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def new_table_current(self) -> List[dict]:
        xs = self.dataframe[self.dataframe.fresh_table != ''].fresh_table.str.split(';').to_list()
        fresh_tables = set([x for ys in xs for x in ys if x != ''])
        selector = self.dataframe.tables.str.split(';').apply(lambda xs: len(fresh_tables.intersection(xs)) > 0)
        df = explode_table(self.dataframe[selector])
        if len(df) == 0:
            return []

        df = df.groupby(['DB', 'table']).Query_time.sum().to_frame('query_time').reset_index()
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def new_digest_query_time(self) -> float:
        digest = self.dataframe[self.dataframe.fresh_sql_digest != ''].fresh_sql_digest
        return self.dataframe[self.dataframe.Digest.isin(digest)].Query_time.sum()

    @property
    def new_database_query_time(self) -> float:
        dbs = self.dataframe[self.dataframe.fresh_db != ''].fresh_db
        return self.dataframe[self.dataframe.DB.isin(dbs)].Query_time.sum()

    @property
    def new_table_query_time(self) -> float:
        xs = self.dataframe[self.dataframe.fresh_table != ''].fresh_table.str.split(';').to_list()
        fresh_tables = set([x for ys in xs for x in ys if x != ''])
        selector = self.dataframe.tables.str.split(';').apply(lambda xs: len(fresh_tables.intersection(xs)) > 0)
        return self.dataframe[selector].Query_time.sum()

    @property
    def total_digest_count(self) -> int:
        return len(self.dataframe[['DB', 'Digest']].drop_duplicates())

    @property
    def tikv_query_count(self) -> int:
        return len(self.dataframe[self.dataframe.cop_path == 'tikv'])

    @property
    def tiflash_query_count(self) -> int:
        return len(self.dataframe[self.dataframe.cop_path == 'tiflash'])

    @property
    def mixed_query_count(self) -> int:
        selector = (self.dataframe.cop_path == 'tiflash;tikv') | (self.dataframe.cop_path == 'tikv;tiflash')
        return len(self.dataframe[selector])

    @property
    def total_query_time(self) -> float:
        return self.dataframe.Query_time.sum()

    @property
    def new_digest_current(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        digest = self.dataframe[self.dataframe.fresh_sql_digest != ''].fresh_sql_digest
        df = self.dataframe[self.dataframe.Digest.isin(digest)].groupby(['DB', 'Digest']).apply(agg).reset_index()
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def digest_statistics(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest']).apply(agg).reset_index()
        return {
            'order_by_time': df.sort_values('query_time', ascending=False).to_dict('records'),
            'order_by_count': df.sort_values('count', ascending=False).to_dict('records'),
        }

    @property
    def new_database_current(self) -> List[dict]:
        dbs = self.dataframe[self.dataframe.fresh_db != ''].fresh_db
        df = self.dataframe[self.dataframe.DB.isin(dbs)]
        qt = df.groupby('DB').Query_time.sum().to_frame('query_time').reset_index()
        fa = df.groupby('DB').Time.min().astype(str).to_frame('first_appearance').reset_index()
        return pd.merge(qt, fa).sort_values('query_time', ascending=False).to_dict('records')

    @property
    def new_table_current(self) -> List[dict]:
        xs = self.dataframe[self.dataframe.fresh_table != ''].fresh_table.str.split(';').to_list()
        fresh_tables = set([x for ys in xs for x in ys if x != ''])
        selector = self.dataframe.tables.str.split(';').apply(lambda xs: len(fresh_tables.intersection(xs)) > 0)
        df = explode_table(self.dataframe[selector])
        if len(df) == 0:
            return []

        qt = df.groupby(['DB', 'table']).Query_time.sum().to_frame('query_time').reset_index()
        fa = df.groupby(['DB', 'table']).Time.min().astype(str).to_frame('first_appearance').reset_index()
        return pd.merge(qt, fa).sort_values('query_time', ascending=False).to_dict('records')

    @property
    def new_digest_query_time(self) -> float:
        digest = self.dataframe[self.dataframe.fresh_sql_digest != ''].fresh_sql_digest
        return self.dataframe[self.dataframe.Digest.isin(digest)].Query_time.sum()

    @property
    def new_database_query_time(self) -> float:
        dbs = self.dataframe[self.dataframe.fresh_db != ''].fresh_db
        return self.dataframe[self.dataframe.DB.isin(dbs)].Query_time.sum()

    @property
    def new_table_query_time(self) -> float:
        xs = self.dataframe[self.dataframe.fresh_table != ''].fresh_table.str.split(';').to_list()
        fresh_tables = set([x for ys in xs for x in ys if x != ''])
        selector = self.dataframe.tables.str.split(';').apply(lambda xs: len(fresh_tables.intersection(xs)) > 0)
        return self.dataframe[selector].Query_time.sum()

    @property
    def new_partition_table_query_time(self) -> float:
        xs = self.dataframe[self.dataframe.fresh_partition_table != ''].fresh_partition_table.str.split(';').to_list()
        fresh_partition_tables = set([x for ys in xs for x in ys if x != ''])
        selector = self.dataframe.tables.str.split(';').apply(
            lambda xs: len(fresh_partition_tables.intersection(xs)) > 0)
        return self.dataframe[selector].Query_time.sum()

    @property
    def total_digest_count(self) -> int:
        return len(self.dataframe[['DB', 'Digest']].drop_duplicates())

    @property
    def tikv_query_count(self) -> int:
        return len(self.dataframe[self.dataframe.cop_path == 'tikv'])

    @property
    def tiflash_query_count(self) -> int:
        return len(self.dataframe[self.dataframe.cop_path == 'tiflash'])

    @property
    def mixed_query_count(self) -> int:
        selector = (self.dataframe.cop_path == 'tiflash;tikv') | (self.dataframe.cop_path == 'tikv;tiflash')
        return len(self.dataframe[selector])

    # 过去一个周期内有哪些 SQL 的 Plan 发生了变化, 变化最多的 Top-N 条 SlowQuery 是哪些
    @property
    def digest_with_multi_plan(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df[df.plan_count > 1].sort_values('query_time', ascending=False).to_dict('records')

    # 过去一段周期内有哪些 SlowQuery 使用的 Index 发生了变化，变化最多的 Top-N SlowQuery 有哪些
    @property
    def digest_with_multi_index(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df[df.index_count > 1].sort_values('query_time', ascending=False).to_dict('records')

    @property
    def internal_digests(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.internal_dataframe
        if len(df) == 0:
            return []
        df = df.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df[df.index_count > 1].sort_values('query_time', ascending=False).to_dict('records')

    @property
    def digests_on_tikv(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe[self.dataframe.cop_path == 'tikv']
        if len(df) == 0:
            return []

        df = df.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def digests_on_tiflash(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe[self.dataframe.cop_path == 'tiflash']
        if len(df) == 0:
            return []

        df = df.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def digests_on_tikv_and_tiflash(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        selector = (self.dataframe.cop_path == 'tiflash;tikv') | (self.dataframe.cop_path == 'tikv;tiflash')

        df = self.dataframe[selector]
        if len(df) == 0:
            return []

        df = df.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    # 过去一段周期内内部 SlowQuery 总耗时
    @property
    def internal_query_time(self) -> float:
        return self.internal_dataframe.Query_time.sum()

    @property
    def multi_plan_digest_query_time(self) -> float:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df[df.plan_count > 1].query_time.sum()

    @property
    def multi_index_digest_query_time(self) -> float:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df[df.index_count > 1].query_time.sum()

    @property
    def execution_at_tikv_query_time(self) -> float:
        df = self.dataframe
        return df[df.cop_path == 'tikv'].Query_time.sum()

    @property
    def execution_at_tiflash_query_time(self) -> float:
        df = self.dataframe
        return df[df.cop_path == 'tiflash'].Query_time.sum()

    @property
    def execution_mixed_query_time(self) -> float:
        selector = (self.dataframe.cop_path == 'tiflash;tikv') | (self.dataframe.cop_path == 'tikv;tiflash')
        return self.dataframe[selector].Query_time.sum()

    # Database 总耗时/总数量情况
    @property
    def database_statistics(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'query_time': df.Query_time.sum(),
                'count': len(df),
            })

        df = self.dataframe.groupby('DB').apply(agg).reset_index()
        return {
            'order_by_time': df.sort_values('query_time', ascending=False).to_dict('records'),
            'order_by_count': df.sort_values('count', ascending=False).to_dict('records'),
        }

    # Table 总耗时/总数量情况
    @property
    def table_statistics(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'query_time': df.Query_time.sum(),
                'count': len(df),
            })

        df = explode_table(self.dataframe)
        if len(df) == 0:
            return {
                'order_by_time': [],
                'order_by_count': []
            }

        df = df.groupby(['DB', 'table']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return {
            'order_by_time': df.sort_values('query_time', ascending=False).to_dict('records'),
            'order_by_count': df.sort_values('count', ascending=False).to_dict('records'),
        }

    @property
    def data_range_days(self) -> int:
        return (self.dataframe.Time.max() - self.dataframe.Time.min()).days

    @property
    def ad_hoc_digests_on_tikv(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        df = self.dataframe[(self.dataframe.cop_path == 'tikv') & (self.dataframe.Query_time > 3)]
        df = df.set_index('Time').groupby([pd.Grouper(freq='1D'), 'DB', 'Digest']).size().to_frame('cnt').reset_index()
        df = df[df.cnt <= 5].groupby(['DB', 'Digest']).size().to_frame('cnt').reset_index()
        selector = (self.dataframe.Digest + self.dataframe.DB).isin(df[df.cnt == 7].Digest + df[df.cnt == 7].DB)

        df = self.dataframe[selector]
        if df.empty:
            return []
        df = df.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def worst_sql_by_plan_digest(self) -> pd.DataFrame:
        def agg(df):
            return pd.Series({
                'Time': df.Time.iloc[0],
                'Count': len(df),
                'Query_time': df.Query_time.sum(),
                'Query_time_std': df.Query_time.std() if len(df) > 1 else np.sqrt(df.Query_time.iloc[0]),
                'Mem_max': df.Mem_max.max(),
                'Mem_max_std': df.Mem_max.std() if len(df) > 1 else np.sqrt(df.Mem_max.iloc[0]),
                'sql_statement': df.sql_statement.iloc[0],
                'tables': df.tables.iloc[0],
            })

        df = self.dataframe.groupby(['DB', 'Digest', 'Plan_digest']).apply(agg).reset_index()
        df = df[df.tables != '<UNKNOWN>']
        return df.sort_values(['DB', 'Time', 'Plan_digest'], ascending=True)

    @classmethod
    def worst_sql_by_digest(cls, data: pd.DataFrame) -> pd.DataFrame:
        def agg(df):
            return pd.Series({
                'query_time_std': df.Query_time_std.std() if len(df) > 1 else np.sqrt(df.Query_time_std.iloc[0]),
                'mem_max_std': df.Mem_max_std.std() if len(df) > 1 else np.sqrt(df.Mem_max_std.iloc[0]),
                'plan_count': len(df),
                'sql_statement': df.sql_statement.iloc[0],
                'tables': df.tables.iloc[0],
            })

        df = data.groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.tables != '<UNKNOWN>']
        return df.sort_values(['DB', 'Digest'], ascending=False)

    @property
    def worse_digests(self) -> List[dict]:
        df = self.worst_sql_by_plan_digest
        rt = SlowlogAnalyzer.worst_sql_by_digest(df)
        return {
            'order_by_query_time_std': rt.sort_values('query_time_std', ascending=False).to_dict('records'),
            'order_by_mem_max_std': rt.sort_values('mem_max_std', ascending=False).to_dict('records'),
            'order_by_plan_count': rt.sort_values('plan_count', ascending=False).to_dict('records'),
        }

    def execute(self) -> dict:
        return {
            'data_range_days': self.data_range_days,
            'query_time_statistics': {
                'total': self.total_query_time,
                'fresh_digest': self.new_digest_query_time,
                'fresh_database': self.new_database_query_time,
                'fresh_table': self.new_table_query_time,
                'internal': self.internal_query_time,
                'multi_plan_digest': self.multi_plan_digest_query_time,
                'multi_index_digest': self.multi_index_digest_query_time,
                'execution_at_tikv': self.execution_at_tikv_query_time,
                'execution_at_tiflash': self.execution_at_tiflash_query_time,
                'execution_mixed': self.execution_mixed_query_time,
            },
            'query_statistics': {
                'digest_count': self.total_digest_count,
                'query_count': len(self.dataframe),
                'internal_count': len(self.internal_dataframe),
                'multi_plan_digest_count': len(self.digest_with_multi_plan),
                'multi_index_digest_count': len(self.digest_with_multi_index),
                'tikv_query_count': self.tikv_query_count,
                'tiflash_query_count': self.tiflash_query_count,
                'mixed_query_count': self.mixed_query_count,
            },
            'new_digest_current': self.new_digest_current,
            'new_database_current': self.new_database_current,
            'new_table_current': self.new_table_current,
            'digest_with_multi_plan': self.digest_with_multi_plan,
            'digest_with_multi_index': self.digest_with_multi_index,
            'database_statistics': self.database_statistics,
            'table_statistics': self.table_statistics,
            'digest_statistics': self.digest_statistics,
            'internal_digests': self.internal_digests,
            'digests_on_tikv': self.digests_on_tikv,
            'digests_on_tiflash': self.digests_on_tiflash,
            'digests_on_tikv_and_tiflash': self.digests_on_tikv_and_tiflash,
            'ad_hoc_digests_on_tikv': self.ad_hoc_digests_on_tikv,
            'worse_digest_stat': self.worse_digests,
        }


class SlowlogComparator:
    def __init__(self, filepathlist: list, prev_filepathlist: list):
        dfs = []
        for filepath in filepathlist:
            conn = sqlite3.connect(filepath)
            cur = conn.cursor()
            # get table names
            cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
            tables = cur.fetchall()
            for t in tables:
                table_name = t[0]
                if table_name == 'slowlog':
                    tempdf = pd.read_sql_query("select * from slowlog", conn)
                    dfs.append(tempdf)
        self.dataframe = fix_data_type(pd.concat(dfs)).drop_duplicates()
        self.internal_dataframe = self.dataframe[self.dataframe.Is_internal]
        self.dataframe = self.dataframe[~self.dataframe.Is_internal]

        prev_dfs = []
        for prev_filepath in prev_filepathlist:
            conn = sqlite3.connect(prev_filepath)
            cur = conn.cursor()
            # get table names
            cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
            tables = cur.fetchall()
            for t in tables:
                table_name = t[0]
                if table_name == 'slowlog':
                    tempdf = pd.read_sql_query("select * from slowlog", conn)
                    prev_dfs.append(tempdf)
        self.prev_dataframe = fix_data_type(pd.concat(prev_dfs)).drop_duplicates()
        self.prev_internal_dataframe = self.prev_dataframe[self.prev_dataframe.Is_internal]
        self.prev_dataframe = self.prev_dataframe[~self.prev_dataframe.Is_internal]

    # 总耗时变化情况: 当前周期总耗时
    @property
    def query_time(self) -> float:
        return self.dataframe.Query_time.sum()

    # 总耗时变化情况: 上一周期总耗时
    @property
    def prev_query_time(self) -> float:
        return self.prev_dataframe.Query_time.sum()

    # 总数量变化情况: 当前周期总数量
    @property
    def count(self) -> float:
        return len(self.dataframe)

    # 总数量变化情况: 上一周期总数量
    @property
    def prev_count(self) -> float:
        return len(self.prev_dataframe)

    # 产生了哪些新 SlowQuery
    @property
    def new_digest(self) -> List[str]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        prev_query = self.prev_dataframe.Digest + self.prev_dataframe.DB
        selector = ~(self.dataframe.Digest + self.dataframe.DB).isin(prev_query)
        df = self.dataframe[selector].groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    @property
    def new_digest_query_time(self) -> float:
        prev_query = self.prev_dataframe.Digest + self.prev_dataframe.DB
        selector = ~(self.dataframe.Digest + self.dataframe.DB).isin(prev_query)
        return self.dataframe[selector].Query_time.sum()

    # Database 总耗时变化与总数量变化
    @property
    def database_statistics_diff(self) -> List[dict]:
        def agg(df):
            db = df.DB.iloc[0]
            nf = self.dataframe[self.dataframe.DB == db]
            return pd.Series({
                'old_query_time': df.Query_time.sum(),
                'new_query_time': nf.Query_time.sum(),
                'query_time_diff': nf.Query_time.sum() - df.Query_time.sum(),
                'old_query_count': len(df),
                'new_query_count': len(nf),
                'query_count_diff': len(nf) - len(df),
            })

        df = self.prev_dataframe[self.prev_dataframe.DB.isin(self.dataframe.DB)]
        df = df.groupby('DB').apply(agg).reset_index()
        return df.sort_values('query_time_diff', ascending=False).to_dict('records')

    @property
    def new_database_query_time(self) -> float:
        selector = ~(self.dataframe.DB.isin(self.prev_dataframe.DB))
        return self.dataframe[selector].Query_time.sum()

    @property
    def new_database_query_count(self) -> int:
        selector = ~(self.dataframe.DB.isin(self.prev_dataframe.DB))
        return len(self.dataframe[selector])

    # Table 总耗时变化与总数量变化
    @property
    def table_statistics_diff(self) -> List[dict]:
        cdf = explode_table(self.dataframe)
        pdf = explode_table(self.prev_dataframe)

        def agg(df):
            db = df.DB.iloc[0]
            table = df.table.iloc[0]
            nf = cdf[(cdf.DB == db) & (cdf.table == table)]
            return pd.Series({
                'old_query_time': df.Query_time.sum(),
                'new_query_time': nf.Query_time.sum(),
                'query_time_diff': nf.Query_time.sum() - df.Query_time.sum(),
                'old_query_count': len(df),
                'new_query_count': len(nf),
                'query_count_diff': len(nf) - len(df),
            })

        df = pdf[(pdf.DB + '|' + pdf.table).isin(cdf.DB + '|' + cdf.table)]
        if len(df) == 0:
            return []
        df = df.groupby(['DB', 'table']).apply(agg).reset_index()
        return df.sort_values('query_time_diff', ascending=False).to_dict('records')

    # 有哪些新的 Database
    @property
    def new_database(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'query_time': df.Query_time.sum(),
                'count': len(df),
            })

        df = self.dataframe[~self.dataframe.DB.isin(self.prev_dataframe.DB)]
        if df.empty:
            return []
        df = df.groupby('DB').apply(agg).reset_index()
        return df.sort_values('query_time', ascending=False).to_dict('records')

    # 有哪些新的 Table
    @property
    def new_table(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'query_time': df.Query_time.sum(),
                'count': len(df),
            })

        df = explode_table(self.dataframe)
        pdf = explode_table(self.prev_dataframe)
        df = df[~(df.DB + '|' + df.table).isin(pdf.DB + '|' + pdf.table)]
        if df.empty:
            return []
        df = df.groupby(['DB', 'table']).apply(agg).reset_index()
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    # 有哪些 SQL 的 Plan 发生了变化，变化最多的 Top-N 条 SlowQuery 是哪些
    @property
    def digest_with_plan_diff(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        prev_query = self.prev_dataframe.Digest + self.prev_dataframe.Plan_digest + self.prev_dataframe.DB
        query = self.dataframe.Digest + self.dataframe.Plan_digest + self.dataframe.DB
        selector = ~query.isin(prev_query)
        df = self.dataframe[selector].groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[(df.Digest + df.DB).isin(self.prev_dataframe.Digest + self.prev_dataframe.DB)]
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    # 有哪些 SlowQuery 使用的 Index 发生了变化，变化最多的 Top-N SlowQuery 有哪些
    @property
    def digest_with_index_diff(self) -> List[dict]:
        def agg(df):
            return pd.Series({
                'sql_text': df.sql_statement.iloc[0],
                'count': len(df),
                'query_time': df.Query_time.sum(),
                'mem_max': df.Mem_max.max(),
                'plan_count': df.Plan_digest.nunique(),
                'index_count': df.Index_names.nunique(),
                'table': df.tables.iloc[0],
            })

        prev_query = self.prev_dataframe.Digest + self.prev_dataframe.Index_names + self.prev_dataframe.DB
        query = self.dataframe.Digest + self.dataframe.Index_names + self.dataframe.DB
        selector = ~query.isin(prev_query)
        df = self.dataframe[selector].groupby(['DB', 'Digest']).apply(agg).reset_index()
        df = df[(df.Digest + df.DB).isin(self.prev_dataframe.Digest + self.prev_dataframe.DB)]
        df = df[df.table != '<UNKNOWN>']
        return df.sort_values('query_time', ascending=False).to_dict('records')

    def execute(self) -> dict:
        return {
            'query_time_diff': {
                'current': self.query_time,
                'prev': self.prev_query_time,
            },
            'query_count_diff': {
                'current': self.count,
                'prev': self.prev_count,
            },
            'new_digest': self.new_digest,
            'database_statistics_diff': self.database_statistics_diff,
            'table_statistics_diff': self.table_statistics_diff,
            'new_database': self.new_database,
            'new_table': self.new_table,
            'digest_with_plan_diff': self.digest_with_plan_diff,
            'digest_with_index_diff': self.digest_with_index_diff,
            'new_digest_query_time': self.new_digest_query_time,
            'new_database_query_time': self.new_database_query_time,
            'new_database_query_count': self.new_database_query_count,
        }


if __name__ == '__main__':
    analyzer = SlowlogAnalyzer(['./temp.sqlite','./local/slow_query/1372813089209061633/1372813089454525346/1379661944646413143/slowquery/sqlite/2023122403/1703386800_1703390400_slowlog.sqlite']).execute()

    # compare_result = SlowlogComparator(['./cache/2024-01-11.sqlite'], ['./cache/2024-01-10.sqlite']).execute()
    #
    # json_dict = {
    #     **analyzer_result,
    #     **compare_result,
    # }
    # with open('./slowquery/result.html', "w") as f:
    #     f.write(render_html(json_dict))