spider_self_asin_detail.py

import copy
import json
import os
import re
import ast
import sys
import threading
import time
import logging
import traceback
import zlib
import pandas as pd
import numpy as np
import redis
from datetime import datetime
sys.path.append("/opt/module/spark-3.2.0-bin-hadoop3.2/demo/py_demo/")
sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from sqlalchemy import create_engine
from utils.templates import Templates
# from ..utils.templates import Templates
from utils.templates_mysql import TemplatesMysql
# from ..utils.templates_mysql import TemplatesMysql
from pyspark.sql.types import IntegerType
from pyspark.sql import functions as F
from pyspark.sql.types import *
from psycopg2.errors import NumericValueOutOfRange
from sqlalchemy.exc import OperationalError, DataError, PendingRollbackError
from utils.mysql_db import sql_connect, sql_update_many, sql_delete, get_country_engine
from pyspark.sql import SparkSession


class SpiderAsinDetail(Templates):

    def __init__(self, site_name='us', date_type="day", date_info='2022-10-01', consumer_type='lastest', topic_name="us_asin_detail", batch_size_history=100000):
        super(SpiderAsinDetail, self).__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.consumer_type = consumer_type  # 消费实时还是消费历史
        # 通过date_type 获取 topic
        self.get_topic_name()
        # 通过date_type 获取 schema
        self.init_schema()
        # self.topic_name = topic_name  # 主题名字
        self.batch_size_history = batch_size_history
        self.db_save = f'spider_asin_detail'
        self.spark = self.create_spark_object(app_name=f"{self.db_save}: {self.site_name},{self.date_type}, {self.date_info}")
        # self.schema = self.init_schema()

        # 连接mysql
        self.engine = get_country_engine(self.site_name)
        self.pg14_engine = self.get_14pg_country_engine(self.site_name)
        sql_connect(self.site_name)
        logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s %(message)s',
                            level=logging.INFO)
        self.df_type_dict = {
            "asin_vartion_list": '',
            "img_list": '',
            "asin_detail": '',
        }

    def judge_spider_asin_detail_is_finished(self):
        while True:
            try:
                sql = f'SELECT * from workflow_progress WHERE page="ASIN详情" and site_name="{self.site_name}" and date_type="{self.date_type}" and date_info="{self.date_info}" and status_val=3'
                df = pd.read_sql(sql, con=self.engine)
                if df.shape[0] == 1:
                    print(f"ASIN详情状态为3, 抓取完成并终止程序, site_name:{self.site_name}, date_type:{self.date_type}, date_info:{self.date_info}")
                    self.spark.stop()
                    quit()  # 退出程序
                break
            except Exception as e:
                print(e, traceback.format_exc())
                time.sleep(10)
                self.engine = self.get_connection()

    def init_schema(self):
        if self.date_type == "month":
            self.schema = StructType([
                StructField("asin", StringType(), True),
                StructField("week", StringType(), True),
                StructField("month", StringType(), True),
                StructField("asin_vartion_list", StringType(), True),
                StructField("img_list", StringType(), True),
                StructField("title", StringType(), True),
                StructField("img_url", StringType(), True),
                StructField("rating", StringType(), True),
                StructField("total_comments", StringType(), True),
                StructField("price", FloatType(), True),
                StructField("rank", StringType(), True),
                StructField("category", StringType(), True),
                StructField("launch_time", StringType(), True),
                StructField("volume", StringType(), True),
                StructField("weight", StringType(), True),
                StructField("page_inventory", IntegerType(), True),
                StructField("buy_box_seller_type", IntegerType(), True),
                StructField("title_len", IntegerType(), True),
                StructField("img_num", IntegerType(), True),
                StructField("img_type", StringType(), True),
                StructField("activity_type", StringType(), True),
                StructField("one_two_val", StringType(), True),
                StructField("three_four_val", StringType(), True),
                StructField("eight_val", StringType(), True),
                StructField("qa_num", IntegerType(), True),
                StructField("five_star", IntegerType(), True),
                StructField("four_star", IntegerType(), True),
                StructField("three_star", IntegerType(), True),
                StructField("two_star", IntegerType(), True),
                StructField("one_star", IntegerType(), True),
                StructField("low_star", IntegerType(), True),
                StructField("together_asin", StringType(), True),
                StructField("brand", StringType(), True),
                StructField("ac_name", StringType(), True),
                StructField("material", StringType(), True),
                StructField("node_id", StringType(), True),
                StructField("data_type", IntegerType(), True),
                StructField("sp_num", StringType(), True),
                StructField("describe", StringType(), True),
                StructField("date_info", StringType(), True),
                StructField("weight_str", StringType(), True),
                StructField("package_quantity", StringType(), True),
                StructField("pattern_name", StringType(), True),
                StructField("seller_id", StringType(), True),
                StructField("variat_num", IntegerType(), True),
                StructField("site_name", StringType(), True),
                StructField("best_sellers_rank", StringType(), True),
                StructField("best_sellers_herf", StringType(), True),
                StructField("account_url", StringType(), True),
                StructField("account_name", StringType(), True),
                StructField("parentAsin", StringType(), True),
                StructField("asinUpdateTime", StringType(), True),
                StructField("spider_int", StringType(), True),
                StructField("follow_sellers", StringType(), True),
            ])
            if self.site_name == "us":
                self.detail_col = [
                    'asin', 'img_url', 'title', 'title_len', 'price', 'rating', 'total_comments', 'buy_box_seller_type',
                    'page_inventory', 'category', 'volume', 'weight', 'rank', 'launch_time', 'img_num',
                    'img_type', 'activity_type', 'one_two_val', 'three_four_val', 'eight_val', 'qa_num',
                    'one_star', 'two_star', 'three_star', 'four_star', 'low_star', 'together_asin', 'brand', 'ac_name',
                    'material', 'node_id', 'data_type', 'sp_num', 'asinUpdateTime',
                    'describe', 'date_info', 'five_star', 'weight_str', 'package_quantity', 'pattern_name', 'spider_int',
                    'follow_sellers'
                ]
            else:
                self.detail_col = [
                    'asin', 'img_url', 'title', 'title_len', 'price', 'rating', 'total_comments', 'buy_box_seller_type',
                    'page_inventory', 'category', 'volume', 'weight', 'rank', 'launch_time', 'img_num',
                    'img_type', 'activity_type', 'one_two_val', 'three_four_val', 'five_six_val', 'eight_val', 'qa_num',
                    'one_star', 'two_star', 'three_star', 'four_star', 'low_star', 'together_asin', 'brand', 'ac_name',
                    'material', 'node_id', 'data_type', 'sp_num', 'asinUpdateTime',
                    'describe', 'date_info', 'five_star', 'weight_str', 'package_quantity', 'pattern_name', 'spider_int',
                    'follow_sellers'
                ]
        elif self.date_type == 'week':
            self.schema = StructType([
                StructField("asin", StringType(), True),
                StructField("img_url", StringType(), True),
                StructField("week", StringType(), True),
                StructField("month", StringType(), True),
                StructField("asin_vartion_list", StringType(), True),
                StructField("img_list", StringType(), True),
                StructField("title", StringType(), True),
                StructField("title_len", IntegerType(), True),
                StructField("price", FloatType(), True),
                StructField("rating", StringType(), True),

                StructField("total_comments", StringType(), True),
                StructField("buy_box_seller_type", IntegerType(), True),
                StructField("page_inventory", IntegerType(), True),
                StructField("category", StringType(), True),
                StructField("volume", StringType(), True),
                StructField("weight", StringType(), True),
                StructField("rank", StringType(), True),
                StructField("launch_time", StringType(), True),

                StructField("category_state", IntegerType(), True),
                StructField("img_num", IntegerType(), True),
                StructField("img_type", StringType(), True),
                StructField("activity_type", StringType(), True),
                StructField("one_two_val", StringType(), True),
                StructField("three_four_val", StringType(), True),
                StructField("five_six_val", StringType(), True),
                StructField("eight_val", StringType(), True),

                StructField("qa_num", IntegerType(), True),
                StructField("one_star", IntegerType(), True),
                StructField("two_star", IntegerType(), True),
                StructField("three_star", IntegerType(), True),
                StructField("four_star", IntegerType(), True),
                StructField("low_star", IntegerType(), True),
                StructField("together_asin", StringType(), True),
                StructField("brand", StringType(), True),

                StructField("ac_name", StringType(), True),
                StructField("material", StringType(), True),
                StructField("node_id", StringType(), True),
                StructField("data_type", IntegerType(), True),
                StructField("sp_num", StringType(), True),
                StructField("describe", StringType(), True),
                StructField("date_info", StringType(), True),
                StructField("five_star", IntegerType(), True),

                StructField("weight_str", StringType(), True),
                StructField("package_quantity", StringType(), True),
                StructField("pattern_name", StringType(), True),
                StructField("asinUpdateTime", StringType(), True),
                StructField("follow_sellers", StringType(), True),
            ])
            self.detail_col = [
                'asin', 'img_url', 'title', 'title_len', 'price', 'rating', 'total_comments', 'buy_box_seller_type',
                'page_inventory', 'category', 'volume', 'weight', 'rank', 'launch_time', 'category_state', 'img_num',
                'img_type', 'activity_type', 'one_two_val', 'three_four_val', 'five_six_val', 'eight_val', 'qa_num',
                'one_star', 'two_star', 'three_star', 'four_star', 'low_star', 'together_asin', 'brand', 'ac_name',
                'material', 'node_id', 'data_type', 'sp_num','describe', 'date_info', 'five_star', 'weight_str',
                'package_quantity', 'pattern_name', 'asinUpdateTime', 'follow_sellers'
            ]
        elif self.date_type == "day":
            self.schema = StructType([
                StructField("asin_vartion_list", StringType(), True),
                StructField("img_list", StringType(), True),
                StructField("asin", StringType(), True),
                StructField("img_url", StringType(), True),
                StructField("title", StringType(), True),
                StructField("title_len", IntegerType(), True),
                StructField("price", StringType(), True),
                StructField("rating", StringType(), True),
                StructField("total_comments", StringType(), True),
                StructField("buy_box_seller_type", IntegerType(), True),
                StructField("page_inventory", IntegerType(), True),
                StructField("category", StringType(), True),
                StructField("volume", StringType(), True),
                StructField("weight", StringType(), True),
                StructField("rank", StringType(), True),
                StructField("launch_time", StringType(), True),

                StructField("video_url", StringType(), True),
                StructField("add_url", StringType(), True),
                StructField("material", StringType(), True),

                StructField("img_num", IntegerType(), True),
                StructField("img_type", StringType(), True),
                StructField("qa_num", StringType(), True),
                StructField("brand", StringType(), True),
                StructField("ac_name", StringType(), True),
                StructField("node_id", StringType(), True),
                StructField("sp_num", StringType(), True),

                StructField("mpn", StringType(), True),
                StructField("online_time", StringType(), True),
                StructField("describe", StringType(), True),
                StructField("one_star", StringType(), True),
                StructField("two_star", StringType(), True),
                StructField("three_star", StringType(), True),
                StructField("four_star", StringType(), True),
                StructField("five_star", StringType(), True),
                StructField("low_star", IntegerType(), True),

                StructField("asin_type", StringType(), True),
                StructField("is_coupon", StringType(), True),
                StructField("search_category", StringType(), True),
                StructField("weight_str", StringType(), True),
                StructField("date_info", StringType(), True),
                StructField("site", StringType(), True),

                StructField("account_name", StringType(), True),
                StructField("other_seller_name", StringType(), True),
                StructField("bsr_date_info", StringType(), True),
                StructField("account_id", StringType(), True),
                StructField("package_quantity", StringType(), True),
                StructField("pattern_name", StringType(), True),
                StructField("together_asin", StringType(), True),
                StructField("activity_type", StringType(), True),
                StructField("one_two_val", StringType(), True),
                StructField("three_four_val", StringType(), True),
                StructField("five_six_val", StringType(), True),
                StructField("eight_val", StringType(), True),
                StructField("product_description", StringType(), True),
                StructField("asinUpdateTime", StringType(), True),
                StructField("follow_sellers", StringType(), True),
            ])
            self.detail_col = [
                'asin', 'img_url', 'title', 'title_len', 'price', 'rating', 'total_comments', 'buy_box_seller_type',
                'page_inventory', 'category', 'volume', 'weight', 'rank', 'launch_time', 'video_url', 'add_url',
                'material', 'img_num', 'img_type', 'qa_num', 'brand', 'ac_name', 'node_id', 'sp_num', 'mpn',
                'online_time', 'describe', 'one_star', 'two_star', 'three_star', 'four_star', 'five_star',
                'low_star', 'asin_type', 'is_coupon', 'search_category', 'weight_str', 'date_info', 'site',
                'account_name', 'other_seller_name', 'bsr_date_info', 'account_id', 'package_quantity',
                'pattern_name', 'together_asin', 'activity_type', 'one_two_val', 'three_four_val', 'five_six_val',
                'eight_val', 'product_description', 'asinUpdateTime', 'follow_sellers'
            ]

    def get_topic_name(self):
        # 需要注意表名问题
        if self.date_type == "month":
            # 月表主题
            self.topic_name = f"{self.site_name}_asin_detail_month"
        elif self.date_type == "week":
            # 周表主题
            self.topic_name = f"{self.site_name}_asin_detail"
        elif self.date_type == "day":
            # 天表主题
            self.topic_name = f"{self.site_name}_self_asin_detail"
        else:
            print("date_type传参有问题,中断程序")
            quit()

    def get_14pg_country_engine(self, site_name="us"):
        h14_pg_us = {
            "user": "postgres",
            "password": "fazAqRRVV9vDmwDNRNb593ht5TxYVrfTyHJSJ3BS",
            # "host": "61.145.136.61",
            "host": "192.168.10.223",
            "port": "5432",
            # "port": 54328,
            "database": "selection",
        }
        if site_name == 'us' or site_name == 'mx' or site_name == 'ca':
            h14_pg_us["database"] = f"selection"
            db_ = 'postgresql+psycopg2://{}:{}@{}:{}/{}'.format(*h14_pg_us.values())
        # elif site_name == "keepa":
        #     db_ = 'mysql+pymysql://{}:{}@{}:{}/{}?charset={}'.format(*h6_pg_us.values())
        else:
            h14_pg_us["database"] = f"selection_{site_name}"
            db_ = 'postgresql+psycopg2://{}:{}@{}:{}/{}'.format(*h14_pg_us.values())
        engine = create_engine(db_, encoding='utf-8')  # , pool_recycle=3600
        return engine

    def field_length_dispose(self, df):
        df.price = df.price.apply(lambda x: round(x, 2) if x is not None else 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)  # 截取字符
        if self.date_type in ["month", "week"]:
            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)  # 截取字符
        return df

    def img_save(self, df):
        logging.info("img处理")
        # 获取对应表字段
        if "site" not in df.keys():
            df["site"] = self.site_name
            logging.info("site is not null")
        df["site"] = df['site'].fillna(self.site_name)
        # df.drop_duplicates(subset=["asin", "site"], inplace=True)
        for name, group in df.groupby(['site']):
            asins = list(set(group["asin"]))
            logging.info(f"img处理 站点{name[0]}  ")
            if name[0] not in ['us', 'de', 'uk', 'it', 'es', 'fr', 'mx', 'ca']:
                logging.info("非8大站点跳过")
                continue
            if name[0] != "us":
                chunk_size = 1000
                split_list = [asins[i:i + chunk_size] for i in range(0, len(asins), chunk_size)]
                with self.pg14_engine.begin() as conn:
                    # Printing the split chunks
                    for chunk in split_list:
                        if len(chunk) == 1:
                            sql_del = f"delete from {name[0]}_asin_image where asin in ('{tuple(chunk)[0]}');"
                        else:
                            sql_del = f"delete from {name[0]}_asin_image where asin in {tuple(chunk)};"
                        logging.info(f"sql: {sql_del[0:100]}")
                        conn.execute(sql_del)
                        logging.info(f"清理{name[0]}_asin_image 表中数据   {chunk[0:10]} 清理{name[0]}_asin_image 表中数据")
            del group["site"]
            logging.info(f"数量为：{group.shape}")
            try:
                group.to_sql(name=f'{name[0]}_asin_image', con=self.pg14_engine, if_exists='append', index=False)
                logging.info(f"入库{name[0]}_asin_image成功 {group.head(10)}")
            except DataError as e:
                logging.info(f"img入库字段超过长度 {e}")
                group.to_csv(f"/root/{name[0]}_asin_image_{time.time()}.csv")

    def variat_save(self, df):
        df.drop_duplicates(subset=["asin", "parent_asin"], inplace=True)
        asins = list(set(df["parent_asin"]))
        logging.info(f"{df}")
        table = f'{self.site_name}_variat'
        if asins:
            chunk_size = 1000
            split_list = [asins[i:i + chunk_size] for i in range(0, len(asins), chunk_size)]
            for chunk in split_list:
                if len(chunk) == 1:
                    sql_del = f"delete from `{table}` where parent_asin in ('{tuple(chunk)[0]}');"
                else:
                    sql_del = f"delete from `{table}` where parent_asin in {tuple(chunk)};"
                logging.info(f"sql: {sql_del[0:100]}")
                for i in range(5):
                    row_id = sql_delete(sql_del)
                    if row_id == -1:
                        logging.info(f"删除失败 {table} 表中数据 {chunk}")
                        continue
                    else:
                        logging.info(f"清理 {table} 表中数据   {chunk[0:10]} 清理 {table} 表中数据")
                        break
        df['color'] = df['color'].apply(lambda x: x.encode('utf-8', 'ignore').decode('utf-8')[:180] if x else None)
        df['size'] = df['size'].apply(lambda x: x.encode('utf-8', 'ignore').decode('utf-8')[:180] if x else None)
        df['style'] = df['style'].apply(lambda x: x.encode('utf-8', 'ignore').decode('utf-8')[:180] if x else None)
        df['column_2'] = df['column_2'].apply(lambda x: x.encode('utf-8', 'ignore').decode('utf-8')[:180] if x else None)
        logging.info(f"数量为：{df.shape}")
        for i in range(3):
            try:
                df.to_sql(name=f'{table}', con=self.engine, if_exists='append', index=False)
                logging.info(f"入库 {table} 成功 {df.head(10)}")
                break
            except PendingRollbackError as e:
                logging.info(f"链接错误  重试{e}")
                continue

    def handle_data_df(self, df=pd.DataFrame, df_type='asin_vartion_list', columns=[]):
        # 根据不同表类型解析df对象
        df[df_type] = df[df_type].apply(json.loads)
        # 对对应数据进行处理,将df_type内列表展开
        exploded_list = df[df_type].explode()
        # 展开后转换为一个大列表
        df_type_list = [i for i in exploded_list.tolist() if not isinstance(i, float)]
        df_type_list = [i for i in df_type_list if isinstance(i, list)]
        if df_type_list:
            df = pd.DataFrame(df_type_list, columns=columns)
            return df
        else:
            return None

    def save_data_asin_detail(self, df):
        df.drop_duplicates(['asin'], inplace=True)
        # 这个要看 self.date_type是否有其他类型   并且数据中有这个字段  可能有坑
        for name, group in df.groupby([self.date_type]):
            logging.info(f"需要处理的 data_info {name[0]}")
            # 获取年
            # y = str(time.localtime().tm_year)
            y = self.date_info.split("-")[0]
            data_time = y + "_" + name[0]

            asins = list(group["asin"])
            # 详情入库表名
            detail_table_data_info = f"{self.site_name}_asin_detail_month_{data_time}" if self.date_type == "month" else f"{self.site_name}_asin_detail_{data_time}"
            logging.info(f"表名：{detail_table_data_info}")
            if asins:
                if self.date_type == "month":
                    logging.info("month data not delete")
                else:
                    chunk_size = 5000
                    split_list = [asins[i:i + chunk_size] for i in range(0, len(asins), chunk_size)]
                    with self.pg14_engine.begin() as conn:
                        for chunk in split_list:
                            if len(chunk) == 1:
                                sql_del = f"delete from {detail_table_data_info} where asin= '{chunk[0]}';"
                            else:
                                sql_del = f"delete from {detail_table_data_info} where asin in {tuple(chunk)};"
                            for i in range(5):
                                try:
                                    start_time = time.time()
                                    conn.execute(sql_del)
                                    end_time = time.time()
                                    logging.info(f"清理 {detail_table_data_info} 表中 {chunk[0:10]} 数据, 耗时：{end_time-start_time}s")
                                    break
                                except OperationalError as e:
                                    logging.info(f"数据库链接 失败{e}")
                                    time.sleep(3)
                                    continue
                # 测试报错代码
                logging.info(f"detail keys {group.keys()}")
                logging.info(f"{self.detail_col}")
                logging.info(f"{group.shape} {detail_table_data_info}")
                group = copy.deepcopy(group)
                group = group[self.detail_col]
                group.rename(columns={"asinUpdateTime": "created_time"}, inplace=True)
                group = self.field_length_dispose(group)
                logging.info(f"{group.keys()}")
                logging.info(f"{group.shape}")
                # df.rename(columns={"asinUpdateTime": "created_at"}, inplace=True)
                try:
                    group.to_sql(name=f'{detail_table_data_info}', con=self.pg14_engine, if_exists='append', index=False)
                    logging.info(f"入库 {detail_table_data_info} 成功 {group.head(10)}")
                except DataError as e:
                    logging.info(f"详情入库字段超过长度：{e}")
                    group.to_csv(f"/root/{detail_table_data_info}_{time.time()}.csv")


    def save_data_common(self, df, df_type):
        if df_type == 'asin_vartion_list':
            logging.info(f"asin_vartion_list 处理")
            if df.shape[0]:
                vartion_columns = ['asin', 'color', 'parent_asin', 'size', 'state', 'style', 'column_2']
                vartion_df = self.handle_data_df(df, df_type='asin_vartion_list', columns=vartion_columns)
                if vartion_df.shape[0]:
                    self.variat_save(df=vartion_df)
        elif df_type == 'img_list':
            logging.info(f"img_list 处理")
            if df.shape[0]:
                img_columns = ['asin', 'img_url', 'img_order_by', 'data_type']
                img_df = self.handle_data_df(df, df_type='img_list', columns=img_columns)
                if img_df.shape[0]:
                    self.img_save(df=img_df)
        elif df_type == 'asin_detail':
            logging.info(f"asin_detail 处理")
            self.save_data_asin_detail(df=df)

    def save_data(self, df):
        threads = []
        for df_type in self.df_type_dict.keys():
            thread = threading.Thread(target=self.save_data_common, args=(df, df_type))
            threads.append(thread)
            thread.start()
        for thread in threads:
            thread.join()
        logging.info("线程处理完成")

    def data_save(self, df):
        if not isinstance(df, pd.DataFrame):
            logging.info("df 不是一个 DataFrame 对象")
            df = df.toPandas()
        if df.shape[0]:
            logging.info(f"{df.keys()}")
            logging.info(f"----------------------------")
            if self.date_type == "day":
                logging.info(f"天数据处理")
                img_columns = ['asin', 'img_url', 'img_order_by', 'data_type']
                img_df = self.handle_data_df(df, df_type='asin_vartion_list', columns=img_columns)
                if img_df.shape[0]:
                    self.img_save(df=img_df)
                vartion_columns = ['asin', 'color', 'parent_asin', 'size', 'state', 'style', 'column_2']
                vartion_df = self.handle_data_df(df, df_type='asin_vartion_list', columns=vartion_columns)
                if vartion_df.shape[0]:
                    self.variat_save(df=vartion_df)
                df = df[self.detail_col]
                df['site'] = df['site'].fillna(self.site_name)
                df.drop_duplicates(['asin', 'site'], inplace=True)
                now_date = time.strftime("%Y-%m-%d", time.gmtime(time.time()))
                detail_table_data_info = f"{self.site_name}_self_asin_detail"
                for name, group in df.groupby(['site']):
                    asins = list(group["asin"])
                    # 详情入库表名
                    if asins:
                        if len(asins) == 1:
                            sql_del = f"delete from `{detail_table_data_info}` where `asin`= '{asins[0]}' and `site`='{name[0]}' and created_at>='{now_date}';"
                        else:
                            sql_del = f"delete from `{detail_table_data_info}` where `asin` in {tuple(asins)} and `site`='{name[0]}' and created_at>='{now_date}';"
                        logging.info(f"{name}, {sql_del}")
                        sql_delete(sql_del)
                        logging.info(f"清理 {detail_table_data_info} 表中 {asins[0:10]} 数据")
                df.to_sql(name=f'{detail_table_data_info}', con=self.engine, if_exists='append', index=False)
                logging.info(f"入库 {detail_table_data_info} 成功 {df.head(10)}")
            else:
                # 过滤date_info不符合的
                # new_df = df[df[self.date_type] == self.date_info.split("-")[-1]]  # self.date_type week
                # logging.info(f"过滤{self.date_type} 不为： {self.date_info.split('-')[-1]} \n 过滤后 {new_df.shape}")
                new_df = df
                if new_df.shape[0]:
                    self.save_data(new_df)
                else:
                    logging.info(f"过滤后 无数据处理{new_df}")
        else:
            logging.info(f"{df.shape}")

    def handle_kafka_history(self, kafka_df):
        self.data_save(kafka_df)

    def handle_kafka_stream(self, kafka_df, epoch_id):
        self.data_save(kafka_df)


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    date_type = sys.argv[2]  # 参数2：类型：week/4_week/month/quarter/day
    date_info = sys.argv[3]  # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
    consumer_type = sys.argv[4]  # 参数4：实时 lastest 历史 history
    # us day date_info 2023-11-07
    handle_obj = SpiderAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info, consumer_type=consumer_type, batch_size_history=10000)
    handle_obj.run_kafka()

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 1 --num-executors 1 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_self_asin_detail.py uk week 2023-46 lastest > amazon_week_uk.log 2>&1 &

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 4 --num-executors 2 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_self_asin_detail.py us month 2023-12 lastest

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 1 --num-executors 1 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_self_asin_detail.py uk week 2023-46 history

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 1 --num-executors 1 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_self_asin_detail.py us day 2023-11-16 lastest

# for i in `ps -ef|grep "spider_asin_detail.py" |awk '{print $2}' `; do kill -9 $i ; done;

# 历史
# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 20g --executor-cores 4 --num-executors 2 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_asin_detail.py de week 2023-46 history > amazon_week_history_de.log 2>&1 &
# 实时
# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 4 --num-executors 2 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_asin_detail.py de week 2023-48 lastest > amazon_week_de.log 2>&1 &