流量选品-上架时间算法调整

Pyspark_job/dwt/dwt_flow_asin.py

@@ -56,7 +56,11 @@ class DwtFlowAsin(Templates):
         self.df_save = self.spark.sql(f"select 1+1;")
         self.partitions_by = ['site_name', 'date_type', 'date_info']
         self.reset_partitions(60)
-        self.launch_time_interval_dict = self.get_launch_time_interval_dict()
+
+        launch_time_base_date = self.spark.sql(
+            f"""SELECT max(`date`) AS last_day FROM dim_date_20_to_30 WHERE year_month = '{self.date_info}'"""
+        ).collect()[0]['last_day']
+        self.launch_time_interval_dict = self.get_launch_time_interval_dict(launch_time_base_date)
         # 初始化全局df
         self.df_asin_detail = self.spark.sql(f"select 1+1;")
         self.df_asin_measure = self.spark.sql(f"select 1+1;")
@@ -123,15 +127,15 @@ class DwtFlowAsin(Templates):
             return str(datetime.now().date())
 
     @staticmethod
-    def get_launch_time_interval_dict():
-        cur_date = datetime.now().date()
+    def get_launch_time_interval_dict(base_date):
+        base_date = datetime.strptime(base_date, '%Y-%m-%d')
         return {
-            "one_month": (cur_date + timedelta(days=-30)).strftime('%Y-%m-%d'),
-            "three_month": (cur_date + timedelta(days=-90)).strftime('%Y-%m-%d'),
-            "six_month": (cur_date + timedelta(days=-180)).strftime('%Y-%m-%d'),
-            "twelve_month": (cur_date + timedelta(days=-360)).strftime('%Y-%m-%d'),
-            "twenty_four_month": (cur_date + timedelta(days=-720)).strftime('%Y-%m-%d'),
-            "thirty_six_month": (cur_date + timedelta(days=-1080)).strftime('%Y-%m-%d')
+            "one_month": (base_date + timedelta(days=-30)).strftime('%Y-%m-%d'),
+            "three_month": (base_date + timedelta(days=-90)).strftime('%Y-%m-%d'),
+            "six_month": (base_date + timedelta(days=-180)).strftime('%Y-%m-%d'),
+            "twelve_month": (base_date + timedelta(days=-360)).strftime('%Y-%m-%d'),
+            "twenty_four_month": (base_date + timedelta(days=-720)).strftime('%Y-%m-%d'),
+            "thirty_six_month": (base_date + timedelta(days=-1080)).strftime('%Y-%m-%d')
         }
 
     @staticmethod