ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Python 数据处理进阶(二十四)把脚本变成工具:argparse、logging 与定时任务

Python 数据处理进阶(二十四)把脚本变成工具:argparse、logging 与定时任务

写在前面

这是第二批的最后一篇。前 23 篇把「怎么处理数据」讲完了,这一篇讲怎么让脚本被别人用起来。

写给自己用的脚本,改参数、print调试、手动运行都没问题;但只要交给同事、或者要每周自动跑,就必须补三件事:参数(不改代码就能换输入输出)、日志(出问题能查)、定时(不用人盯着)。

这篇把这三件事做成一个完整的「报表生成器」,你抄走就能用。

先确认示例数据在手边:

importosimportmkdataifnotos.path.exists("data/sales.csv"):mkdata.main()print("示例数据就绪:",os.path.exists("data/sales.csv"))

这段会检查第 1 篇生成的data/目录在不在,不在就自动补一次;已经有数据时只会打印一行示例数据就绪: True,不影响后面的结果。

一、argparse:把参数从代码里挪出来

最朴素的写法是「改代码里的路径」,麻烦且容易改错。argparse让你在命令行传参,还能自动生成--help。

importargparsedefbuild_parser():parser=argparse.ArgumentParser(prog="report.py",description="从销售明细生成 Excel 报表")parser.add_argument("input",help="输入 CSV 路径")parser.add_argument("--top",type=int,default=5,help="每个维度取前 N 名")parser.add_argument("--out",default="output/report.xlsx",help="输出 xlsx 路径")parser.add_argument("--verbose",action="store_true",help="输出调试日志")returnparser parser=build_parser()# 文章里演示用显式参数,实际运行时直接 python report.py data/sales.csvargs=parser.parse_args(["data/sales.csv","--top","3","--out","output/report.xlsx"])print("解析结果:")print(" input =",args.input)print(" top =",args.top)print(" out =",args.out)print(" verbose=",args.verbose)print("----")print("帮助信息:")parser.print_help()
解析结果: input = data/sales.csv top = 3 out = output/report.xlsx verbose= False ---- 帮助信息: usage: report.py [-h] [--top TOP] [--out OUT] [--verbose] input 从销售明细生成 Excel 报表 positional arguments: input 输入 CSV 路径 options: -h, --help show this help message and exit --top TOP 每个维度取前 N 名 --out OUT 输出 xlsx 路径 --verbose 输出调试日志

三个设计要点:

  • 位置参数(input)给必需项,可选参数(--top)给默认值,命令行里少写几个也能跑;
  • type=int让 argparse 自动转换并校验,传abc会直接报错,不用自己判断;
  • action="store_true"是布尔开关,--verbose出现即为 True。

二、logging:print 是调试,logging 是工程

print的问题:没法分级、没法关、没法带时间戳和行号。logging四行配置就解决。

importloggingimportos os.makedirs("output",exist_ok=True)logger=logging.getLogger("report")logger.setLevel(logging.INFO)logger.handlers.clear()# 避免重复添加 handler(Jupyter 里尤其容易)fmt=logging.Formatter("%(asctime)s | %(levelname)-7s | %(message)s","%Y-%m-%d %H:%M:%S")console=logging.StreamHandler()console.setFormatter(fmt)logger.addHandler(console)file_handler=logging.FileHandler("output/report.log",encoding="utf-8")file_handler.setFormatter(fmt)logger.addHandler(file_handler)logger.info("开始生成报表")logger.warning("示例数据只有 16 行,仅供演示")try:1/0exceptZeroDivisionError:logger.exception("计算失败,已捕获异常并继续")# exception 会带上 tracebacklogger.info("流程结束,日志已写入 output/report.log")withopen("output/report.log",encoding="utf-8")asf:print("---- 日志文件内容 ----")print(f.read())
2026-10-21 19:42:10 | INFO | 开始生成报表 2026-10-21 19:42:10 | WARNING | 示例数据只有 16 行,仅供演示 2026-10-21 19:42:10 | ERROR | 计算失败,已捕获异常并继续 Traceback (most recent call last): File "<article>", line 23, in <module> ZeroDivisionError: division by zero 2026-10-21 19:42:10 | INFO | 流程结束,日志已写入 output/report.log ---- 日志文件内容 ---- 2026-10-21 19:42:10 | INFO | 开始生成报表 2026-10-21 19:42:10 | WARNING | 示例数据只有 16 行,仅供演示 2026-10-21 19:42:10 | ERROR | 计算失败,已捕获异常并继续 Traceback (most recent call last): File "<article>", line 23, in <module> ZeroDivisionError: division by zero 2026-10-21 19:42:10 | INFO | 流程结束,日志已写入 output/report.log

四个级别够用:DEBUG(调试细节)、INFO(正常进度)、WARNING(可继续的问题)、ERROR(出错了)。logger.exception()要写在except里,它会自动把 traceback 写进日志——这是排查线上问题最值钱的一行。

三、合起来:一个能交付的报表生成器

把参数、日志、数据处理、导出串起来,每个环节都留日志,出错不崩。

