from __future__ import annotations import csv import io import os import subprocess from typing import Any from .settings import AppConfig from .sql import POSTGRES_QUERY def query_postgres(config: AppConfig) -> dict[str, Any]: 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 ({POSTGRES_QUERY}) TO STDOUT WITH CSV HEADER", ] 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 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))) return {"row_count": len(rows), "rows": rows}