代码修改说明

• 用标准库 logging 替换所有 print，统一输出到文件，日志级别默认 DEBUG。
• 为关键路径添加性能监控，记录每个阶段的耗时（毫秒）：read_csv、transform、aggregate、write_json 以及整体流水线。
• 在可能失败的I/O操作（读CSV、写JSON）中加入异常日志（logger.exception）并继续抛出，保证行为与原逻辑一致但可观测性更强。
• 对数据质量问题（transform 时的坏行）记录 WARNING 级别日志，并附带简化后的行内容与异常栈，避免敏感信息泄露。
• 统一详细（detailed）日志格式，包含时间、级别、进程/线程、模块、函数、行号等上下文信息。
• 保持原有业务逻辑和返回值不变，日志记录避免额外的性能开销（采用惰性格式化、限制大对象输出预览）。

增强后的完整代码

import csv
import json
import logging
import time
from pathlib import Path
from statistics import mean


logger = logging.getLogger(__name__)


def configure_logging(log_file: str = "logs/app.log", level: int = logging.DEBUG) -> None:
    """
    基础文件日志配置：DEBUG 级别、详细格式、仅文件输出。
    """
    # 避免重复添加 handler（例如被多次调用或被导入执行）
    root = logging.getLogger()
    if root.handlers:
        return

    Path(log_file).parent.mkdir(parents=True, exist_ok=True)

    fmt = (
        "%(asctime)s | %(levelname)s | pid=%(process)d tid=%(thread)d | "
        "%(name)s:%(funcName)s:%(lineno)d | %(message)s"
    )
    datefmt = "%Y-%m-%d %H:%M:%S"

    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    file_handler.setLevel(level)
    file_handler.setFormatter(logging.Formatter(fmt=fmt, datefmt=datefmt))

    root.setLevel(level)
    root.addHandler(file_handler)


def _elapsed_ms(start: float) -> float:
    return (time.perf_counter() - start) * 1000.0


def read_csv(path):
    t0 = time.perf_counter()
    logger.debug("Reading CSV file: %s", path)
    rows = []
    try:
        with open(path, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                rows.append(row)
        logger.info("Loaded rows: %d (%.2f ms)", len(rows), _elapsed_ms(t0))
        return rows
    except Exception:
        logger.exception("Failed to read CSV: %s (%.2f ms)", path, _elapsed_ms(t0))
        raise


def transform(rows):
    t0 = time.perf_counter()
    logger.debug("Start transform with %d rows", len(rows))
    converted = []
    error_count = 0
    for r in rows:
        try:
            amount = float((r.get("amount", "0").strip()) or 0)
            category = (r.get("category") or "unknown").strip().lower()
            converted.append({"category": category, "amount": amount})
        except Exception:
            # 仅记录必要字段，避免潜在敏感信息泄漏
            sample = {"category": r.get("category"), "amount": r.get("amount")}
            logger.warning("Bad row skipped: %s", sample, exc_info=True)
            error_count += 1
    logger.info(
        "Transform done: in=%d out=%d errors=%d (%.2f ms)",
        len(rows),
        len(converted),
        error_count,
        _elapsed_ms(t0),
    )
    return converted


def aggregate(rows):
    t0 = time.perf_counter()
    logger.debug("Start aggregate with %d rows", len(rows))
    sums = {}
    for r in rows:
        sums.setdefault(r["category"], []).append(r["amount"])
    result = [
        {"category": k, "total": sum(v), "avg": round(mean(v), 2)}
        for k, v in sums.items()
    ]
    preview = result if len(result) <= 5 else result[:5]
    logger.info(
        "Aggregate done: categories=%d out=%d (%.2f ms) preview=%s%s",
        len(sums),
        len(result),
        _elapsed_ms(t0),
        preview,
        " ... (truncated)" if len(result) > 5 else "",
    )
    return result


def write_json(data, out_path):
    t0 = time.perf_counter()
    logger.debug("Writing JSON file: %s", out_path)
    try:
        out_path = Path(out_path)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        with open(out_path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        size = out_path.stat().st_size if out_path.exists() else 0
        items = len(data) if isinstance(data, list) else 1
        logger.info(
            "Wrote JSON: items=%d size=%d bytes path=%s (%.2f ms)",
            items,
            size,
            str(out_path),
            _elapsed_ms(t0),
        )
    except Exception:
        logger.exception("Failed to write JSON: %s (%.2f ms)", out_path, _elapsed_ms(t0))
        raise


def main():
    configure_logging(log_file="logs/app.log", level=logging.DEBUG)

    pipeline_t0 = time.perf_counter()
    src = "data/sales.csv"
    dst = "build/sales_summary.json"
    logger.info("Pipeline started: src=%s dst=%s", src, dst)

    step_t0 = time.perf_counter()
    rows = read_csv(src)
    logger.debug("Step read_csv took %.2f ms", _elapsed_ms(step_t0))

    step_t0 = time.perf_counter()
    rows = transform(rows)
    logger.debug("Step transform took %.2f ms", _elapsed_ms(step_t0))

    step_t0 = time.perf_counter()
    report = aggregate(rows)
    logger.debug("Step aggregate took %.2f ms", _elapsed_ms(step_t0))

    step_t0 = time.perf_counter()
    write_json(report, dst)
    logger.debug("Step write_json took %.2f ms", _elapsed_ms(step_t0))

    logger.info("Pipeline finished in %.2f ms", _elapsed_ms(pipeline_t0))


if __name__ == "__main__":
    main()

日志配置示例

• 基于 dictConfig 的“详细（detailed）+ 文件输出 + DEBUG 级别”示例，可直接替换 configure_logging 使用：

import logging
import logging.config
from pathlib import Path

def setup_logging_dictconfig(log_file="logs/app.log", level="DEBUG"):
    Path(log_file).parent.mkdir(parents=True, exist_ok=True)
    logging.config.dictConfig(
        {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {
                "detailed": {
                    "format": "%(asctime)s | %(levelname)s | pid=%(process)d tid=%(thread)d | "
                              "%(name)s:%(funcName)s:%(lineno)d | %(message)s",
                    "datefmt": "%Y-%m-%d %H:%M:%S",
                }
            },
            "handlers": {
                "file": {
                    "class": "logging.FileHandler",
                    "level": level,
                    "formatter": "detailed",
                    "filename": log_file,
                    "encoding": "utf-8",
                }
            },
            "root": {"level": level, "handlers": ["file"]},
        }
    )

使用建议

• 日志级别调整
  • 开发/调试：保持 DEBUG，获得完整的性能与数据质量信息。
  • 生产：建议切换为 INFO 或 WARNING，减少日志量；仅在故障排查时临时开启 DEBUG。
• 日志轮转
  • 当前示例使用 FileHandler。生产环境建议改用 RotatingFileHandler 或 TimedRotatingFileHandler，并配置最大大小/保留个数，避免磁盘占满。
• 性能监控
  • 已对每个阶段和整体耗时进行记录。若耗时过长，可进一步在 transform 内部按批次计时，或为特定大数据集引入采样日志。
• 数据与隐私
  • 对异常行仅输出必要字段（category、amount）。请避免在日志中写入密码、密钥或个人隐私信息。
• 可扩展性
  • 若日志过于冗长，可对聚合结果和中间结果仅输出前 N 条预览，或通过环境变量控制预览长度与是否输出数据样本。