import ast
import csv
import functools
import os
import re
from datetime import datetime, timedelta, timezone
from io import BytesIO
from urllib.parse import urlparse

import boto3
import pandas as pd
import requests
from google.cloud import bigquery

SESSION_SECONDS = 60 * 60 * 1


# no cache for this function
def dbaas_cluster_components(cluster_ids):
    client = bigquery.Client()
    r = client.query(
        f"SELECT * FROM `gcp-digital-bi.metadb.dbaas_cluster_components` WHERE cluster_id IN ({', '.join(cluster_ids)})"
    )
    return pd.DataFrame.from_records([dict(row) for row in r])


@functools.lru_cache
def dbaas_paied_cluster_ids():
    client = bigquery.Client()
    r = client.query(
        """
        SELECT DISTINCT(JSON_EXTRACT_SCALAR(labels, '$.clusterID')) as cluster_id
        FROM `gcp-digital-bi.metadb.dbaas_billing`
        WHERE consumed_source = 'invoice'
    """
    )
    return pd.DataFrame.from_records([dict(row) for row in r])


@functools.lru_cache
def activeqps(begin, threshold=100):
    records = requests.get(
        "https://clinic.pingcap.com/clinic/api/v1/dashboard/activeqps",
        params={
            "threshold": threshold,
            "begin_ts": int(begin.timestamp()),
            "end_ts": int(datetime.now().timestamp()),
        },
        headers={"Authorization": f"Bearer {os.getenv('CLINIC_API_KEY')}"},
    ).json()["items"]

    activeqps_df = pd.DataFrame.from_records(records)
    activeqps_df["time"] = pd.to_datetime(activeqps_df.timestamp, unit="s")

    return activeqps_df


@functools.lru_cache
def pingcap_emails():
    S3_BUCKET = os.getenv("S3_BUCKET")
    s3 = boto3.client("s3")

    content = s3.get_object(Bucket=S3_BUCKET, Key=f"data/pingcap_users.csv")["Body"].read()
    return [u.decode() for u in content.split(b"\n") if u]


@functools.lru_cache
def tenants_view():
    client = bigquery.Client()
    r = client.query("SELECT * FROM `gcp-digital-bi.shared_view.uniq_tidb_port_id_view`")
    return pd.DataFrame.from_records([dict(row) for row in r])


@functools.lru_cache
def salesforce_country_list():
    S3_BUCKET = os.getenv("S3_BUCKET")
    s3 = boto3.client("s3")

    return pd.read_csv(
        s3.get_object(Bucket=S3_BUCKET, Key=f"data/salesforce_country_list.csv")["Body"],
        keep_default_na=False,
    )


