import sys
import os

sys.path.append(os.path.dirname(sys.path[0]))
import re
import time
import json
import traceback
import threading
from queue import Queue, Empty
from datetime import datetime

import pandas as pd
from lxml import etree
from curl_cffi import requests

from utils.db_connect import BaseUtils
from utils.requests_param import Requests_param_val
from utils.asin_parse import ParseAsinUs


_PATTERN_1_DICT = {
    "us": "See Top 100 in ".lower(),
    "uk": "See Top 100 in ".lower(),
    "de": "Siehe Top 100 in ".lower(),
    "es": "Ver el Top 100 en ".lower(),
    "fr": "Voir les 100 premiers en ".lower(),
    "it": "Visualizza i Top 100 nella categoria ".lower(),
}
_PATTERN_CURRENT_DICT = {
    "us": r"#(\d+) ",
    "uk": r"(\d+) in ",
    "de": r"(\d+) in ",
    "es": r"(\d+) en ",
    "fr": r"(\d+) en ",
    "it": r"(\d+) in ",
}


def udf_parse_bs_category(asin_bs_sellers_rank_lower, last_herf, all_best_sellers_href,
                          cate_current_pattern, cate_1_pattern, node_id):
    """解析 BS 榜单字符串，返回 (cate_1_id, cate_current_id, cate_1_rank, cate_current_rank)"""
    if str(all_best_sellers_href).lower() not in ['', 'none', 'null']:
        bs_href = all_best_sellers_href
    elif str(last_herf).lower() not in ['', 'none', 'null']:
        bs_href = last_herf
    else:
        bs_href = ''
    href_list = bs_href.replace("?tf=1", "").split("&&&&")

    rank_flag = None
    while True:
        if '/climate-pledge' in href_list[-1] and len(href_list) >= 2:
            href_list.pop()
            rank_flag = True
        else:
            break

    cate_1_id, cate_current_id, cate_1_rank, cate_current_rank = None, None, None, None
    if node_id and len(href_list) > 1:
        node_id_str = str(node_id)
        matched_idx = None
        for i, href in enumerate(href_list):
            if node_id_str in href:
                matched_idx = i
                break
        if matched_idx is not None:
            cate_current_id = re.findall('bestsellers/(.*)/ref', href_list[matched_idx])
            cate_current_id = cate_current_id[0].split("/")[-1] if cate_current_id else None
            cate_1_id = re.findall('bestsellers/(.*)/ref', href_list[0])
            cate_1_id = cate_1_id[0].split("/")[0] if cate_1_id else None
            asin_bs_sellers_rank_lower2 = (asin_bs_sellers_rank_lower or '').replace(".", "").replace(",", "").replace(" 100 ", "")
            rank_list = [int(r) for r in re.findall(cate_current_pattern, asin_bs_sellers_rank_lower2)]
            if matched_idx < len(rank_list):
                cate_current_rank = rank_list[matched_idx]
            if rank_list and cate_1_pattern in (asin_bs_sellers_rank_lower or ''):
                cate_1_rank = rank_list[0]
            return cate_1_id, cate_current_id, cate_1_rank, cate_current_rank

    if href_list:
        if len(href_list) == 1:
            cate_list = re.findall('bestsellers/(.*)/ref', href_list[0])
            if cate_list:
                if "/" in cate_list[0]:
                    cate_1_id, cate_current_id = cate_list[0].split("/")[0], cate_list[0].split("/")[-1]
                else:
                    cate_1_id, cate_current_id = cate_list[0].split("/")[0], None
        else:
            m1 = re.findall('bestsellers/(.*)/ref', href_list[0])
            mc = re.findall('bestsellers/(.*)/ref', href_list[-1])
            cate_1_id = m1[0] if m1 else None
            cate_current_id = mc[0] if mc else None
            if cate_1_id and "/" in cate_1_id:
                cate_1_id = cate_1_id.split("/")[0]
            if cate_current_id and "/" in cate_current_id:
                cate_current_id = cate_current_id.split("/")[-1]

    asin_bs_sellers_rank_lower2 = (asin_bs_sellers_rank_lower or '').replace(".", "").replace(",", "").replace(" 100 ", "")
    rank_list = [int(r) for r in re.findall(cate_current_pattern, asin_bs_sellers_rank_lower2)]
    if rank_flag and len(rank_list) > len(href_list):
        rank_list = rank_list[:len(href_list)]
    if rank_list:
        if len(rank_list) == 1:
            if cate_1_pattern in (asin_bs_sellers_rank_lower or ''):
                cate_1_rank, cate_current_rank = rank_list[0], None
            else:
                cate_1_rank, cate_current_rank = None, rank_list[0]
        else:
            if cate_1_pattern in (asin_bs_sellers_rank_lower or ''):
                cate_1_rank, cate_current_rank = rank_list[0], rank_list[-1]
            else:
                cate_1_rank, cate_current_rank = None, rank_list[0]
    return cate_1_id, cate_current_id, cate_1_rank, cate_current_rank


