import sys
import os

sys.path.append(os.path.dirname(sys.path[0]))
import threading
import time
import random
import traceback
from datetime import datetime

import pandas as pd
from func_timeout.exceptions import FunctionTimedOut

from amazon_spider.asin_detail_pg import async_asin_pg
from amazon_save_db.save_asin_detail_pg import Save_asin_detail
from amazon_spider.VPS_IP import is_internet_available


class RealtimeSaveAsinDetail(Save_asin_detail):
    """覆盖存储目标表（写流量选品表）和状态更新表（realtime_crawl_asin）"""

    def __init__(self, site_name, task_id, month=None, spider_int=None, skip_read_init=True):
        self.task_id = task_id
        super().__init__(site_name=site_name, month=month, spider_int=spider_int,
                         skip_read_init=skip_read_init)
        self.db_syn_realtime = 'realtime_crawl_asin'
        self.realtime_task_table = 'realtime_crawl_task'

    def db_change_state_common(self, state=None, asin_list=None):
        print(f"==================== 实时任务 存储状态 {state} 数据 ========== {len(asin_list)} ========")
        df = self.df_read.loc[self.df_read.asin.isin(asin_list)]
        if state == 3:
            df = self.df_read.loc[
                (self.df_read.asin.isin(asin_list)) &
                ~(self.df_read.asin.isin(self.asin_not_foot_list)) &
                ~(self.df_read.asin.isin(self.asin_not_buyBox_list))]
        id_tuple = tuple(df.id)
        while True:
            try:
                if not is_internet_available():
                    self.engine = self.mysql_connect()
                    self.engine_pg = self.pg_connect()
                with self.engine_pg.begin() as conn:
                    if id_tuple:
                        if len(id_tuple) == 1:
                            sql = (f"update {self.db_syn_realtime} set state={state} "
                                   f"where id in ({id_tuple[0]}) and state=2;")
                        else:
                            sql = (f"update {self.db_syn_realtime} set state={state} "
                                   f"where id in {id_tuple} and state=2;")
                        conn.execute(sql)
                break
            except Exception as e:
                self.engine = self.mysql_connect()
                self.engine_pg = self.pg_connect()
                time.sleep(random.uniform(10, 20.5))
                print(f"更改 realtime_crawl_asin state={state} 出错", e,
                      f"\n{traceback.format_exc()}")
                continue

    def save_data(self, requests_error_asin_list, asin_list_update, asin_not_found_list,
                  asin_not_sure_list, asin_not_foot_list, asin_not_foot2_list,
                  asin_not_buyBox_list, asin_not_response_list, asin_not_redirect_list,
                  asin_not_div_id_dp_list, star_list, add_cart_asin_list,
                  bs_category_asin_list, week_):
        self.asin_not_buyBox_list = asin_not_buyBox_list
        self.asin_not_foot_list = asin_not_foot_list
        df_asin_detail = pd.DataFrame(data=self.asin_detail_list, columns=self.cols)
        self.asin_list_update = list(df_asin_detail.asin)

        target_table = f'{self.site_name}_asin_detail_realtime'
        while True:
            try:
                if not is_internet_available():
                    self.engine = self.mysql_connect()
                    self.engine_pg = self.pg_connect()
                print(f"===============存储实时详情数据=={target_table}=====")
                df = df_asin_detail.loc[:, self.cols]
                df = df.where(pd.notnull(df), None)
                df.ac_name = df.ac_name.apply(lambda x: str(x)[:100] if x is not None else None)
                df.brand = df.brand.apply(lambda x: str(x)[:100] if x is not None else None)
                df.title = df.title.apply(lambda x: str(x)[:400] if x is not None else None)
                df.category = df.category.apply(lambda x: str(x)[:400] if x is not None else None)
                df.img_url = df.img_url.apply(lambda x: str(x)[:400] if x is not None else None)
                df.material = df.material.apply(lambda x: str(x)[:150] if x is not None else None)
                df.volume = df.volume.apply(lambda x: str(x)[:50] if x is not None else None)
                df.package_quantity = df.package_quantity.apply(
                    lambda x: str(x)[:50] if x is not None else None)
                df.pattern_name = df.pattern_name.apply(
                    lambda x: str(x)[:50] if x is not None else None)
                df.weight_str = df.weight_str.apply(
                    lambda x: str(x)[:250] if x is not None else None)
                self.engine_pg.to_sql(df, target_table, if_exists='append')
                break
            except Exception as e:
                traceback.print_exc()
                self.engine = self.mysql_connect()
                self.engine_pg = self.pg_connect()
                time.sleep(random.uniform(10, 20.5))
                print(f"存储实时详情数据失败，重试", e, f"\n{traceback.format_exc()}")
                continue

        success_count = len(self.asin_list_update)
        fail_count = (len(requests_error_asin_list) + len(asin_not_found_list) +
                      len(asin_not_sure_list) + len(asin_not_foot_list) +
                      len(asin_not_foot2_list) + len(asin_not_buyBox_list) +
                      len(asin_not_response_list) + len(asin_not_redirect_list) +
                      len(asin_not_div_id_dp_list))

        if requests_error_asin_list:
            self.db_change_state(state=1, asin_list=requests_error_asin_list)
        if self.asin_list_update:
            self.db_change_state(state=3, asin_list=self.asin_list_update)
            self.asin_list_update = []
        if asin_not_found_list:
            self.db_change_state(state=4, asin_list=asin_not_found_list)
        if asin_not_sure_list:
            self.db_change_state(state=6, asin_list=asin_not_sure_list)
        if asin_not_foot_list:
            self.db_change_state(state=7, asin_list=asin_not_foot_list)
        if asin_not_foot2_list:
            self.db_change_state(state=8, asin_list=asin_not_foot2_list)
        if asin_not_buyBox_list:
            self.db_change_state(state=9, asin_list=asin_not_buyBox_list)
        if asin_not_response_list:
            self.db_change_state(state=10, asin_list=asin_not_response_list)
        if asin_not_redirect_list:
            self.db_change_state(state=12, asin_list=asin_not_redirect_list)
        if asin_not_div_id_dp_list:
            self.db_change_state(state=13, asin_list=asin_not_div_id_dp_list)

        self._update_task_stats(success_count, fail_count)

    def _update_task_stats(self, success_count, fail_count):
        try:
            with self.engine.begin() as conn:
                sql = (f"UPDATE {self.realtime_task_table} "
                       f"SET success_count = success_count + {success_count}, "
                       f"    fail_count = fail_count + {fail_count}, "
                       f"    updated_at = NOW() "
                       f"WHERE task_id = '{self.task_id}'")
                conn.execute(sql)
            print(f"回写 realtime_crawl_task 成功: task_id={self.task_id}, "
                  f"+success={success_count}, +fail={fail_count}")
        except Exception as e:
            print(f"回写 realtime_crawl_task 失败: {e}\n{traceback.format_exc()}")


