import sys
import os

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
import time
from redis.exceptions import LockError
from threading_spider.db_connectivity import connect_db
from threading_spider.post_to_dolphin import DolphinschedulerHelper
from utils.db_connect import BaseUtils
from utils.secure_db_client import get_remote_engine


def db_engine_us(site_name,db_type):
    engine_mysql = get_remote_engine(
        site_name=site_name,  # -> database "selection"
        db_type=db_type,  # -> 服务端 alias "mysql"
    )
    return engine_mysql


def db_cursor_connect_update(sql, site):
    for i in range(3):
        try:
            engine_us_mysql = db_engine_us('us','mysql')
            print('更新sql：', sql)
            with engine_us_mysql.begin() as conn:
                conn.execute(sql)
            break
        except:
            print(site, 'db_cursor_connect 报错：', sql)


def db_cursor_connect_msyql_read(select_state1_sql):
    for i in range(3):
        try:
            engine_us_mysql = db_engine_us('us','mysql')
            print(select_state1_sql)
            df = engine_us_mysql.read_sql(select_state1_sql)
            return df
        except Exception as e:
            import traceback
            traceback.print_exc()  # ★ 打印完整栈到终端
            print(e, 'db_cursor_connect_msyql_read 报错：', select_state1_sql)


def db_cursor_connect_pg_read(site,select_state1_sql):
    for i in range(3):
        try:
            engine_pg = db_engine_us(site,'postgresql_14_outer')
            df = engine_pg.read_sql(select_state1_sql)
            return df
        except:
            print('db_cursor_connect_pg_read 报错：', select_state1_sql)