def udf_parse_amazon_orders(asin_amazon_orders_str):
    """解析月销字段，兼容 '500+' / '1K+' / '1.5K+' / 'week' 单位"""
    if not asin_amazon_orders_str:
        return None
    pattern = r"([\d.]+[k]{0,})\+"
    results_list = re.findall(pattern, str(asin_amazon_orders_str).lower())
    if len(results_list) == 1:
        raw = results_list[0]
        if 'k' in raw:
            num = float(raw.replace('k', '')) * 1000
        else:
            num = float(raw)
        result = int(num)
        if 'week' in asin_amazon_orders_str.lower():
            result = result * 4
        return result
    return None


class AsinCompetitorSpider(BaseUtils):
    """竞品 ASIN 详情轻量爬虫：us_asin_competitor_syn (state=1) → 抓详情 → us_asin_competitor_detail"""

    def __init__(self, site_name='us'):
        super().__init__()
        self.site_name = site_name
        self.read_size = 200
        self.thread_num = 10
        self.asin_queue = Queue()
        self.data_queue = Queue()
        self.df_read = pd.DataFrame()
        self._init_state_lists()
        self.delete_cookies_list = []
        self.cookie_dict_delete_id = {}
        self.cookie_refill_lock = threading.Lock()
        self.init_db_names()
        self.reuests_para_val = Requests_param_val(site_name=site_name)
        self.site_url, self.host = self.reuests_para_val.get_site_url(site_name)
        self.cookies_queue = Queue()
        self._load_cookies()

    def _init_state_lists(self):
        self.requests_error_asin_list = []     # 1: 请求异常/超时/验证码（回滚）
        self.asin_list_update = []             # 3: 成功
        self.asin_not_found_list = []          # 4: Page Not Found / 下架
        self.asin_not_response_list = []       # 10: response 解析异常
        self.asin_not_redirect_list = []       # 12: 被重定向
        self.asin_not_div_id_dp_list = []      # 13: HTML 没有 div@id='dp' 容器

    def init_db_names(self):
        self.engine = self.mysql_connect()
        self.db_syn = f'{self.site_name}_asin_competitor_syn'
        self.db_target = f'{self.site_name}_asin_competitor_detail'

    def _load_cookies(self):
        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)

    def db_change_state_common(self, state, asin_competitor_list):
        """按 asin 回写 syn 表 state"""
        if not asin_competitor_list:
            return
        asin_set = set(asin_competitor_list)
        in_clause = "(" + ",".join(f"'{a}'" for a in asin_set) + ")"
        while True:
            try:
                with self.engine.begin() as conn:
                    sql = (f"UPDATE {self.db_syn} SET state={state} "
                           f"WHERE asin IN {in_clause} AND state=2")
                    conn.execute(sql)
                print(f"=== 回写竞品 state={state} 共 {len(asin_set)} 个 ASIN ===")
                break
            except Exception as e:
                print(f"更改 {self.db_syn} state={state} 报错：", e, f"\n{traceback.format_exc()}")
                time.sleep(20)
                self.engine = self.mysql_connect()
                continue

    def request_amazon(self):
        """子线程：从 asin_queue 取 asin_competitor → 抓 HTML → 分类落状态列表 → data_queue"""
        while True:
            try:
                asin_competitor = self.asin_queue.get(block=False)
            except Empty:
                break
            try:
                with self.cookie_refill_lock:
                    if self.cookies_queue.empty():
                        self._load_cookies()
                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=asin_competitor, scraper_url=None)
                # headers=headers_nlist[0]
                # n=headers_nlist[1]
                headers["cookie"] = cookie_str
                scraper_url = self.site_url + 'dp/' + asin_competitor + '?th=1&psc=1'
                print('scraper_url::', scraper_url)
                try:
                    resp = requests.get(scraper_url, headers=headers,
                                        timeout=30, verify=False, impersonate=f"chrome")
                    if self.reuests_para_val.check_amazon_yzm(resp):
                        print('出验证码 → state=1：', asin_competitor)
                        self.requests_error_asin_list.append(asin_competitor)
                        continue
                except Exception as e:
                    print('请求异常 → state=1：', e)
                    if 'Received response with content-encoding: gzip' in str(e):
                        self.asin_not_found_list.append(asin_competitor)
                    else:
                        self.requests_error_asin_list.append(asin_competitor)
                    continue
                response = resp.text
                response_url = resp.url
                if 'validateCaptcha' in response:
                    print('无图片验证码 → state=1：', asin_competitor)
                    self.requests_error_asin_list.append(asin_competitor)
                    continue
                if self.reuests_para_val.check_amazon_not_page(response):
                    print('页面不存在 → state=4：', asin_competitor)
                    self.asin_not_found_list.append(asin_competitor)
                    continue
                # 12: 重定向
                if self.reuests_para_val.check_amazon_page(response, response_url):
                    self.asin_not_redirect_list.append(asin_competitor)
                    continue
                if self.reuests_para_val.check_amazon_allow_redirects(response_url, asin_competitor):
                    self.asin_not_redirect_list.append(asin_competitor)
                    continue
                tree = etree.HTML(response)
                # 邮编检查
                try:
                    ingress = tree.xpath("//span[@id='glow-ingress-line2']/text()")
                except Exception:
                    self.asin_not_response_list.append(asin_competitor)
                    continue
                try:
                    ingress = ingress[0].strip()
                except:
                    ingress = None
                if ingress:
                    ingress = re.sub('[​-‏ - ⁠﻿]', '', ingress)
                print(ingress, ' 邮编 ', resp.url)
                if ingress:
                    if self.reuests_para_val.check_amazon_ingress(ingress):
                        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 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])
                        print(ingress, '邮编错误：：', '#' * 80)
                        self.requests_error_asin_list.append(asin_competitor)
                        continue
                else:
                    if self.site_name == 'us':
                        if 'Currently unavailable.' in response:
                            self.asin_not_found_list.append(asin_competitor)
                            continue
                    elif self.site_name == 'de':
                        if 'Die eingegebene Webadresse ist keine funktionsfähige ' in response:
                            self.asin_not_found_list.append(asin_competitor)
                            continue
                    self.requests_error_asin_list.append(asin_competitor)
                    continue
                # 13: 没有 dp 容器
                div_dp = tree.xpath('//div[@id="dp"]')
                if not div_dp:
                    print('无 div#dp → state=13：', asin_competitor)
                    self.asin_not_div_id_dp_list.append(asin_competitor)
                    continue
                # 10: 解析异常
                parsed = self.parse_html(tree, asin_competitor, response)
                if parsed is None:
                    print('解析异常 → state=10：', asin_competitor)
                    self.asin_not_response_list.append(asin_competitor)
                    continue
                self.data_queue.put(parsed)
            except Exception as e:
                print('request_amazon 异常 → state=1：', e, f"\n{traceback.format_exc()}")
                try:
                    self.requests_error_asin_list.append(asin_competitor)
                except Exception:
                    pass

    def parse_html(self, tree, asin_competitor, response_text):
        """复用 ParseAsinUs 解析 title/price/coupon，单独解析 amazon_orders_str"""
        try:
            items = ParseAsinUs(
                resp=response_text, asin=asin_competitor,
                month=datetime.now().strftime('%m'),
                date_info=datetime.now().strftime('%Y-%m'),
                site_name=self.site_name, response_s=tree
            ).xpath_html()

            title = items.get('title')
            price_raw = items.get('price')
            try:
                price = float(price_raw) if price_raw not in (None, '', '0') else None
            except Exception:
                price = None
            img_url = items.get('img_url')
            coupon_list = []
            # 限定到首个 apex_desktop（主商品价格区），兼容两种页面结构：
            # 1) 单商品页（data-csa-c-asin 为空）
            # 2) 带推荐区页面（顶层 apex_desktop 是主商品，下面嵌套的是推荐 ASIN）
            base = "(//div[@id='apex_desktop'])[1]"

            # 1) Limited time deal：主商品区首个 savingsPercentage
            deal_pct_list = tree.xpath(
                f"({base}//span[contains(@class,'savingsPercentage')])[1]/text()"
            )
            for dpct in deal_pct_list:
                dpct = dpct.strip()
                if dpct and '%' in dpct:
                    coupon_list.append({"name": "Limited time deal", "value": dpct})
                    break

            # 2) Promotion：greenBadge label 文本 + 紧邻 promoMessage 描述
            seen_promo = set()
            for label_elem in tree.xpath(f"{base}//label[contains(@id,'greenBadge')]"):
                label_text = label_elem.xpath("string(.)").strip()
                if not label_text or label_text in seen_promo:
                    continue
                seen_promo.add(label_text)
                label_id_suffix = label_elem.get('id', '').replace('greenBadge', '')
                desc = None
                desc_elems = tree.xpath(f"//span[@id='promoMessage{label_id_suffix}']")
                if desc_elems:
                    desc = desc_elems[0].xpath("string(.)").strip()
                    desc = re.sub(r'\s+Shop items.*$', '', desc)
                    desc = re.sub(r'\s+\|\s+Terms.*$', '', desc)
                    desc = re.sub(r'\s+Terms.*$', '', desc).strip()
                item = {"name": label_text}
                if desc:
                    item["desc"] = desc
                coupon_list.append(item)

            # 3) Coupon：couponText 直接用页面原文
            seen_coupon = set()
            for ct_elem in tree.xpath(f"{base}//span[contains(@id,'couponText')]"):
                ct_text = ct_elem.xpath("string(.)").strip()
                if not ct_text or ct_text in seen_coupon:
                    continue
                seen_coupon.add(ct_text)
                ct_text = re.sub(r'\s+Shop items.*$', '', ct_text)
                ct_text = re.sub(r'\s+\|\s+Terms.*$', '', ct_text)
                ct_text = re.sub(r'\s+Terms.*$', '', ct_text).strip()
                coupon_list.append({"name": ct_text})

            coupon_json = json.dumps(
                coupon_list, separators=(',', ':'), ensure_ascii=False
            ) if coupon_list else None

            orders_str = None
            orders_xpaths = [
                "//span[@id='social-proofing-faceout-title-tk_bought']",
                "//div[@id='social-proofing-faceout-title']",
            ]
            for xp in orders_xpaths:
                elements = tree.xpath(xp)
                if elements:
                    full_text = elements[0].xpath('string(.)').strip()
                    m = re.search(r'[\d.]+[Kk]?\+?\s*bought\s+in\s+past\s+(month|week)', full_text)
                    if m:
                        orders_str = m.group(0)
                        break
            amazon_orders = udf_parse_amazon_orders(orders_str) if orders_str else None

            # === BS 榜单字段（复用 ParseAsinUs + udf 解析） ===
            best_sellers_rank = items.get('best_sellers_rank') or ''
            _, _, cate_1_rank, cate_current_rank = udf_parse_bs_category(
                str(best_sellers_rank).lower(),
                items.get('best_sellers_herf') or '',
                items.get('all_best_sellers_herf') or '',
                _PATTERN_CURRENT_DICT.get(self.site_name, _PATTERN_CURRENT_DICT['us']),
                _PATTERN_1_DICT.get(self.site_name, _PATTERN_1_DICT['us']),
                items.get('node_id') or '',
            )

            result = {
                'asin_competitor': asin_competitor,
                'amazon_orders_str': orders_str[:255] if orders_str else None,
                'amazon_orders': amazon_orders,
                'price': price,
                'coupon_json': coupon_json,
                'title': title[:500] if title else None,
                'img_url': img_url[:500] if img_url else None,
                'best_sellers_rank': best_sellers_rank or None,
                'cate_1_rank': cate_1_rank,
                'cate_current_rank': str(cate_current_rank) if cate_current_rank is not None else None,
                'spider_date': datetime.now().date(),
            }
            print(f"=== 解析结果 {asin_competitor} ===")
            print(f"  title:             {result['title']}")
            print(f"  img_url:           {result['img_url']}")
            print(f"  price:             {result['price']}")
            print(f"  amazon_orders_str: {result['amazon_orders_str']}")
            print(f"  amazon_orders:     {result['amazon_orders']}")
            print(f"  coupon_json:       {result['coupon_json']}")
            print(f"  best_sellers_rank: {result['best_sellers_rank']}")
            print(f"  cate_1_rank:       {result['cate_1_rank']}")
            print(f"  cate_current_rank: {result['cate_current_rank']}")
            print(f"=========================")
            return result
        except Exception as e:
            print('parse_html 异常：', e, f"\n{traceback.format_exc()}")
            return None

    def process_item(self):
        """从 data_queue 取数据 → to_sql 入库 → 按 asin_competitor 分类回写 state"""
        rows = []
        success_asins = []
        while not self.data_queue.empty():
            try:
                d = self.data_queue.get(block=False)
                rows.append(d)
                success_asins.append(d['asin_competitor'])
            except Empty:
                break
        if rows:
            df = pd.DataFrame(rows, columns=[
                'asin_competitor', 'amazon_orders_str', 'amazon_orders',
                'price', 'coupon_json', 'title', 'img_url',
                'best_sellers_rank', 'cate_1_rank', 'cate_current_rank',
                'spider_date'
            ])
            df = df.where(pd.notnull(df), None)
            while True:
                try:
                    self.engine.to_sql(df, self.db_target, if_exists='append')
                    print(f"=== 存储竞品详情 {len(df)} 条到 {self.db_target} ===")
                    break
                except Exception as e:
                    print('to_sql 失败：', e, f"\n{traceback.format_exc()}")
                    time.sleep(20)
                    self.engine = self.mysql_connect()
                    continue
            self.asin_list_update = success_asins
        if self.requests_error_asin_list:
            self.db_change_state_common(state=1, asin_competitor_list=self.requests_error_asin_list)
        if self.asin_list_update:
            self.db_change_state_common(state=3, asin_competitor_list=self.asin_list_update)
        if self.asin_not_found_list:
            self.db_change_state_common(state=4, asin_competitor_list=self.asin_not_found_list)
        if self.asin_not_response_list:
            self.db_change_state_common(state=10, asin_competitor_list=self.asin_not_response_list)
        if self.asin_not_redirect_list:
            self.db_change_state_common(state=12, asin_competitor_list=self.asin_not_redirect_list)
        if self.asin_not_div_id_dp_list:
            self.db_change_state_common(state=1, asin_competitor_list=self.asin_not_div_id_dp_list)

    def run_with_data(self, asin_competitor_list):
        """接收主进程分配的 asin_competitor 列表（已去重），抓取并入库"""
        if not asin_competitor_list:
            print(f"{self.site_name} 子进程未分配数据")
            return
        for asin_competitor in asin_competitor_list:
            self.asin_queue.put(asin_competitor)
        threads = []
        for _ in range(self.thread_num):
            t = threading.Thread(target=self.request_amazon)
            t.start()
            threads.append(t)
        for t in threads:
            t.join()
        try:
            self.process_item()
        except Exception as e:
            print(f'process_item 异常：{e}\n{traceback.format_exc()}')
            self.db_change_state_common(state=1, asin_competitor_list=self.requests_error_asin_list)
            self.db_change_state_common(state=4, asin_competitor_list=self.asin_not_found_list)
            self.db_change_state_common(state=10, asin_competitor_list=self.asin_not_response_list)
            self.db_change_state_common(state=12, asin_competitor_list=self.asin_not_redirect_list)
            self.db_change_state_common(state=1, asin_competitor_list=self.asin_not_div_id_dp_list)
        self.reuests_para_val.delete_china_cookie(self.delete_cookies_list)


if __name__ == '__main__':
    print('请通过 threading_spider/Poll_asin_competitor.py 启动竞品爬虫')
