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 utils.secure_db_client import stream_load_df
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

from curl_cffi import requests

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 = time.strftime("%m")
        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.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.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.not_sp_url_kw_list = []
        self.nums_no_sp = 0
        self.delete_cookies_list = []  # 存储出现中国邮编的cookie

        self.next_page_lock = Lock()
        # 列表操作锁，保证线程安全
        self.list_lock = Lock()
        # Cookie补充锁，防止多线程踩踏
        self.cookie_refill_lock = Lock()
        # ===== 新增：预加载Cookie，避免爬取时阻塞 =====
        self._preload_cookies()

    def _preload_cookies(self):
        """启动时预加载Cookie，避免爬取时阻塞"""
        cookies_dict = self.reuests_para_val.get_cookie(num=200)
        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.doris_connect_adv()
        self.db_search_term = f'{self.site_name}_sp_search_term_syn'
        self.db_search_term_save_sp = f'{self.site_name}_sp_search_term_rank'

    def get_search_kw(self, t_num):
        while True:
            try:
                keywords_scraper_url = self.search_term_queue.get(timeout=5)
            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]
            time_batch = keywords_scraper_url_list[3]
            print('当前请求：', scraper_url)
            page = int(re.findall("&page=(\d+)", scraper_url)[0]) # 手动构造的url。有page参数

            # ===== 修复：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
            resp = None
            try:
                resp = requests.get(scraper_url, headers=headers,
                                timeout=30, impersonate="chrome")
                # with open(rf'D:\新建文件夹\html_selenium_files\{self.site_name}_211123333_{keyword}_{page}.html', 'w', encoding='utf-8')as f:
                #     f.write(resp.text)
                if self.reuests_para_val.check_amazon_yzm(resp):
                    self.search_term_priority_list.append(keywords_id)
                    continue
            except Exception as e:
                print(f'请求报错：', e)
                self.search_term_priority_list.append(keywords_id)
                continue

            if resp is None:
                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 not in ('es', '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)
                continue
            with self.next_page_lock:
                no_results = 'No results for'
                if no_results in response:
                    self.asin_not_sure_list.append(keywords_id)
                    continue
            # 解析
            try:
                self.parse_html_page(response=response, keywords=keyword, scraper_url=scraper_url, page=page,
                                     keywords_id=keywords_id, time_batch=time_batch, etree_html=etree_html)
            except Exception as e:
                print(f"解析异常，回滚ID {keywords_id}：", e)
                self.search_term_priority_list.append(keywords_id)

    def init_list(self):
        print("=======清空变量==========")
        self.search_term_queue = Queue()  # 需要爬取的asin队列
        self.delete_cookies_list = []  # 存储出现中国邮编的cookie
        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.not_sp_url_kw_list = []
        self.nums_no_sp = 0
        self.search_term_priority_list = []

    def run_pol(self):
        search_term_list = self.db_read_data_common()
        if not search_term_list:
            print("没有更多数据，进程退出")
            return
        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('当前批次抓取结束')
        # 存储数据
        print("存储数据")
        self.db_save_data()
        # 成功的改 state=3
        self.db_change_state_common(3, [])
        # 失败的改回 state=1
        if self.search_term_priority_list:
            self.db_change_state_common(1, list(set(self.search_term_priority_list)))
        # 无结果的改 state=6
        if self.asin_not_sure_list:
            self.db_change_state_common(6, list(set(self.asin_not_sure_list)))
        print("删除cookie：", len(self.delete_cookies_list))
        self.reuests_para_val.delete_china_cookie(list(set(self.delete_cookies_list)))
        self.init_list()
        print("run_pol 正常结束")

    def run_pol_with_data(self, search_term_list):
        """接收主进程分发的数据列表，不自己读数据库"""
        if not search_term_list:
            print("没有分配到数据，进程退出")
            return
        # 构造 df_read 用于后续 db_change_state
        rows = []
        for item in search_term_list:
            parts = item.split('|-|')
            rows.append({'id': int(parts[0]), 'search_term': parts[1], 'url': parts[2], 'time_batch': parts[3]})
        self.df_read = pd.DataFrame(rows)

        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('当前批次抓取结束')
        print("存储数据")
        self.db_save_data()
        self.db_change_state_common(3, [])
        if self.search_term_priority_list:
            self.db_change_state_common(1, list(set(self.search_term_priority_list)))
        if self.asin_not_sure_list:
            self.db_change_state_common(6, list(set(self.asin_not_sure_list)))
        print("删除cookie：", len(self.delete_cookies_list))
        self.reuests_para_val.delete_china_cookie(list(set(self.delete_cookies_list)))
        self.init_list()
        print("run_pol_with_data 正常结束")

    def parse_html_page(self, response=None, keywords=None, scraper_url=None, page=None,
                        time_batch=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,
                                              time_batch=time_batch, sp_all=False, 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)

    def db_read_data_common(self):
        while True:
            try:
                self.engine_pg = self.doris_connect_adv()
                sql_read = f"""SELECT id, search_term, url, time_batch FROM {self.db_search_term} where state=1 LIMIT {self.read_size};"""
                print(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"],
                )
                if self.df_read.shape[0] > 0:
                    search_term_list = list(
                        self.df_read.id.astype(
                            "U") + '|-|' + self.df_read.search_term + '|-|' + self.df_read.url + '|-|' + self.df_read.time_batch)
                    return search_term_list
                else:
                    return []
            except Exception as e:
                time.sleep(random.uniform(5, 17.5))
                print("读取数据出bug并等待5s继续", e)
                continue

    def db_change_state(self):
        self.db_change_state_common(3, [])
        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)} ========")
        print(f"  修改syn表 id数量: {len(set(id_tuple))}, 总行数: {len(id_tuple)}")
        while True:
            try:
                if id_tuple:
                    conn = self.doris_adv_direct_connect()
                    cursor = conn.cursor()
                    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}"
                    cursor.execute(sql_update)
                    conn.commit()
                    cursor.close()
                    conn.close()
                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))
            if cate_type not in ['zr', 'sp']:
                print(f"未知 cate_type: {cate_type}，跳过存储")
                return
            df = pd.DataFrame(data=data_list, columns=self.columns)
            df['cate_type'] = cate_type
            df = df.loc[:, ['search_term', 'asin', 'page', 'page_row', 'time_batch', 'cate_type']]
            df.drop_duplicates(['search_term', 'asin', 'page', 'cate_type'], inplace=True)  # 去重
            while True:
                try:
                    if df.shape[0] > 0:
                        print("df 前10行数据：")
                        pd.set_option('display.max_columns', None)
                        pd.set_option('display.max_colwidth', None)
                        pd.set_option('display.width', None)
                        print(df.head(10).to_string(index=False))
                        print("db_name:", db_name)
                        df['asin'] = df['asin'].str.replace('/', '')
                        # self.engine_pg = self.doris_connect_adv()
                        # if df.shape[0] > 0:
                        #     print("db_name 表名称 :", db_name)
                        #     self.engine_pg.to_sql(df, db_name, if_exists="append")
                        resp = stream_load_df(
                            df=df,
                            database="advertising_manager",
                            table="us_sp_search_term_rank"
                        )
                        print(resp)
                    break
                except Exception as e:
                    print(e, f"\n{traceback.format_exc()}")
                    time.sleep(5)
                    continue

    def db_save_data(self):
        print('=========================  准备存储   ============================')
        if self.zr_all_list or self.sp_all_list:
            self.db_save_common(cate_type='zr', data_list=self.zr_all_list,
                                db_name=f"{self.db_search_term_save_sp}")
            self.db_save_common(cate_type='sp', data_list=self.sp_all_list,
                                db_name=f"{self.db_search_term_save_sp}")
if __name__ == '__main__':
    search_temp_pg(site_name='us', read_size=50).run_pol()
