import math
import os
import sys
import time
import traceback

import pandas as pd

sys.path.append(os.path.dirname(sys.path[0]))
from utils.secure_db_client import get_remote_engine


def update_workflow_manager(site_name, date_type, date_info):
    print(f"当前执行的参数: {site_name, date_type, date_info}")
    if date_type == "month":
        while True:
            try:
                site_name_pri_dict = {
                    "us": 2,
                    "uk": 4,
                    "de": 6,
                }
                engine_mysql = get_remote_engine(
                    site_name='us',
                    db_type='mysql'
                )
                spider_script = f'ansible dabing_all -f 10 -m shell -a "nohup /usr/local/bin/python3 /mnt/py_spider/threading_spider/Poll_site_spider_month.py {site_name} {date_type} {date_info} >/dev/null 2>&1 &";'
                priority = site_name_pri_dict[site_name]
                update_sql_workflow_spider = f"""
                                                    INSERT INTO selection.workflow_manager
                                    (
                                        workflow_name,
                                        site_name,
                                        date_type,
                                        date_info,
                                        priority,
                                        spider_name,
                                        spider_script,
                                        spider_is_ready,
                                        spider_state,
                                        bg_name,
                                        bg_dol_name, 
                                        bg_dol_state,
                                        finished_count
                                    )
                                    VALUES
                                    (
                                        '月全流程',
                                        '{site_name}',
                                        'month',
                                        '{date_info}',
                                         {priority}, 
                                        '{site_name}_spider_asin',
                                        '{spider_script}',
                                        'yes',
                                        1,
                                        '{site_name}_all_cal', 
                                        'ALL-月流程-ABA+反查+流量选品', 
                                        1,
                                        0
                                    )
                                    ON DUPLICATE KEY UPDATE
                                        spider_is_ready = VALUES(spider_is_ready),
                                        spider_script = VALUES(spider_script),
                                        spider_state = VALUES(spider_state),
                                        bg_dol_name = VALUES(bg_dol_name);
                            """
                print(f"workflow_manager进度表---重置爬虫的asin抓取进度: {update_sql_workflow_spider}")
                engine_mysql.execute(update_sql_workflow_spider)

                update_sql_workflow_bg = f"""update selection.workflow_manager 
set bg_dol_state=3, finished_count=COALESCE(finished_count, 0) + 1
WHERE workflow_name='月全流程' and site_name='{site_name}' and date_type='{date_type}' and date_info='{date_info}' and priority={priority}"""
                print(f"workflow_manager进度表---更新asin导出进度和完成次数: {update_sql_workflow_bg}")
                engine_mysql.execute(update_sql_workflow_bg)
                break
            except Exception as e:
                print(f"失败：workflow_manager进度表---重置爬虫的asin抓取进度: {update_sql_workflow_spider}, {e}, 报错信息: {traceback.format_exc()}")
                time.sleep(300)
                continue


def export_data(site_name, date_type, date_info):
    engine = get_remote_engine(
        site_name=site_name,  # -> database "selection"
        db_type="postgresql_14",  # -> 服务端 alias "mysql"
        # user="fangxingjun",         # -> 服务端 alias "mysql"
        # user_token="5f1b2e9c3a4d7f60"   # 可不传，走默认
    )
    partitions = {
        'site_name': site_name,
        'date_type': date_type,
        'date_info': date_info,
    }
    cols_list = ['asin', 'asin_is_variation', 'volume', 'weight_str', 'date_info']

    import_table = f'{site_name}_all_syn_st_{date_type}_{date_info.replace("-", "_")}'

    hive_table = 'dwt_asin_sync'
    print(f"import_table: {import_table}, hive_table: {hive_table}")
    print(f"partitions: {partitions}")

    engine.sqoop_raw_export(
        hive_table=hive_table,
        import_table=import_table,
        partitions=partitions,
        m=1,
        cols=','.join(cols_list)
    )


def get_minid_maxid(site_name, date_type, date_info):
    engine = get_remote_engine(
        site_name=site_name,  # -> database "selection"
        db_type="postgresql_14",  # -> 服务端 alias "mysql"
        # user="fangxingjun",         # -> 服务端 alias "mysql"
        # user_token="5f1b2e9c3a4d7f60"   # 可不传，走默认
    )
    engine_mysql = get_remote_engine(
        site_name=site_name,  # -> database "selection"
        db_type="mysql",  # -> 服务端 alias "mysql"
        # user="fangxingjun",         # -> 服务端 alias "mysql"
        # user_token="5f1b2e9c3a4d7f60"   # 可不传，走默认
    )
    # 查询最小和最大 id
    print(site_name, ' 查询最小和最大 id')
    query = f"SELECT MIN(id) AS min_id, MAX(id) AS max_id FROM {site_name}_all_syn_st_month_{date_info.replace('-', '_')} where state in (1,2)"
    print(query)
    # result = pd.read_sql(query, self.engine_pg)
    result = engine.read_sql(query)
    if result.shape[0] > 0:
        min_id = result['min_id'].values[0]
        max_id = result['max_id'].values[0]
        # 分组数
        num_groups = 150
        group_size = math.ceil((max_id - min_id + 1) / num_groups)
        # 创建分组范围
        group_ranges = [(i, i + group_size - 1) for i in range(min_id, max_id + 1, group_size)]
        id_list = []
        # 打印分组范围
        for group_start, group_end in group_ranges:
            print(f"Group: {group_start} - {group_end}")
            if site_name == 'us':
                id_list.append([f'{group_start}-{group_end}', date_info])
            else:
                id_list.append([f'{group_start}-{group_end}', date_info])
        print(id_list)
        df_asin_img_video = pd.DataFrame(data=id_list, columns=['minid_maxid', 'yaer_month'])

        delete_sql = f'DELETE from {site_name}_syn_asin_all_minid_maxid where state <4'
        print('delete_sql::', delete_sql)
        engine_mysql.execute(delete_sql)
        engine_mysql.to_sql(df_asin_img_video, f'{site_name}_syn_asin_all_minid_maxid', if_exists='append')


if __name__ == '__main__':
    # site_name = 'us'
    # date_type = 'month'
    # date_info = '2026-01'
    site_name = sys.argv[1]  # 参数1：站点
    date_type = sys.argv[2]  # 参数2：类型：week/4_week/month/quarter/day
    date_info = sys.argv[3]  # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
    export_data(site_name, date_type, date_info)
    if site_name in ["us", "uk", "de"] and date_type == "month":
        get_minid_maxid(site_name, date_type, date_info)
        update_workflow_manager(site_name, date_type, date_info)