def select_sate_mysql(site=None, num=None, page=None, month=None, week=None, spider_int=None):
    db_class = connect_db(site)
    print('当前抓取月：：', month)
    if int(week) < 10:
        week = f'0{int(week)}'
    year_week = f'2025-{week}'
    year_month = f'2025-{month}'
    if num is None:
        if site in ['fr', 'es', 'it']:
            print('查询 周工作进度 工作进度 状态')
            select_state1_sql = f"select status_val from workflow_progress where site_name='{site}' and date_info='{year_week}' and date_type='week' and page='反查搜索词'"
            print(select_state1_sql)
            df = db_cursor_connect_msyql_read(select_state1_sql)
            if not df.empty:
                if df.status_val[0] == 3:
                    print(f"{site} 站点 {week} 周 搜索词 已完成 执行下一步")
                    select_sate_sql = f"select status_val from workflow_progress where site_name='{site}' and date_info='{year_week}' and date_type='week' and page='ASIN详情'"
                    print(select_sate_sql)
                    df1 = db_cursor_connect_msyql_read(select_state1_sql)
                    if df1.status_val[0] != 3:
                        update_workflow_progress = f"update workflow_progress set status_val=2,status='抓取中' where page='ASIN详情' and date_info='2025-{week}' and site_name='{site}' and date_type='week' and status_val in(1,0)"
                        db_cursor_connect_update(update_workflow_progress, site)
                        print(f"{site} 站点 未全部完成 数据 抓取")
                        return True
                    else:
                        return False
                else:
                    return False
        else:
            print('查询 工作进度 工作进度 状态')
            # 定义锁的键
            redis_client = BaseUtils().redis_db()
            lock_key = f"{year_week}_{site}_lock"
            lock = redis_client.lock(lock_key, timeout=55)  # 10秒超时
            select_sql = f"select status_val from workflow_progress WHERE date_info='{year_week}' and date_type='week' and site_name='{site}' and page='ASIN详情'"
            print(select_sql)
            df_state = db_cursor_connect_msyql_read(select_sql)
            print('df_state.status_val::', df_state.status_val)
            if int(df_state.status_val[0]) != 3:
                try:
                    if lock.acquire(blocking=True):
                        spider_state_sql = f"select id from workflow_progress where date_info='{year_week}' and date_type='week' and site_name='{site}' and page='ASIN详情' and up_spider_state=1"
                        print(spider_state_sql)
                        df_id = db_cursor_connect_msyql_read(spider_state_sql)
                        if not df_id.empty:
                            update_up_spider_state_list = f"update workflow_progress set up_spider_state=3 where date_info='{year_week}' and date_type='week' and site_name='{site}' and page='ASIN详情' and up_spider_state=1"
                            db_cursor_connect_update(update_up_spider_state_list, site)
                            update_month_spider_state = f"update workflow_progress set spider_state=2,spider_int={spider_int} WHERE site_name='{site}' and date_type='month' and date_info='{year_month}' and page='ASIN详情'"
                            print(update_month_spider_state)
                            db_cursor_connect_update(update_month_spider_state, site)
                            DolphinschedulerHelper.start_process_instance_common(
                                project_name="big_data_selection",
                                process_df_name='ALL-实时-插件+流量选品',
                                startParams={
                                    "site_name": f"{site}",
                                    "date_type": "month",
                                    "date_info": f'2025-{month}'
                                }
                            )
                            print('发送消息通知')
                            account = 'pengyanbing,chenyuanjie,hezhe,wangrui4,fangxingjun,zhouyuchen'
                            title = site + f' {site} 开启 kafka实时消费 {year_month}'
                            content = f' {site} 开始 调度 ALL-实时-插件+流量选品 抓取asin'
                            db_class.send_mg(account, title, content)
                        else:
                            print("获取锁失败，另一个进程正在运行.")
                except LockError:
                    print("获取锁失败")
                except Exception as e:
                    print(f"错误: {e}")
                finally:
                    if lock.locked():
                        lock.release()
                n = 0
                while True:
                    try:
                        sql_read = f"select id from workflow_progress where date_info='{year_month}' and date_type='month' and page='ASIN详情' and site_name='{site}' and kafka_flow_state=3 and spider_state=2"
                        print(sql_read, '232323')
                        df_report_date = db_cursor_connect_msyql_read(sql_read)
                        if n > 20:
                            up_state_sql = f"update workflow_progress set up_spider_state=1 where date_info='{year_week}' and date_type='week' and site_name='{site}' and page='ASIN详情'"
                            print('up_state_sql::', up_state_sql)
                            db_cursor_connect_update(up_state_sql, site)
                            return False
                        if df_report_date.id[0]:
                            return True
                        else:
                            n += 1
                            time.sleep(120)
                    except:
                        time.sleep(10)
    if num == 1:
        if site in ['fr', 'es', 'it']:
            print(f' {site}  _all_syn_st 完成')
        else:
            sql_select_ = f"select status_val from workflow_progress where date_info='{year_week}' and date_type='week' and page='ASIN详情' and site_name='{site}'"
            print(sql_select_)
            df_status_dict = db_cursor_connect_msyql_read(sql_select_)
            print('查询kafka是否开启', df_status_dict)
            if int(df_status_dict.status_val[0]) in (1, 2):
                update_workflow_progress = f"update workflow_progress set status_val=3,status='ASIN爬取完成',up_spider_state=3 where page='ASIN详情' and date_info='{year_week}' and site_name='{site}' and date_type='week'"
                print(update_workflow_progress)
                db_cursor_connect_update(update_workflow_progress, site)
                db_class.send_mg('pengyanbing', '修改进度表', update_workflow_progress)
                ii = 0
                for i in range(10):
                    time.sleep(180)
                    ii += 1
                    if ii > 10:
                        break
                # SELECT * from workflow_progress WHERE site_name='us' and page='asin详情' and date_type='month' and status_val=1 and status='月ASIN导出完成  and date_info='
                update_month_asin_state = f"update workflow_progress set status_val=3,status='月ASIN抓取完成' WHERE site_name='{site}' and page='asin详情' and date_type='month' and status_val=1 and status='月ASIN导出完成' and date_info='{year_month}'"
                print(update_month_asin_state)
                db_cursor_connect_update(update_month_asin_state, site)
                db_class.send_mg('pengyanbing', '修改 月 维度 进度表', update_month_asin_state)
                update_month_spider_state = f"update workflow_progress set  kafka_flow_state=1,spider_state=3,spider_int={spider_int} WHERE site_name='{site}' and date_type='month' and date_info='{year_month}' and page='ASIN详情'"
                db_cursor_connect_update(update_month_spider_state, site)
                DolphinschedulerHelper.start_process_instance_common(
                    project_name="big_data_selection",
                    process_df_name='ALL站点-启动30day/月流程',
                    startParams={
                        "site_name": f"{site}",
                        "date_type": "month",
                        "date_info": f'2025-{month}'
                    }
                )
                time_strftime = time.strftime("%Y-%m-%d %X", time.localtime())
                # print('发送消息通知')
                account = 'pengyanbing,chenyuanjie,hezhe,wangrui4,fangxingjun,chenjianyun,zhouyuchen'
                title = site + '站点 asin 详情'
                content = str(month) + ' 月  asin 详情 已结束,请确认下一步流程!时间:' + time_strftime
                db_class.send_mg(account, title, content)
    if num == 2:
        sql_read = f"SELECT status_val FROM workflow_progress where page='{page}' and date_info='{year_month}' and site_name='{site}' and date_type='month'"
        print(sql_read)
        df_status_val_tuple = db_cursor_connect_msyql_read(sql_read)
        if df_status_val_tuple.status_val[0] == 3:
            print(site, page, '已完成')
            return False
        else:
            print(f'开始 {page}')
            return True
    if num == 3:
        for i in range(5):
            try:
                sql_read = f'SELECT id FROM {site}_all_syn_st_month_2025_{month} where state in (1,2) LIMIT 1'
                print("sql_read:", sql_read)
                df = db_cursor_connect_pg_read(site,sql_read)
                if not df.empty:
                    id_tuple = [1]
                else:
                    id_tuple = None
                if id_tuple is None:
                    if site in ['us', 'de', 'uk']:
                        select_sate_mysql(site, num=1, month=month, week=week, spider_int=spider_int)
                    else:
                        select_sate_mysql(site, num=1, week=week)
                    return False
                else:
                    return True
            except Exception as e:
                print('查询状态 1 2 报错',e)
# if __name__ == '__main__':
#     DolphinschedulerHelper.start_process_instance_common(
#         project_name="big_data_selection",
#         process_df_name='ALL站点-启动30day/月流程',
#         startParams={
#            "site_name": "uk",
#             "date_type": "month",
#            "date_info": '2025-10'
#         }
#     )