调整 日志输出格式

inventory_analytics/batch_inventory_report/service.py

@@ -4,8 +4,10 @@
 from __future__ import annotations
 
 import argparse
+from dataclasses import dataclass
 from datetime import datetime
 import logging
+import os
 from pathlib import Path
 import time
 from typing import Any
@@ -16,6 +18,7 @@ import psycopg2
 from ..config import DbConfig, build_app_config, load_analytics_db_config, load_raw_db_config
 from ..feishu.client import FeishuApiError, FeishuClient
 from ..logging_utils import configure_logging as configure_app_logging
+from ..logging_utils import exception_fields
 from ..logging_utils import log_event as emit_log_event
 from ..paths import LOG_DIR, REPORT_DIR
 
@@ -190,18 +193,19 @@ LOG_FIELD_LABELS = {
     "database": "数据库",
     "deleted_records": "删除记录数",
     "elapsed_ms": "耗时毫秒",
-    "env_path": "环境文件",
+    "error": "错误",
+    "error_type": "异常类型",
     "existing_records": "已有记录数",
     "filtered_rows": "筛选后行数",
     "join_key": "关联字段",
-    "log_file": "日志文件",
     "matched_rows": "匹配行数",
-    "output": "输出文件",
+    "missing_join_key_rows": "缺少关联字段行数",
+    "report_output": "本地报表输出",
     "raw_rows": "Raw行数",
     "report_rows": "报表行数",
     "rows": "行数",
+    "stage": "阶段",
     "table": "表",
-    "table_id": "表ID",
     "table_name": "表名",
     "unmatched_rows": "未匹配行数",
     "warehouse_count": "仓库数量",
@@ -387,13 +391,18 @@ def ensure_batch_inventory_fields(
 
 
 def sync_report_to_feishu(df: pd.DataFrame, table_name: str = BATCH_FEISHU_TABLE_NAME) -> dict[str, Any]:
+    started_at = time.monotonic()
     app_config = build_app_config()
     client = FeishuClient(app_config)
+    log_event(logging.INFO, "开始获取飞书访问令牌", stage="feishu_auth")
     access_token = client.get_tenant_access_token()
+    log_event(logging.INFO, "飞书访问令牌获取完成", stage="feishu_auth")
+    log_event(logging.INFO, "开始解析飞书Base信息", stage="feishu_base")
     base_info = client.parse_base_info(client.build_base_url(), access_token)
     app_token = base_info["app_token"] or ""
     client.enable_advanced_permission(access_token, app_token)
     tables = client.list_tables(access_token, app_token)
+    log_event(logging.INFO, "飞书表列表读取完成", stage="feishu_tables", rows=len(tables))
     target_table = next((table for table in tables if table.get("name") == table_name), None)
     if not target_table:
         raise FeishuApiError(f"未找到目标飞书表: {table_name}")
@@ -401,7 +410,7 @@ def sync_report_to_feishu(df: pd.DataFrame, table_name: str = BATCH_FEISHU_TABLE
     if not table_id:
         raise FeishuApiError(f"目标飞书表缺少 table_id: {table_name}")
 
-    log_event(logging.INFO, "开始同步飞书批次库存表", table_name=table_name, table_id=table_id, report_rows=len(df))
+    log_event(logging.INFO, "开始同步飞书批次库存表", stage="feishu_sync", table_name=table_name, report_rows=len(df))
     available_fields = ensure_batch_inventory_fields(client, access_token, app_token, table_id)
     existing_records = client.list_records(access_token, app_token, table_id)
     existing_record_ids = [record.get("record_id") for record in existing_records if record.get("record_id")]
@@ -417,12 +426,13 @@ def sync_report_to_feishu(df: pd.DataFrame, table_name: str = BATCH_FEISHU_TABLE
     log_event(
         logging.INFO,
         "飞书批次库存表同步完成",
+        stage="feishu_sync",
         table_name=table_name,
-        table_id=table_id,
         existing_records=len(existing_record_ids),
         deleted_records=deleted_count,
         created_records=created_count,
         available_fields=", ".join(available_fields),
+        elapsed_ms=int((time.monotonic() - started_at) * 1000),
     )
     return {
         "table_name": table_name,
@@ -436,7 +446,7 @@ def sync_report_to_feishu(df: pd.DataFrame, table_name: str = BATCH_FEISHU_TABLE
 
 def query_dataframe(db_config: DbConfig, sql: str, *, database_name: str) -> pd.DataFrame:
     started_at = time.monotonic()
-    log_event(logging.INFO, "开始查询数据库", database=database_name)
+    log_event(logging.INFO, "开始查询数据库", stage="database", database=database_name)
     with psycopg2.connect(
         host=db_config.host,
         port=db_config.port,
@@ -450,6 +460,7 @@ def query_dataframe(db_config: DbConfig, sql: str, *, database_name: str) -> pd.
     log_event(
         logging.INFO,
         "数据库查询完成",
+        stage="database",
         database=database_name,
         rows=len(df),
         elapsed_ms=int((time.monotonic() - started_at) * 1000),
@@ -476,7 +487,7 @@ def extract_raw_codes(df: pd.DataFrame) -> list[str]:
         .drop_duplicates()
         .tolist()
     )
-    log_event(logging.INFO, "已提取Raw商家编码集合", rows=len(codes), join_key="商家编码=统一规格编码")
+    log_event(logging.INFO, "已提取Raw商家编码集合", stage="prepare", rows=len(codes), join_key="商家编码=统一规格编码")
     return codes
 
 
@@ -504,8 +515,10 @@ def filter_batch_inventory(df: pd.DataFrame) -> pd.DataFrame:
     log_event(
         logging.INFO,
         "Raw批次库存筛选完成",
+        stage="prepare",
         raw_rows=len(df),
         filtered_rows=len(filtered),
+        missing_join_key_rows=int(missing_join_key),
         warehouse_count=len(WAREHOUSE_FILTER),
     )
     return filtered
@@ -514,7 +527,7 @@ def filter_batch_inventory(df: pd.DataFrame) -> pd.DataFrame:
 def prepare_analytics_stock(df: pd.DataFrame) -> pd.DataFrame:
     prepared = df.copy()
     prepared["关联编码"] = prepared["统一规格编码"].map(normalize_code)
-    log_event(logging.INFO, "店铺锁定库存整理完成", rows=len(prepared), join_key="统一规格编码")
+    log_event(logging.INFO, "店铺锁定库存整理完成", stage="prepare", rows=len(prepared), join_key="统一规格编码")
     return prepared
 
 
@@ -589,6 +602,7 @@ def build_report(batch_df: pd.DataFrame, analytics_df: pd.DataFrame) -> pd.DataF
     log_event(
         logging.INFO,
         "批次库存与店铺锁定库存汇总完成",
+        stage="build_report",
         report_rows=len(report),
         matched_rows=int(matched_rows),
         unmatched_rows=int(unmatched_rows),
@@ -597,7 +611,7 @@ def build_report(batch_df: pd.DataFrame, analytics_df: pd.DataFrame) -> pd.DataF
     return report
 
 
-def save_report(df: pd.DataFrame, output_path: Path) -> None:
+def save_report(df: pd.DataFrame, output_path: Path, *, log_output_path: bool = True) -> None:
     output_path.parent.mkdir(parents=True, exist_ok=True)
     started_at = time.monotonic()
     suffix = output_path.suffix.lower()
@@ -605,13 +619,14 @@ def save_report(df: pd.DataFrame, output_path: Path) -> None:
         df.to_csv(output_path, index=False, encoding="utf-8-sig")
     else:
         df.to_excel(output_path, index=False)
-    log_event(
-        logging.INFO,
-        "报表输出完成",
-        output=output_path,
-        report_rows=len(df),
-        elapsed_ms=int((time.monotonic() - started_at) * 1000),
-    )
+    fields: dict[str, Any] = {
+        "report_rows": len(df),
+        "elapsed_ms": int((time.monotonic() - started_at) * 1000),
+    }
+    if log_output_path:
+        fields["output"] = output_path
+    fields["stage"] = "report_output"
+    log_event(logging.INFO, "报表输出完成", **fields)
 
 
 def default_output_path() -> Path:
@@ -636,23 +651,33 @@ def parse_args() -> argparse.Namespace:
         action="store_true",
         help="只生成本地报表，不同步到飞书",
     )
+    parser.add_argument(
+        "--skip-report-output",
+        action="store_true",
+        help="不生成本地报表文件，仅保留飞书同步等后续处理",
+    )
+    parser.add_argument(
+        "--safe-logs",
+        action="store_true",
+        help="隐藏环境文件路径、日志文件路径、飞书表ID等配置细节",
+    )
     return parser.parse_args()
 
 
 def main() -> int:
     configure_logging()
     started_at = time.monotonic()
+    try:
         args = parse_args()
-    output_path = Path(args.output).expanduser().resolve()
-    raw_config, env_path = load_raw_db_config()
+        safe_logs = args.safe_logs or os.getenv("INVENTORY_SAFE_LOGS") == "1"
+        output_path = None if args.skip_report_output else Path(args.output).expanduser().resolve()
+        report_output = "skipped" if output_path is None else "enabled"
+        log_event(logging.INFO, "批次库存统计任务启动", stage="start", report_output=report_output)
+
+        raw_config, _ = load_raw_db_config()
         analytics_config, _ = load_analytics_db_config()
-    log_event(
-        logging.INFO,
-        "批次库存统计任务启动",
-        env_path=env_path,
-        output=output_path,
-        log_file=LOG_FILE,
-    )
+        log_event(logging.INFO, "配置加载完成", stage="config")
+
         raw_df = query_dataframe(raw_config, RAW_BATCH_SQL, database_name="Raw")
         raw_codes = extract_raw_codes(raw_df)
         analytics_sql = build_analytics_lock_sql(raw_codes)
@@ -660,19 +685,33 @@ def main() -> int:
         filtered_batch_df = filter_batch_inventory(raw_df)
         prepared_analytics_df = prepare_analytics_stock(analytics_df)
         report_df = build_report(filtered_batch_df, prepared_analytics_df)
-    save_report(report_df, output_path)
+
+        if output_path is None:
+            log_event(logging.INFO, "已跳过本地报表输出", stage="report_output", report_rows=len(report_df))
+        else:
+            save_report(report_df, output_path, log_output_path=not safe_logs)
         if args.skip_feishu_sync:
-        log_event(logging.INFO, "已跳过飞书批次库存表同步", table_name=args.feishu_table)
+            log_event(logging.INFO, "已跳过飞书批次库存表同步", stage="feishu_sync", table_name=args.feishu_table)
         else:
             sync_report_to_feishu(report_df, table_name=args.feishu_table)
+
         log_event(
             logging.INFO,
             "批次库存统计任务完成",
-        output=output_path,
+            stage="finish",
             report_rows=len(report_df),
             elapsed_ms=int((time.monotonic() - started_at) * 1000),
         )
         return 0
+    except Exception as exc:  # noqa: BLE001
+        log_event(
+            logging.ERROR,
+            "批次库存统计任务失败",
+            stage="failed",
+            **exception_fields(exc),
+            elapsed_ms=int((time.monotonic() - started_at) * 1000),
+        )
+        return 1
 
 
 if __name__ == "__main__":

inventory_analytics/batch_upload/web.py

@@ -113,13 +113,8 @@ class RawDbConfig:
 
     @classmethod
     def from_env(cls) -> "RawDbConfig":
-        config, env_path = load_raw_db_config()
-        log_event(
-            logging.INFO,
-            "已加载Raw数据库配置",
-            env_path=env_path,
-            database=config.name,
-        )
+        config, _ = load_raw_db_config()
+        log_event(logging.INFO, "已加载Raw数据库配置")
         return cls(
             host=config.host,
             port=config.port,
@@ -129,11 +124,7 @@ class RawDbConfig:
         )
 
     def connect(self):
-        log_event(
-            logging.INFO,
-            "开始连接Raw数据库",
-            database=self.name,
-        )
+        log_event(logging.INFO, "开始连接Raw数据库")
         return psycopg2.connect(
             host=self.host,
             port=self.port,

inventory_analytics/channel_inventory_sync/entrypoint.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import logging
+import os
 from pathlib import Path
 import sys
 import time
 
 from .cli import main as channel_inventory_sync_main
 from ..logging_utils import configure_logging, log_event
+from ..logging_utils import exception_fields
 from ..paths import LOG_DIR
 
 
@@ -18,36 +20,43 @@ LOG_FIELD_LABELS = {
     "argv": "参数",
     "cwd": "工作目录",
     "elapsed_ms": "耗时毫秒",
+    "error": "错误",
+    "error_type": "异常类型",
     "exit_code": "退出码",
     "log_file": "日志文件",
+    "stage": "阶段",
 }
 
 
 def main() -> int:
     configure_logging(LOG_FILE, stream=sys.stderr)
     started_at = time.monotonic()
-    log_event(
-        logger,
-        LOG_FIELD_LABELS,
-        logging.INFO,
-        "飞书同步入口启动",
-        argv=sys.argv[1:],
-        cwd=Path.cwd(),
-        log_file=LOG_FILE,
-    )
+    start_fields = {"argv": sys.argv[1:], "stage": "start"}
+    if os.getenv("INVENTORY_SAFE_LOGS") != "1":
+        start_fields.update(cwd=Path.cwd(), log_file=LOG_FILE)
+    log_event(logger, LOG_FIELD_LABELS, logging.INFO, "飞书同步入口启动", **start_fields)
     try:
         exit_code = channel_inventory_sync_main()
     except SystemExit as exc:
         exit_code = int(exc.code or 0) if isinstance(exc.code, int) else 1
-    except Exception:
-        logger.exception("事件=%s", "飞书同步入口异常")
+    except Exception as exc:  # noqa: BLE001
+        log_event(
+            logger,
+            LOG_FIELD_LABELS,
+            logging.ERROR,
+            "渠道库存同步入口异常",
+            stage="failed",
+            **exception_fields(exc),
+            elapsed_ms=int((time.monotonic() - started_at) * 1000),
+        )
         return 1
 
     log_event(
         logger,
         LOG_FIELD_LABELS,
         logging.INFO if exit_code == 0 else logging.ERROR,
-        "飞书同步入口完成",
+        "渠道库存同步入口完成" if exit_code == 0 else "渠道库存同步入口失败",
+        stage="finish" if exit_code == 0 else "failed",
         exit_code=exit_code,
         elapsed_ms=int((time.monotonic() - started_at) * 1000),
     )

inventory_analytics/logging_utils.py

@@ -8,15 +8,65 @@ import sys
 from typing import Any
 
 
-SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
+SENSITIVE_KEYWORDS = (
+    "app_id",
+    "app_token",
+    "authorization",
+    "base_url",
+    "db_host",
+    "db_name",
+    "db_password",
+    "db_port",
+    "db_user",
+    "env_path",
+    "field_id",
+    "host",
+    "password",
+    "passwd",
+    "secret",
+    "selected_table_id",
+    "selected_view_id",
+    "table_id",
+    "token",
+    "url",
+    "user",
+)
+SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization|app_id|app_token)\s*[:=]\s*([^\s,;]+)")
+FEISHU_APP_PATH_PATTERN = re.compile(r"(/bitable/v1/apps/)[^/?\s]+")
+FEISHU_QUERY_TOKEN_PATTERN = re.compile(r"([?&]token=)[^&\s]+")
+
+
+def is_sensitive_key(key: str) -> bool:
+    normalized = key.lower()
+    return any(keyword in normalized for keyword in SENSITIVE_KEYWORDS)
 
 
 def sanitize_value(value: Any) -> Any:
+    if isinstance(value, Path):
+        return value.name
+    if isinstance(value, BaseException):
+        return sanitize_value(str(value))
+    if isinstance(value, dict):
+        return {
+            key: "***" if is_sensitive_key(str(key)) else sanitize_value(item)
+            for key, item in value.items()
+        }
+    if isinstance(value, (list, tuple, set)):
+        return [sanitize_value(item) for item in value]
     if isinstance(value, str):
-        return SENSITIVE_PATTERN.sub(r"\1=***", value)
+        value = SENSITIVE_PATTERN.sub(r"\1=***", value)
+        value = FEISHU_APP_PATH_PATTERN.sub(r"\1***", value)
+        return FEISHU_QUERY_TOKEN_PATTERN.sub(r"\1***", value)
     return value
 
 
+def exception_fields(exc: BaseException) -> dict[str, Any]:
+    return {
+        "error_type": exc.__class__.__name__,
+        "error": sanitize_value(str(exc)),
+    }
+
+
 def configure_logging(log_file: Path, *, stream = None, level: int = logging.INFO) -> None:
     log_file.parent.mkdir(parents=True, exist_ok=True)
     formatter = logging.Formatter(
@@ -37,7 +87,8 @@ def configure_logging(log_file: Path, *, stream = None, level: int = logging.INF
 def log_event(logger: logging.Logger, labels: dict[str, str], level: int, event: str, **fields: Any) -> None:
     safe_event = sanitize_value(event)
     details = " ".join(
-        f"{labels.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items()
+        f"{labels.get(key, key)}={sanitize_value(value)!r}"
+        for key, value in fields.items()
+        if value is not None and not is_sensitive_key(key)
     )
     logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "")
-

inventory_sync/cli.py

@@ -3,10 +3,14 @@ from __future__ import annotations
 import argparse
 import json
 import logging
-import re
+import os
 import time
 from typing import Any
 
+from inventory_analytics.logging_utils import exception_fields
+from inventory_analytics.logging_utils import log_event as emit_log_event
+from inventory_analytics.logging_utils import sanitize_value
+
 from .feishu import FeishuClient
 from .postgres import query_postgres
 from .service import run_feishu_summary, run_sync_brand
@@ -14,33 +18,24 @@ from .settings import build_app_config
 
 
 logger = logging.getLogger("inventory_sync.cli")
-SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
 
 LOG_FIELD_LABELS = {
     "action": "动作",
     "elapsed_ms": "耗时毫秒",
-    "env_path": "环境文件",
     "error": "错误",
+    "error_type": "异常类型",
     "fallback": "兜底方式",
+    "loaded": "已加载",
     "row_count": "行数",
     "source_row_count": "源数据行数",
+    "stage": "阶段",
     "table_count": "表数量",
     "target_count": "目标数量",
 }
 
 
-def sanitize_value(value: Any) -> Any:
-    if isinstance(value, str):
-        return SENSITIVE_PATTERN.sub(r"\1=***", value)
-    return value
-
-
 def log_event(level: int, event: str, **fields: Any) -> None:
-    safe_event = sanitize_value(event)
-    details = " ".join(
-        f"{LOG_FIELD_LABELS.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items()
-    )
-    logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "")
+    emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields)
 
 
 def parse_args() -> argparse.Namespace:
@@ -54,60 +49,90 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
+def safe_result_summary(result: dict[str, object]) -> dict[str, object]:
+    summary: dict[str, object] = {}
+    feishu_result = result.get("feishu")
+    if isinstance(feishu_result, dict):
+        summary["feishu"] = {"table_count": feishu_result.get("table_count")}
+    postgres_result = result.get("postgres")
+    if isinstance(postgres_result, dict):
+        summary["postgres"] = {"row_count": postgres_result.get("row_count")}
+    sync_result = result.get("sync_brand")
+    if isinstance(sync_result, dict):
+        summary["sync_brand"] = {
+            "target_count": sync_result.get("target_count"),
+            "source_row_count": sync_result.get("source_row_count"),
+        }
+    return summary
+
+
+def public_result(result: dict[str, object]) -> dict[str, object]:
+    return sanitize_value(result)
+
+
 def main() -> int:
     started_at = time.monotonic()
     try:
         args = parse_args()
-        log_event(logging.INFO, "飞书同步动作开始", action=args.action)
+        safe_logs = os.getenv("INVENTORY_SAFE_LOGS") == "1"
+        log_event(logging.INFO, "渠道库存同步任务启动", action=args.action, stage="start")
         config = build_app_config()
         if config.env_path:
-            log_event(logging.INFO, "已加载环境变量文件", env_path=config.env_path)
+            log_event(logging.INFO, "配置加载完成", loaded=True, stage="config")
         else:
-            log_event(logging.WARNING, "未找到环境变量文件", fallback="当前Shell环境变量")
+            log_event(logging.WARNING, "未找到环境变量文件，使用当前Shell环境变量", fallback="shell", stage="config")
 
         client = FeishuClient(config)
         result: dict[str, object] = {}
 
         if args.action in {"feishu", "all"}:
-            log_event(logging.INFO, "开始读取飞书Base信息")
+            log_event(logging.INFO, "开始读取飞书表信息", stage="feishu")
             result["feishu"] = run_feishu_summary(client)
             log_event(
                 logging.INFO,
-                "飞书Base信息读取完成",
+                "飞书表信息读取完成",
+                stage="feishu",
                 table_count=result["feishu"].get("table_count") if isinstance(result["feishu"], dict) else None,
             )
         if args.action in {"postgres", "all"}:
-            log_event(logging.INFO, "开始查询PostgreSQL")
+            log_event(logging.INFO, "开始查询渠道库存数据", stage="database")
             result["postgres"] = query_postgres(config)
             log_event(
                 logging.INFO,
-                "PostgreSQL查询完成",
+                "渠道库存数据查询完成",
+                stage="database",
                 row_count=result["postgres"].get("row_count") if isinstance(result["postgres"], dict) else None,
             )
         if args.action in {"sync-brand", "all"}:
-            log_event(logging.INFO, "开始同步品牌数据")
+            log_event(logging.INFO, "开始同步渠道库存到飞书", stage="sync")
             postgres_result = result.get("postgres")
             result["sync_brand"] = run_sync_brand(config, client, postgres_result if isinstance(postgres_result, dict) else None)
             log_event(
                 logging.INFO,
-                "品牌数据同步完成",
+                "渠道库存同步到飞书完成",
+                stage="sync",
                 target_count=result["sync_brand"].get("target_count") if isinstance(result["sync_brand"], dict) else None,
                 source_row_count=result["sync_brand"].get("source_row_count") if isinstance(result["sync_brand"], dict) else None,
             )
 
-        print(json.dumps(result, ensure_ascii=False, indent=2))
+        if safe_logs:
+            print(json.dumps(safe_result_summary(result), ensure_ascii=False, indent=2))
+        else:
+            print(json.dumps(public_result(result), ensure_ascii=False, indent=2))
         log_event(
             logging.INFO,
-            "飞书同步动作完成",
+            "渠道库存同步任务完成",
             action=args.action,
+            stage="finish",
             elapsed_ms=int((time.monotonic() - started_at) * 1000),
         )
         return 0
     except Exception as exc:  # noqa: BLE001
         log_event(
             logging.ERROR,
-            "飞书同步动作失败",
-            error=exc,
+            "渠道库存同步任务失败",
+            stage="failed",
+            **exception_fields(exc),
             elapsed_ms=int((time.monotonic() - started_at) * 1000),
         )
         return 1

inventory_sync/service.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import logging
-import re
 from typing import Any
 
+from inventory_analytics.logging_utils import log_event as emit_log_event
+
 from .feishu import FeishuApiError, FeishuClient
 from .postgres import query_postgres
 from .settings import AppConfig
@@ -11,7 +12,6 @@ from .specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME, PRIMARY_
 
 
 logger = logging.getLogger("inventory_sync.service")
-SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
 
 LOG_FIELD_LABELS = {
     "advanced_permission_enabled": "高级权限已开启",
@@ -20,26 +20,16 @@ LOG_FIELD_LABELS = {
     "deleted_records": "删除记录数",
     "existing_records": "已有记录数",
     "rows": "行数",
-    "selected_table_id": "选中表ID",
     "source_rows": "源数据行数",
+    "stage": "阶段",
     "table_count": "表数量",
     "table_name": "表名",
     "target_count": "目标数量",
 }
 
 
-def sanitize_value(value: Any) -> Any:
-    if isinstance(value, str):
-        return SENSITIVE_PATTERN.sub(r"\1=***", value)
-    return value
-
-
 def log_event(level: int, event: str, **fields: Any) -> None:
-    safe_event = sanitize_value(event)
-    details = " ".join(
-        f"{LOG_FIELD_LABELS.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items()
-    )
-    logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "")
+    emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields)
 
 
 def build_feishu_record(row: dict[str, Any]) -> dict[str, Any]:
@@ -64,34 +54,24 @@ def build_feishu_record(row: dict[str, Any]) -> dict[str, Any]:
 
 
 def run_feishu_summary(client: FeishuClient) -> dict[str, Any]:
-    log_event(logging.INFO, "开始获取飞书租户访问令牌")
+    log_event(logging.INFO, "开始获取飞书访问令牌", stage="feishu_auth")
     access_token = client.get_tenant_access_token()
-    log_event(logging.INFO, "飞书租户访问令牌获取完成")
+    log_event(logging.INFO, "飞书访问令牌获取完成", stage="feishu_auth")
     base_info = client.parse_base_info(client.build_base_url(), access_token)
-    log_event(logging.INFO, "飞书Base信息解析完成", selected_table_id=base_info["selected_table_id"])
+    log_event(logging.INFO, "飞书Base信息解析完成", stage="feishu_base")
     advanced_permission_enabled = client.enable_advanced_permission(access_token, base_info["app_token"] or "")
     tables = client.list_tables(access_token, base_info["app_token"] or "")
     log_event(
         logging.INFO,
         "飞书Base表列表读取完成",
+        stage="feishu_tables",
         table_count=len(tables),
         advanced_permission_enabled=advanced_permission_enabled,
     )
     return {
-        "base_url": base_info["base_url"],
-        "app_token": base_info["app_token"],
         "auth_mode": "tenant_access_token",
-        "selected_table_id": base_info["selected_table_id"],
-        "selected_view_id": base_info["selected_view_id"],
         "advanced_permission_enabled": advanced_permission_enabled,
-        "folder": {
-            "name": base_info["app_token"],
-            "type": "base",
-            "tables": [
-                {"table_id": table.get("table_id"), "name": table.get("name"), "revision": table.get("revision")}
-                for table in tables
-            ],
-        },
+        "tables": [{"name": table.get("name"), "revision": table.get("revision")} for table in tables],
         "table_count": len(tables),
     }
 
@@ -107,6 +87,7 @@ def sync_brand_target(
     log_event(
         logging.INFO,
         "开始同步目标飞书表",
+        stage="sync_target",
         brand_name=target.brand_name,
         table_name=target.table_name,
         source_rows=len(postgres_rows or []),
@@ -128,6 +109,7 @@ def sync_brand_target(
     log_event(
         logging.INFO,
         "目标飞书表已有记录读取完成",
+        stage="sync_target",
         brand_name=target.brand_name,
         table_name=target.table_name,
         existing_records=len(existing_record_ids),
@@ -142,6 +124,7 @@ def sync_brand_target(
     log_event(
         logging.INFO,
         "目标飞书表同步完成",
+        stage="sync_target",
         brand_name=target.brand_name,
         table_name=target.table_name,
         deleted_records=deleted_count,
@@ -162,13 +145,13 @@ def sync_brand_target(
 
 
 def run_sync_brand(config: AppConfig, client: FeishuClient, postgres_result: dict[str, Any] | None = None) -> dict[str, Any]:
-    log_event(logging.INFO, "开始准备品牌同步", target_count=len(BRAND_SYNC_TARGETS))
+    log_event(logging.INFO, "开始准备品牌同步", stage="sync_prepare", target_count=len(BRAND_SYNC_TARGETS))
     access_token = client.get_tenant_access_token()
     base_info = client.parse_base_info(client.build_base_url(), access_token)
     app_token = base_info["app_token"] or ""
     client.enable_advanced_permission(access_token, app_token)
     tables = client.list_tables(access_token, app_token)
-    log_event(logging.INFO, "品牌同步准备完成", table_count=len(tables))
+    log_event(logging.INFO, "品牌同步准备完成", stage="sync_prepare", table_count=len(tables))
 
     results = []
     source_row_count = 0
@@ -181,6 +164,7 @@ def run_sync_brand(config: AppConfig, client: FeishuClient, postgres_result: dic
         log_event(
             logging.INFO,
             "目标品牌PostgreSQL数据已准备",
+            stage="sync_prepare",
             brand_name=target.brand_name,
             table_name=target.table_name,
             rows=len(postgres_rows or []),

prefect.yaml

@@ -6,7 +6,7 @@ push: []
 pull: []
 
 deployments:
-  - name: inventory-channel-inventory-sync-daily-0915
+  - name: channel-inventory-sync-0915
     version: "1.0.0"
     description: "每天 09:15 执行渠道库存统计并写入飞书多维表格。"
     entrypoint: prefect_inventory_scheduler.py:inventory_channel_inventory_sync_flow
@@ -22,14 +22,15 @@ deployments:
     work_pool:
       name: inventory-analytics-pool
 
-  - name: inventory-batch-inventory-report-daily-0915
+  - name: batch-inventory-repor-0915
     version: "1.0.0"
-    description: "每天 09:15 生成批次库存统计报表并写入飞书多维表格。"
+    description: "每天 09:15 生成批次库存统计数据并写入飞书多维表格，不输出本地报表文件。"
     entrypoint: prefect_inventory_scheduler.py:inventory_batch_inventory_report_flow
     parameters:
       output: ""
       feishu_table: "批次库存"
       skip_feishu_sync: false
+      skip_report_output: true
     tags:
       - inventory
       - feishu

prefect_inventory_scheduler.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import argparse
+import logging
 import os
 import sys
 from pathlib import Path
+import time
 
 
 def bootstrap_prefect_env() -> None:
@@ -22,28 +24,65 @@ def bootstrap_prefect_env() -> None:
 
 bootstrap_prefect_env()
 
-from prefect import flow, task
+from prefect import flow, get_run_logger, task
 
+from inventory_analytics.logging_utils import exception_fields
 from inventory_analytics.channel_inventory_sync.entrypoint import main as channel_inventory_sync_main
 from inventory_analytics.batch_inventory_report.cli import main as batch_inventory_report_main
 
 
 @task(name="run-channel-inventory-sync")
 def run_channel_inventory_sync_task() -> None:
+    logger = get_run_logger()
+    started_at = time.monotonic()
+    logger.info("渠道库存统计开始，action=all，安全日志模式=enabled")
     original_argv = sys.argv[:]
+    original_safe_logs = os.environ.get("INVENTORY_SAFE_LOGS")
+    root_logger = logging.getLogger()
+    original_handlers = root_logger.handlers[:]
+    original_level = root_logger.level
     try:
+        os.environ["INVENTORY_SAFE_LOGS"] = "1"
         sys.argv = ["sync_channel_inventory.py", "--action", "all"]
         exit_code = channel_inventory_sync_main()
+    except Exception as exc:  # noqa: BLE001
+        logger.error("渠道库存统计异常，%s", exception_fields(exc))
+        raise
     finally:
         sys.argv = original_argv
+        if original_safe_logs is None:
+            os.environ.pop("INVENTORY_SAFE_LOGS", None)
+        else:
+            os.environ["INVENTORY_SAFE_LOGS"] = original_safe_logs
+        root_logger.handlers = original_handlers
+        root_logger.setLevel(original_level)
     if exit_code != 0:
+        logger.error("渠道库存统计失败，退出码=%s", exit_code)
         raise RuntimeError(f"渠道库存统计执行失败，退出码: {exit_code}")
+    logger.info("渠道库存统计完成，耗时毫秒=%s", int((time.monotonic() - started_at) * 1000))
 
 
 @task(name="run-batch-inventory-report")
-def run_batch_inventory_report_task(output: str = "", feishu_table: str = "批次库存", skip_feishu_sync: bool = False) -> None:
+def run_batch_inventory_report_task(
+    output: str = "",
+    feishu_table: str = "批次库存",
+    skip_feishu_sync: bool = False,
+    skip_report_output: bool = True,
+) -> None:
+    logger = get_run_logger()
+    started_at = time.monotonic()
+    logger.info(
+        "批次库存统计开始，飞书同步=%s，本地报表输出=%s，安全日志模式=enabled",
+        "disabled" if skip_feishu_sync else "enabled",
+        "skipped" if skip_report_output else "enabled",
+    )
     original_argv = sys.argv[:]
+    original_safe_logs = os.environ.get("INVENTORY_SAFE_LOGS")
+    root_logger = logging.getLogger()
+    original_handlers = root_logger.handlers[:]
+    original_level = root_logger.level
     try:
+        os.environ["INVENTORY_SAFE_LOGS"] = "1"
         sys.argv = ["generate_batch_inventory_report.py"]
         if output:
             sys.argv.extend(["--output", output])
@@ -51,11 +90,25 @@ def run_batch_inventory_report_task(output: str = "", feishu_table: str = "批
             sys.argv.extend(["--feishu-table", feishu_table])
         if skip_feishu_sync:
             sys.argv.append("--skip-feishu-sync")
+        if skip_report_output:
+            sys.argv.append("--skip-report-output")
+        sys.argv.append("--safe-logs")
         exit_code = batch_inventory_report_main()
+    except Exception as exc:  # noqa: BLE001
+        logger.error("批次库存统计异常，%s", exception_fields(exc))
+        raise
     finally:
         sys.argv = original_argv
+        if original_safe_logs is None:
+            os.environ.pop("INVENTORY_SAFE_LOGS", None)
+        else:
+            os.environ["INVENTORY_SAFE_LOGS"] = original_safe_logs
+        root_logger.handlers = original_handlers
+        root_logger.setLevel(original_level)
     if exit_code != 0:
+        logger.error("批次库存统计失败，退出码=%s", exit_code)
         raise RuntimeError(f"批次库存统计报表执行失败，退出码: {exit_code}")
+    logger.info("批次库存统计完成，耗时毫秒=%s", int((time.monotonic() - started_at) * 1000))
 
 
 @flow(name="inventory-channel-inventory-sync")
@@ -68,8 +121,14 @@ def inventory_batch_inventory_report_flow(
     output: str = "",
     feishu_table: str = "批次库存",
     skip_feishu_sync: bool = False,
+    skip_report_output: bool = True,
 ) -> None:
-    run_batch_inventory_report_task(output=output, feishu_table=feishu_table, skip_feishu_sync=skip_feishu_sync)
+    run_batch_inventory_report_task(
+        output=output,
+        feishu_table=feishu_table,
+        skip_feishu_sync=skip_feishu_sync,
+        skip_report_output=skip_report_output,
+    )
 
 
 def parse_args() -> argparse.Namespace:
@@ -81,6 +140,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--output", default="", help="批次库存统计输出路径，可选")
     parser.add_argument("--feishu-table", default="批次库存", help="批次库存统计同步目标飞书表名")
     parser.add_argument("--skip-feishu-sync", action="store_true", help="批次库存统计只生成本地文件，不写飞书")
+    parser.add_argument("--write-report-output", action="store_true", help="Prefect 执行批次库存统计时同时生成本地报表文件")
     return parser.parse_args()
 
 
@@ -90,17 +150,18 @@ def main() -> None:
         "output": args.output,
         "feishu_table": args.feishu_table,
         "skip_feishu_sync": args.skip_feishu_sync,
+        "skip_report_output": not args.write_report_output,
     }
     if args.serve_channel_inventory_sync:
         inventory_channel_inventory_sync_flow.serve(
-            name="inventory-channel-inventory-sync-daily-0915",
+            name="channel-inventory-sync-0915",
             cron="15 9 * * *",
             tags=["inventory", "feishu", "channel-inventory-sync"],
         )
         return
     if args.serve_batch_inventory_report:
         inventory_batch_inventory_report_flow.serve(
-            name="inventory-batch-inventory-report-daily-0915",
+            name="batch-inventory-repor-0915",
             cron="15 9 * * *",
             parameters=flow_kwargs,
             tags=["inventory", "feishu", "batch-inventory-report"],