prefect日志标准化

import_logistics_to_feishu_bitable.py

@@ -4,7 +4,7 @@ import os
 import time
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional, Tuple
+from typing import Callable, Dict, Iterable, List, Optional, Tuple
 from urllib import error, parse, request
 
 try:
@@ -98,12 +98,22 @@ FIELD_TYPE_HINTS = {
     "商品数量": FIELD_TYPE_NUMBER,
 }
 
+LOGGER_WRITER: Optional[Callable[[str, str], None]] = None
+
 
 def log(message: str, *, level: str = "INFO") -> None:
+    if LOGGER_WRITER is not None:
+        LOGGER_WRITER(level, message)
+        return
     timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
     print(f"[{timestamp}] [{level}] {message}")
 
 
+def set_logger_writer(writer: Optional[Callable[[str, str], None]]) -> None:
+    global LOGGER_WRITER
+    LOGGER_WRITER = writer
+
+
 def load_env(path: Path) -> None:
     if not path.exists():
         return

prefect_logistics_scheduler.py

@@ -9,6 +9,17 @@ from prefect import flow, get_run_logger, task
 import import_logistics_to_feishu_bitable
 
 
+def prefect_log_writer(level: str, message: str) -> None:
+    logger = get_run_logger()
+    level_name = level.upper()
+    if level_name == "ERROR":
+        logger.error(message)
+    elif level_name == "WARN":
+        logger.warning(message)
+    else:
+        logger.info(message)
+
+
 def send_feishu_group_message(text: str) -> None:
     import_logistics_to_feishu_bitable.load_env(import_logistics_to_feishu_bitable.ENV_PATH)
 
@@ -41,6 +52,7 @@ def run_logistics_sync_task() -> dict:
     logger = get_run_logger()
     logger.info("开始执行物流数据同步任务")
     started_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    import_logistics_to_feishu_bitable.set_logger_writer(prefect_log_writer)
 
     try:
         summary = import_logistics_to_feishu_bitable.execute_sync()
@@ -54,6 +66,8 @@ def run_logistics_sync_task() -> dict:
         )
         send_feishu_group_message(failure_text)
         raise
+    finally:
+        import_logistics_to_feishu_bitable.set_logger_writer(None)
 
     finished_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
     logger.info("物流数据同步任务完成")