judge.py

import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.templates_mysql import TemplatesMysql
# from ..utils.templates_mysql import TemplatesMysql
import time
import datetime
import pandas as pd


class Judge(TemplatesMysql):

    def __init__(self, site_name='us', data_type=1):
        super().__init__()
        self.site_name = site_name
        # 1: 关键词, 2: asin详情, 3, 4: 店铺产品, 5, 6, 7: ao_val,
        # 11: last_week, 12: last_4_week, 13: asin_last_4_week, 14: st_month_年_月(每个月最后一周), 15: asin_st
        # 21： brand_analytics_week, 22: brand_analytics_month
        self.data_type = data_type
        self.engine = self.mysql_connect()
        self.year, self.week, self.week_day = datetime.datetime.now().isocalendar()
        self.current_date = time.strftime("%Y-%m-%d", time.localtime())
        print("current_date:", self.current_date)
        print("site_name, year, week, week_day:", self.site_name, self.year, self.week, self.week_day)
        self.df_judge = pd.DataFrame()
        self.db_name = "selection" if self.site_name == "us" else f"selection_{self.site_name}"

    def read_workflow(self):
        # 1.1 data_type的值为1,2,4时, 监听workflow_crawling表的search_term和asin_detail和店铺产品抓取情况
        table = "selection.workflow_crawling"
        # 1.2 data_type的值为7时, 监听workflow_exhibition表的ao_va计算情况
        if self.data_type in [7]:
            table = "selection.workflow_exhibition"
        sql = f"select * from {table} WHERE week='{self.year}_{self.week}' and site_name='{self.site_name}' and data_type={self.data_type} and status=3;"
        print("sql_read:", sql)
        self.df_judge = pd.read_sql(sql=sql, con=self.engine)

    def judge(self):
        if self.data_type in [1, 2]:
            if self.data_type == 1:
                print_words = "关键词已经抓完, 结束判断, 进行下一步操作"
            else:
                print_words = "asin详情已经抓完, 结束判断, 进行下一步操作"
        elif self.data_type in [4]:
            print_words = "店铺产品已经抓完, 结束判断, 进行下一步操作"
        elif self.data_type in [7]:
            print_words = "关键词的ao_val已经计算完成, 结束判断, 进行下一步操作"
        else:
            print_words = "输入类型有误, 请重新检查"
            print(print_words)
        while True:
            if self.df_judge.shape[0] == 1:
                print("current_time:", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                print("site_name, year, week:", self.site_name, self.year, self.week, print_words)
                break
            else:
                print("current_time:", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                print("site_name, year, week:", self.site_name, self.year, self.week, "当前流程尚未完成,睡眠10min之后继续判断")
                time.sleep(600)
                self.read_workflow()
                continue

    def delete_and_rename_and_create_asin_last_4_week(self):
        num = 0
        while True:
            num += 1
            print("开始检查数据")
            sql_read = f"""SELECT min(id) as counts FROM {self.db_name}.{self.site_name}_asin_last_4_week_copy1 union
                    SELECT max(id) as counts FROM {self.db_name}.{self.site_name}_asin_last_4_week_copy1;"""
            df = pd.read_sql(sql_read, con=self.engine)
            count = list(df.counts)[1] - list(df.counts)[0] + 1
            print("copy表数据量：", count)
            if count > 1000000:
                print("3. 删除正式表，将copy表改成正式表")
                with self.engine.begin() as conn:
                    sql_drop = f"drop table if exists {self.db_name}.{self.site_name}_asin_last_4_week;"
                    sql_rename = f"ALTER TABLE {self.db_name}.{self.site_name}_asin_last_4_week_copy1 RENAME {self.db_name}.{self.site_name}_asin_last_4_week;"
                    print("sql_drop:", sql_drop)
                    conn.execute(sql_drop)
                    print("sql_rename:", sql_rename)
                    conn.execute(sql_rename)
                break
            else:
                if num > 3:
                    print("检查次数超过3次异常，停止")
                    break
                else:
                    print(f"第{num}次检查异常，继续删除+改表面")
                    continue
        # self.update_table_state(data_type=3)  # last_4_week
        with self.engine.begin() as conn:
            sql_drop = f"drop table if exists {self.db_name}.{self.site_name}_asin_last_4_week_copy1;"
            print("sql_drop:", sql_drop)
            conn.execute(sql_drop)
            sql_create = f"""CREATE TABLE {self.db_name}.{self.site_name}_asin_last_4_week_copy1 (
  'id' int(11) NOT NULL AUTO_INCREMENT,
  'asin' varchar(30) DEFAULT NULL,
  'ao_val' decimal(10,3) DEFAULT '0.000' COMMENT 'SP广告词+品牌广告词+视频广告词）/ asin自然词总数',
  'zr_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的zr类型的asin总数',
  'sp_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的sp类型的asin总数',
  'sb_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的sb(底部、头部)类型的asin总数',
  'vi_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的视频词类型的asin总数',
  'bs_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的Best Seller词类型的asin总数',
  'ac_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的AC词类型的asin总数',
  'tr_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的TR词类型的asin总数',
  'er_counts' int(10) DEFAULT NULL COMMENT '搜索词对应的ER类型的asin总数',
  'bsr_orders' int(10) DEFAULT '0' COMMENT 'bsr销量(统一使用当前月该asin最后一周销量,不用求总和)',
  'orders' int(10) DEFAULT NULL COMMENT '流量预估销量(统一使用当前月该asin最后一周销量,不用求总和)',
  'sales' decimal(20,4) DEFAULT '0.0000' COMMENT 'bsr销量*asin价格=销售额',
  'is_self' smallint(2) DEFAULT '1' COMMENT '是否为公司内部asin(1否,2是,3brand表没有抓到的asin)',
  'pt_category_id' int(2) DEFAULT NULL COMMENT 'bsr该asin的上级分类id',
  'one_category_id' int(2) DEFAULT NULL COMMENT 'bsr该asin的一级分类id',
  'img_url' varchar(200) DEFAULT NULL COMMENT 'asin的URL',
  'title' varchar(400) DEFAULT NULL COMMENT '标题',
  'title_len' int(5) DEFAULT NULL COMMENT '标题长度',
  'price' decimal(10,3) DEFAULT '0.000' COMMENT '价格',
  'rating' decimal(10,1) DEFAULT NULL COMMENT '评分',
  'total_comments' int(20) DEFAULT NULL COMMENT '评论数',
  'buy_box_seller_type' double DEFAULT NULL COMMENT 'buybox卖家类型(1:Amazon,2:FBA,3FBM,4无BB卖家)',
  'page_inventory' double DEFAULT NULL COMMENT '页面库存(1:充足，2:1-20(正常),3:缺货)',
  'volume' varchar(30) DEFAULT NULL COMMENT '体积',
  'weight' double DEFAULT NULL COMMENT '重量',
  'rank' int(10) DEFAULT NULL COMMENT '排名',
  'launch_time' date DEFAULT NULL COMMENT '上架时间',
  'img_num' smallint(2) DEFAULT '0' COMMENT 'aisn的图片数量',
  'img_type' varchar(10) DEFAULT NULL COMMENT '所有图片类型(1:普通图片，2:视频,3:A+图片)',
  'activity_type' varchar(10) DEFAULT NULL COMMENT '1.coupon(with coupon、with coupon) 百分比 2.coupon(with coupon、with coupon) 金额 3.Join Prime(Prime Exclusive Discounts) 百分比 4.Join Prime(Prime Exclusive Discounts) 金额 5.Deal(Lightning Deal，7-day Deal，Deal OF The Day，Outlet deals,With  Deal)  6.Top Deal(Top Deal) 7.Extra Savings(绑定促销,Extra Savings)  可允许该ASIN同时做两种促销',
  'one_two_val' decimal(10,3) DEFAULT NULL COMMENT 'coupon额度 activity_type 1、2对应的值',
  'three_four_val' decimal(10,3) DEFAULT NULL COMMENT 'Prime Exclusive Discounts额度 activity_type 3、4对应的值',
  'five_six_val' decimal(10,3) DEFAULT NULL COMMENT 'Join Prime额度activity_type对应5,6的值',
  'eight_val' decimal(10,0) DEFAULT NULL COMMENT '9：降低幅度的值 %',
  'qa_num' int(10) DEFAULT NULL COMMENT 'QA数量',
  'brand_name' varchar(50) DEFAULT NULL COMMENT '品牌名称',
  'variation_num' int(10) DEFAULT NULL COMMENT '变体数量',
  'one_star' int(2) DEFAULT NULL COMMENT '1星比例',
  'two_star' int(2) DEFAULT NULL COMMENT '2星比例',
  'three_star' int(2) DEFAULT NULL COMMENT '3星比例',
  'four_star' int(2) DEFAULT NULL COMMENT '4星比例',
  'five_star' int(2) DEFAULT NULL COMMENT '5星比例',
  'low_star' int(2) DEFAULT NULL COMMENT 'one_star + two_star + three_star 比例总和',
  'together_asin' varchar(255) DEFAULT NULL COMMENT '一起购买的asin，多个逗号拼接',
  'account_name' varchar(255) DEFAULT NULL COMMENT '亚马逊卖家账号名称',
  'account_id' bigint(2) DEFAULT NULL COMMENT '卖家账号id',
  'site_name' varchar(100) DEFAULT NULL COMMENT '国家名称',
  'updated_at' timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  'created_at' timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (id),
  UNIQUE KEY 'sel_asin' (asin) USING BTREE,
  KEY 'sel_updat' (updated_at) USING BTREE,
  KEY 'sel_ao' (ao_val) USING BTREE,
  KEY 'sel_is_self' (is_self) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;"""
            conn.execute(sql_create)
        print(f"site_name: {self.site_name}站点的_asin_last_4_week已经跑完,copy表已经建立")

    def delete_and_rename_and_create_st_last_4_week(self):
        num = 0
        while True:
            num += 1
            print("开始检查数据")
            sql_read = f"""SELECT min(id) as counts FROM {self.db_name}.{self.site_name}_last_4_week_st_copy1 union
            SELECT max(id) as counts FROM {self.db_name}.{self.site_name}_last_4_week_st_copy1;"""
            df = pd.read_sql(sql_read, con=self.engine)
            count = list(df.counts)[1] - list(df.counts)[0] + 1
            print("copy表数据量：", count)
            if count > 1000000:
                print("3. 删除正式表，将copy表改成正式表")
                with self.engine.begin() as conn:
                    sql_drop = f"drop table if exists {self.db_name}.{self.site_name}_last_4_week_st;"
                    sql_rename = f"ALTER TABLE {self.db_name}.{self.site_name}_last_4_week_st_copy1 RENAME {self.db_name}.{self.site_name}_last_4_week_st;"
                    print("sql_drop:", sql_drop)
                    conn.execute(sql_drop)
                    print("sql_rename:", sql_rename)
                    conn.execute(sql_rename)
                break
            else:
                if num > 3:
                    print("检查次数超过3次异常，停止")
                    break
                else:
                    print(f"第{num}次检查异常，继续删除+改表面")
                    continue
        self.update_table_state(data_type=3)  # last_4_week
        with self.engine.begin() as conn:
            sql_drop = f"drop table if exists {self.db_name}.{self.site_name}_last_4_week_st_copy1;"
            print("sql_drop:", sql_drop)
            conn.execute(sql_drop)
            sql_create = f"""CREATE TABLE {self.db_name}.{self.site_name}_last_4_week_st_copy1 (
  id int(11) NOT NULL AUTO_INCREMENT,
  week int(10) NOT NULL COMMENT '周',
  asin varchar(30) NOT NULL,
  search_term varchar(500) NOT NULL COMMENT '搜索词',
  ao_val decimal(10,3) DEFAULT '0.000' COMMENT '该asin的搜索词的AO值：当前asin的(sp总数+sb总数+视频总数)/自然总数',
  zr_counts int(10) DEFAULT NULL COMMENT '搜索词对应的zr类型的asin总数',
  sp_counts int(10) DEFAULT NULL COMMENT '搜索词对应的sp类型的asin总数',
  orders int(20) DEFAULT '0' COMMENT '预估销量',
  orders_sum int(20) DEFAULT NULL COMMENT '总销量',
  flow double(10,4) DEFAULT NULL COMMENT '流量占比',
  search_num int(10) DEFAULT NULL COMMENT '搜索量',
  search_rank int(10) DEFAULT NULL COMMENT '搜索词排名',
  adv_compet int(1) DEFAULT '0' COMMENT '广告竞品数',
  compet int(1) DEFAULT '0' COMMENT '竞品数',
  quantity_being_sold bigint(10) DEFAULT NULL COMMENT '关键词搜索量总条数',
  zr_page_rank int(1) DEFAULT NULL COMMENT 'zr_词类型具体位置(所有页所有行的具体位置)',
  zr_page int(1) DEFAULT NULL COMMENT 'zr_词类型具体页',
  zr_page_row int(1) DEFAULT NULL COMMENT 'zr_词类型具体页面的行数',
  zr_updated_at timestamp NULL DEFAULT NULL COMMENT 'zr_词类型抓取时间',
  sp_page int(1) DEFAULT NULL COMMENT 'sp_词类型具体页',
  sp_page_rank int(1) DEFAULT NULL COMMENT 'sp_词类型具体位置(所有页所有行的具体位置)',
  sp_page_row int(1) DEFAULT NULL COMMENT 'sp_词类型具体页面的行数',
  sp_updated_at timestamp NULL DEFAULT NULL COMMENT 'sp_词类型抓取时间',
  sb1_page int(1) DEFAULT NULL COMMENT 'sb_词类型具体页(头部)',
  sb1_updated_at timestamp NULL DEFAULT NULL COMMENT 'sb_词类型抓取时间(头部)',
  sb2_page int(1) DEFAULT NULL COMMENT 'sb_词类型具体页(尾部)',
  sb2_updated_at timestamp NULL DEFAULT NULL COMMENT 'sb_词类型抓取时间(尾部)',
  sb3_page int(1) DEFAULT NULL COMMENT 'sb_词类型具体页(视频)',
  sb3_updated_at timestamp NULL DEFAULT NULL COMMENT 'sb_词类型抓取时间(视频)',
  ac_page int(1) DEFAULT NULL COMMENT 'ac_词类型具体页',
  ac_updated_at timestamp NULL DEFAULT NULL COMMENT 'ac_词类型抓取时间',
  bs_page int(1) DEFAULT NULL COMMENT 'bs_词类型具体页',
  bs_updated_at timestamp NULL DEFAULT NULL COMMENT 'bs_词类型抓取时间',
  er_page int(1) DEFAULT NULL COMMENT 'er_词类型具体页',
  er_updated_at timestamp NULL DEFAULT NULL COMMENT 'er_词类型抓取时间',
  tr_page int(1) DEFAULT NULL COMMENT 'tr_词类型具体页',
  tr_updated_at timestamp NULL DEFAULT NULL COMMENT 'tr_词类型抓取时间',
  search_term_type varchar(50) DEFAULT NULL COMMENT '搜索词类型(1.主要流量词;2.精准长尾词;3.精准流量词;4.转化优质词;5.转化平稳词;6.转化流失词;7.出单词;8.无效曝光词)',
  created_at timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (id),
  KEY sel_st (search_term),
  KEY sel_asin (asin),
  KEY sel_week (week) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1020421895 DEFAULT CHARSET=utf8mb4;"""
            conn.execute(sql_create)
        print(f"site_name: {self.site_name}站点的last_4_week_st已经跑完,copy表已经建立")

    def update_table_state(self, data_type=1):
        table = "selection.workflow_integration"
        with self.engine.begin() as conn:
            sql_update = f"update {table} set status=3 where week='{self.year}_{self.week}' and site_name='{self.site_name}' and data_type={data_type}"
            print("sql_update:", sql_update)
            conn.execute(sql_update)

    def delete_and_rename_and_create_seller_asin_account_detail(self):
        num = 0
        while True:
            num += 1
            print("开始检查数据")
            sql_read = f"""SELECT min(id) as counts FROM {self.db_name}.{self.site_name}_seller_asin_account_detail_copy1 union
                            SELECT max(id) as counts FROM {self.db_name}.{self.site_name}_seller_asin_account_detail_copy1;"""
            df = pd.read_sql(sql_read, con=self.engine)
            count = list(df.counts)[1] - list(df.counts)[0] + 1
            print("copy表数据量：", count)
            if count > 1000000:
                print("3. 删除正式表，将copy表改成正式表")
                with self.engine.begin() as conn:
                    sql_drop = f"drop table if exists {self.db_name}.{self.site_name}_seller_asin_account_detail;"
                    sql_rename = f"ALTER TABLE {self.db_name}.{self.site_name}_seller_asin_account_detail_copy1 RENAME {self.db_name}.{self.site_name}_seller_asin_account_detail;"
                    print("sql_drop:", sql_drop)
                    conn.execute(sql_drop)
                    print("sql_rename:", sql_rename)
                    conn.execute(sql_rename)
                break
            else:
                if num > 3:
                    print("检查次数超过3次异常，停止")
                    break
                else:
                    print(f"第{num}次检查异常，继续删除+改表面")
                    continue
        with self.engine.begin() as conn:
            sql_drop = f"drop table if exists {self.db_name}.{self.site_name}_seller_asin_account_detail_copy1;"
            print("sql_drop:", sql_drop)
            conn.execute(sql_drop)
            sql_create = f"""CREATE TABLE {self.db_name}.{self.site_name}_seller_asin_account_detail_copy1 (
                  id int(11) NOT NULL AUTO_INCREMENT,
                  account_id int(11) NULL DEFAULT NULL COMMENT 'us_seller_account_syn表的 主键id',
                  account_name varchar(200) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '亚马逊卖家账号名称',
                  asin varchar(20) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
                  launch_time date NULL DEFAULT NULL COMMENT '上架日期',
                  days_diff int(10) NULL DEFAULT NULL COMMENT '距离计算日期的间隔天数',
                  is_asin_new int(2) NULL DEFAULT NULL COMMENT '是否是新品(间隔天数<=180天为新品，否则为老品)0:老品,1:新品',
                  created_at timestamp(0) NULL DEFAULT CURRENT_TIMESTAMP(0),
                  updated_at timestamp(0) NULL DEFAULT CURRENT_TIMESTAMP(0) ON UPDATE CURRENT_TIMESTAMP(0),
                  PRIMARY KEY (id) USING BTREE,
                  INDEX sel_is_asin_new(account_name, is_asin_new) USING BTREE,
                  INDEX asin(asin) USING BTREE,
                  INDEX sel_account_id(account_id, is_asin_new) USING BTREE
                ) ENGINE = InnoDB AUTO_INCREMENT = 72102642 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;"""
            conn.execute(sql_create)
        print(f"site_name: {self.site_name}站点的_seller_asin_account_detail已经跑完,copy表已经建立")

    def run(self):
        if self.data_type in [1, 2, 4, 7]:
            self.read_workflow()
            self.judge()
        elif self.data_type in range(11, 21, 1):
            if self.data_type in [11]:
                self.update_table_state(data_type=2)  # week_st
                print(f"site_name: {self.site_name}站点的week_st已经跑完")
            if self.data_type in [12]:
                self.delete_and_rename_and_create_st_last_4_week()
            if self.data_type in [13]:
                self.delete_and_rename_and_create_st_last_4_week()
        elif self.data_type in range(21, 31, 1):
            if self.data_type in [21]:
                self.delete_and_rename_and_create_seller_asin_account_detail()
            pass


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    data_type = int(sys.argv[2])  # 参数1：站点
    handle_obj = Judge(site_name=site_name, data_type=data_type)
    handle_obj.run()