import os
import sys
import traceback

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.requests_param import Requests_param_val
from utils.db_connect import BaseUtils
import random
import time
from lxml import etree
import urllib3
from queue import Queue
import threading
import pandas as pd
import datetime
from amazon_spider.VPS_IP import pppoe_ip
from curl_cffi import requests
from utils.secure_db_client import get_remote_engine
from threading_spider.post_to_dolphin import DolphinschedulerHelper
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
import json
import uuid

"""抓取亚马逊 bsr分类，和 bsr asin top100"""


class bsr_category(BaseUtils):

    _SITE_CONFIG = {
        'us': ('https://www.amazon.com', 'www.amazon.com'),
        'uk': ('https://www.amazon.co.uk', 'www.amazon.co.uk'),
        'de': ('https://www.amazon.de', 'www.amazon.de'),
        'fr': ('https://www.amazon.fr', 'www.amazon.fr'),
        'es': ('https://www.amazon.es', 'www.amazon.es'),
        'it': ('https://www.amazon.it', 'www.amazon.it'),
    }
    _BSR_TITLE_PREFIX = {
        'us': 'Best Sellers in',
        'uk': 'Best Sellers in',
        'de': 'Bestseller in',
        'fr': 'Les meilleures ventes en',
        'es': 'Los más vendidos en',
        'it': 'Bestseller in',
    }

    def __init__(self, site_name):
        super().__init__()
        self.site_name = site_name
        self.week = int(time.strftime("%W"))
        self.year = int(time.strftime("%Y"))
        today = datetime.datetime.today()
        self.time_strftime_ = time.strftime("%Y-%m-%d", time.localtime())
        self.month = today.month
        self.name_path_queue = Queue()
        self.insert_list = []
        self.insert_id_pid_list = []
        self.id_and_en_name_list = []
        self.init_db(site_name)
        self.name_id_queue = Queue()
        self.cookies_queue = Queue()
        self.item_list = []
        self.delect_asin_list = []
        self.catgory_next_path_list = []
        self.asin_dict_ = {}
        self.item_queue = Queue()
        self.redirect_first_id_list = []
        self.path_en_name_list = []
        self.headers_num_int = 0
        self.columns = ['cate_current_id', 'asin', 'bsr_rank', 'price', 'rating', 'total_comments', "week",
                        "year_month", 'category_id']
        self._ip_switch_lock = threading.Lock()
        self._last_ip_switch_time = 0

    def _safe_switch_ip(self, min_interval=60):
        """线程安全的IP切换，避免多线程重复切换"""
        with self._ip_switch_lock:
            now = time.time()
            if now - self._last_ip_switch_time < min_interval:
                print(f'距离上次切换IP不到{min_interval}秒，跳过')
                return False
            print('切换IP...')
            pppoe_ip()
            self._last_ip_switch_time = time.time()
        time.sleep(5)  # sleep 移到锁外，避免长时间阻塞其他线程
        return True

    def db_engine_us(self, site_name, db_type):
        engine_mysql = get_remote_engine(
            site_name=site_name,
            db_type=db_type,
        )
        return engine_mysql

    def db_cursor_connect_update(self, sql, site):
        for i in range(3):
            try:
                engine_us_mysql = self.db_engine_us(site, '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(self, site=None, select_state1_sql=None):
        for i in range(3):
            try:
                if site:
                    engine_us_mysql = self.db_engine_us(site, 'mysql')
                else:
                    engine_us_mysql = self.db_engine_us(self.site_name, '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 init_db(self, site_name):
        self.engine = self.mysql_connect()
        self.engine_pg = self.pg_connect()
        self.engine_pg6 = self.pg_connect_6()
        self.site_url, self.host = self._SITE_CONFIG[site_name]
        self.reuests_para_val = Requests_param_val(site_name=self.site_name)
        self.year_month = f'{self.year}_{self.month}'
        sele_sql = f"SELECT `week` FROM week_20_to_30 WHERE `year_month`='{self.year}_{self.month}'"
        print(sele_sql)
        df_year_week = self.db_cursor_connect_msyql_read(site='us', select_state1_sql=sele_sql)
        self.year_week = list(df_year_week['week'])[-1]
        print(self.year_week, '====当前周===1232333')

    def safeIndex(self, list: list, index: int, default: object = None):
        """安全获取list的索引对应的值"""
        if index <= len(list) - 1:
            return list[index]
        return default

    def _build_headers(self, url):
        """构建随机请求头，防反爬"""
        n = random.randint(70, 114)
        ua = (f'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
              f'(KHTML, like Gecko) Chrome/{n}.0.{random.randint(1000, 5000)}'
              f'.{random.randint(1, 181)} Safari/537.36')
        headers = {
            'connection': 'close',
            'authority': self.host,
            'accept': 'text/html,*/*',
            'accept-language': 'zh-CN,zh;q=0.9',
            'cache-control': 'no-cache',
            'content-type': 'application/x-www-form-urlencoded;charset=UTF-8',
            'origin': url,
            'referer': url,
            'sec-ch-ua-mobile': '?0',
            'user-agent': ua,
        }
        alphabet = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n']
        k = ""
        for i in (0, random.randint(0, 5)):
            k += random.choice(alphabet)
        headers[k] = str(uuid.uuid4())
        return headers

    def _is_captcha(self, text):
        """检测页面是否出现验证码"""
        return any(kw in text for kw in [
            "Enter the characters you see below",
            "Geben Sie die Zeichen unten ein",
            "Introduce los caracteres que se muestran",
            "Saisissez les caractères que vous voyez",
            "Inserisci i caratteri visualizzati nello spazio",
        ])

    def _parse_h1_name(self, response):
        """从页面 h1 解析 BSR 分类名称"""
        prefix = self._BSR_TITLE_PREFIX.get(self.site_name, '')
        if not prefix:
            return None
        h1_list = response.xpath(f"//h1[contains(text(),'{prefix}')]/text()")
        if h1_list:
            return h1_list[0].replace(prefix, '').strip()
        return None

    def _get_sibling_first_id(self, response, bsr_class_name, bum, bsr_url_category_id_dict):
        """获取同级节点的 first_id 映射，用于 redirect_first_id 更新"""
        if (bum - 1) < 2:
            return []
        xpath_herf_list = [
            f'//div[@role="treeitem"]/span[contains(text(),"{bsr_class_name}")]/parent::div/parent::div//a/@href',
            f'//div[@role="treeitem"]/span[contains(text(),"{bsr_class_name}")]/parent::div/parent::div/parent::div//a/@href',
            f'//div[@role="treeitem"]/span[contains(text(),"{bsr_class_name[0:5]}")]/parent::div/parent::div/parent::div//a/@href',
        ]
        for xpath_herf in xpath_herf_list:
            a_herf_list = response.xpath(xpath_herf)
            if a_herf_list:
                a_url_dict = self.parse_url(bum, self.site_url + a_herf_list[0])
                return [[bsr_url_category_id_dict['category_first_id'], a_url_dict['category_first_id']]]
        return []

    def parse_url(self, nodes_num: int, url: str):
        """统一解析链接获取 bsr 分类id等数据"""
        arr = url.split("/")
        if "ref=" not in url:
            ref_suffix = None
            category_id = self.safeIndex(arr, len(arr) - 1, None)
            category_first_id = self.safeIndex(arr, len(arr) - 2, None)
        else:
            ref_suffix = self.safeIndex(arr, len(arr) - 1, None)
            category_first_id = self.safeIndex(arr, len(arr) - 3, None)
            category_id = self.safeIndex(arr, len(arr) - 2, None)
        if nodes_num == 1:
            level = 1
        elif url.endswith("_0"):
            level = 2
        elif url.endswith(f"{category_first_id}_1"):
            level = 3
        else:
            level = 4

        # 获取 parent id
        if level == 1:
            # 根节点
            category_id = "0"
            category_first_id = None
            category_parent_id = None
        elif level == 2:
            # 一级节点
            category_id = category_id
            category_first_id = category_id
            category_parent_id = "0"
        elif level == 3:
            # 一级节点下的次级节点
            category_id = category_id
            category_first_id = category_first_id
            category_parent_id = category_first_id
        elif level == 4:
            category_id = category_id
            category_first_id = category_first_id
            if ref_suffix is not None:
                category_parent_id = ref_suffix[ref_suffix.rfind("_") + 1:]
            else:
                category_parent_id = None
        else:
            category_id = None
            category_first_id = None
            category_parent_id = None

        items = {
            "category_id": category_id,
            "category_first_id": category_first_id,
            "category_parent_id": category_parent_id
        }
        return items

    def html_4(self, bum):
        print('bum 当前请求的层级：', bum)
        while True:
            if not self.name_path_queue.empty():
                querys = self.name_path_queue.get()
                print('请求=====：', querys)
                bsr_url = querys[2]
                bsr_class_name = querys[1]
                bsr_id = querys[0]
                category_id = querys[3]
                cookie_str = self.get_cookies()
                headers = self._build_headers(bsr_url)
                headers["cookie"] = cookie_str
                num = 0
                while True:
                    try:
                        resp = requests.get(bsr_url, headers=headers,
                                            timeout=20, impersonate="chrome")
                        if self._is_captcha(resp.text):
                            print('请求出现验证码：')
                            num += 1
                            if num > 20:
                                self._safe_switch_ip()
                                num = 0
                            continue

                        if resp.status_code == 200 or resp.status_code == 201:
                            response = etree.HTML(resp.text)
                            response.xpath("//span[@id='glow-ingress-line2']/text()")
                            break
                    except Exception as e:
                        print("报错 继续 请求", e)
                        time.sleep(random.uniform(1.5, 5.5))
                        continue
                # with open(rf'D:\新建文件夹\requests_files/{self.site_name}_bsr.html', 'w', encoding='utf-8')as f:
                #     f.write(resp.text)
                a_2_list = response.xpath(
                    f'//div[@role="treeitem"]/span[contains(text(),"{bsr_class_name}")]/parent::div/following-sibling::div[@role="group"]/div[@role="treeitem"]/a')
                print(
                    f'//div[@role="treeitem"]/span[contains(text(),"{bsr_class_name}")]/parent::div/following-sibling::div[@role="group"]/div[@role="treeitem"]/a')
                h1_bsr_name = self._parse_h1_name(response)
                # self.id_and_en_name_list 1 代表还有 子分类 2代表没有子分类，为最终节点
                bsr_url_category_id_dict = self.parse_url(bum - 1, bsr_url)
                print('bsr_url_category_id_dict', bsr_url_category_id_dict)
                print('获取下一级数量：', len(a_2_list))
                if a_2_list:
                    self.id_and_en_name_list.append((bsr_id, h1_bsr_name, 1, bsr_url_category_id_dict['category_id'],
                                                     bsr_url_category_id_dict['category_parent_id']))
                    print(self.id_and_en_name_list, '23434343434343434343434')
                    redirect_first_id_list = self._get_sibling_first_id(
                        response, bsr_class_name, bum, bsr_url_category_id_dict)
                    name_2_href_list = []
                    for a_2 in a_2_list:
                        name_2 = a_2.xpath('./text()')[0].strip()
                        name_2_href = a_2.xpath('./@href')[0].strip()
                        if 'com/-/zh/-' in name_2_href:
                            int_time = int(time.time() * 1000)
                            with open(f'{int_time}_{name_2}.html', 'w', encoding='utf-8') as f:
                                f.write(resp.text)
                        href = self.site_url + name_2_href
                        id_pid_data_dict = self.parse_url(bum, href)
                        print('id_pid_data_dict: ', id_pid_data_dict)
                        self.insert_id_pid_list.append(
                            [name_2, href, id_pid_data_dict['category_id'], id_pid_data_dict['category_parent_id']])
                        # 父级类目自增id  类目名称 二级类目， 链接
                        self.insert_list.append((bsr_id, name_2, bum, href, id_pid_data_dict['category_id'],
                                                 id_pid_data_dict['category_parent_id']))
                        # 存储当前分类id，和页面获取的下一级url
                        name_2_href_list.append(href)
                        # 存储path 和 名称
                        self.path_en_name_list.append([href, name_2])
                    self.catgory_next_path_list.append(
                        [bsr_url_category_id_dict['category_id'], '|-|'.join(name_2_href_list)])
                else:
                    redirect_first_id_list = self._get_sibling_first_id(
                        response, bsr_class_name, bum, bsr_url_category_id_dict)
                    self.id_and_en_name_list.append((bsr_id, h1_bsr_name, 2, bsr_url_category_id_dict['category_id'],
                                                     bsr_url_category_id_dict['category_parent_id']))
                if redirect_first_id_list:
                    # 获取当前请求url 分类id，对 first_id 拼接
                    self.redirect_first_id_list.append([bsr_url, '|-|'.join(redirect_first_id_list[0])])
                print(self.year_week, '====当前周===', self.week)
                ele_next = response.xpath(
                    '//a[contains(text(), "Weiter")]/@href|//a[contains(text(), "Next")]/@href|//a[contains(text(), "Page suivante")]/@href|//a[contains(text(), "Suivant")]/@href|//a[contains(text(), "Avanti")]/@href|//a[contains(text(), "Siguiente")]/@href|//a[contains(text(), "Seite")]/@href|//a[contains(text(), "siguiente")]/@href|//a[contains(text(), "Pagina")]/@href')
                data_client_recs_list = response.xpath(
                    "//div[@class='p13n-desktop-grid']/@data-client-recs-list")
                if data_client_recs_list:
                    data_list = json.loads(data_client_recs_list[0])
                    print('第一页获取asin:', len(data_list))
                    for data in data_list:
                        asin = data['id']
                        rank = data['metadataMap']['render.zg.rank']
                        self.item_list.append(
                            (bsr_id, asin, int(rank), None, None, None, self.week, self.year_month,
                             bsr_url_category_id_dict['category_id']))
                if ele_next and data_client_recs_list:
                    self.get_bsr_asin_data(ele_next, bsr_id, bsr_url_category_id_dict['category_id'], asin_dict_={})
            else:
                break

    def get_cookies(self):
        if self.cookies_queue.empty():
            cookies_dict = self.reuests_para_val.get_cookie()
            self.cookie_dict_delete_id = cookies_dict
            for ck in cookies_dict.values():
                self.cookies_queue.put(ck)
        while 1:
            cookie_str = self.cookies_queue.get()
            if len(cookie_str) > 50:
                try:
                    cookie_lsit = json.loads(cookie_str)
                except:
                    cookie_lsit = eval(cookie_str)
                cookie_dic = {}
                try:
                    for i in cookie_lsit:
                        if i:
                            cookie_dic[i["name"]] = i["value"]
                        else:
                            continue
                    cookie_str = ''
                    for k, v in cookie_dic.items():
                        cookie_str = cookie_str + str(k) + '=' + str(v) + ';'
                    return cookie_str
                except:
                    cookie_str = ''
                    for k, v in cookie_lsit.items():
                        cookie_str = cookie_str + str(k) + '=' + str(v) + ';'
                    return cookie_str
            else:
                return ''

    def get_bsr_asin_data(self, ele_next, c_id, bsr_url_category_id, asin_dict_):
        next_url = self.site_url + ele_next[0]
        num = 0
        while True:
            try:
                cookie_str = self.get_cookies()
                if not cookie_str:
                    cookie_str = self.get_cookies()
                headers = self._build_headers(next_url)
                headers['Sec-Fetch-Site'] = 'same-origin'
                headers['Sec-Fetch-Mode'] = 'cors'
                headers['Sec-Fetch-Dest'] = 'empty'
                headers['Accept'] = '*/*'
                headers['X-Requested-With'] = 'XMLHttpRequest'
                headers["cookie"] = cookie_str

                print('请求第二页：：', next_url)
                resp = requests.get(next_url, headers=headers, timeout=15, impersonate="chrome")

                if self._is_captcha(resp.text):
                    print("出现验证码")
                    num += 1
                    if num > 20:
                        self._safe_switch_ip()
                        num = 0
                    time.sleep(random.uniform(3.5, 5.5))
                    continue
                if resp.status_code == 200 or resp.status_code == 201:
                    response = etree.HTML(resp.text)
                    response.xpath("//span[@id='glow-ingress-line2']/text()")
                    break
                else:
                    print('resp.status_code:L:', resp.status_code)
                    num += 1
                    if resp.status_code == 503:
                        self._safe_switch_ip()
                        num = 0
            except Exception as e:
                print("报错 继续 请求", e)
                time.sleep(random.uniform(5.5, 15.5))
                num += 1
                if num > 20:
                    self._safe_switch_ip()
                    num = 0
                continue

        data_client_recs_list = response.xpath("//div[@class='p13n-desktop-grid']/@data-client-recs-list")
        if data_client_recs_list:
            data_list = json.loads(data_client_recs_list[0])
            print('第二页获取asin:', len(data_list))
            for data in data_list:
                asin = data['id']
                rank = data['metadataMap']['render.zg.rank']
                self.item_list.append(
                    (c_id, asin, int(rank), None, None, None, self.week, self.year_month, bsr_url_category_id))
        else:
            print('没有获取到第二页获取asin：：', next_url)

    def save_asin_data(self):
        while True:
            try:
                self.engine_pg = self.pg_connect()
                print('存储====::', len(self.item_list))
                df = pd.DataFrame(data=self.item_list, columns=self.columns)
                df['date_info'] = self.time_strftime_
                df.drop_duplicates(['asin', 'bsr_rank', 'cate_current_id'], inplace=True)  # 去重
                print(df.shape)
                self.engine_pg.to_sql(df, f'{self.site_name}_bs_top100_asin_2026', if_exists="append")
                self.item_list = []
            except Exception as e:
                print(e, '报错存储 pg')
                time.sleep(10)
                self.engine_pg = self.pg_connect()
                continue
            break

    def init_list(self):
        self.insert_list = []
        self.insert_id_pid_list = []
        self.name_path_queue = Queue()
        self.id_and_en_name_list = []
        self.catgory_next_path_list = []
        self.name_id_queue = Queue()
        self.cookies_queue = Queue()
        self.asin_dict_ = {}
        self.item_list = []
        self.headers_num_int = 0
        self.path_en_name_list = []
        self.redirect_first_id_list = []

    def update_delete_time(self, redirect_flag=None):
        try:
            if self.catgory_next_path_list:
                for category_path in self.catgory_next_path_list:
                    print('category_path::', category_path)
                    category_url_list = category_path[1].split('|-|')
                    category_id = category_path[0]
                    path_sql = f"""
                           select id, category_id, category_parent_id, en_name, `path`, delete_time
                       from {self.site_name}_bs_category
                       where (category_parent_id = "{category_id}")
                       order by {self.site_name}_bs_category.category_id, category_parent_id;
                       """
                    print('path_sql:', path_sql)
                    df_exist_rows = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=path_sql)
                    exist_rows = df_exist_rows.values.tolist()
                    group1_id = []
                    group2_id = []
                    for row in exist_rows:
                        print('rowrow::', row)
                        if row[4] in category_url_list:
                            group1_id.append(row[0])
                        else:
                            group2_id.append(row[0])
                    _strftime_ = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                    print(_strftime_)
                    if group1_id:
                        if len(group1_id) == 1:
                            sql1 = f'update {self.site_name}_bs_category set delete_time = null where id in ({group1_id[0]})'
                        else:
                            sql1 = f"update {self.site_name}_bs_category set delete_time = null where id in {tuple(group1_id)}"
                            print('sql1::', sql1)
                        self.db_cursor_connect_update(sql1, self.site_name)
                    if group2_id:
                        if len(group2_id) == 1:
                            sql2 = f"update {self.site_name}_bs_category set delete_time = '{_strftime_}' where id in ({group2_id[0]})"
                        else:
                            sql2 = f"update {self.site_name}_bs_category set delete_time = '{_strftime_}' where id in {tuple(group2_id)}"
                        print('sql2::', sql2)
                        self.db_cursor_connect_update(sql2, self.site_name)
        except Exception as e:
            print('更新 delete_time 报错', traceback.format_exc())
        try:
            if self.redirect_first_id_list:
                for redirect_first_id in self.redirect_first_id_list:
                    print('redirect_first_id:', redirect_first_id)
                    category_bsr_url = redirect_first_id[0]
                    redirect_first_id_name_id_list = redirect_first_id[1].split('|-|')
                    len_first_name = list(set(redirect_first_id_name_id_list))
                    print('len_first_name', len_first_name)
                    if redirect_flag:
                        print('重定向url：', category_bsr_url)
                        update_first_id_sql = f'''update {self.site_name}_bs_category set redirect_first_id="{redirect_first_id_name_id_list[1]}" where `path`="{category_bsr_url}"'''
                        print('update_first_id_sql::', update_first_id_sql)
                        self.db_cursor_connect_update(update_first_id_sql, self.site_name)
                    else:
                        if len(len_first_name) == 2:
                            print('重定向url：', category_bsr_url)
                            update_first_id_sql = f'''update {self.site_name}_bs_category set redirect_first_id="{redirect_first_id_name_id_list[1]}" where `path`="{category_bsr_url}"'''
                            print('update_first_id_sql::', update_first_id_sql)
                            self.db_cursor_connect_update(update_first_id_sql, self.site_name)
        except Exception as e:
            print('更新 redirect_first_id_name_id_list 报错', traceback.format_exc())
            self.init_db(self.site_name)

    def save_date(self):
        print('存储：：save_date')
        self.init_db(self.site_name)
        for name_num_path in self.insert_list:
            print('name_num_path：', name_num_path)
            save_name_num_list = []
            while True:
                #  不存在就插入
                try:
                    select_sql_id = f'''SELECT id FROM {self.site_name}_bs_category  WHERE `path`="{name_num_path[3]}"'''
                    df_id = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=select_sql_id)
                    if not df_id.empty:
                        save_name_num_list.append(name_num_path)
                    else:
                        select_sql_name = f'''SELECT en_name FROM {self.site_name}_bs_category  WHERE `path`="{name_num_path[3]}" order by id desc '''
                        df_en_name = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=select_sql_name)
                        print('en_name::', df_en_name.values)
                        if df_en_name['en_name'][0] == name_num_path[1]:
                            pass
                        else:
                            update_name_sql = f'''update {self.site_name}_bs_category set delete_time = '2023-06-19 00:00:00' WHERE `path`="{name_num_path[3]}" and delete_time is null'''
                            print('更新 en_name：', update_name_sql)
                            self.db_cursor_connect_update(update_name_sql, self.site_name)
                            save_name_num_list.append(name_num_path)

                        select_sql_name_1 = f'''SELECT en_name,id FROM {self.site_name}_bs_category  WHERE `path`="{name_num_path[3]}" and delete_time is null'''
                        df_en_name_1 = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=select_sql_name_1)
                        print('en_name::', df_en_name_1.values)
                        if df_en_name_1['en_name'][0] == name_num_path[1]:
                            pass
                        else:
                            update_name_sql_1 = f'''update {self.site_name}_bs_category set delete_time = '2023-06-19 00:00:00' WHERE id={df_en_name_1['id'][0]} and delete_time is null'''
                            print('更新 en_name：', update_name_sql_1)
                            self.db_cursor_connect_update(update_name_sql_1, self.site_name)
                    break
                except Exception as e:
                    print(e)
                    time.sleep(10)
                    self.init_db(self.site_name)
                    continue
            # # 插入新的数据
            print('save_name_num_list::', save_name_num_list)
            if save_name_num_list:
                with self.engine.begin() as conn:
                    conn.execute(
                        f"insert into {self.site_name}_bs_category (p_id, en_name, nodes_num,path, category_id, category_parent_id) values (%s, %s,%s, %s,%s, %s)",
                        save_name_num_list)
        self.init_db(self.site_name)
        for id_en_mane in self.id_and_en_name_list:
            while True:
                try:
                    # self.id_and_en_name_list.append((bsr_id, h1_bsr_name, 2,bsr_url_category_id_dict['category_id'],bsr_url_category_id_dict['category_parent_id']))
                    # 更新 and_en_name， 根据 id 更新 分类名称
                    update_sql = f'update {self.site_name}_bs_category set and_en_name="{id_en_mane[1]}",leaf_node={id_en_mane[2]} WHERE category_id="{id_en_mane[3]}" and category_parent_id="{id_en_mane[4]}"'
                    print("更新 and_en_name：", update_sql)
                    self.db_cursor_connect_update(update_sql, self.site_name)
                    break
                except Exception as e:
                    print('更新失败：', e)
                    time.sleep(10)
                    self.init_db(self.site_name)
                    continue

    def updata_category_first_id(self):
        sql = f"""
            select id, path,nodes_num from {self.site_name}_bs_category where category_first_id is null and category_parent_id != '0'
              and delete_time is null;
            """
        df_nodes_num = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=sql)

        if not df_nodes_num.empty:
            id_path_list = df_nodes_num.values.tolist()
            for id_path in id_path_list:
                data_dict = self.parse_url(id_path[2], id_path[1])
                print(data_dict)
                updata_sql = f"""UPDATE {self.site_name}_bs_category set category_first_id="{data_dict['category_first_id']}" WHERE id={id_path[0]}"""
                print(updata_sql)
                self.db_cursor_connect_update(updata_sql, self.site_name)
        updata_sql_none = f"UPDATE {self.site_name}_bs_category set and_en_name=NULL WHERE and_en_name='None'"
        self.db_cursor_connect_update(updata_sql_none, self.site_name)

    def run(self):
        print(" run 函数 是抓取 分类节点，只新增，不删除")
        if self.site_name == 'us':
            self.dele_self_real_spider()
        num = 1
        while True:
            print(num)
            b_num = num
            name_1_list = self.db_read_data(num, state_sql=True)
            if name_1_list:
                if num < 15:
                    for name_1 in name_1_list:
                        self.name_path_queue.put(name_1)
                    if self.cookies_queue.empty():
                        cookies_dict = self.reuests_para_val.get_cookie()
                        self.cookie_dict_delete_id = cookies_dict
                        for ck in cookies_dict.values():
                            self.cookies_queue.put(ck)
                    html_thread = []
                    for i in range(50):
                        thread2 = threading.Thread(target=self.html_4, args=(b_num + 1,))
                        html_thread.append(thread2)
                    for ti in html_thread:
                        ti.start()
                        time.sleep(1)
                    for t2 in html_thread:
                        t2.join()
                    print(f"当前 {num} 级分类 全部请求完毕", "获取二级类目：", len(self.insert_list))
                else:
                    break
                self.save_date()
                self.update_delete_time()
                self.save_asin_data()
                self.init_list()
                pppoe_ip()
                time.sleep(5)
                self.init_db(self.site_name)
            elif num > 15:
                break
            num += 1
        self.select_id_1()

    def run_update_redirect_flag(self):
        print("run_update_redirect_flag 单独抓取 更新是否重定向的分类")
        num = 1
        while True:
            print(num)
            b_num = num
            name_1_list = self.db_read_data(num, state_sql=False)
            if num < 15:
                for name_1 in name_1_list:
                    self.name_path_queue.put(name_1)
                if self.cookies_queue.empty():
                    cookies_dict = self.reuests_para_val.get_cookie()
                    self.cookie_dict_delete_id = cookies_dict
                    for ck in cookies_dict.values():
                        self.cookies_queue.put(ck)
                html_thread = []
                for i in range(30):
                    thread2 = threading.Thread(target=self.html_4, args=(b_num + 1,))
                    html_thread.append(thread2)
                for ti in html_thread:
                    ti.start()
                    time.sleep(1)
                for t2 in html_thread:
                    t2.join()
                print(f"当前 {num} 级分类 全部请求完毕", "获取二级类目：", len(self.insert_list))
            elif num > 15:
                break
            num += 1
            self.update_delete_time(redirect_flag=True)
            self.init_list()

    def select_id_1(self):
        # 查询 子节点的顶级父类id
        select_sql_1 = f'select id from {self.site_name}_bs_category where nodes_num=2'
        df_id = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=select_sql_1)
        df_id_list = df_id.values.tolist()
        for id_ in df_id_list:
            en_name_id_list = []
            select_p_id = f"select t3.id,t4.en_name from (select t1.id,t1.parent_id,if(find_in_set(parent_id, @pids) > 0, @pids := concat(@pids, ',',id), 0) as ischild from (select id,p_id as parent_id from {self.site_name}_bs_category t  order by p_id,id) t1,(select @pids := {id_[0]}) t2) t3 LEFT JOIN {self.site_name}_bs_category t4 on t3.id = t4.id where ischild != 0;"
            print('select_p_id::', select_p_id)
            df_all_id = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=select_p_id)
            if not df_all_id.empty:
                all_id_list = df_all_id.values.tolist()
                for en_name_id in all_id_list:
                    en_name_id_list.append(en_name_id[0])
                id_tuple = tuple(en_name_id_list)
                print(len(id_tuple))
                try:
                    if len(id_tuple) == 1:
                        update_sql = f"""UPDATE {self.site_name}_bs_category set one_category_id={id_[0]} where id in  ('{id_tuple[0]}')"""
                    else:
                        update_sql = f'update {self.site_name}_bs_category set one_category_id={id_[0]} where id in {id_tuple}'
                    self.db_cursor_connect_update(update_sql, self.site_name)
                except Exception as e:
                    print(e)
                    time.sleep(10)
                    self.init_db(self.site_name)
                    continue

    def db_read_data(self, i, state_sql):
        while True:
            try:
                sql_read = f'select id,en_name,`path`,category_id from {self.site_name}_bs_category where nodes_num={i} and category_state=1 and delete_time is NULL'
                print(sql_read)
                df_read = self.engine.read_sql(sql_read)
                df_read.drop_duplicates(subset='category_id', inplace=True)
                if df_read.shape[0] == 0:
                    return []
                with self.engine.begin() as conn:
                    index_tuple = tuple(df_read['category_id'])
                    id_tuple = tuple(df_read['id'])
                    if state_sql == True:
                        if len(index_tuple) == 1:
                            sql_update = f"""UPDATE {self.site_name}_bs_category a set category_state=2 where a.category_id in ('{index_tuple[0]}')"""
                        else:
                            sql_update = f"""UPDATE {self.site_name}_bs_category a set category_state=2 where a.category_id in {index_tuple}"""
                    else:
                        if len(id_tuple) == 1:
                            sql_update = f"""UPDATE {self.site_name}_bs_category a set category_state=2 where a.id in ('{id_tuple[0]}')"""
                        else:
                            sql_update = f"""UPDATE {self.site_name}_bs_category a set category_state=2 where a.id in {id_tuple}"""
                    print('sql_update:', sql_update)
                    conn.execute(sql_update)

                categorys_list = list(df_read.id.astype(
                    "U") + '|-|' + df_read.en_name + '|-|' + df_read.path + '|-|' + df_read.category_id)
                category_url_list = []
                for i in categorys_list:
                    category_url_list.append(i.split('|-|'))
                return category_url_list
            except Exception as e:
                print(e)
                time.sleep(10)
                self.init_db(self.site_name)
                continue

    def update_category_state(self):
        id_sql = f"""
        select id
        from {self.site_name}_bs_category
        where category_id in (
                select category_id
                from (
                                 select category_id,
                                                count(distinct coalesce(redirect_first_id, category_first_id)) count
                                 from {self.site_name}_bs_category
                                 where delete_time is null
                                 group by category_id
                                 having count > 1) tmp
        )
          and delete_time is null
        order by category_id,category_first_id
        """
        df_id_tuple = self.db_cursor_connect_msyql_read(site=None, select_state1_sql=id_sql)
        id_tuple = df_id_tuple.values.tolist()
        id_list = []
        for id_ in id_tuple:
            id_list.append(id_[0])
        if id_list:
            try:
                self.init_db(self.site_name)
                update_category_state_sql = f"""
                    update {self.site_name}_bs_category set category_state=1 where id in {tuple(id_list)}"""
                print('update_sql::', update_category_state_sql)
                self.db_cursor_connect_update(update_category_state_sql, self.site_name)
            except:
                print('报错：update_category_state 函数')
                self.init_db(self.site_name)
                time.sleep(10)

    def send_big_data_selection(self):
        for i in range(3):
            try:
                DolphinschedulerHelper.start_process_instance_common(
                    project_name="big_data_selection",
                    process_df_name='日-bsr+nsr+asin导出',
                    startParams={
                        "site_name": self.site_name,
                        "date_type": "day",
                        "date_info": self.time_strftime_
                    }
                )
                break
            except:
                time.sleep(30)
                continue


    def updata_category_state(self):
        for i in range(3):
            try:
                _day = ((datetime.datetime.now()) + datetime.timedelta()).strftime("%Y-%m-%d")
                update_sql = f'UPDATE {self.site_name}_bs_category set category_state=1;'
                self.db_cursor_connect_update(update_sql, self.site_name)
                self.engine_pg = self.pg_connect()

                inset_sql = f"""INSERT INTO {self.site_name}_bs_category_reprot (
                            id, p_id, ch_name, en_name, nodes_num, `path`, created_at, updated_at, is_show,
                            one_category_id, and_en_name, leaf_node, full_name, category_id, category_parent_id,
                            category_first_id, category_state, redirect_flag, redirect_first_id, delete_time, date_info
                        )
                        SELECT
                            id, p_id, ch_name, en_name, nodes_num, `path`, created_at, updated_at, is_show,
                            one_category_id, and_en_name, leaf_node, full_name, category_id, category_parent_id,
                            category_first_id, category_state, redirect_flag, redirect_first_id, delete_time,
                            '{_day}' as date_info
                        FROM {self.site_name}_bs_category;"""
                print(inset_sql)
                self.db_cursor_connect_update(inset_sql, self.site_name)
                _0_days = ((datetime.datetime.now()) + datetime.timedelta(days=0)).strftime("%Y-%m-%d")
                select_sql = f"select count(id) as cnt FROM {self.site_name}_bs_top100_asin_2026 WHERE date_info = '{_0_days}';"
                print(select_sql)
                df_count_data_num = self.engine_pg.read_sql(select_sql)
                count_data_num = df_count_data_num['cnt'][0]
                print('count_data_num::', count_data_num)
                self.send_ms_count_data_num(self.site_name, count_data_num, _0_days)
                break
            except Exception as e:
                self.init_db(self.site_name)
                print(e, '222222222')
                time.sleep(20)
                continue

    def sele_msyql_category(self, site):
        engine_mysql = self.db_engine_us(site, 'mysql')
        sql = f"select path, nodes_num, id from {site}_bs_category where nodes_num>1"
        df = engine_mysql.read_sql(sql)
        values_list = df.values.tolist()
        try:
            with engine_mysql.begin() as conn_6:
                for value in values_list:
                    items = self.parse_url(value[1], value[0])
                    items['id'] = value[2]
                    ai_sql1 = f"update {site}_bs_category set category_id = '{items['category_id']}',category_parent_id='{items['category_parent_id']}',category_first_id='{items['category_first_id']}' where id={items['id']}"
                    try:
                        conn_6.execute(ai_sql1)
                    except Exception:
                        self.try_sql(site, ai_sql1)
        except Exception as e:
            print("事务退出时报错：", repr(e))

    def try_sql(self, site, ai_sql):
        for i in range(5):
            try:
                engine_mysql = self.db_engine_us(site, 'mysql')
                with engine_mysql.begin() as conn_6:
                    conn_6.execute(ai_sql)
                return
            except:
                time.sleep(10)

    def dele_self_real_spider(self):
        print('每天晚上定时删除贺哲的抓取表。用户已经取消收藏店铺')
        select_sql = 'select data_id  from user_collection_syn  where data_type =2'
        df_data_id = self.engine_pg6.read_sql(select_sql)
        print('df_data_id::', df_data_id)
        for i in range(2):
            try:
                with self.engine.begin() as conn:
                    if len(set(df_data_id.data_id)) == 1:
                        sql_delete = f"delete from us_self_real_spider where account_id not in ('{tuple(df_data_id.data_id)[0]}') and data_type=9;"
                    else:
                        sql_delete = f"delete from us_self_real_spider where account_id not in {tuple(set(df_data_id.data_id))} and data_type=9;"
                    print(sql_delete)
                    conn.execute(sql_delete)
                break
            except Exception as e:
                print('dele_self_real_spider', e)
                time.sleep(10)
                self.init_db('us')

    def send_ms_count_data_num(self, site, num, day_time):
        account = 'pengyanbing,chenjianyun,wujicang'
        title = site + ' bsr 榜单'
        content = f'{day_time} bsr 榜单 抓取完成 当日抓取总数 {num}'
        url = 'http://47.112.96.71:8082/selection/sendMessage'
        data = {
            'account': account,
            'title': title,
            'content': content
        }
        requests.post(url=url, data=data, timeout=15)


if __name__ == '__main__':
    # pppoe_ip()
    # time.sleep(5)
    print("如果运行 run 函数 有个别类目 别名没有抓取到，结束之后 运行 run_start 抓取，获取所有 and_en_name 为空的类目id，path")
    for site in ['us', 'de', 'uk']:
        spider_us = bsr_category(site_name=site)
        spider_us.run()
        spider_us.update_category_state()
        spider_us.run_update_redirect_flag()
        spider_us.updata_category_first_id()
        spider_us.send_big_data_selection()
    for site in ['us', 'de', 'uk']:
        spider_us = bsr_category(site_name=site)
        spider_us.updata_category_state()
        spider_us.sele_msyql_category(site)
