修复 prefect中psql报错

inventory_sync/postgres.py

@@ -2,10 +2,10 @@ from __future__ import annotations
 
 import csv
 import io
-import os
-import subprocess
 from typing import Any
 
+import psycopg2
+
 from .settings import AppConfig
 from .specs import BrandSyncTarget
 from .sql import DEFAULT_SYNC_TARGET, POSTGRES_QUERY, build_postgres_query
@@ -27,40 +27,21 @@ def query_postgres(
             warehouse_keys=active_warehouse_keys,
         )
     )
-    env = os.environ.copy()
-    env["PGPASSWORD"] = config.analytics_db_password
-    command = [
-        "/usr/local/bin/psql",
-        "-h",
-        config.analytics_db_host,
-        "-p",
-        config.analytics_db_port,
-        "-U",
-        config.analytics_db_user,
-        "-d",
-        config.analytics_db_name,
-        "-v",
-        "ON_ERROR_STOP=1",
-        "-c",
-        "SET TIME ZONE 'Asia/Shanghai';",
-        "-c",
-        f"COPY ({query}) TO STDOUT WITH CSV HEADER",
-    ]
-
+    output = io.StringIO()
     try:
-        completed = subprocess.run(
-            command,
-            check=True,
-            capture_output=True,
-            text=True,
-            env=env,
-        )
-    except subprocess.CalledProcessError as exc:
-        detail = exc.stderr.strip() or exc.stdout.strip()
-        raise RuntimeError(f"PostgreSQL 查询失败: {detail}") from exc
+        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,
+        ) as conn:
+            with conn.cursor() as cur:
+                cur.execute("SET TIME ZONE 'Asia/Shanghai';")
+                cur.copy_expert(f"COPY ({query}) TO STDOUT WITH CSV HEADER", output)
+    except psycopg2.Error as exc:
+        raise RuntimeError(f"PostgreSQL 查询失败: {exc}") from exc
 
-    cleaned_output = "\n".join(
-        line for line in completed.stdout.splitlines() if line.strip() and line.strip() != "SET"
-    )
-    rows = list(csv.DictReader(io.StringIO(cleaned_output)))
+    output.seek(0)
+    rows = list(csv.DictReader(output))
     return {"row_count": len(rows), "rows": rows}

prefect_inventory_scheduler.py

@@ -28,7 +28,19 @@ 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.channel_inventory_sync.entrypoint import LOG_FILE as CHANNEL_INVENTORY_SYNC_LOG_FILE
 from inventory_analytics.batch_inventory_report.cli import main as batch_inventory_report_main
+from inventory_analytics.batch_inventory_report.service import LOG_FILE as BATCH_INVENTORY_REPORT_LOG_FILE
+
+
+def read_log_tail(log_file: Path, line_count: int = 80) -> str:
+    try:
+        lines = log_file.read_text(encoding="utf-8", errors="replace").splitlines()
+    except FileNotFoundError:
+        return f"日志文件不存在: {log_file.name}"
+    except Exception as exc:  # noqa: BLE001
+        return f"读取日志文件失败: {exc.__class__.__name__}: {exc}"
+    return "\n".join(lines[-line_count:]) if lines else f"日志文件为空: {log_file.name}"
 
 
 @task(name="run-channel-inventory-sync")
@@ -58,6 +70,7 @@ def run_channel_inventory_sync_task() -> None:
         root_logger.setLevel(original_level)
     if exit_code != 0:
         logger.error("渠道库存统计失败，退出码=%s", exit_code)
+        logger.error("渠道库存统计日志尾部:\n%s", read_log_tail(CHANNEL_INVENTORY_SYNC_LOG_FILE))
         raise RuntimeError(f"渠道库存统计执行失败，退出码: {exit_code}")
     logger.info("渠道库存统计完成，耗时毫秒=%s", int((time.monotonic() - started_at) * 1000))
 
@@ -107,6 +120,7 @@ def run_batch_inventory_report_task(
         root_logger.setLevel(original_level)
     if exit_code != 0:
         logger.error("批次库存统计失败，退出码=%s", exit_code)
+        logger.error("批次库存统计日志尾部:\n%s", read_log_tail(BATCH_INVENTORY_REPORT_LOG_FILE))
         raise RuntimeError(f"批次库存统计报表执行失败，退出码: {exit_code}")
     logger.info("批次库存统计完成，耗时毫秒=%s", int((time.monotonic() - started_at) * 1000))