代码修改说明
- 用标准库 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 条预览，或通过环境变量控制预览长度与是否输出数据样本。
