print('抓取 月 店铺数据')
import os
import sys
from datetime import datetime, timedelta

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
import time
import random
import multiprocessing
from amazon_spider.seller_account_feedback import async_account_feedback
from amazon_spider.products_asin import async_account_name_products
from amazon_spider.VPS_IP import pppoe_ip
from threading_spider.db_connectivity import connect_db
from utils.secure_db_client import get_remote_engine
from utils.db_connect import BaseUtils
from threading_spider.post_to_dolphin import DolphinschedulerHelper


def db_engine(site_name, db_type):
    print(site_name, db_type,'222222site_name, db_typesite_name, db_type')
    return get_remote_engine(site_name=site_name, db_type=db_type)


def db_execute(sql, site=None):
    print(sql,site,'-----------------------23232')
    for _ in range(3):
        try:
            with db_engine('us', 'mysql').begin() as conn:
                conn.execute(sql)
            return
        except Exception as e:
            print(site, 'db_execute 报错：', e, sql)


def seller_syn_table(site):
    return f'{site}_seller_account_syn_distinct'


def has_pending_feedback(site):
    """feedback sync 表是否还有 state in (1,2)"""
    print('feedback sync 表是否还有 state in (1,2)')
    table = seller_syn_table(site)
    print(table,'tabletabletabletable')
    for _ in range(5):
        try:
            df = db_engine(site, 'mysql').read_sql(
                f"SELECT id FROM {table} WHERE state IN (1,2) LIMIT 1")
            print(df)
            return not df.empty
        except Exception as e:
            print(e, '查 feedback state 报错')
            time.sleep(3)
    return True  # 查询失败时保守返回 True，避免误判完成


def has_pending_products(site):
    """products sync 表是否还有 product_state in (1,2)"""
    print("""products sync 表是否还有 product_state in (1,2)""")
    table = seller_syn_table(site)
    for _ in range(5):
        try:
            df = db_engine(site, 'mysql').read_sql(
                f"SELECT id FROM {table} WHERE product_state IN (1,2) LIMIT 1")
            return not df.empty
        except Exception as e:
            print(e, '查 products state 报错')
            time.sleep(3)
    return True


def has_any_pending(site):
    """两个阶段任意一个还有数据则返回 True"""
    return has_pending_feedback(site) or has_pending_products(site)


def mark_started(site, date_info, spider_name):
    """workflow_manager state 1→2，spider_is_ready='yes' 保护"""
    sql = (f"UPDATE workflow_manager SET spider_state=2 "
           f"WHERE site_name='{site}' AND date_type='month' AND date_info='{date_info}' "
           f"AND spider_is_ready='yes' AND spider_name='{spider_name}' AND spider_state=1")
    db_execute(sql, site)
    print(f'[启动] {site} {date_info} {spider_name} 推进 state=2')


def finalize_task(site, date_info, spider_name, db_class, title, body):
    """两阶段全部完成后：state 2→3 + NX 锁防多机重复触发"""
    redis_client = BaseUtils().redis_db()

    if has_any_pending(site):
        print(f'[结束] {site} {spider_name} 同步表还有数据，暂不结束')
        return False

    # state 必须为 2 才允许推进
    try:
        sql_check = (f"SELECT spider_state FROM workflow_manager "
                     f"WHERE site_name='{site}' AND date_type='month' AND date_info='{date_info}' "
                     f"AND spider_name='{spider_name}'")
        df = db_engine('us', 'mysql').read_sql(sql_check)
        if df.empty:
            print(f'[结束] {site} {spider_name} workflow_manager 无匹配行，跳过')
            return False
        if df.spider_state.iloc[0] != 2:
            print(f'[结束] {site} {spider_name} state={df.spider_state.iloc[0]} 非2，跳过')
            return False
    except Exception as e:
        print('[结束] 查 spider_state 异常', e)
        return False

    # NX 锁（多台 VPS 同时完成时只允许 1 台推进）
    nx_key = f"seller_{spider_name}_done_{site}_{date_info}"
    if not redis_client.set(nx_key, 1, nx=True, ex=1800):
        print(f'[结束] {site} {spider_name} NX 已锁，跳过')
        return False

    # 复核等待，防止数据回流误触发
    time.sleep(60)
    if has_any_pending(site):
        redis_client.delete(nx_key)
        print(f'[结束] {site} {spider_name} 复核发现数据回流，归还 NX')
        return False

    sql_done = (f"UPDATE workflow_manager SET spider_state=3, spider_is_ready='no' "
                f"WHERE site_name='{site}' AND date_type='month' AND date_info='{date_info}' "
                f"AND spider_is_ready='yes' AND spider_name='{spider_name}' AND spider_state=2")
    db_execute(sql_done, site)
    db_class.send_mg(
        'pengyanbing,chenyuanjie,fangxingjun,zhouyuchen,guohuanjie',
        title,
        body + str(datetime.now())
    )
    print(f'[结束] {site} {date_info} {spider_name} 完成，已发送通知')
    DolphinschedulerHelper.start_process_instance_common(
        project_name="big_data_selection",
        process_df_name='店铺数据导入-月流程',
        startParams={"site_name": site, "date_type": "month", "date_info": date_info}
    )
    return True


