月asin利润率计算窗口调整

Pyspark_job/script/export_need_profit_rate.py

@@ -2,7 +2,7 @@
 author: CT
 description: 导出待计算利润率的 ASIN(修复漏洞: 不再仅依赖 keepa 近 1 天更新)
              【数据流】
-              flow_asin 池: Hive dwt_flow_asin 最近 3 月 ∪ Doris dwt.{site}_flow_asin_30day
+              flow_asin 池: Hive dwt_flow_asin 最近 1 月 ∪ Doris selection.{site}_flow_asin_30day
                             → (asin, price) 去重保留 asin_crawl_date 最新
               两路合并:
                 A) keepa 变动重算: flow_asin INNER JOIN keepa(updated_time >= date_info-1天)
@@ -10,9 +10,8 @@ description: 导出待计算利润率的 ASIN(修复漏洞: 不再仅依赖 keep
                                                 INNER JOIN keepa(全量,无时间过滤)
               A ∪ B → 按 (asin, price) 去重保留 asin_crawl_date 最新
               → LEFT JOIN 分类 → 派生 source_month / part_key → 写 PG {site}_asin_profit_rate_calc
-             【近 3 月窗口】
-              通过 MySQL workflow_everyday.MAX(report_date) WHERE date_type='month' AND page='流量选品'
-              向前推 3 个月(参考 export_keepa_asin_del.py)
+             【近 1 月窗口】
+              取 MySQL workflow_everyday.MAX(report_date) WHERE date_type='month' AND page='流量选品' 对应月份
              【已计算判断】
               hive dim_asin_profit_rate_info 全量(只保留最新分区,按 site_name 过滤即可),
               按 (asin, price) LEFT ANTI 剔除
@@ -23,7 +22,6 @@ description: 导出待计算利润率的 ASIN(修复漏洞: 不再仅依赖 keep
 import os
 import sys
 from datetime import datetime, timedelta
-from dateutil.relativedelta import relativedelta
 
 sys.path.append(os.path.dirname(sys.path[0]))
 
@@ -49,10 +47,10 @@ class ExportNeedProfitRate(object):
         df_export = self.build_export_df()
         self.write_to_pg(df_export)
 
-    # ---------------- 公共数据池: flow_asin 近 3 月 ----------------
+    # ---------------- 公共数据池: flow_asin 近 1 月 ----------------
 
-    def _get_recent_3_months(self):
-        """通过 MySQL workflow_everyday 拿最大 report_date 向前推 3 月(参考 export_keepa_asin_del.py)"""
+    def _get_recent_month(self):
+        """通过 MySQL workflow_everyday 拿最大 report_date 对应月份"""
         sql_max_month = (
             f"select MAX(report_date) as date_info from workflow_everyday "
             f"where site_name = '{self.site_name}' and date_type = 'month' and page = '流量选品'"
@@ -64,18 +62,16 @@ class ExportNeedProfitRate(object):
             pwd=mysql_con['pwd'], username=mysql_con['username'], query=sql_max_month,
         ).collect()[0]['date_info']
         print(f"workflow_everyday 最大 report_date: {max_date_info}")
-        base_dt = datetime.strptime(str(max_date_info), '%Y-%m')
-        months_3 = [(base_dt - relativedelta(months=i)).strftime('%Y-%m') for i in range(3)]
-        print(f"近 3 月窗口: {months_3}")
-        return months_3
+        recent_month = str(max_date_info)
+        print(f"近 1 月窗口: {recent_month}")
+        return recent_month
 
     def _build_flow_asin_pool(self):
-        """flow_asin 池: Hive dwt_flow_asin 近 3 月 ∪ Doris {site}_flow_asin_30day
+        """flow_asin 池: Hive dwt_flow_asin 近 1 月 ∪ Doris selection.{site}_flow_asin_30day
            → (asin, price) 去重保留 asin_crawl_date 最新"""
-        months_3 = self._get_recent_3_months()
-        months_in = ",".join([f"'{m}'" for m in months_3])
+        recent_month = self._get_recent_month()
 
-        # 1. Hive dwt_flow_asin 近 3 月
+        # 1. Hive dwt_flow_asin 近 1 月
         sql_dwt = f"""
             SELECT asin,
                    asin_price       AS price,
@@ -84,7 +80,7 @@ class ExportNeedProfitRate(object):
             FROM dwt_flow_asin
             WHERE site_name = '{self.site_name}'
               AND date_type = 'month'
-              AND date_info IN ({months_in})
+              AND date_info = '{recent_month}'
               AND asin_price > 0
         """
         print(f"sql_dwt =\n{sql_dwt}")
@@ -92,8 +88,8 @@ class ExportNeedProfitRate(object):
             .withColumn('price', F.round(F.col('price'), 2).cast('decimal(20,2)')) \
             .withColumn('asin_crawl_date', F.to_timestamp(F.col('asin_crawl_date')))
 
-        # 2. Doris dwt.{site}_flow_asin_30day(保留: 月数据可能延迟,30day 兜底)
-        table_identifier = f"dwt.{self.site_name}_flow_asin_30day"
+        # 2. Doris selection.{site}_flow_asin_30day(保留: 月数据可能延迟,30day 兜底)
+        table_identifier = f"selection.{self.site_name}_flow_asin_30day"
         read_fields = "asin,price,category_first_id,asin_crawl_date"
         df_doris = DorisHelper.spark_import_with_connector(self.spark, table_identifier, read_fields) \
             .filter(F.col('price') > 0) \