# This file contains all the logics generate the data for agg_user_daily table.
import base64
from hashlib import md5

import pandas as pd
import snowflake.snowpark as snowpark
from snowflake.snowpark.session import Session
from user_agents import parse

user_agent_metadata = [
    "hashed_user_agent_id",
    "user_agent",
    "browser_family",
    "browser_version",
    "os_family",
    "os_version",
    "device_family",
    "device_brand",
    "device_model",
    "is_mobile",
    "is_tablet",
    "is_touch_capable",
    "is_pc",
    "is_bot",
]

def get_sql_result(session: snowpark.Session, schema: str):
    df = session.sql(schema)
    local_df = df.collect()
    if len(local_df) == 0:
        return None
    return pd.DataFrame(local_df)


# process the two_hour_ago data only
def get_user_agent_metadata(user_agents: list[str]):
    result = []
    for user_agent_value in user_agents:
        user_agent = parse(user_agent_value)
        result.append(
            {
                "hashed_user_agent_id": base64.b64encode(
                    md5(user_agent_value.encode("utf-8")).digest()
                ).decode(),
                "user_agent": user_agent_value,
                "browser_family": user_agent.browser.family,
                "browser_version": user_agent.browser.version_string,
                "os_family": user_agent.os.family,
                "os_version": user_agent.os.version_string,
                "device_family": user_agent.device.family,
                "device_brand": user_agent.device.brand,
                "device_model": user_agent.device.model,
                "is_mobile": user_agent.is_mobile,
                "is_tablet": user_agent.is_tablet,
                "is_touch_capable": user_agent.is_touch_capable,
                "is_pc": user_agent.is_pc,
                "is_bot": user_agent.is_bot,
            },
        )

    return pd.DataFrame(result)


def insert_data(session: snowpark.Session, user_agents: list[str]):
    df = get_user_agent_metadata(user_agents=user_agents)
    session.create_dataframe(df).write.save_as_table("user_agent_metadata", mode="append")


def get_user_agent(session: snowpark.Session, p_date: str):
    sql = """
        with a as (
            select distinct user_agent
            from web_artist_actions
            where P_DATE = '{p_date}'
            union 
            select distinct user_agent
            from web_audio_actions
            where P_DATE = '{p_date}'
            union 
            select distinct user_agent
            from web_audio_creation
            where P_DATE = '{p_date}'
            union 
            select distinct user_agent
            from web_audio_player_actions 
            where P_DATE = '{p_date}'
            union 
            select distinct user_agent
            from web_general_event
            where P_DATE = '{p_date}'
            union
            select distinct user_agent
            from web_playlist_actions
            where P_DATE = '{p_date}'
            union
            select distinct user_agent
            from web_user_event
            where P_DATE = '{p_date}'
        ), b as (
            select  
                GET_MD5_ID(user_agent) as HASHED_USER_AGENT_ID,
                user_agent as USER_AGENT,
            from a
        ), c as (
            select 
                b.HASHED_USER_AGENT_ID,
                b.USER_AGENT
            from b
            left join user_agent_metadata as d on b.HASHED_USER_AGENT_ID = d.HASHED_USER_AGENT_ID
            where d.HASHED_USER_AGENT_ID is null
        )
        select distinct user_agent from c;
    """.format(p_date = p_date)
    user_agents = get_sql_result(session, sql);
    if  user_agents is not None:
        user_agents = user_agents["USER_AGENT"].to_list()
        insert_data(session, user_agents)

    return len(user_agents)