class RealtimeReportSpider(async_asin_pg):
    """实时抓取爬虫：接收 id|asin|task_id 格式列表，抓取写入流量选品表并发 Kafka"""

    def __init__(self, site_name='us', task_id=None):
        current_month = datetime.now().strftime('%m')
        super().__init__(site_name=site_name, month=current_month, spider_int=0,
                         skip_read_init=True)
        self.task_id = task_id
        # 替换存储层为实时版本
        self.save_asin_detail = RealtimeSaveAsinDetail(
            site_name=site_name, task_id=task_id,
            month=current_month, spider_int=0, skip_read_init=True
        )
        # 同步更新引用，避免指向被丢弃的旧 Save_asin_detail 实例
        self.reuests_para_val = self.save_asin_detail.reuests_para_val
        self.redis14 = self.save_asin_detail.redis_db14
        # 覆盖 Kafka topic 为实时专用，与月表 topic 隔离
        self.topic_detail_month = f'{site_name}_asin_realtime_{current_month}'
        self.topic_asin_html = f'asin_html_realtime_{current_month}'

    def run_with_data(self, asin_list):
        """接收 id|asin|task_id 格式列表，覆盖父类的7字段格式"""
        if not asin_list:
            print("实时任务：没有分配到数据，进程退出")
            return
        current_date = datetime.now().strftime('%Y-%m')
        rows = []
        asin_data_list = []
        for item in asin_list:
            parts = item.split('|')
            id_, asin = parts[0], parts[1]
            rows.append({'id': int(id_), 'asin': asin})
            # 转换为 get_asin 需要的6字段格式：asin|date_info|is_variation|data_type|volume|weight_str
            asin_data_list.append(f'{asin}|{current_date}|0|1|null|null')
        self.df_read = pd.DataFrame(rows)
        # 注入 df_read 到存储层，db_change_state_common 依赖此
        self.save_asin_detail.df_read = self.df_read
        self.save_asin_detail.index_tuple = tuple(self.df_read['id'])

        for asin_data in asin_data_list:
            self.queries_asin_queue.put(asin_data)
        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(20):
            t = threading.Thread(target=self.get_asin)
            t.start()
            html_thread.append(t)
        for t in html_thread:
            t.join()

        self._flush_running = False
        self._flush_thread.join(timeout=20)
        print('实时任务 最后刷新 kafka flush')
        try:
            self.reuests_para_val.kafuka_producer_str.flush(timeout=35)
        except Exception as e:
            print("item producer flush 超时，跳过：", e)
        try:
            self.kafuka_producer_html.flush(timeout=35)
        except Exception as e:
            print("html producer flush 超时，跳过：", e)

        while True:
            try:
                print('存储 asin bsr 文本 存储pg')
                self.save_asin_detail.save_bs_category_asin_detail(self.bs_category_asin_list_pg)
                break
            except FunctionTimedOut:
                print('实时任务 bsr 存储断网，重试')
        while True:
            try:
                self.save_asin_detail.save_buyBoxname_url(self.buyBox_list)
                break
            except FunctionTimedOut:
                print('实时任务 save_buyBoxname_url 断网，重试')
        while True:
            try:
                self.save_asin_detail.save_buyBoxname_asin(self.buyBoxname_asin_list)
                break
            except FunctionTimedOut:
                print('实时任务 save_buyBoxname_asin 断网，重试')

        while True:
            try:
                self.save_asin_detail.process_item(
                    self.item_queue, self.requests_error_asin_list,
                    self.asin_list_update, self.asin_not_found_list,
                    self.asin_not_sure_list, self.asin_not_foot_list,
                    self.asin_not_foot2_list, self.asin_not_buyBox_list,
                    self.asin_not_response_list, self.asin_not_redirect_list,
                    self.asin_not_div_id_dp_list, self.star_list,
                    self.add_cart_asin_list, self.bs_category_asin_list, self.month_)
                break
            except FunctionTimedOut:
                print('实时任务 process_item 断网，重试')

        self.reuests_para_val.delete_china_cookie(self.delete_cookies_list)
        new_date = datetime.now().strftime("%Y-%m-%d")
        site_new_date = f'{self.site_name}_realtime_' + str(new_date)
        try:
            if self.yzm_err_total_list:
                self.redis14.rpush(site_new_date, *self.yzm_err_total_list)
            if self.asin_request_errp_total_list:
                self.redis14.rpush(site_new_date, *self.asin_request_errp_total_list)
            if self.success_asin_total_list:
                self.redis14.rpush(site_new_date, *self.success_asin_total_list)
            if self.request_total_count_list:
                self.redis14.rpush(site_new_date, *self.request_total_count_list)
        except Exception:
            pass
        print("实时任务 run_with_data 正常结束")
