流量选品-利润率计算适配小站点

Pyspark_job/doris_handle/dwt_user_selection_pattern.py

@@ -3,7 +3,7 @@
    @Description : 按站点增量追平用户筛选模式选品结果，写入 Doris selection.user_selection_pattern
                   - 传入 site_name，只处理该站点的 active filter
                   - 纯增量追平：水位线落后于该站点最新月表时补齐缺失月份，不做刷新判断
-                    （us 站点利润率/Keepa 字段每日刷新型 filter，最新月差异化刷新见 dws_flow_asin_refresh.py 第5步）
+                    （利润率/Keepa 字段每日刷新型 filter，最新月差异化刷新见 dws_flow_asin_refresh.py 第5步）
                   - 支持断点续算：latest_computed_month 作为水位线，跳过已算月份
                   - Doris AGGREGATE KEY(filter_id, asin) + MIN(date_info) 自动保留首次入选月份
                   - 每月执行一次即可（只有新月表出现时才会产生缺口）

Pyspark_job/script/export_keepa_asin_del.py

@@ -2,7 +2,7 @@
 author: CT
 description: 从 Hive dim_keepa_asin_info 读取 keepa 数据，过滤 updated_time 超过 3 个月的 asin（keepa 过期），
              并关联近期有月销的 asin（dwt_flow_asin 最近 3 个月 ∪ Doris dwt.{site}_flow_asin_30day），
-             只导出"过期 + 近期有月销"的 asin 到 PG us_asin_profit_keepa_add，触发爬虫重新抓取 Keepa。
+             只导出"过期 + 近期有月销"的 asin 到 PG {site}_asin_profit_keepa_add，触发爬虫重新抓取 Keepa。
              导出前 LEFT ANTI 剔除 PG 表中已存在的 asin，避免重复触发。
 执行示例: spark-submit export_keepa_asin_del.py us
 """
@@ -48,7 +48,7 @@ if __name__ == '__main__':
     # 3. 读取 Hive dwt_flow_asin 最近 3 个月（基于 MySQL workflow_everyday 最大 report_date 向前推 3 个月）、月销 > 0 的 asin
     sql_max_month = f"select MAX(report_date) as date_info from workflow_everyday where site_name = '{site_name}' and date_type = 'month' and page = '流量选品'"
     print(f"sql_max_month=\n{sql_max_month}")
-    mysql_con = DBUtil.get_connection_info('mysql', site_name)
+    mysql_con = DBUtil.get_connection_info('mysql', 'us')
     max_date_info = SparkUtil.read_jdbc_query(
         session=spark, url=mysql_con['url'],
         pwd=mysql_con['pwd'], username=mysql_con['username'], query=sql_max_month,
@@ -97,7 +97,7 @@ if __name__ == '__main__':
 
     # 7. 读取 PG 已导出 asin（去重），用于 LEFT ANTI 剔除
     con_info = DBUtil.get_connection_info(db_type='postgresql_cluster', site_name=site_name)
-    table_name = 'us_asin_profit_keepa_add'
+    table_name = f'{site_name}_asin_profit_keepa_add'
     df_pg_existing = spark.read.format("jdbc") \
         .option("url", con_info["url"]) \
         .option("dbtable", table_name) \

Pyspark_job/script/export_need_profit_rate.py

@@ -22,6 +22,7 @@ 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]))
 
@@ -45,6 +46,7 @@ class ExportNeedProfitRate(object):
 
     def run(self):
         df_export = self.build_export_df()
+        self.ensure_partitions(df_export)
         self.write_to_pg(df_export)
 
     # ---------------- 公共数据池: flow_asin 近 1 月 ----------------
@@ -56,7 +58,7 @@ class ExportNeedProfitRate(object):
             f"where site_name = '{self.site_name}' and date_type = 'month' and page = '流量选品'"
         )
         print(f"sql_max_month=\n{sql_max_month}")
-        mysql_con = DBUtil.get_connection_info('mysql', self.site_name)
+        mysql_con = DBUtil.get_connection_info('mysql', 'us')
         max_date_info = SparkUtil.read_jdbc_query(
             session=self.spark, url=mysql_con['url'],
             pwd=mysql_con['pwd'], username=mysql_con['username'], query=sql_max_month,
@@ -234,6 +236,20 @@ class ExportNeedProfitRate(object):
         df_need.unpersist()
         return df_result
 
+    def ensure_partitions(self, df_export):
+        """按本批 source_month 自动补建 PG 月子表"""
+        engine    = DBUtil.get_db_engine('postgresql_cluster', self.site_name)
+        master_tb = f"{self.site_name}_asin_profit_rate_calc"
+        months = sorted(r['source_month'] for r in df_export.select('source_month').distinct().collect())
+        print(f"本批涉及 source_month: {months}")
+        for month in months:
+            next_month = (datetime.strptime(month, '%Y-%m') + relativedelta(months=1)).strftime('%Y-%m')
+            DBUtil.engine_exec_sql(engine, f"""
+                create table if not exists {master_tb}_{month.replace('-', '_')} partition of {master_tb}
+                    for values from ('{month}') to ('{next_month}');
+            """)
+        engine.dispose()
+
     def write_to_pg(self, df_export):
         con_info   = DBUtil.get_connection_info(db_type='postgresql_cluster', site_name=self.site_name)
         table_name = f"{self.site_name}_asin_profit_rate_calc"