From 02b34700753b1acf4a52fcc759bb08bfcdbbc427 Mon Sep 17 00:00:00 2001 From: BXY0414 <1079823339@qq.com> Date: Wed, 29 Apr 2026 16:28:23 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=20=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E8=BE=93=E5=87=BA=E6=A0=BC=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../batch_inventory_report/service.py | 135 +++++++++++------- inventory_analytics/batch_upload/web.py | 15 +- .../channel_inventory_sync/entrypoint.py | 33 +++-- inventory_analytics/logging_utils.py | 59 +++++++- inventory_sync/cli.py | 79 ++++++---- inventory_sync/service.py | 46 ++---- prefect.yaml | 7 +- prefect_inventory_scheduler.py | 71 ++++++++- 8 files changed, 303 insertions(+), 142 deletions(-) diff --git a/inventory_analytics/batch_inventory_report/service.py b/inventory_analytics/batch_inventory_report/service.py index 4c06932..a1e24cd 100644 --- a/inventory_analytics/batch_inventory_report/service.py +++ b/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,43 +651,67 @@ 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() - args = parse_args() - output_path = Path(args.output).expanduser().resolve() - raw_config, env_path = 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, - ) - 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) - analytics_df = query_dataframe(analytics_config, analytics_sql, database_name="Analytics") - 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 args.skip_feishu_sync: - log_event(logging.INFO, "已跳过飞书批次库存表同步", table_name=args.feishu_table) - else: - sync_report_to_feishu(report_df, table_name=args.feishu_table) - log_event( - logging.INFO, - "批次库存统计任务完成", - output=output_path, - report_rows=len(report_df), - elapsed_ms=int((time.monotonic() - started_at) * 1000), - ) - return 0 + try: + args = parse_args() + 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, "配置加载完成", 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) + analytics_df = query_dataframe(analytics_config, analytics_sql, database_name="Analytics") + 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) + + 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, "已跳过飞书批次库存表同步", stage="feishu_sync", table_name=args.feishu_table) + else: + sync_report_to_feishu(report_df, table_name=args.feishu_table) + + log_event( + logging.INFO, + "批次库存统计任务完成", + 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__": diff --git a/inventory_analytics/batch_upload/web.py b/inventory_analytics/batch_upload/web.py index 6ef96a8..a5a555c 100644 --- a/inventory_analytics/batch_upload/web.py +++ b/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, diff --git a/inventory_analytics/channel_inventory_sync/entrypoint.py b/inventory_analytics/channel_inventory_sync/entrypoint.py index ae8699d..4ee62e7 100644 --- a/inventory_analytics/channel_inventory_sync/entrypoint.py +++ b/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), ) diff --git a/inventory_analytics/logging_utils.py b/inventory_analytics/logging_utils.py index 5c9f485..9b96f1f 100644 --- a/inventory_analytics/logging_utils.py +++ b/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 "") - diff --git a/inventory_sync/cli.py b/inventory_sync/cli.py index 16cebaa..203c0af 100644 --- a/inventory_sync/cli.py +++ b/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 diff --git a/inventory_sync/service.py b/inventory_sync/service.py index 1c6f0f0..bd121db 100644 --- a/inventory_sync/service.py +++ b/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 []), diff --git a/prefect.yaml b/prefect.yaml index 16cdb1d..59ae39c 100644 --- a/prefect.yaml +++ b/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 diff --git a/prefect_inventory_scheduler.py b/prefect_inventory_scheduler.py index 7aeae27..7229647 100644 --- a/prefect_inventory_scheduler.py +++ b/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"],