import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.db_connect import BaseUtils
from utils.requests_param import Requests_param_val
from utils.parse_search_term_xpath import ParseSearchTermUs
from amazon_params.params import DB_SEARCH_TERM_PARAMS_SPIDER
from lxml import etree
from queue import Queue, Empty
import time
import random
import re
import pandas as pd
from threading import Lock
import threading
import urllib3
import traceback
import gzip
from curl_cffi import requests
from datetime import datetime

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class search_temp_pg(BaseUtils):
    def __init__(self, site_name='us', read_size=100, proxy_name=None, week=None, month=None):
        super().__init__()
        self.site_name = site_name  # 站点
        self.month = month
        print(site_name, self.month, 'proxy_name-proxy_name:', proxy_name)
        self.reuests_para_val = Requests_param_val(site_name=self.site_name, spider="search_term",
                                                   proxy_name=proxy_name)
        self.read_size = read_size
        self.cookies_queue = Queue()  # cookie队列
        self.search_term_queue = Queue()  # 需要爬取的asin队列
        self.search_term_priority_list = []  # 优先抓取列表
        self.search_term_not_found = []  # 变狗页面
        self.asin_not_sure_list = []  # 没有关键词相关结果的列表
        # 返回 对应站点的host，首页链接
        self.site_url, self.host = self.reuests_para_val.get_site_url(self.site_name)
        self.search_term_not_next_page_list = []  # 存放找不到下一页关键词的url队列
        self.stop_item_queue = True  # 用于是否退出循环存储的条件
        # 关键词相关
        # 初始化数据库相关参数
        self.db_name_change_common()  # 初始化对应站点的表名 + 初始化站点的周期
        self.df_read = pd.DataFrame()
        # 页面解析(zr和sp才有page和page_row)
        self.zr_all_list = []
        self.sp_all_list = []
        self.sb_all_list = []
        self.ac_all_list = []
        self.bs_all_list = []
        self.er_all_list = []
        self.tr_all_list = []
        self.buy_text_list = []
        self.hr_list = []
        self.sort_all_list = []
        self.columns = [
            'search_term', 'asin', 'page', 'page_row', 'data_type',
            'time_batch',
            'title', 'img_url', 'price', 'rating', 'total_comments'
        ]
        self.next_page_lock = Lock()
        # 新增：列表操作锁，保证线程安全
        self.list_lock = Lock()
        # 新增：Cookie补充锁，防止多线程踩踏
        self.cookie_refill_lock = Lock()
        self.not_sp_url_kw_list = []
        self.nums_no_sp = 0
        self.delete_cookies_list = []  # 存储出现中国邮编的cookie
        self.headers_num_int_s = 0
        # 存储html 主题
        self.search_term_html_topic = f'search_term_html_2026_{self.month}'
        # 存储数据详情
        self.search_term_month_topic = f'{self.site_name}_search_term_month_2026_{self.month}'
        print('存储html 主题:', self.search_term_html_topic)
        print('存储数据详情 主题:', self.search_term_month_topic)

        # ===== 新增：线程局部存储，每个线程独立Session =====
        self._thread_local = threading.local()

        # ===== 新增：预加载Cookie，避免爬取时阻塞 =====
        self._preload_cookies()



    def _preload_cookies(self):
        """启动时预加载Cookie，避免爬取时阻塞"""
        cookies_dict = self.reuests_para_val.get_cookie(num=500)
        self.cookie_dict_delete_id = cookies_dict
        for ck in cookies_dict.values():
            self.cookies_queue.put(ck)
        print(f'预加载Cookie完成，共 {len(cookies_dict)} 个')

    def db_name_change_common(self):
        self.engine_pg = self.pg_connect()
        self.kafuka_producer = self.kafuka_connect(acks=True)
        self.db_search_term = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term"][2:] + '_month_syn'
        self.db_search_term_zr = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_zr"][2:] + '_month'
        self.db_search_term_sp = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_sp"][2:] + '_month'
        self.db_search_term_sb = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_sb"][2:] + '_month'
        self.db_search_term_ac = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_ac"][2:] + '_month'
        self.db_search_term_bs = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_bs"][2:] + '_month'
        self.db_search_term_er = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_er"][2:] + '_month'
        self.db_search_term_tr = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_tr"][2:] + '_month'
        self.db_search_term_hr = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_search_term_hr"][2:] + '_month'
        self.db_other_search_term = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER["db_other_search_term"][2:] + '_month'
        self.db_brand_analytics = self.site_name + DB_SEARCH_TERM_PARAMS_SPIDER[f"us_brand_analytics"][2:] + '_month'

    def get_search_kw(self, t_num):
        # ===== 修复：获取线程独立Session，避免线程竞争 =====
        while True:
            try:
                keywords_scraper_url = self.search_term_queue.get(timeout=10)
            except Empty:
                print(f"当前线程-{t_num} 已完成-爬取-跳出循环")
                break
            keywords_scraper_url_list = keywords_scraper_url.split('|-|')
            keywords_id = int(keywords_scraper_url_list[0])
            keyword = keywords_scraper_url_list[1]
            scraper_url = keywords_scraper_url_list[2]
            if scraper_url not in self.search_term_not_next_page_list:
                page = int(re.findall("&page=(\d+)", scraper_url)[0])

                # ===== 修复：Cookie不足时补充，加锁防止多线程踩踏 =====
                if self.cookies_queue.qsize() < 20:
                    with self.cookie_refill_lock:
                        if self.cookies_queue.qsize() < 20:  # 二次检查，拿到锁后可能已被其他线程补充
                            cookies_dict = self.reuests_para_val.get_cookie(num=200)
                            self.cookie_dict_delete_id.update(cookies_dict)
                            for ck in cookies_dict.values():
                                self.cookies_queue.put(ck)

                # 获取组装cookie
                cookie_str = self.reuests_para_val.get_cookie_str(self.cookies_queue)
                headers = self.reuests_para_val.requests_amazon_headers(host=self.host, site_url=self.site_url,
                                                                        asin=None, scraper_url=scraper_url)
                headers["cookie"] = cookie_str
                try:
                    # ===== 修复：使用线程独立Session =====
                    resp = requests.get(scraper_url, headers=headers,
                                    timeout=10, verify=False,impersonate="chrome")
                    if self.reuests_para_val.check_amazon_yzm(resp):
                        self.search_term_priority_list.append(keywords_id)
                        continue
                    # with open(rf'D:\新建文件夹\html_selenium_files\{self.site_name}_211123333_{keyword}_{page}.html', 'w', encoding='utf-8')as f:
                    #     f.write(resp.text)
                except Exception as e:
                    print('请求报错：', e)
                    self.search_term_priority_list.append(keywords_id)
                    continue
                response = resp.text
                etree_html = etree.HTML(response)
                try:
                    ingress = etree_html.xpath("//span[@id='glow-ingress-line2']/text()")
                    print("**************  邮编  ***************", ingress)
                    ingress = ingress[0].strip()
                except:
                    ingress = None
                    print("获取邮编錯誤:")
                print('***** 邮编: ', ingress, "关键词：", keyword, '  请求url: ', scraper_url)
                if ingress:
                    if self.reuests_para_val.check_amazon_ingress(ingress):
                        self.search_term_priority_list.append(keywords_id)
                        if self.site_name != 'es' and self.site_name != 'it':
                            ubid_list = re.findall(r'ubid-main=(.*?);', cookie_str)
                            if ubid_list:
                                cookie_ubid_main_id = ubid_list[0]
                            else:
                                session_list = re.findall(r'session-id=(.*?);', cookie_str)
                                cookie_ubid_main_id = session_list[0] if session_list else None
                            for cookie_key_value in list(self.cookie_dict_delete_id.items()):
                                if cookie_ubid_main_id and cookie_ubid_main_id in cookie_key_value[1]:
                                    with self.list_lock:
                                        self.delete_cookies_list.append(cookie_key_value[0])
                        continue
                else:
                    print("没有获取到邮编，", keyword, scraper_url)
                    self.search_term_priority_list.append(keywords_id)
                    continue
                if (self.reuests_para_val.check_amazon_not_page(response)) and (
                        '404' not in keyword and 'page not found' not in keyword):
                    print("***** Page Not Found  关键词搜索出现问题：", keyword, scraper_url)
                    self.search_term_priority_list.append(keywords_id)
                    # self.search_term_not_found.extend([keywords_id])
                    continue
                # 判断是否有下一页
                no_results = 'No results for'
                if no_results in response:
                    self.asin_not_sure_list.append(keywords_id)
                    continue
                num = random.randint(1, 100)
                if num < 7:
                    new_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    response_gzip = self.compress_string(response)
                    md5_hex_digest = self.reuests_para_val.hex_md5(keyword)
                    html_data = f'{md5_hex_digest}|-||=|-|=||-|{self.site_name}|-||=|-|=||-|{keyword}|-||=|-|=||-|{response_gzip}|-||=|-|=||-|{new_date}|-||=|-|=||-|{page}'
                    self.send_kafka_html(html_data=html_data)
                self.parse_html_page(response=response, keywords=keyword, scraper_url=scraper_url, page=page,
                                     keywords_id=keywords_id, etree_html=etree_html)

    # 压缩字符串
    def compress_string(self, input_string):
        return gzip.compress(input_string.encode())

    # 解压缩字符串
    def decompress_string(self, input_bytes):
        # 在Python中，字节对象和它们的字符串表示形式是不同的。例如，字节对象 b'hello' 和字符串 'b'hello'' 是不同的，尽管它们看起来很相似。
        # 当你从数据库中读取压缩的数据时，如果数据被存储为二进制格式，那么你需要以二进制形式读取它，然后解压缩。但是，如果数据被存储为字符串（
        # 可能是因为数据库的限制或其他原因），那么你需要先将字符串转换为字节对象，然后再解压缩。
        import ast
        if isinstance(input_bytes, str):
            # html_str = json.loads(html_str)
            input_bytes = ast.literal_eval(input_bytes)
        return gzip.decompress(input_bytes).decode('utf-8')

    def init_list(self):
        print("=======清空变量==========")
        self.search_term_queue = Queue()  # 需要爬取的asin队列
        self.delete_cookies_list = []  # 存储出现中国邮编的cookie
        self.search_term_not_found = []  # 没有关键词相关结果的列表
        self.asin_not_sure_list = []  # 没有关键词相关结果的列表
        self.search_term_not_next_page_list = []  # 存放找不到下一页关键词的url队列
        self.zr_all_list = []
        self.sp_all_list = []
        self.sb_all_list = []
        self.ac_all_list = []
        self.bs_all_list = []
        self.er_all_list = []
        self.tr_all_list = []
        self.buy_text_list = []
        self.hr_list = []
        self.sort_all_list = []
        self.kafuka_producer.close()

    def run_pol(self):
        search_term_list = self.db_read_data_common()
        if search_term_list:
            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)
            for search_url in search_term_list:
                self.search_term_queue.put(search_url)
            html_thread = []
            for i in range(10):
                thread2 = threading.Thread(target=self.get_search_kw, args=(i,))
                html_thread.append(thread2)
            for ti in html_thread:
                ti.start()
            for t2 in html_thread:
                t2.join()
            print('最后刷新kafka flush')
            self.kafuka_producer.flush()
            print('当前线程抓取结束')
            # 存储数据
            print("存储数据")
            self.db_save_data()
            self.db_change_state()
            # 删除cookie
            print("删除cookie：", len(self.delete_cookies_list))
            self.reuests_para_val.delete_china_cookie(list(set(self.delete_cookies_list)))
            # 清空变量，
            self.init_list()

    def parse_html_page(self, response=None, keywords=None, scraper_url=None, page=None, keywords_id=None, etree_html=None):
        parse_search_term = ParseSearchTermUs(page_source=response, driver=None, search_term=keywords,
                                              page=page, site_name=self.site_name, etree_html=etree_html)
        st_list = parse_search_term.run()
        zr_list, sp_list, sb_list, ac_list, bs_list, er_list, tr_list, sort_list, buy_text_list, hr_list = st_list
        # 直接添加到结果列表（修复：不再使用 self.st_list 中转）
        # 修复：使用锁保证线程安全
        with self.list_lock:
            self.zr_all_list.extend(zr_list)
            self.sp_all_list.extend(sp_list)
            self.sb_all_list.extend(sb_list)
            self.ac_all_list.extend(ac_list)
            self.bs_all_list.extend(bs_list)
            self.er_all_list.extend(er_list)
            self.tr_all_list.extend(tr_list)
            self.buy_text_list.extend(buy_text_list)
            self.hr_list.extend(hr_list)
            if parse_search_term.page == 1:
                self.sort_all_list.extend(sort_list)

            # if (len(sp_list) == 0) and (not self.search_term_queue.empty()):
            #     self.nums_no_sp += 1
            #     self.not_sp_url_kw_list.append(str(keywords_id) + '|-|' + keywords + '|-|' + scraper_url)
            #     if self.nums_no_sp > 3:
            #         print("************** 没有sq：", self.not_sp_url_kw_list)
            #         for st_url_queue in self.not_sp_url_kw_list:
            #             self.search_term_queue.put(st_url_queue)
            #         ## 重试后才清空
            #         self.not_sp_url_kw_list = []
            #         self.nums_no_sp = 0

    def db_read_data_common(self):
        current_read_size = self.read_size
        while True:
            try:
                self.date_info = f'2026-{self.month}'
                self.engine_pg = self.pg_connect()
                sql_read = f"""SELECT id, search_term, url FROM {self.db_search_term} where state=1 and date_info='{self.date_info}' LIMIT {current_read_size} for update;"""
                print(sql_read)
                # self.df_read = self.engine_pg.read_sql(sql_read)
                self.df_read = self.engine_pg.read_then_update(
                    select_sql=sql_read,
                    update_table=self.db_search_term,
                    set_values={"state": 2},  # 把库存清零
                    where_keys=["id"],  # WHERE sku = :sku
                )
                if self.df_read.shape[0] > 0:
                    print('date_info::', self.date_info, ' 月：', self.month)
                    search_term_list = list(
                        self.df_read.id.astype("U") + '|-|' + self.df_read.search_term + '|-|' + self.df_read.url)
                    return search_term_list
                else:
                    self.stop_item_queue = False
                    return []
            except Exception as e:
                time.sleep(random.uniform(5, 17.5))
                current_read_size = 100
                print("读取数据出bug并等待5s继续", e)
                continue

    def db_change_state(self):
        self.db_change_state_common(3, [])
        if self.search_term_not_found:
            self.db_change_state_common(4, list(set(self.search_term_not_found)))
        if self.asin_not_sure_list:
            self.db_change_state_common(6, list(set(self.asin_not_sure_list)))
        if self.search_term_priority_list:
            self.db_change_state_common(1, list(set(self.search_term_priority_list)))

    def db_change_state_common(self, state, search_term_list):
        if state == 3:
            df = self.df_read.loc[
                ~self.df_read.id.isin(self.search_term_priority_list)]  # 找出id不再search_term_list里面的修改状态 3
            id_tuple = tuple(df.id)
        else:
            df = self.df_read.loc[self.df_read.id.isin(search_term_list)]  # 找出id再search_term_list里面的修改状态 1
            id_tuple = tuple(df.id)
        print(f"== 存储状态 {state} 数据 ========== {len(id_tuple)} ========")
        while True:
            try:
                self.engine_pg = self.pg_connect()
                with self.engine_pg.begin() as conn:
                    if id_tuple:
                        if len(id_tuple) == 1:
                            sql_update = f"update {self.db_search_term} set state={state} where id in ({id_tuple[0]});"
                        else:
                            sql_update = f"update {self.db_search_term} set state={state} where id in {id_tuple};"
                        conn.execute(sql_update)
                break
            except Exception as e:
                print(f"更改{self.db_search_term}表的state={state}出错", e, f"\n{traceback.format_exc()}")
                time.sleep(15)
                continue

    def db_save_common(self, cate_type=None, data_list=None, db_name=None):
        if data_list:
            print(db_name, "存储详细数据：", cate_type, len(data_list))
            """
            仅仅当传过来的data_list不为空的情况下，才会执行下面的数据
            """
            if cate_type in ['buy']:
                df = pd.DataFrame(data=data_list,
                                  columns=['search_term', 'asin', 'page', 'buy_data', 'label', 'asin_brand',
                                           'time_batch'])
                df = df.loc[:, ['search_term', 'asin', 'page', 'buy_data', 'label','asin_brand']]
                df.label = df.label.apply(lambda x: str(x)[:200] if x is not None else None)  # 截取字符
                df.buy_data = df.buy_data.apply(lambda x: str(x)[:200] if x is not None else None)  # 截取字符
            else:
                df = pd.DataFrame(data=data_list, columns=self.columns)
            df['date_info'] = self.date_info
            if cate_type in ['zr', 'sp']:
                df = df.loc[:, ['search_term', 'asin', 'page', 'page_row', 'date_info']]
                df.drop_duplicates(['search_term', 'asin', 'page', 'page_row'], inplace=True)
            elif cate_type in ['buy']:
                df = df.loc[:, ['search_term', 'asin', 'page', 'buy_data', 'date_info', 'label']]
                df.drop_duplicates(['search_term', 'asin', 'page', 'buy_data', 'label'], inplace=True)
            else:
                if cate_type in ['sb', 'tr']:
                    df = df.loc[:, ['search_term', 'asin', 'page', 'data_type', 'date_info']]
                    df.drop_duplicates(['search_term', 'asin', 'page', 'data_type'], inplace=True)
                else:
                    df = df.loc[:, ['search_term', 'asin', 'page', 'date_info']]
                    df.drop_duplicates(['search_term', 'asin', 'page'], inplace=True)
            while True:
                try:
                    self.engine_pg = self.pg_connect()
                    if df.shape[0] > 0:
                        print("db_name:", db_name)
                        df['asin'] = df['asin'].str.replace('/', '')
                        self.engine_pg.to_sql(df, db_name, if_exists="append")
                    break
                except Exception as e:
                    print(e, f"\n{traceback.format_exc()}")
                    time.sleep(5)
                    continue

    def db_update_brand(self):
        if self.sort_all_list:
            while True:
                try:
                    self.engine_pg = self.pg_connect()
                    df_being_sold = pd.DataFrame(data=self.sort_all_list,
                                                 columns=['search_term', 'quantity_being_sold',
                                                          'quantity_being_sold_str', 'result_count', 'departments'])
                    # 获取成功抓取的搜索词来更改状态 3
                    df_being_sold['month'] = self.month
                    df_being_sold['date_info'] = self.date_info
                    year_moth_list = self.date_info.split('-')
                    print('year_moth_list::,', year_moth_list)
                    print(f'存储表：：{self.db_brand_analytics}_{year_moth_list[0]}')
                    print(len(self.sort_all_list))
                    df_being_sold.drop_duplicates(['search_term', 'quantity_being_sold'], inplace=True)  # 去重
                    if df_being_sold.shape[0] > 0:
                        self.engine_pg.to_sql(df_being_sold, f'{self.db_brand_analytics}_{year_moth_list[0]}',
                                              if_exists='append')
                    break
                except Exception as e:
                    print('db_update_brand::', e, f"\n{traceback.format_exc()}")
                    time.sleep(5)
                    continue

    def on_send_success(self, record_metadata):
        # print(f"消息发送成功: {record_metadata.topic}-{record_metadata.partition}-{record_metadata.offset}")
        pass

    def on_send_error(self, excp):
        print("消息发送失败", excp)

    def send_kafka_html(self, html_data=None, items=None):
        if not html_data:
            return
        try:
            future = self.kafuka_producer.send(self.search_term_html_topic, html_data)
            future.add_callback(self.on_send_success).add_errback(self.on_send_error)
            # ===== 修复：移除阻塞的 future.get(30)，改为异步发送 =====
            self.headers_num_int_s += 1
            if self.headers_num_int_s % 50 == 0:  # 每50条flush一次
                self.kafuka_producer.flush()
        except Exception as e:
            print("kafka发送失败", e)

    def db_save_data(self):
        print('=========================  准备存储   ============================')
        if self.zr_all_list or self.sp_all_list:
            year_moth = self.date_info.replace('-', '_')
            self.db_save_common(cate_type='zr', data_list=self.zr_all_list,
                                db_name=f"{self.db_search_term_zr}_{year_moth}")
            self.db_save_common(cate_type='sp', data_list=self.sp_all_list,
                                db_name=f"{self.db_search_term_sp}_{year_moth}")
            self.db_save_common(cate_type='sb', data_list=self.sb_all_list,
                                db_name=f"{self.db_search_term_sb}_{year_moth}")
            self.db_save_common(cate_type='ac', data_list=self.ac_all_list,
                                db_name=f"{self.db_search_term_ac}_{year_moth}")
            self.db_save_common(cate_type='bs', data_list=self.bs_all_list,
                                db_name=f"{self.db_search_term_bs}_{year_moth}")
            self.db_save_common(cate_type='er', data_list=self.er_all_list,
                                db_name=f"{self.db_search_term_er}_{year_moth}")
            self.db_save_common(cate_type='tr', data_list=self.tr_all_list,
                                db_name=f"{self.db_search_term_tr}_{year_moth}")
            self.db_save_common(cate_type='buy', data_list=self.buy_text_list,
                                db_name=f"{self.db_other_search_term}_{year_moth}")
            self.db_save_common(cate_type='hr', data_list=self.hr_list,
                                db_name=f"{self.db_search_term_hr}_{year_moth}")
            self.db_update_brand()
if __name__ == '__main__':
    search_temp_pg(site_name='us', read_size=20, proxy_name=None, month='03').run_pol()
