diff --git a/inventory_analytics/batch_inventory_report/service.py b/inventory_analytics/batch_inventory_report/service.py index a1e24cd..cebcf0a 100644 --- a/inventory_analytics/batch_inventory_report/service.py +++ b/inventory_analytics/batch_inventory_report/service.py @@ -18,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 configure_logging_with_extra_handler from ..logging_utils import exception_fields from ..logging_utils import log_event as emit_log_event from ..paths import LOG_DIR, REPORT_DIR @@ -212,13 +213,38 @@ LOG_FIELD_LABELS = { } def configure_logging() -> None: - configure_app_logging(LOG_FILE) + if os.getenv("INVENTORY_PREFECT_LOG_BRIDGE") == "1": + handler = PrefectLogHandler() + handler.setFormatter( + logging.Formatter( + fmt="%(asctime)s %(levelname)s %(name)s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + ) + configure_logging_with_extra_handler(LOG_FILE, handler) + else: + configure_app_logging(LOG_FILE) def log_event(level: int, event: str, **fields: Any) -> None: emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields) +class PrefectLogHandler(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + try: + from prefect.logging import get_run_logger + + prefect_logger = get_run_logger() + except Exception: + return + message = self.format(record) + log_method = getattr(prefect_logger, record.levelname.lower(), None) + if log_method is None: + log_method = prefect_logger.info + log_method(message) + + @dataclass(frozen=True) class FeishuFieldSpec: field_name: str # 飞书字段名 diff --git a/inventory_analytics/channel_inventory_sync/entrypoint.py b/inventory_analytics/channel_inventory_sync/entrypoint.py index 4ee62e7..9bd6616 100644 --- a/inventory_analytics/channel_inventory_sync/entrypoint.py +++ b/inventory_analytics/channel_inventory_sync/entrypoint.py @@ -8,6 +8,7 @@ import time from .cli import main as channel_inventory_sync_main from ..logging_utils import configure_logging, log_event +from ..logging_utils import configure_logging_with_extra_handler from ..logging_utils import exception_fields from ..paths import LOG_DIR @@ -28,8 +29,33 @@ LOG_FIELD_LABELS = { } +class PrefectLogHandler(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + try: + from prefect.logging import get_run_logger + + prefect_logger = get_run_logger() + except Exception: + return + message = self.format(record) + log_method = getattr(prefect_logger, record.levelname.lower(), None) + if log_method is None: + log_method = prefect_logger.info + log_method(message) + + def main() -> int: - configure_logging(LOG_FILE, stream=sys.stderr) + if os.getenv("INVENTORY_PREFECT_LOG_BRIDGE") == "1": + handler = PrefectLogHandler() + handler.setFormatter( + logging.Formatter( + fmt="%(asctime)s %(levelname)s %(name)s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + ) + configure_logging_with_extra_handler(LOG_FILE, handler, stream=sys.stderr) + else: + configure_logging(LOG_FILE, stream=sys.stderr) started_at = time.monotonic() start_fields = {"argv": sys.argv[1:], "stage": "start"} if os.getenv("INVENTORY_SAFE_LOGS") != "1": diff --git a/inventory_analytics/logging_utils.py b/inventory_analytics/logging_utils.py index 9b96f1f..7e81c1c 100644 --- a/inventory_analytics/logging_utils.py +++ b/inventory_analytics/logging_utils.py @@ -84,6 +84,18 @@ def configure_logging(log_file: Path, *, stream = None, level: int = logging.INF root_logger.addHandler(file_handler) +def configure_logging_with_extra_handler( + log_file: Path, + extra_handler: logging.Handler, + *, + stream = None, + level: int = logging.INFO, +) -> None: + configure_logging(log_file, stream=stream, level=level) + extra_handler.setLevel(level) + logging.getLogger().addHandler(extra_handler) + + def log_event(logger: logging.Logger, labels: dict[str, str], level: int, event: str, **fields: Any) -> None: safe_event = sanitize_value(event) details = " ".join( diff --git a/inventory_sync/postgres.py b/inventory_sync/postgres.py index e83981f..02ff049 100644 --- a/inventory_sync/postgres.py +++ b/inventory_sync/postgres.py @@ -2,21 +2,41 @@ from __future__ import annotations import csv import io +import logging +import time from typing import Any import psycopg2 +from inventory_analytics.logging_utils import log_event as emit_log_event + from .settings import AppConfig from .specs import BrandSyncTarget from .sql import DEFAULT_SYNC_TARGET, POSTGRES_QUERY, build_postgres_query +logger = logging.getLogger("inventory_sync.postgres") + +LOG_FIELD_LABELS = { + "elapsed_ms": "耗时毫秒", + "row_count": "行数", + "stage": "阶段", + "statement_timeout_ms": "查询超时毫秒", +} + + +def log_event(level: int, event: str, **fields: Any) -> None: + emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields) + + def query_postgres( config: AppConfig, brand_name: str | None = None, warehouse_keys: tuple[int | str, ...] | None = None, target: BrandSyncTarget | None = None, ) -> dict[str, Any]: + started_at = time.monotonic() + statement_timeout_ms = 300_000 active_brand_name = brand_name or (target.brand_name if target else DEFAULT_SYNC_TARGET.brand_name) active_warehouse_keys = warehouse_keys or (target.warehouse_keys if target else DEFAULT_SYNC_TARGET.warehouse_keys) query = ( @@ -29,19 +49,49 @@ def query_postgres( ) output = io.StringIO() try: + log_event(logging.INFO, "开始连接PostgreSQL", stage="database_connect") with psycopg2.connect( host=config.analytics_db_host, port=config.analytics_db_port, user=config.analytics_db_user, password=config.analytics_db_password, dbname=config.analytics_db_name, + connect_timeout=10, ) as conn: + log_event( + logging.INFO, + "PostgreSQL连接完成", + stage="database_connect", + elapsed_ms=int((time.monotonic() - started_at) * 1000), + ) with conn.cursor() as cur: cur.execute("SET TIME ZONE 'Asia/Shanghai';") + cur.execute("SET statement_timeout = %s;", (statement_timeout_ms,)) + copy_started_at = time.monotonic() + log_event( + logging.INFO, + "开始执行渠道库存SQL", + stage="database_query", + statement_timeout_ms=statement_timeout_ms, + ) cur.copy_expert(f"COPY ({query}) TO STDOUT WITH CSV HEADER", output) + log_event( + logging.INFO, + "渠道库存SQL执行完成", + stage="database_query", + elapsed_ms=int((time.monotonic() - copy_started_at) * 1000), + ) except psycopg2.Error as exc: raise RuntimeError(f"PostgreSQL 查询失败: {exc}") from exc + parse_started_at = time.monotonic() output.seek(0) rows = list(csv.DictReader(output)) + log_event( + logging.INFO, + "渠道库存查询结果解析完成", + stage="database_parse", + row_count=len(rows), + elapsed_ms=int((time.monotonic() - parse_started_at) * 1000), + ) return {"row_count": len(rows), "rows": rows} diff --git a/prefect_inventory_scheduler.py b/prefect_inventory_scheduler.py index 8bc5adf..e4c56af 100644 --- a/prefect_inventory_scheduler.py +++ b/prefect_inventory_scheduler.py @@ -50,11 +50,13 @@ def run_channel_inventory_sync_task() -> None: logger.info("渠道库存统计开始，action=all，安全日志模式=enabled") original_argv = sys.argv[:] original_safe_logs = os.environ.get("INVENTORY_SAFE_LOGS") + original_prefect_log_bridge = os.environ.get("INVENTORY_PREFECT_LOG_BRIDGE") root_logger = logging.getLogger() original_handlers = root_logger.handlers[:] original_level = root_logger.level try: os.environ["INVENTORY_SAFE_LOGS"] = "1" + os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = "1" sys.argv = ["sync_channel_inventory.py", "--action", "all"] exit_code = channel_inventory_sync_main() except Exception as exc: # noqa: BLE001 @@ -66,6 +68,10 @@ def run_channel_inventory_sync_task() -> None: os.environ.pop("INVENTORY_SAFE_LOGS", None) else: os.environ["INVENTORY_SAFE_LOGS"] = original_safe_logs + if original_prefect_log_bridge is None: + os.environ.pop("INVENTORY_PREFECT_LOG_BRIDGE", None) + else: + os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = original_prefect_log_bridge root_logger.handlers = original_handlers root_logger.setLevel(original_level) if exit_code != 0: @@ -91,11 +97,13 @@ def run_batch_inventory_report_task( ) original_argv = sys.argv[:] original_safe_logs = os.environ.get("INVENTORY_SAFE_LOGS") + original_prefect_log_bridge = os.environ.get("INVENTORY_PREFECT_LOG_BRIDGE") root_logger = logging.getLogger() original_handlers = root_logger.handlers[:] original_level = root_logger.level try: os.environ["INVENTORY_SAFE_LOGS"] = "1" + os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = "1" sys.argv = ["generate_batch_inventory_report.py"] if output: sys.argv.extend(["--output", output]) @@ -116,6 +124,10 @@ def run_batch_inventory_report_task( os.environ.pop("INVENTORY_SAFE_LOGS", None) else: os.environ["INVENTORY_SAFE_LOGS"] = original_safe_logs + if original_prefect_log_bridge is None: + os.environ.pop("INVENTORY_PREFECT_LOG_BRIDGE", None) + else: + os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = original_prefect_log_bridge root_logger.handlers = original_handlers root_logger.setLevel(original_level) if exit_code != 0: