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

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

"""店铺 feedback 抓取"""


class async_account_feedback(BaseUtils):
    def __init__(self, site_name='us', read_size=300, proxy_name=None, date_info=None):
        super().__init__()
        self.site_name = site_name  # 站点
        print(site_name, '代理 async_account_feedback: ', proxy_name)
        self.reuests_para_val = Requests_param_val(site_name=self.site_name, spider="seller_account",
                                                   proxy_name=proxy_name)
        self.date_info = date_info  # 外部传入，如 '2026-05'
        self.read_size = read_size
        self.init_db_names()
        self.ip_num_proxy = 0
        self.ip_num = 0
        self.cookies_queue = Queue()  # cookie队列
        self.item_queue = Queue()  # 存数据队列
        self.queries_account_url_queue = Queue()  # 需要爬取的asin队列
        self.account_id_not_found_list = []  # 存储状态4的卖家id
        self.requests_error_asin_list = []  # 存储状态请求错误的卖家id 修改状态1
        self.account_list_update = []
        self.stop_item_queue = True  # 用于是否退出循环存储的条件
        self.request_param_val_dict = {}
        self.account_detail_list = []
        self.products_link_account_name_list = []
        # 需要存到数据库的字段
        self.cols_db = ['seller_id', 'count_30_day', 'count_1_year', 'count_lifetime', "site_name", 'date_info', 'seller_address', 'seller_rating', 'feedback_histogram']
        # 返回 对应站点的host，首页链接
        self.site_url, self.host = self.reuests_para_val.get_site_url(self.site_name)
        self.headers_num_int = 0
        self.cookie_dict_delete_id = {}
        self.delete_cookies_list = []

    def init_db_names(self):
        self.engine = self.mysql_connect()
        self.engine_pg = self.pg_connect()
        self.year_month = self.date_info.replace('-', '_')  # '2026_05'，用于表名拼接
        self.year_week = self.date_info                     # 传给 notify_finish
        print(self.date_info)
        self.db_seller_account_syn = self.site_name + DB_REQUESTS_ASIN_PARAMS['db_seller_account_syn'][2:] + '_distinct'
        self.db_seller_account_feedback = self.site_name + DB_REQUESTS_ASIN_PARAMS['db_seller_account_feedback'][2:]
        self.db_seller_account_product_syn = self.site_name + DB_REQUESTS_ASIN_PARAMS['db_seller_account_product_syn'][
                                                              2:]
        self.db_search_term = self.site_name + DB_REQUESTS_ASIN_PARAMS["db_search_term"][2:]

    def get_account_url(self, t_num):
        while True:
            try:
                querys = self.queries_account_url_queue.get_nowait()
            except Empty:
                print(f"当前线程-{t_num} 已完成-爬取-跳出循环")
                break
            else:
                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)
                # 获取组装cookie
                cookie_str = self.reuests_para_val.get_cookie_str(self.cookies_queue)
                query = querys.split('|-|')
                if "seller=" in query[0]:
                    seller = re.findall("seller=(.*)&", query[0])
                    if len(seller) == 0:
                        seller = re.findall("seller=(.*)", query[0])
                    scraper_url = "{}sp?ie=UTF8&seller={}".format(self.site_url, seller[0])
                else:
                    scraper_url = query[0]  # 卖家页面url
                account_name = query[1]  # 卖家名称
                seller_id = query[2]  # 卖家id
                headers = self.reuests_para_val.requests_amazon_headers(host=self.host, site_url=self.site_url,
                                                                        asin=None, scraper_url=scraper_url)
                print(self.headers_num_int, account_name, scraper_url)
                headers["cookie"] = cookie_str
                try:
                    print('请求url: ', scraper_url)
                    resp = requests.get(scraper_url, headers=headers,
                                    timeout=10, verify=False, impersonate="chrome")
                    if self.reuests_para_val.check_amazon_yzm(resp):
                        print(f"{self.site_name}  站点  +   使用代理ip出现验证码：{scraper_url}")
                        self.requests_error_asin_list.append(seller_id)
                        self.headers_num_int += 1
                        time.sleep(random.uniform(1.5, 5.5))
                        continue
                except Exception as e:
                    print("{} 请求错误错误".format(account_name), e)
                    self.requests_error_asin_list.append(seller_id)
                    time.sleep(random.uniform(2, 3.5))
                    continue
                response_url = scraper_url
                response = resp.text
                response_s = etree.HTML(response)
                try:
                    ingress = response_s.xpath("//span[@id='glow-ingress-line2']/text()")
                except Exception as e:
                    print(seller_id, "html 获取失败，返回空 html", response_url)
                    self.requests_error_asin_list.append(seller_id)
                    continue
                if self.reuests_para_val.check_amazon_not_page(response):
                    print(f" 页面变狗 {resp.url}")
                    self.headers_num_int += 1
                    self.requests_error_asin_list.append(seller_id)
                    time.sleep(random.uniform(4, 10.5))
                    continue
                # 获取邮编
                try:
                    ingress = ingress[0].strip()
                    print('ingress:',ingress)
                    if self.reuests_para_val.check_amazon_ingress(ingress):
                        try:
                            cookie_ubid_main_id = re.findall(r'ubid-main=(.*?);', cookie_str)[0]
                        except Exception:
                            cookie_ubid_main_id = re.findall(r'session-id=(.*?);', cookie_str)[0]
                        for cookie_key_value in self.cookie_dict_delete_id.items():
                            if cookie_ubid_main_id in cookie_key_value[1]:
                                self.delete_cookies_list.append(cookie_key_value[0])
                        self.requests_error_asin_list.append(seller_id)
                        time.sleep(random.uniform(4, 10.5))
                        continue
                except Exception:
                    ingress = None
                    print("获取邮编錯誤:")
                    self.headers_num_int += 1
                all_account_list = response_s.xpath(
                    '//table[@id="feedback-summary-table"]//tr[last()]//td[@class="a-text-right"]/span//text()')
                all_time_list = response_s.xpath('//table[@id="feedback-summary-table"]//th/text()')
                site_name_list = response_s.xpath(
                    '//ul[@class="a-unordered-list a-nostyle a-vertical"]//li/span/text()|//div[contains (@id,"detail-seller")]//span/text()')
                # 30天有效评级说明
                rating_thirty_list = response_s.xpath('//span[contains(@id,"-timeperiod-rating-thirty")]/text()')
                rating_thirty =rating_thirty_list[0] if rating_thirty_list else -1
                # 90天有效评级说明
                rating_ninety_list =  response_s.xpath('//span[contains(@id,"-timeperiod-rating-ninety")]/text()')
                rating_ninety =rating_ninety_list[0] if rating_ninety_list else -1
                # 年度有效评级说明 12个月
                rating_year_list =  response_s.xpath('//span[contains(@id,"-timeperiod-rating-year")]/text()')
                rating_year =rating_year_list[0] if rating_year_list else -1
                # 终身有效评级说明 所有时间
                rating_lifetime_list =  response_s.xpath('//span[contains(@id,"-timeperiod-rating-lifetime")]/text()')
                rating_lifetime =rating_lifetime_list[0] if rating_lifetime_list else -1
                seller_rating = f"{rating_thirty}|-|{rating_ninety}|-|{rating_year}|-|{rating_lifetime}".replace(',','.')
                feedback_histogram = self.parse_feedback_histogram(response_s)

                if site_name_list:
                    print('完整地址拼接：：','|-|'.join(site_name_list))
                    seller_address = '|-|'.join(site_name_list)
                    if len(site_name_list[-1]) == 2:
                        site_name_ = site_name_list[-1]
                    else:
                        site_name_ = None

                else:
                    seller_address = None
                    site_name_ = None
                if len(all_account_list) == len(all_time_list) and len(all_time_list) == 4:
                    count_30_day = all_account_list[0].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                        ' ', '').strip()
                    count_90_day = all_account_list[1].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                        ' ', '').strip()
                    count_1_year = all_account_list[2].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                        ' ', '').strip()
                    count_lifetime = all_account_list[3].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                        ' ', '').strip()
                    site_name = site_name_
                else:
                    if site_name_ is None and ingress is None:
                        print("获取数据失败")
                        self.requests_error_asin_list.append(seller_id)
                        continue
                    else:
                        day_30 = response_s.xpath(
                            "//div[@id='rating-thirty-num']/span[@class='ratings-reviews-count']/text()")
                        if day_30:
                            count_30_day = day_30[0].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                                ' ', '').strip()
                        else:
                            count_30_day = 0
                        day_90 = response_s.xpath(
                            "//div[@id='rating-90-num']/span[@class='ratings-reviews-count']/text()")
                        if day_90:
                            count_90_day = day_90[0].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                                ' ', '').strip()
                        else:
                            count_90_day = 0
                        day_365 = response_s.xpath(
                            "//div[@id='rating-365d-num']/span[@class='ratings-reviews-count']/text()")
                        if day_365:
                            count_1_year = day_365[0].replace(',', '').replace('\xa0', '').replace('.', '').replace(
                                ' ', '').strip()
                        else:
                            count_1_year = 0
                        all_lifetime = response_s.xpath(
                            "//div[@id='rating-lifetime-num']/span[@class='ratings-reviews-count']/text()")
                        if all_lifetime:
                            count_lifetime = all_lifetime[0].replace(',', '').replace('\xa0', '').replace('.',
                                                                                                          '').replace(
                                ' ', '').strip()
                        else:
                            count_lifetime = 0
                        print(day_30, day_90, day_365, all_lifetime)
                        site_name = site_name_
                item = {'seller_id': seller_id, 'site_name': site_name, 'account_name': account_name,
                        'count_30_day': int(count_30_day), 'count_90_day': int(count_90_day),
                        'count_1_year': int(count_1_year), 'count_lifetime': int(count_lifetime),
                        'seller_address': seller_address, 'seller_rating': seller_rating,
                        'feedback_histogram': feedback_histogram,
                        }
                print(item)
                self.item_queue.put(item)

    def parse_feedback_histogram(self, response_s):
        # 解析卖家评价星级分布，来源：页面 <script type="a-state"> 标签内 JSON
        # 输出示例：
        # {
        #   "30d": {
        #     "star1": 7,  "star2": 0,  "star3": 0,  "star4": 7,  "star5": 86,       # 各星占比 (0-100)
        #     "star1Count": 1, "star2Count": 0, "star3Count": 0, "star4Count": 1, "star5Count": 12,  # 各星绝对数量
        #     "star1CountWithText": 1, "star2CountWithText": 0, "star3CountWithText": 0,              # 各星有文字评价数
        #     "star4CountWithText": 1, "star5CountWithText": 12,
        #     "ratingCount": 14, "ratingCountWithText": 14                             # 总评价数 / 有文字总数
        #   },
        #   "90d":      { ... },   # 近 90 天
        #   "365d":     { ... },   # 近 12 个月
        #   "lifetime": { ... }    # 全部时间
        # }
        # Amazon script a-state key → 存储时段名映射
        period_key_map = {
            'oneMonthRatingsData':    '30d',
            'threeMonthRatingsData':  '90d',
            'twelveMonthRatingsData': '365d',
            'lifetimeRatingsData':    'lifetime',
        }
        # 固定 17 个 key，页面新增字段不影响后端，取不到返回 None
        fixed_keys = [
            'star1', 'star2', 'star3', 'star4', 'star5',                          # 各星占比 (0-100)
            'star1Count', 'star2Count', 'star3Count', 'star4Count', 'star5Count',  # 各星绝对数量
            'star1CountWithText', 'star2CountWithText', 'star3CountWithText',      # 各星有文字评价数
            'star4CountWithText', 'star5CountWithText',
            'ratingCount', 'ratingCountWithText',                                  # 总评价数 / 有文字总数
        ]
        result = {}
        scripts = response_s.xpath('//script[@type="a-state"]')
        for s in scripts:
            try:
                key = json.loads(s.get('data-a-state', '{}')).get('key', '')
                if key in period_key_map and s.text:
                    raw = json.loads(s.text)
                    result[period_key_map[key]] = {k: raw.get(k) for k in fixed_keys}
            except Exception:
                continue
        # 页面结构异常时返回 None，存库为 NULL
        return json.dumps(result, ensure_ascii=False) if result else None

    def init_list(self):
        print("清空变量")
        self.item_queue = Queue()  # 存数据队列
        self.queries_account_url_queue = Queue()  # 需要爬取的asin队列
        self.account_id_not_found_list = []  # 存储状态4的卖家id
        self.requests_error_asin_list = []  # 存储状态请求错误的卖家id 修改状态1
        self.account_detail_list = []  # 存储详情数据列表
        self.account_list_update = []
        self.products_link_account_name_list = []
        self.headers_num_int = 0
        self.delete_cookies_list = []


    def run(self):
        account_querys_url_list = self.read_db_data()
        if account_querys_url_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 account_name_url in account_querys_url_list:
                self.queries_account_url_queue.put(account_name_url)
            html_thread = []
            for i in range(10):
                thread2 = threading.Thread(target=self.get_account_url, args=(i,))
                html_thread.append(thread2)
            for ti in html_thread:
                ti.start()
            for t2 in html_thread:
                t2.join()
            # 存储数据
            print("存储数据")
            self.process_item()
            self.reuests_para_val.delete_china_cookie(list(set(self.delete_cookies_list)))
            # 清空变量，
            self.init_list()
        else:
            if not self.stop_item_queue:
                self.notify_finish(
                    page='店铺Feedback',
                    year_week=self.year_week,
                    year_month=self.year_month,
                    site_name=self.site_name,
                    state_field='state',
                    syn_table=self.db_seller_account_syn,
                    target_table=f'{self.site_name}_seller_account_feedback',
                    title_suffix='站点 feedback',
                    body_text=' 月  feedback 已结束,请确认下一步流程!时间:',
                )

    def process_item(self):
        print("=================开始存储数据======================")
        while True:
            if not self.item_queue.empty():
                item = self.item_queue.get()
                item_list = []
                # 需要存到数据库的字段
                item_list.append(item['seller_id'])
                item_list.append(item['count_30_day'])
                item_list.append(item['count_1_year'])
                item_list.append(item['count_lifetime'])
                # item_list.append(log_time)
                item_list.append(item["site_name"])
                item_list.append(self.year_month.replace('_', '-'))
                item_list.append(item['seller_address'])
                item_list.append(item['seller_rating'])
                item_list.append(item.get('feedback_histogram'))
                self.account_detail_list.append(item_list)
            else:
                self.save_data()
                self.account_detail_list = []
                # self.save_products_link_account_name()
                self.products_link_account_name_list = []
                print("----队列空-----跳出--存储---")
                break

    def read_db_data(self):
        while True:
            try:
                self.engine = self.mysql_connect()
                sql_read = f'SELECT url, account_name, id,seller_id FROM {self.db_seller_account_syn} WHERE STATE=1 LIMIT {self.read_size} for update;'
                self.df_read = self.engine.read_then_update(
                    select_sql=sql_read,
                    update_table=f"{self.db_seller_account_syn}",
                    set_values={"state": 2},
                    where_keys=["id"],
                )
                if self.df_read.shape[0] > 0:
                    account_name_url_list = list(
                        self.df_read.url + '|-|' + self.df_read.account_name + '|-|' + self.df_read.seller_id)
                    return account_name_url_list
                elif self.df_read.shape[0] == 0:
                    self.stop_item_queue = False
                    print(f"**************** {self.site_name} feedback 抓取完毕 **********************")
                    return []
            except Exception as e:
                print("读取数据出bug并等待5s继续", e, f"\n{traceback.format_exc()}")
                time.sleep(5)
                continue

    def save_data(self):
        while True:
            try:
                self.engine_pg = self.pg_connect()
                df_account_detail = pd.DataFrame(data=self.account_detail_list, columns=self.cols_db)
                df_account_detail.drop_duplicates(['seller_id'], inplace=True)  # 去重
                self.account_list_update = list(df_account_detail.seller_id)
                if df_account_detail.shape[0] > 0:
                    with self.engine_pg.begin() as conn:
                        db_seller_account_feedback = f"{self.db_seller_account_feedback + '_' + self.year_month}"
                        if len(set(df_account_detail.seller_id)) == 1:
                            sql_delete = f"delete from {db_seller_account_feedback} where seller_id in ('{tuple(df_account_detail.seller_id)[0]}');"
                        else:
                            sql_delete = f"delete from {db_seller_account_feedback} where seller_id in {tuple(set(df_account_detail.seller_id))};"
                        conn.execute(sql_delete)
                    print(f"feedback 信息 {db_seller_account_feedback}")
                    self.engine_pg.to_sql(df_account_detail, db_seller_account_feedback,
                                             if_exists='append')
                self.account_detail_list = []
                break
            except Exception as e:
                time.sleep(10)
                print(f"存储'{self.db_seller_account_feedback}'失败，等待5s继续", e, f"\n{traceback.format_exc()}")
                continue
        if self.requests_error_asin_list:
            self.db_change_state(state=1)
            self.requests_error_asin_list = []
        if self.account_list_update:
            self.db_change_state(state=3)
            self.account_list_update = []
        if self.account_id_not_found_list:
            self.db_change_state(state=4)
            self.account_id_not_found_list = []

    def db_change_state(self, state=2):
        if state == 1:
            self.db_change_state_common(state=state, account_id_list=self.requests_error_asin_list)
        elif state == 3:
            self.db_change_state_common(state=state, account_id_list=self.account_list_update)
        elif state == 4:
            self.db_change_state_common(state=state, account_id_list=self.account_id_not_found_list)

    def db_change_state_common(self, state, account_id_list):
        account_id_list = list(set(account_id_list))
        print(f"==================== 存储状态 {state} 数据 ========== {len(account_id_list)} ========")
        df = self.df_read.loc[self.df_read.seller_id.isin(account_id_list)]
        id_tuple = tuple(df.id)
        while True:
            try:
                self.engine = self.mysql_connect()
                with self.engine.begin() as conn:
                    # 1,3：1--回滚；3--成功
                    if id_tuple:
                        if len(id_tuple) == 1:
                            sql_update = f"update {self.db_seller_account_syn} set state={state} where id in ({id_tuple[0]}) and state=2;"
                        else:
                            sql_update = f"update {self.db_seller_account_syn} set state={state} where id in {id_tuple} and state=2;"
                        conn.execute(sql_update)
                break
            except Exception as e:
                print(f"更改{self.db_seller_account_syn}表的state={state}出错", e, f"\n{traceback.format_exc()}")
                time.sleep(5)
                continue

if __name__ == '__main__':
    async_account_feedback('us', read_size=15, proxy_name=None, date_info='2026-04').run()