importargparseimportloggingimportosimportpandasaspddefsetup_logger(verbose=False):logger=logging.getLogger("report")logger.setLevel(logging.DEBUGifverboseelselogging.INFO)logger.handlers.clear()fmt=logging.Formatter("%(asctime)s | %(levelname)-7s | %(message)s","%H:%M:%S")sh=logging.StreamHandler()sh.setFormatter(fmt)logger.addHandler(sh)returnloggerdefbuild_report(path,top=5,out="output/report.xlsx",logger=None):logger=loggerorsetup_logger()logger.info("读取数据:%s",path)df=pd.read_csv(path,encoding="utf-8-sig")df["金额"]=df["金额"].fillna((df["数量"]*df["单价"]).round(2))clean=df[df["金额"]>0].drop_duplicates()logger.info("原始 %d 行,清洗后 %d 行",len(df),len(clean))by_city=(clean.groupby("城市",dropna=False)["金额"].sum().sort_values(ascending=False).head(top).round(2).reset_index())by_product=(clean.groupby("产品")["金额"].sum().sort_values(ascending=False).head(top).round(2).reset_index())os.makedirs(os.path.dirname(out)or".",exist_ok=True)withpd.ExcelWriter(out,engine="openpyxl")aswriter:by_city.to_excel(writer,sheet_name="城市Top",index=False)by_product.to_excel(writer,sheet_name="产品Top",index=False)logger.info("已写出:%s",out)returnout,by_city,by_product parser=argparse.ArgumentParser(prog="report.py",description="生成销售报表")parser.add_argument("input")parser.add_argument("--top",type=int,default=3)parser.add_argument("--out",default="output/report.xlsx")parser.add_argument("--verbose",action="store_true")args=parser.parse_args(["data/sales.csv","--top","3","--verbose"])log=setup_logger(args.verbose)try:out,city,product=build_report(args.input,args.top,args.out,log)exceptFileNotFoundError:log.error("输入文件不存在:%s",args.input)raiseSystemExit(1)print("城市 Top:")print(city.to_string(index=False))print("产品 Top:")print(product.to_string(index=False))
19:42:10 | INFO | 读取数据:data/sales.csv 19:42:10 | INFO | 原始 16 行,清洗后 12 行 19:42:10 | INFO | 已写出:output/report.xlsx 城市 Top: 城市 金额 杭州 3404.1 北京 1425.9 上海 772.5 产品 Top: 产品 金额 D-显示器 4193.0 C-键盘 1157.0 A-笔记本 548.9

图 1:脚本工具化的五段流水线——参数、日志、处理、导出、定时。前四段写完就是「能交付」,第五段让它自己跑起来。

注意raise SystemExit(1):工具化脚本的退出码很重要,定时任务和 CI 靠它判断成功失败,崩了要返回非 0。

四、定时运行:让它自己跑起来

系统方式最小配置
Windows任务计划程序触发器选「每周一 8:00」,操作填python.exe,参数填脚本全路径
Linux / macOScrontab0 8 * * 1 cd /path/to/project && python3 report.py data/sales.csv >> cron.log 2>&1
任意GitHub Actions定时 workflow,schedule: - cron: "0 0 * * 1"
服务器系统服务(systemd / supervisor)适合常驻任务,不适合「跑一次就退出」

三条工程习惯,比工具本身重要:

  1. 幂等:同一天跑两次,结果应该一样(文件名带日期、if_exists="replace");
  2. 日志落盘:定时任务没人看着,日志是唯一的现场;
  3. 失败要能知道:最简单的做法是失败时发一封邮件 / 一条 webhook,别等下周才发现没跑。

五、第二批 12 篇回顾

#主题一句话
13多层索引set_index造、loc/xs取、unstack转
14向量化列运算比 for 循环快上千倍
15多种数据源JSON、网页表格、数据库各有各的 read_*
16内存优化降级 + category,55 MB 压到 4 MB
17自动周报groupby → 多 sheet → openpyxl 格式
18批量处理glob → 循环读加来源列 → concat
19Pandas 与 ExcelPandas 算,Excel 呈现和协作
20数据质量10 项校验做成函数,挂到交付流程
21相关性与分组corr / crosstab / groupby / 热力图
22用户行为分析留存、复购、月活,都是分组建透视
23Seaborn统计图:误差棒、箱线、热力、pairplot
24工具化argparse + logging + 定时(本篇)

三批方向,按需选:

  • 往深了学:polars(更快的新一代 DataFrame)、duckdb(用 SQL 查 Parquet/CSV,本地分析神器);
  • 往宽了学:SQL(GROUP BY和groupby是同一套思维)、统计学基础、scikit-learn;
  • 往工程走:把脚本包成 API(FastAPI)、上调度(Airflow / Dagster)、加测试(pytest + 样例数据)。

常见问题

Q1:argparse在 Jupyter 里报SystemExit: 2。

Jupyter 传了它不认识的参数(比如-f kernel.json)。文章里的写法是在代码里显式parse_args([...]),不要用parse_args()读sys.argv。

Q2:日志重复输出两遍。

logger.addHandler调了两次(比如脚本被 import 两次)。加logger.handlers.clear(),或者在addHandler前判断if not logger.handlers:。

Q3:python report.py报找不到data/sales.csv。

相对路径是相对于「你运行命令时所在的目录」。定时任务里尤其容易踩:crontab 里先cd到项目目录,或者干脆在代码里用绝对路径。

小结

工具化的三件套:argparse 管输入、logging 管过程、定时管触发。判断一个脚本是否「可交付」,就看三点:别人能不能不改代码就用、出错时能不能查到原因、能不能无人值守地跑。做到这三点,你就从「会写脚本」跨到了「会做工具」。

第二批到此结束。数据这条路很长,但你已经把最常用的 80% 走完了——剩下的,遇到问题再查,比从头看教程快得多。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进