def run_feedback_worker(site, proxy_name, date_info):
    spider = async_account_feedback(site, read_size=250, proxy_name=proxy_name,
                                    date_info=date_info)
    spider.run()


def run_products_worker(site, proxy_name, date_info):
    print(34244444444)
    spider = async_account_name_products(site, read_size=250, proxy_name=proxy_name,
                                         date_info=date_info)
    spider.run()


def run_phase(site, date_info, proxy_name, worker_fn, check_pending_fn, phase_name):
    """单阶段抓取循环：多进程 + IP 切换 + 无数据自动退出"""
    print('单阶段抓取循环：多进程 + IP 切换 + 无数据自动退出')
    if not check_pending_fn(site):
        print(f'[{phase_name}] {site} 无待抓数据，跳过本阶段')
        return

    while True:
        five_min_later = datetime.now() + timedelta(minutes=3)
        print('232323232323232323232323232323232323232323')
        processes = []
        for _ in range(6):
            p = multiprocessing.Process(target=worker_fn,
                                        args=(site, proxy_name, date_info))
            processes.append(p)
            p.start()
        for p in processes:
            p.join()

        print(f'[{phase_name}] 所有进程运行完毕，准备切换 IP')
        if datetime.now() > five_min_later:
            pppoe_ip()
        else:
            time.sleep(random.uniform(150, 220))
            pppoe_ip()

        if not check_pending_fn(site):
            print(f'[{phase_name}] {site} 数据清空，退出本阶段')
            break


if __name__ == '__main__':
    # 调用示例: /usr/local/bin/python3 /mnt/py_spider/threading_spider/Poll_site_seller_account.py us 2026-05
    pppoe_ip()
    site = sys.argv[1]      # 'us'
    date_info = sys.argv[2]   # '2026-05'
    spider_name = f'{site}_spider_fd'
    proxy_name = None
    db_class = connect_db(site)

    print(f'站点: {site}, date_info: {date_info}, spider_name: {spider_name}')

    # 整体启动：workflow_manager state 1→2（spider_is_ready='yes' 保护，误启动不影响）
    mark_started(site, date_info, spider_name)
    print('eeeeeeeeeeeeeeeeew')
    # Phase 1：店铺 Feedback
    run_phase(site, date_info, proxy_name,
              run_feedback_worker, has_pending_feedback, 'Feedback')

    # Phase 2：店铺产品 Top20
    run_phase(site, date_info, proxy_name,
              run_products_worker, has_pending_products, 'Products')

    # 两阶段均完成：state 2→3 + 通知（NX 锁保证多机只触发一次）
    finalize_task(
        site=site, date_info=date_info, spider_name=spider_name,
        db_class=db_class,
        title=f'{site} 店铺数据 {date_info}',
        body=f'{date_info} 月 店铺数据已结束，请确认下一步流程！时间：',
    )