@functools.lru_cache
def dbaas_tenants(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.dbaas_tenants`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_tenants(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_users(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.dbaas_users`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_users(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_clusters(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.dbaas_clusters`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_clusters(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_imports(epoch=None):
    if epoch is None:
        cluster_df = dbaas_clusters()
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.dbaas_imports`")
        df = pd.DataFrame.from_records([dict(row) for row in r])
        df.loc[df.cluster_id.isin(cluster_df[cluster_df.free == 1].id), "cluster_type"] = "devtier"
        df.loc[
            df.cluster_id.isin(cluster_df[cluster_df.free == 0].id), "cluster_type"
        ] = "dedicated"
        return df
    df = dbaas_imports(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_trials(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.dbaas_trials`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_trials(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_cluster_usedsize(begin):
    client = bigquery.Client()
    date = begin.strftime("%Y-%m-%d 00:00:00")
    r = client.query(
        f'SELECT * FROM `gcp-digital-bi.metadb.dbaas_cluster_usedsize` WHERE created_at >= "{date}"'
    )
    return pd.DataFrame.from_records([dict(row) for row in r])


@functools.lru_cache
def dbaas_tenant_plans(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.dbaas_tenant_plans`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_tenant_plans(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_msg_bot_interact_records(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.msg_bot_interact_records`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_msg_bot_interact_records(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_sqleditor_sql_editor_messages(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.sqleditor_sql_editor_messages`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_sqleditor_sql_editor_messages(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_sqleditor_sql_editor_oplogs(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.sql-editor-oplogs`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_msg_bot_interact_records(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def tenant_daily_billing_cost(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query(
            """ select billing_tenant_id, billing_cost_date, unit_category, sum(billing_cost_price) cost_price,sum(billing_raw_price) raw_price, sum(billing_cost_unit_num) raw_cost
from `model_dbaas_billing.dws_cloud_tenant_daily_billing_cost_distribution` 
where unit_category like '%Serverless%'
group by billing_tenant_id,billing_cost_date, unit_category
order by billing_cost_date desc"""
        )
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_msg_bot_interact_records(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def dbaas_spend_limit(epoch=None):
    if epoch is None:
        client = bigquery.Client()
        r = client.query("SELECT * FROM `gcp-digital-bi.metadb.spend_limit`")
        return pd.DataFrame.from_records([dict(row) for row in r])
    df = dbaas_spend_limit(epoch=None)
    return df[df.created_at < epoch]


@functools.lru_cache
def events(begin, epoch=None):
    if epoch is None:
        S3_BUCKET = os.getenv("S3_BUCKET")
        s3 = boto3.client("s3")

        event_df = pd.DataFrame()

        d = begin
        while d < datetime.now(timezone.utc):
            try:
                df = pd.read_csv(
                    s3.get_object(Bucket=S3_BUCKET, Key=f"data/mixpanel/{d.date()}.csv")["Body"],
                    low_memory=False,
                )
                event_df = pd.concat([event_df, df])
            except Exception as e:
                pass

            d += timedelta(days=1)

        event_df["$user_id"] = event_df["$user_id"].astype(int, errors="ignore")
        event_df["time"] = pd.to_datetime(event_df["mp_processing_time_ms"], unit="ms", utc=True)
        event_df.loc[
            event_df.event.str.match("[0-9].* Page View"), "event"
        ] = "Data Import Page View"

        event_meta = event_df.groupby("event").time.min().to_frame()
        return (event_meta, event_df)

    event_meta, event_df = events(begin, epoch=None)
    event_df = event_df[event_df.time < epoch]
    return event_meta, event_df


@functools.lru_cache
def connections(begin, epoch=None):
    if epoch is None:
        S3_BUCKET = os.getenv("S3_BUCKET")
        s3 = boto3.client("s3")

        connection_df = pd.DataFrame()

        d = begin
        while d < datetime.now(timezone.utc):
            try:
                df = pd.read_csv(
                    s3.get_object(Bucket=S3_BUCKET, Key=f"data/metrics/{d.date()}_connection.csv")[
                        "Body"
                    ],
                    low_memory=False,
                )
                connection_df = pd.concat([connection_df, df])
            except:
                pass

            d += timedelta(days=1)

        connection_df["time"] = pd.to_datetime(connection_df["time"], unit="s", utc=True)
        return connection_df

    connection_df = connections(begin, epoch=None)
    connection_df = connection_df[connection_df.time < epoch]
    return connection_df


@functools.lru_cache
def qps(begin, epoch=None):
    if epoch is None:
        S3_BUCKET = os.getenv("S3_BUCKET")
        s3 = boto3.client("s3")

        qps_df = pd.DataFrame()

        d = begin
        while d < datetime.now(timezone.utc):
            try:
                df = pd.read_csv(
                    s3.get_object(Bucket=S3_BUCKET, Key=f"data/metrics/{d.date()}_qps.csv")["Body"],
                    low_memory=False,
                )
                qps_df = pd.concat([qps_df, df])
            except:
                pass

            d += timedelta(days=1)

        qps_df["time"] = pd.to_datetime(qps_df["time"], unit="s", utc=True)
        return qps_df

    qps_df = qps(begin, epoch=None)
    qps_df = qps_df[qps_df.time < epoch]
    return qps_df


@functools.lru_cache
def serverless_qps(begin, epoch=None):
    if epoch is None:
        S3_BUCKET = os.getenv("S3_BUCKET")
        s3 = boto3.client("s3")

        qps_df = pd.DataFrame()

        d = begin
        while d < datetime.now(timezone.utc):
            try:
                df = pd.read_csv(
                    s3.get_object(
                        Bucket=S3_BUCKET, Key=f"data/metrics/{d.date()}_serverless_qps.csv"
                    )["Body"],
                    low_memory=False,
                )
                df["time"] = d.date()
                qps_df = pd.concat([qps_df, transform_qps_list_value_to_rows(df)])
            except:
                pass

            d += timedelta(days=1)
        return qps_df

    qps_df = serverless_qps(begin, epoch=None)
    qps_df = qps_df[qps_df.time < epoch]
    return qps_df


@functools.lru_cache
def gw_access(begin, epoch=None):
    if epoch is None:
        S3_BUCKET = os.getenv("S3_BUCKET")
        s3 = boto3.client("s3")

        gw_access_df = pd.DataFrame()

        d = begin
        while d < datetime.now(timezone.utc):
            try:
                df = pd.read_csv(
                    s3.get_object(
                        Bucket=S3_BUCKET,
                        Key=f"data/gw_access_log/{d.date()}.sample.csv",
                    )["Body"],
                    low_memory=False,
                )
                gw_access_df = pd.concat([gw_access_df, df])
            except:
                pass

            d += timedelta(days=1)
        return gw_access_df

    gw_access_df = gw_access(begin, epoch=None)
    gw_access_df = gw_access_df[gw_access_df.time < epoch]
    return gw_access_df


@functools.lru_cache
def gw_first_access_df(begin, epoch=None):
    df = gw_access(begin, epoch)
    return (
        pd.to_datetime(
            df[(df.status == "success") & (~df.gateway_instance.str.contains("-inner-"))]
            .groupby("cluster_id")
            .ts.min(),
            unit="s",
            utc=True,
        )
        .to_frame("first_access")
        .reset_index()
    )


class User:
    def __init__(self, tenant_id, begin, epoch=None):
        self._id = tenant_id
        self._tenant_df = dbaas_tenants(epoch)
        self._user_df = dbaas_users(epoch)
        self._cluster_df = dbaas_clusters(epoch)
        self._trial_df = dbaas_trials(epoch)
        self._import_df = dbaas_imports(epoch)
        self._cluster_usedsize_df = dbaas_cluster_usedsize(begin)
        self._tenant_plan_df = dbaas_tenant_plans(epoch)
        self._msg_bot_interact_records_df = dbaas_msg_bot_interact_records(epoch)
        self._sqleditor_sql_editor_messages_df = dbaas_sqleditor_sql_editor_messages(epoch)
        self._sqleditor_sql_editor_oplogs_df = dbaas_sqleditor_sql_editor_oplogs(epoch)
        self._daily_billing_cost_df = tenant_daily_billing_cost(epoch)
        self._spend_limit_df = dbaas_spend_limit(epoch)

        self._event_meta, self._event_df = events(begin, epoch)
        self._gw_access_df = gw_access(begin, epoch)
        self._connection_df = connections(begin, epoch)
        self._qps_df = qps(begin, epoch)
        self._serverless_qps_df = serverless_qps(begin, epoch)
        self._activeqps_df = activeqps(begin, 1)

        self._gw_first_access_df = gw_first_access_df(begin, epoch)

    def id(self):
        return self._id

    @functools.lru_cache
    def account_create_time(self):
        tenants = self._tenant_df[self._tenant_df.id == self.id()]
        return tenants.created_at.min()

    @functools.lru_cache
    def first_cluster_create_time(self):
        clusters = self._cluster_df[self._cluster_df.tenant_id == self.id()]
        return clusters.created_at.min() if len(clusters) > 0 else None

    @functools.lru_cache
    def first_cluster_ready_time(self):
        clusters = self._cluster_df[self._cluster_df.tenant_id == self.id()]
        return clusters.creation_finished_time.min() if len(clusters) > 0 else None

    @functools.lru_cache
    def create_first_cluster_time(self):
        TZ = timezone(timedelta(hours=8))
        if (
            self.signup_time().astimezone(TZ).date()
            >= datetime.strptime("2023-01-12 00:00:00", "%Y-%m-%d %H:%M:%S").astimezone(TZ).date()
        ):
            df = self.clusters()
            if len(df[df.created_at.notna()]) <= 0:
                return (None, None)
            step = self.first_event("Clusters Page View")[0]
            return (
                step if step is not None else 0,
                df[df.created_at.notna()].created_at.iloc[0].to_pydatetime(),
            )
        else:
            return self.first_event("Created A Cluster")

    @functools.lru_cache
    def create_cluster_count(self):
        TZ = timezone(timedelta(hours=8))
        if (
            self.signup_time().astimezone(TZ).date()
            >= datetime.strptime("2023-01-12 00:00:00", "%Y-%m-%d %H:%M:%S").astimezone(TZ).date()
        ):
            df = self.clusters()
            return len(df[df.created_at.notna()].created_at)
        else:
            return self.count_event("Created A Cluster")

    @functools.lru_cache
    def clusters_alive(self):
        last_access_df = (
            self._connection_df[self._connection_df.value > 0]
            .groupby("cluster_id")
            .time.max()
            .to_frame("last_access")
        )
        clusters = pd.merge(
            self._cluster_df,
            last_access_df,
            how="left",
            left_on=self._cluster_df.id.astype(str),
            right_on="cluster_id",
        )
        clusters = clusters[clusters.tenant_id == self.id()].copy()
        return (
            (pd.to_datetime(clusters.last_access, utc=True) - clusters.created_at)
            .fillna(timedelta(0))
            .sum()
        )

    @functools.lru_cache
    def country(self):
        cs = self._tenant_df[self._tenant_df.id == self.id()].country
        if len(cs) == 0:
            return None
        return cs.iloc[0]

    @functools.lru_cache
    def region(self):
        df = tenants_view()
        df = df[df.TiDBCloudPortalID == str(self.id())]
        return df.Region.iloc[0] if len(df) > 0 else None

    @functools.lru_cache
    def members(self):
        members = self._user_df[self._user_df.tenant_id == self.id()]
        return list(members.id.astype(str).unique())

    @functools.lru_cache
    def signup_time(self):
        return self._tenant_df[self._tenant_df.id == self.id()].created_at.min()

    @functools.lru_cache
    def third_party_account_provider(self):
        return self._tenant_df[self._tenant_df.id == self.id()].third_party_account_provider.iloc[0]

    @functools.lru_cache
    def events(self):
        members = self._user_df[self._user_df.tenant_id == self.id()]
        return self._event_df[self._event_df["$user_id"].isin(members.id)].sort_values("time")

    @functools.lru_cache
    def first_event(self, *events):
        df = self.events().reset_index()
        df = df[df.event.isin(events)]
        if len(df) == 0:
            return (None, None)
        return (df.index[0], df.time.iloc[0].to_pydatetime())

    @functools.lru_cache
    def first_event_with(self, fn):
        df = self.events().reset_index()
        df = df[df.apply(fn, axis=1)]
        if len(df) == 0:
            return (None, None)
        return (df.index[0], df.time.iloc[0].to_pydatetime())

    @functools.lru_cache
    def last_event(self, event):
        df = self.events().reset_index()
        df = df[df.event == event]
        if len(df) == 0:
            return (None, None)
        return (df.index[-1], df.time.iloc[-1].to_pydatetime())

    @functools.lru_cache
    def alive(self):
        events = self.events()
        return events.iloc[-1].time - self.signup_time() if len(events) > 0 else timedelta(0)

    @functools.lru_cache
    def count_event(self, *event_list):
        events = self.events()
        return len(events[events.event.isin(event_list)])

    @functools.lru_cache
    def count_event_with(self, fn):
        events = self.events()
        return len(events[events.apply(fn, axis=1)])

    @functools.lru_cache
    def initial_referrer_raw(self) -> str:
        events = self.events()
        events = events[events["$initial_referrer"].notna()]
        if len(events) == 0:
            return None
        return events["$initial_referrer"].iloc[0]

    @functools.lru_cache
    def initial_referrer(self) -> str:
        ref = self.initial_referrer_raw()
        if ref is None:
            return None
        if ref == "$direct":
            return ref
        ref = urlparse(ref).netloc
        if ref.endswith("safeframe.googlesyndication.com"):
            return "safeframe.googlesyndication.com"
        if ref.endswith("cdn.ampproject.org"):
            return "cdn.ampproject.org"
        if ref.endswith("ampproject.net"):
            return "ampproject.net"
        if ref.endswith("pikiran-rakyat.com"):
            return "pikiran-rakyat.com"
        return ref

    @functools.lru_cache
    def event_types(self, before=None):
        events = self.events()
        if before is not None:
            events = events[events.time < before]
        return len(events.event.unique())

    @functools.lru_cache
    def event_ratio(self, before=None):
        events = self.events()
        if before is not None:
            events = events[events.time < before]
        if len(events) == 0:
            return 0
        last_access_time = events.time.iloc[-1]
        return len(events.event.unique()) / len(
            self._event_meta[self._event_meta.time <= last_access_time]
        )

    @functools.lru_cache
    def is_poc(self) -> bool:
        return (
            len(
                self._tenant_plan_df[
                    (self._tenant_plan_df.tenant_id == self.id())
                    & (self._tenant_plan_df.plan == "poc")
                ]
            )
            > 0
        )

    @functools.lru_cache
    def is_billed(self) -> bool:
        return (
            len(
                self._tenant_plan_df[
                    (self._tenant_plan_df.tenant_id == self.id())
                    & (self._tenant_plan_df.plan == "on_demand")
                ]
            )
            > 0
        )

    @functools.lru_cache
    def has_poc(self) -> bool:
        return (
            len(
                self._trial_df[
                    (self._trial_df.trial_type == "poc") & (self._trial_df.tenant_id == self.id())
                ]
            )
            > 0
        )

    @functools.lru_cache
    def max_qps(self, include_dedicate=True):
        clusters = self._cluster_df[self._cluster_df.tenant_id == self.id()]
        if not include_dedicate:
            clusters = clusters[clusters.deploy_type != "dedicate"]
        qps = self._qps_df[
            (self._qps_df.cluster_id.isin(clusters.id.astype(str)))
            | (self._qps_df.cluster_id.isin("tidb" + clusters.id.astype(str)))
        ].value
        return qps.max() if len(qps) > 0 else 0

    @functools.lru_cache
    def max_activeqps(self, include_dedicate=True):
        qps = self._activeqps_df[self._activeqps_df.tenant_id == str(self.id())]
        if not include_dedicate:
            qps = qps[qps.deploy_type != "dedicated"]
        return qps.max_qps.max() if len(qps) > 0 else 0

    @functools.lru_cache
    def max_connection(self):
        clusters = self._cluster_df[self._cluster_df.tenant_id == self.id()]
        connection = self._connection_df[
            (self._connection_df.cluster_id.isin(clusters.id.astype(str)))
            | (self._connection_df.cluster_id.isin("tidb" + clusters.id.astype(str)))
        ].value
        return connection.max() if len(connection) > 0 else 0

    @functools.lru_cache
    def connections(self):
        clusters = self._cluster_df[self._cluster_df.tenant_id == self.id()]
        return self._connection_df[
            (self._connection_df.cluster_id.isin(clusters.id.astype(str)))
            | (self._connection_df.cluster_id.isin("tidb" + clusters.id.astype(str)))
        ]

    @functools.lru_cache
    def first_connect(self):
        clusters = self._cluster_df[self._cluster_df.tenant_id == self.id()]

        def gw_first_success_seen():
            df = self._gw_first_access_df[self._gw_first_access_df.cluster_id.isin(clusters.id)]
            if len(df) == 0:
                return None
            return df.first_access.iloc[0]

        connection = self._connection_df[
            (self._connection_df.cluster_id.isin(clusters.id.astype(str)))
            | (self._connection_df.cluster_id.isin("tidb" + clusters.id.astype(str)))
        ]
        df = connection[connection.value > 0]

        gw_connect_time = gw_first_success_seen()
        if len(df) == 0:
            return gw_connect_time

        metrics_connect_time = df.time.iloc[0].to_pydatetime()

        return (
            metrics_connect_time
            if gw_connect_time is None or metrics_connect_time < gw_connect_time
            else gw_connect_time
        )

    @functools.lru_cache
    def event_duration_sum(self, *event_list):
        events = self.events().copy().sort_values("time")
        events["duration"] = (events.shift(-1).time - events.time).fillna(timedelta(0))
        events.loc[events.duration > timedelta(seconds=SESSION_SECONDS), "duration"] = timedelta(0)
        return events[events.event.isin(event_list)].duration.sum()

    @functools.lru_cache
    def event_duration_first(self, *event_list):
        events = self.events().copy().sort_values("time")
        events["duration"] = (events.shift(-1).time - events.time).fillna(timedelta(0))
        events.loc[events.duration > timedelta(seconds=SESSION_SECONDS), "duration"] = timedelta(0)
        durations = events[events.event.isin(event_list)].duration
        if len(durations) == 0:
            return timedelta(0)
        return durations.iloc[0]

    @functools.lru_cache
    def has_create_cluster(self):
        return len(self.clusters()) > 0

    @functools.lru_cache
    def has_cluster(self):
        return (
            len(
                self._cluster_df[
                    (self._cluster_df.tenant_id == self.id())
                    & (self._cluster_df.deleted_time.isna())
                ]
            )
            > 0
        )

    @functools.lru_cache
    def clusters(self):
        return self._cluster_df[self._cluster_df.tenant_id == self.id()].sort_values("created_at")

    @functools.lru_cache
    def imports(self):
        return self._import_df[self._import_df.cluster_id.isin(self.clusters().id)]

    @functools.lru_cache
    def serverless_clusters(self):
        return self._cluster_df[
            (self._cluster_df.tenant_id == self.id()) & (self._cluster_df.free == 1)
        ]

    @functools.lru_cache
    def dedicate_clusters(self):
        return self._cluster_df[
            (self._cluster_df.tenant_id == self.id()) & (self._cluster_df.free == 0)
        ]

    @functools.lru_cache
    def serverless_cluster_usedsize(self):
        clusters = self.serverless_clusters()
        return self._cluster_usedsize_df[self._cluster_usedsize_df.cluster_id.isin(clusters.id)]

    @functools.lru_cache
    def dedicate_cluster_usedsize(self):
        clusters = self.dedicate_clusters()
        return self._cluster_usedsize_df[self._cluster_usedsize_df.cluster_id.isin(clusters.id)]

    @functools.lru_cache
    def daily_billing_cost(self):
        return self._daily_billing_cost_df[
            self._daily_billing_cost_df.billing_tenant_id == self.id()
        ]

    @functools.lru_cache
    def daily_ru_cost(self):
        df = self.daily_billing_cost()
        return df[df.unit_category == "Serverless Request Unit Cost"]

    @functools.lru_cache
    def spend_limit(self):
        return self._spend_limit_df[self._spend_limit_df.tenant_id == self.id()]

    @functools.lru_cache
    def daily_billing_cost(self):
        return self._daily_billing_cost_df[
            self._daily_billing_cost_df.billing_tenant_id == self.id()
        ]

    @functools.lru_cache
    def daily_ru_cost(self):
        df = self.daily_billing_cost()
        return df[df.unit_category == "Serverless Request Unit Cost"]

    @functools.lru_cache
    def email(self):
        return self._tenant_df[self._tenant_df.id == self.id()].email.iloc[0]

    @functools.lru_cache
    def is_business_email(self):
        publist = set(
            [
                "gmail.com",
                "qq.com",
                "163.com",
                "126.com",
                "msn.com",
                "outlook.com",
                "hotmail.com",
                "yahoo.com",
                "foxmail.com",
                "yopmail.com",
                "icloud.com",
                "protonmail.com",
                "sohu.com",
                "yeah.net",
                "sina.com",
                "139.com",
                "fastmail.com",
                "snapmail.cc",
            ]
        )

        xs = self.email().split("@")
        mail = xs[1]
        if mail in publist or mail.endswith(".edu.cn") or mail.endswith(".edu"):
            return False
        return True

    @functools.lru_cache
    def emails(self):
        members = self._user_df[self._user_df.tenant_id == self.id()]
        return list(members["email"])

    @functools.lru_cache
    def owner_id(self):
        ids = self._user_df[self._user_df.tenant_id == self.id()].sort_values("id").id
        if len(ids) == 0:
            return None
        return ids.iloc[0]

    @functools.lru_cache
    def has_bind_card(self):
        return (
            len(
                self._tenant_plan_df[
                    (self._tenant_plan_df.tenant_id == self.id())
                    & (self._tenant_plan_df.plan == "on_demand")
                ]
            )
            > 0
        )

    @functools.lru_cache
    def excuted_editor_sql(self):
        return self._sqleditor_sql_editor_messages_df[
            self._sqleditor_sql_editor_messages_df.created_by == self.email()
        ]

    @functools.lru_cache
    def ai_generated_sql(self):
        return self._msg_bot_interact_records_df[
            self._msg_bot_interact_records_df.created_by == self.email()
        ]

    @functools.lru_cache
    def executed_ai_generated_sql_count(self):
        ai_sqls = self.ai_generated_sql().sort_values("created_at")
        editor_sqls = self.excuted_editor_sql().sort_values("created_at")

        result = 0
        for _, ai_sql in ai_sqls.iterrows():
            if ai_sql["response"] == None:
                continue
            for _, editor_sql in editor_sqls.iterrows():
                if editor_sql["sql"] == None:
                    continue
                if editor_sql["created_at"] > ai_sql["created_at"]:
                    if editor_sql["created_at"] > ai_sql["created_at"] + timedelta(minutes=5):
                        break
                    n_ai_sql = normalize(ai_sql["response"])
                    n_editor_sql = normalize(editor_sql["sql"])
                    if n_ai_sql in n_editor_sql or n_editor_sql in n_ai_sql:
                        result += 1
        return result

    @functools.lru_cache
    def sql_editor_info(self):
        return self._sqleditor_sql_editor_oplogs_df[
            self._sqleditor_sql_editor_oplogs_df.created_by == self.email()
        ]

    def is_pingcap(self):
        if self.email().endswith("pingcap.com") or self.email().endswith("snapmail.cc"):
            return True
        df = tenants_view()
        df = df[df.TiDBCloudPortalID == str(self.id())]
        return df.InternalUser.iloc[0] if len(df) > 0 else None

    @functools.lru_cache
    def count_session(self):
        if self.events().empty:
            return 0, []
        count = 0
        events = self.events().copy().sort_values("time")
        events["duration"] = (events.shift(-1).time - events.time).fillna(timedelta(0))
        session_info = []
        start_time = events.time.iloc[0]
        if len(events) == 1:
            return 1, [(start_time, start_time)]
        for _, e in events.iterrows():
            end_time = e.time
            if e.duration > timedelta(seconds=SESSION_SECONDS):
                session_info.append((start_time, end_time))
                count += 1
                start_time = e.time
        if end_time != start_time:
            session_info.append((start_time, end_time))
            count += 1

        return count, session_info

    @functools.lru_cache
    def serverless_max_qps_info(self):
        clusters = self.serverless_clusters()
        qps = self._serverless_qps_df[self._serverless_qps_df.cluster_id.isin(clusters.id)]
        return qps

    @functools.lru_cache
    def serverless_qps_info_until(self, d=timedelta(days=7)):
        begin = self.signup_time().date()
        max_qps = 0
        duration = 0
        smqi = self.serverless_max_qps_info()
        smqi = smqi[(begin <= smqi.time) & (smqi.time <= begin + d)]
        res = pd.DataFrame(columns=["cluster_id", "duration", "max_qps"])

        for cluster_id, v in smqi.groupby("cluster_id"):
            res = pd.concat(
                [
                    res,
                    pd.DataFrame.from_records(
                        [
                            {
                                "cluster_id": cluster_id,
                                "duration": v.duration.sum(),
                                "max_qps": v.max_qps.max(),
                            }
                        ]
                    ),
                ]
            )
        return res

    @functools.lru_cache
    def count_session(self):
        if self.events().empty:
            return 0, []
        count = 0
        events = self.events().copy().sort_values("time")
        events["duration"] = (events.shift(-1).time - events.time).fillna(timedelta(0))
        session_info = []
        start_time = events.time.iloc[0]
        if len(events) == 1:
            return 1, [(start_time, start_time)]
        for _, e in events.iterrows():
            end_time = e.time
            if e.duration > timedelta(seconds=SESSION_SECONDS):
                session_info.append((start_time, end_time))
                count += 1
                start_time = e.time
        if end_time != start_time:
            session_info.append((start_time, end_time))
            count += 1

        return count, session_info


def load_users(begin):
    users = []
    for tenant_id in dbaas_tenants(None).id.unique():
        users.append(User(tenant_id, begin))

    return users


def normalize(query: str) -> str:
    def comma_fix(s):
        # Remove spaces in front of commas
        return s.replace(" , ", ", ")

    def white_space_fix(s):
        # Remove double and triple spaces
        return " ".join(s.split())

    def lower(s):
        # Convert everything except text between (single or double) quotation marks to lower case
        return re.sub(r"\b(?<!['\"])(\w+)(?!['\"])\b", lambda match: match.group(1).lower(), s)

    return comma_fix(white_space_fix(lower(query)))


def interfering_sql(sql):
    if sql is None:
        return True
    elif sql == "":
        return True
    # eliminate "USE xxx" only
    elif sql.startswith("USE") and len(sql) <= 10:
        return True
    elif sql.startswith("use") and len(sql) <= 10:
        return True
    elif sql == ";":
        return True
    return False


def qps_groupby_cluster_id(qf):
    res = pd.DataFrame(columns=["cluster_id", "duration", "max_qps", "time"])
    for cluster_id, v in qf.groupby("cluster_id"):
        res = pd.concat(
            [
                res,
                pd.DataFrame.from_records(
                    [
                        {
                            "cluster_id": cluster_id,
                            "duration": v.duration.sum(),
                            "max_qps": v.max_qps.max(),
                            "time": v.time.min(),
                        }
                    ]
                ),
            ]
        )
    return res


def transform_qps_list_value_to_rows(qps_df):
    qf = pd.DataFrame(columns=["cluster_id", "duration", "max_qps", "time"])
    for _, qps in qps_df.iterrows():
        vlist = []
        for m in eval(qps.value):
            vlist.append(float(m[1]))
        duration = get_continuous_time_slice(vlist)
        qf = pd.concat(
            [
                qf,
                pd.DataFrame.from_records(
                    [
                        {
                            "cluster_id": qps.cluster_id,
                            "duration": duration,
                            "max_qps": max(vlist),
                            "time": qps.time,
                        }
                    ]
                ),
            ]
        )
    qf = qps_groupby_cluster_id(qf)
    return qf


def get_continuous_time_slice(alist, threshold=1, continuous_num=1):
    result = 0
    s = -1
    for i in range(len(alist)):
        if alist[i] > threshold:
            if s < 0:
                s = i
        else:
            if s > 0:
                if i - s + 1 > continuous_num:
                    result += i - s + 1
            s = -1
    if alist[len(alist) - 1] > threshold:
        if len(alist) - s > continuous_num:
            result += len(alist) - s
    return result