etsy版本增加流程记录

Pyspark_job/sqoop_export/dwt_ehunt_product_detail_pg.py

@@ -11,6 +11,7 @@ from sqlalchemy import text
 from utils.db_util import DBUtil
 from utils.ssh_util import SSHUtil
 from utils.common_util import CommonUtil, DateTypes
+from utils.ehunt_util import write_workflow_record
 
 if __name__ == '__main__':
     site_name = CommonUtil.get_sys_arg(1, None)
@@ -185,4 +186,7 @@ if __name__ == '__main__':
         conn.execute(text(sql))
 
     engine.dispose()
+
+    write_workflow_record(site_name, date_type, date_info, page='etsy产品', table_name=base_tb,
+                          remark='Etsy产品详情', db_type=db_type)
     print("success")

Pyspark_job/sqoop_export/dwt_ehunt_store_detail_pg.py

@@ -11,6 +11,7 @@ from sqlalchemy import text
 from utils.db_util import DBUtil
 from utils.ssh_util import SSHUtil
 from utils.common_util import CommonUtil, DateTypes
+from utils.ehunt_util import write_workflow_record
 
 if __name__ == '__main__':
     site_name = CommonUtil.get_sys_arg(1, None)
@@ -170,4 +171,7 @@ if __name__ == '__main__':
         conn.execute(text(sql))
 
     engine.dispose()
+
+    write_workflow_record(site_name, date_type, date_info, page='etsy店铺', table_name=base_tb,
+                          remark='Etsy店铺详情', db_type=db_type)
     print("success")

Pyspark_job/utils/ehunt_util.py

@@ -67,3 +67,30 @@ def yes_no_to_cn(col_name):
     """标签统一为 是/否，无法识别的置空"""
     val = F.lower(F.trim(F.col(col_name)))
     return F.when(val.isin(YES_VALUES), F.lit("是")).when(val.isin(NO_VALUES), F.lit("否")).otherwise(F.lit(None))
+
+
+def write_workflow_record(site_name, date_type, date_info, page, table_name, remark, db_type):
+    """导出PG完成后写 mysql workflow_everyday 流程记录，测试库导出不写，已存在则跳过"""
+    from utils.db_util import DBUtil
+    if db_type != 'postgresql_cluster':
+        print(f"db_type={db_type}，非正式导出，跳过流程记录表更新")
+        return
+    engine_mysql = DBUtil.get_db_engine('mysql', 'us')
+    select_sql = f"""
+        select id from selection.workflow_everyday
+        where site_name='{site_name}' and date_type='{date_type}' and report_date='{date_info}'
+        and page='{page}' and table_name='{table_name}' and status_val=14 and is_end='是'
+    """
+    with engine_mysql.connect() as conn:
+        is_finished = conn.execute(select_sql).first()
+    if is_finished is None:
+        replace_sql = f"""
+            REPLACE INTO selection.workflow_everyday
+            (site_name, report_date, status, status_val, table_name, date_type, page, is_end, remark, export_db_type, freeze_flag)
+            VALUES('{site_name}', '{date_info}', '导出PG数据库完成', 14, '{table_name}', '{date_type}', '{page}', '是', '{remark}', '{db_type}', 'prod_disable');
+        """
+        DBUtil.engine_exec_sql(engine_mysql, replace_sql)
+        print(f"流程记录表 workflow_everyday 已写入：{site_name} {date_info} page={page} table_name={table_name}")
+    else:
+        print(f"流程记录表已存在该记录，跳过：page={page} table_name={table_name}")
+    engine_mysql.dispose()