写在前面
这是第二批的最后一篇。前 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 / macOS | crontab | 0 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) | 适合常驻任务,不适合「跑一次就退出」 |
三条工程习惯,比工具本身重要:
- 幂等:同一天跑两次,结果应该一样(文件名带日期、
if_exists="replace"); - 日志落盘:定时任务没人看着,日志是唯一的现场;
- 失败要能知道:最简单的做法是失败时发一封邮件 / 一条 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 |
| 19 | Pandas 与 Excel | Pandas 算,Excel 呈现和协作 |
| 20 | 数据质量 | 10 项校验做成函数,挂到交付流程 |
| 21 | 相关性与分组 | corr / crosstab / groupby / 热力图 |
| 22 | 用户行为分析 | 留存、复购、月活,都是分组建透视 |
| 23 | Seaborn | 统计图:误差棒、箱线、热力、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% 走完了——剩下的,遇到问题再查,比从头看教程快得多。