很多人学 Python 都会经历一个迷惑阶段语法看完了教程跟完了照着代码敲也能跑但合上编辑器面对一个真实需求大脑就直接卡壳。我一开始也这样。真正让我从能跑代码变成能搞定事的是一轮又一轮的实战练习以及每次练习之后的复盘总结。今天这篇我就拿自己最近做的模拟项目 X——一个从数据采集到报表生成的小系统——当作完整例子把整个构思、技术选型、核心代码、踩坑过程都摊开讲一遍。不管你是准备找工作的初学者还是想巩固基础的后端开发这篇文章的思路和代码都能直接抄作业。1. 项目背景为什么练了一堆代码还是心里没底很多人觉得实战练习就是多刷题、多敲几个小例子比如用 Python 算个阶乘、写个爬虫抓标题、做个猜数字游戏。这类练习不是没用但它只训练了单点技能没有训练完整做事的能力。真实开发场景里一个需求往往要串联好几个模块要拿到数据、清洗数据、按照特定规则处理、再输出成别人能看懂的成果。任何一个环节掉链子整个流程就跑不通。我当时给自己定的练习目标不是一个炫酷的功能而是一条完整的链路。模拟项目 X 的定位很朴素每天从某个数据平台拉取一批清单数据去重、补全、整理格式之后生成一份带汇总统计的 Excel 报表。听起来很简单但真正动手之后才发现这里有网络请求的稳定性问题、编码与类型问题、数据脏乱的问题、Excel 格式处理的问题还有跑完整个流程之后的异常排查问题。我当时选择这个项目有三个原因。第一它覆盖了 Python 日常开发里最常用的几块内容requests 发请求、pandas 做表格处理、openpyxl 写 Excel、logging 记日志。第二它的输出结果可以直接看到做出来有一份实际文件有很强的反馈感。第三它每跑一次都会产生真实数据这就能逼着你处理各种意外而不是像刷题那样所有输入都是老师准备好、规范工整的。项目做完之后我做了一次完整的总结把每一次报错、每一个犹豫过的选择都记了下来这才有了下面这些内容。2. 实战项目的整体设计与技术选型刚开始做这个项目的时候我犯过一个典型的错误上来就写代码没有先考虑结构。结果写到第三天就乱了采集部分的逻辑和处理部分的逻辑混在一起想改动清洗规则还要去翻一堆爬虫代码。后来我把自己拉回正常节奏先花半天时间把项目拆成了三层。采集层负责向数据源发起请求拿到原始数据存成 CSV。处理层负责读取 CSV做去重、格式整理、字段补全产出标准化的纯数据表。展示层负责把处理后的数据写进带格式的 Excel包含汇总统计和简单的数据透视。这个分层结构对我来说最大的价值是每一层都能独立验证。采集层跑完看一眼 CSV 有多少行处理层跑完看一眼输出表的字段是否符合预期展示层跑完打开 Excel 直接检查格式。任何一层出了问题都不需要连带到别的代码里去排查。工具选型这一块我也纠结过一段时间。一开始考虑过用别人封装好的爬虫框架但练实战项目的目的不是为了偷懒而是为了理解底层所以网络请求部分直接用 requests。数据处理部分网上有一些人会用纯 Python 的列表和字典来写数据量小的时候确实没问题但这个项目涉及多表合并和分组统计手写太痛苦所以用 pandas。Excel 输出部分openpyxl 是最直接的选择它能控制单元格样式、列宽、字体颜色比 pandas 自带的 to_excel 输出好看很多。| 功能 | 工具 | 选它而不是另一个的理由 | |--------------|------------|------------------------------------------------------------------------------------------| | 数据获取 | requests | 轻量、直观、文档多相比于重框架更容易理解请求头、超时、重试这些底层行为 | | 数据处理 | pandas | DataFrame 的合并、分组、去重、缺失值处理都提供了现成接口学习性价比极高 | | Excel 写入 | openpyxl | 支持单元格级样式控制生成报表好看且能实现冻结首行、自动列宽、条件颜色等 | | 定时调度 | schedule | 纯 Python 实现的轮询调度器语法简单适合练习场景 | | 运行日志 | logging | 标准库自带能同时输出到控制台和文件排查问题比 print 高级太多 |环境搭建方面我建议统一用虚拟环境别图省事全局 pip install。我见过太多次全局环境装了一堆包结果版本互相冲突项目从 A 电脑搬到 B 电脑就跑不起来的案例。创建环境后用 requirements.txt 锁定版本也方便之后复盘。我的要求文件当时是这样写的。requests2.31.0 pandas2.1.4 openpyxl3.1.2 schedule1.2.1整体设计做完我心里其实踏实了一大半。因为此时这个项目不再是脑子里一团模糊的做个报表而是三个清晰的小问题怎么请求数据、怎么处理数据、怎么写 Excel。拆开之后每个问题单独看都不难。3. 核心模块拆解采集、清洗、报表三步走3.1 采集模块别只写爬虫逻辑先管好稳健性采集模块的目标是从某数据平台获取当天的清单列表。数据平台返回的是 JSON 结构里面有列表数据、总数、页码等字段。我第一次写的时候只用了最朴素的 requests.get正则都没加结果跑了不到半小时就报错——对方平台直接返回超时程序就断了。后来我才意识到网络请求这件事最大的坑根本不在于怎么解析数据而在于怎么保证请求的稳健性。我把采集层的代码重构成了三个关键点。超时设置requests.get 里的 timeout 参数必须写否则一个挂死的连接会一直阻塞。重试机制遇到网络波动或服务端 5xx 错误停顿几秒后重新请求最多重试 3 次。随机间隔每两次请求之间加一个随机延时避免请求频率过高被服务端限流。采集核心代码差不多是这个样子。import requests import time import random import logging logger logging.getLogger(__name__) def fetch_page(page_no: int, retries: int 3) - dict: base_url https://api.sample-data-source.local/v1/list params { page: page_no, page_size: 100, } for attempt in range(1, retries 1): try: resp requests.get(base_url, paramsparams, timeout10) resp.raise_for_status() return resp.json() except (requests.Timeout, requests.ConnectionError) as exc: logger.warning(第 %s 次请求第 %s 页失败: %s, attempt, page_no, exc) time.sleep(2 * attempt random.uniform(0, 2)) except requests.HTTPError as exc: logger.error(第 %s 页返回异常状态码: %s, page_no, exc) break return {} def collect_all(): all_items [] page 1 while True: data fetch_page(page) items data.get(items) or [] all_items.extend(items) total data.get(total, 0) if len(all_items) total or not items: break page 1 time.sleep(random.uniform(1, 3)) logger.info(采集完成共获取 %s 条记录, len(all_items)) return all_items这里有一个非常关键的习惯把每一段逻辑都写在单独的函数里并且主流程用清晰的函数名串联起来。这样做的价值在调试的时候体现得淋漓尽致。当第 5 页请求失败时日志能直接告诉我失败原因和重试次数而不用在几十行代码里打断点。3.2 清洗模块数据真正花费时间的地方数据清洗是整个项目里最花时间、也最体现经验的部分。第一次采集成功后我拿到了一万多条 JSON 数据高高兴兴地转成 DataFrame一看结果就头大了数据里同一个商品出现了多次价格字段有的是数字、有的是带货币符号的字符串日期字段有快照时间和更新时间两种有些记录关键字段直接为空。清洗模块我做了一个很实用的小动作把每一步清洗规则拆开写每一步都用单独的 DataFrame 赋值方便检查中间过程。import pandas as pd def clean_data(df: pd.DataFrame) - pd.DataFrame: # 1. 去重按 item_id snap_date 判断是否重复 df df.drop_duplicates(subset[item_id, snap_date], keeplast) # 2. 删除关键字段全为空的行 df df.dropna(subset[item_id, item_name]) # 3. 价格字段清洗去掉货币符号、千分位逗号转为浮点数 df[price] ( df[price] .astype(str) .str.replace(, ) .str.replace(,, ) .astype(float) ) # 4. 日期统一只保留 YYYY-MM-DD 部分 df[snap_date] pd.to_datetime(df[snap_date]).dt.strftime(%Y-%m-%d) # 5. 缺失分类处理没有分类的记录统一标记为“未归类” df[category] df[category].fillna(未归类) return df这段代码看起来简单但每一条规则背后都有实际的坑在等着。比如keeplast是因为同一个 item_id 在一天内可能被更新多次我要保留的是最后一次更新后的版本。比如价格字段要先把整个列转成字符串再 replace否则数值类型会直接报错。再比如时间字段如果不先转成 datetime直接用字符串截取遇到带时区的时间字符串就会出问题。清洗过程中我还专门做了数据量对比。采集层拿到的原始 10080 条记录经过去重剩下 8735 条再经过缺失值处理剩下 8610 条。这个数字对比并不是成本浪费它让我能准确判断每个环节过滤掉了多少数据写周报的时候也能说清楚数据处理的完整过程。3.3 报表模块让非程序员看得懂如果只是把数据重新导出成 CSV那这个练习的成就感会少一半。报表端我做了一份真正的 Excel 文件包含原始明细、筛选后的汇总透视表还有一个简单的统计标签页。openpyxl 在这个环节发挥了大作用。from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils.dataframe import dataframe_to_rows def build_excel_report(df: pd.DataFrame, output_path: str): wb Workbook() header_font Font(boldTrue, colorFFFFFF) header_fill PatternFill(start_color4472C4, end_color4472C4, fill_typesolid) # 明细页 ws1 wb.active ws1.title 明细数据 for row in dataframe_to_rows(df, indexFalse, headerTrue): ws1.append(row) for cell in ws1[1]: cell.font header_font cell.fill header_fill cell.alignment Alignment(horizontalcenter, verticalcenter) ws1.freeze_panes A2 # 汇总页按分类统计条数和平均价格 ws2 wb.create_sheet(分类汇总) summary df.groupby(category).agg(count(item_id, count), avg_price(price, mean)).reset_index() ws2.append([分类, 条数, 平均价格]) for _, row in summary.iterrows(): ws2.append([row[category], row[count], round(row[avg_price], 2)]) ws2[A1].font header_font ws2[B1].font header_font ws2[C1].font header_font # 自动列宽 for ws in wb.worksheets: for col in ws.columns: max_len max(len(str(c.value)) if c.value is not None else 0 for c in col) ws.column_dimensions[col[0].column_letter].width min(max_len 4, 60) wb.save(output_path)这里有一个细节对新手帮助很大dataframe_to_rows这个工具函数能把 DataFrame 一行一行地取出来然后直接用 ws1.append 塞进 Excel 单元格比手动遍历行和列省心很多。自动列宽这段代码我是从踩坑里学来的。第一次生成的 Excel所有列宽默认都是 8.38 个字符中文字符显示成一片乱码一样的窄条看得人头疼。后来我遍历每一列的最大长度动态设置列宽观感瞬间好了很多。还有一点想提醒openpyxl 写入时空值如果直接写进去Excel 里会显示成 None 字符串这非常丑。所以我处理的时候会把 None 统一替换成空字符串或者直接删除这些行否则报表阅读者会以为那是个叫 None 的异常值。4. 关键模块整合与自动化调度让程序每天自己跑一遍模块分开写完之后整合其实是个体力活但也正是在这个环节我体会到了为什么代码要写函数、要写日志。主入口长这样import logging import schedule import time from collector import collect_all from cleaner import clean_data from reporter import build_excel_report def main_job(): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(run.log, encodingutf-8), logging.StreamHandler(), ], ) logger logging.getLogger(main) logger.info(任务开始) raw_data collect_all() df pd.DataFrame(raw_data) df clean_data(df) build_excel_report(df, freport_{time.strftime(%Y%m%d)}.xlsx) logger.info(任务完成报告已生成) if __name__ __main__: schedule.every().day.at(09:30).do(main_job) while True: schedule.run_pending() time.sleep(1)整合完成后的第一件事不是直接等定时任务触发了而是先手动执行 main_job()。这个习惯一定不能省。因为我第一次直接跑代码时调度器在 while True 循环里空转了好几天回头一看日志才发现 main_job 里有一个拼写错误函数名没对上异常又没捕获整个任务静默失败了。从那以后我要求自己写完任何自动化任务必须手动执行一次确认成功再让它进入计划调度。日志这一块我想多花点篇幅说明。很多初学者喜欢用 print 打输出但这在自动化脚本里完全不够。程序挂在后台跑的时候print 的输出你看不到就算看到了也没有时间戳不知道是哪一次运行打出来的。用 logging 可以同时把日志写到控制台和文件并且在任务失败时记录完整的堆栈信息。更重要的是日志文件能帮我做长期复盘——周一的任务为什么比周三晚跑了十分钟翻日志就能找到真相。我在调度逻辑里其实也踩了一个隐形坑。用schedule.every().day.at(09:30)设定好后我一开始没有加异常兜底结果某一天源数据接口返回的是空列表主流程继续往下走清洗后的 DataFrame 是空表Excel 报表照样生成了但打开后里面只有表头没有任何数据。这种半成功的失败最坑人因为程序退出码是 0日志里也没有异常直到有人发现当天报表数据是空的回过头来排查才发现源端数据异常。所以后来我加了数据量阈值的校验如果清洗后的数据条数小于某个最小值就主动发送日志告警并且不生成报表、不覆盖旧文件。这个做法给我的启发是自动化任务不仅要让程序能跑更要让程序在不该跑的时候停下来。5. 常见问题与排查技巧实录我踩过的那几组坑这个项目做完我积累了一整套高频问题排查表这里挑几个最有代表性的整理出来每个问题背后都是一段真实耗时。现象根因解决方法请求时好时坏偶尔抛超时没设置 timeout有的连接挂死了所有 requests 请求统一加 timeout10并做 3 次重试CSV 文件打开全是乱码pandas 默认用 UTF-8 编码写入但 Excel 默认用系统编码读取保存时指定 encodingutf-8-sig或者在打开时手动选编码报表里的数字求和不对有些价格字段是字符串Excel 当成了文本清洗阶段统一 astype(float)并检查是否有空值残留DataFrame 输出 Excel 慢循环逐行 append 一条一条塞改用 openpyxl 的 dataframe_to_rows大幅提升写入速度定时任务到点没跑while True 循环被前面代码阻塞了把耗时长的任务放到子进程或者把 schedule 的循环放到单独线程程序报错却找不到原因只 print 没日志且没有 except 记录 traceback用 logger.exception 在异常分支打印完整堆栈到文件openpyxl 保存的文件模板被破坏直接打开原模板并另存为丢失了原格式改为 locked 或复制模板副本再写入保留原始文件七条问题里耗时最久的是第一个也就是请求超时。那次我调试到怀疑人生一会儿能通一会儿不能通还以为是服务器的问题直到抓包才发现是自己的连接池里堆积了一堆半开连接。加 timeout 和重试机制之后立竿见影。这里我可以给你的建议是所有需要依赖外部网络资源的代码第一版必须带上超时和重试两层保险否则后续所有的调试都会被这种偶发问题干扰。还有一个值得单独拎出来讲的坑就是中文编码。第一次我用 DataFrame.to_csv 保存结果时程序输出的是默认的 UTF-8 编码用 Excel 打开之后中文完全乱掉。当时我以为是 pandas 写坏了折腾了半天才发现是编码标准不统一。utf-8-sig 这个编码里带了一个特殊的 BOM 头Excel 才能准确识别。这个细节到现在我都提醒自己做任何导出操作时先确认目标软件对编码的预期。排查的时候我总结出一个通用套路遇到报错不要急着看倒数第一行先看完整 traceback 的最顶层定位是哪一个函数出了问题然后检查这一轮的输入数据长什么样。大部分数据处理 Bug问题都不在代码逻辑本身而在输入数据不符合你的假设。举个例子我代码里假设每个 item_id 都是字符串结果源端某一个分类下返回的是数字去重时就会因为类型不一致而无法清洗。类似这种情况用df.dtypes和df.head()打印一下五秒钟就能发现。6. 写在最后的几点体会一口气把这个实战项目从头讲到尾最后分享几个让我特别有感的经验。第一练习项目的难度不在会不会用某个库而在能不能把一个模糊目标拆成可执行的小步骤。没有拆解就不会有后面的层模块、独立函数和边界校验。上来就写主逻辑大概率会在中途被各种意外问题打断然后不了了之。第二所有项目里的技巧都可以用踩坑-总结这个模式掌握。我写采集层时第一次超时错误、写报表时第一次列宽显示混乱、写调度时第一次静默失败每一个都不光彩但每一个都是最深刻的教材。建议你也准备一个 bug 记录本把自己项目里每种问题的根因和方法记下来过两三个月回头看那份记录比教程值钱得多。第三也是最重要的实战练习不要贪全贪大。做一个能把数据从源头处理到产出报表的小闭环远远好过同时开三个半成品项目。单点突破做透一件事这种经验才能迁移到更复杂的场景里。这个项目之后我又陆续给它加了并发请求、配置化规则、数据库存储等扩展。每次改动都是一次新的实战练习。如果你现在正处于什么都会一点但又什么都不敢说会的尴尬期别多想找一个真实的小问题从第一行代码写到输出成果再回来读一遍这篇总结你一定会比听十遍课收获更多。 
长河君 · 2026/10/21 9:29:11 · 阅读 1.2k Python实战：从数据采集到Excel报表的完整项目拆解
很多人学 Python 都会经历一个迷惑阶段语法看完了教程跟完了照着代码敲也能跑但合上编辑器面对一个真实需求大脑就直接卡壳。我一开始也这样。真正让我从能跑代码变成能搞定事的是一轮又一轮的实战练习以及每次练习之后的复盘总结。今天这篇我就拿自己最近做的模拟项目 X——一个从数据采集到报表生成的小系统——当作完整例子把整个构思、技术选型、核心代码、踩坑过程都摊开讲一遍。不管你是准备找工作的初学者还是想巩固基础的后端开发这篇文章的思路和代码都能直接抄作业。1. 项目背景为什么练了一堆代码还是心里没底很多人觉得实战练习就是多刷题、多敲几个小例子比如用 Python 算个阶乘、写个爬虫抓标题、做个猜数字游戏。这类练习不是没用但它只训练了单点技能没有训练完整做事的能力。真实开发场景里一个需求往往要串联好几个模块要拿到数据、清洗数据、按照特定规则处理、再输出成别人能看懂的成果。任何一个环节掉链子整个流程就跑不通。我当时给自己定的练习目标不是一个炫酷的功能而是一条完整的链路。模拟项目 X 的定位很朴素每天从某个数据平台拉取一批清单数据去重、补全、整理格式之后生成一份带汇总统计的 Excel 报表。听起来很简单但真正动手之后才发现这里有网络请求的稳定性问题、编码与类型问题、数据脏乱的问题、Excel 格式处理的问题还有跑完整个流程之后的异常排查问题。我当时选择这个项目有三个原因。第一它覆盖了 Python 日常开发里最常用的几块内容requests 发请求、pandas 做表格处理、openpyxl 写 Excel、logging 记日志。第二它的输出结果可以直接看到做出来有一份实际文件有很强的反馈感。第三它每跑一次都会产生真实数据这就能逼着你处理各种意外而不是像刷题那样所有输入都是老师准备好、规范工整的。项目做完之后我做了一次完整的总结把每一次报错、每一个犹豫过的选择都记了下来这才有了下面这些内容。2. 实战项目的整体设计与技术选型刚开始做这个项目的时候我犯过一个典型的错误上来就写代码没有先考虑结构。结果写到第三天就乱了采集部分的逻辑和处理部分的逻辑混在一起想改动清洗规则还要去翻一堆爬虫代码。后来我把自己拉回正常节奏先花半天时间把项目拆成了三层。采集层负责向数据源发起请求拿到原始数据存成 CSV。处理层负责读取 CSV做去重、格式整理、字段补全产出标准化的纯数据表。展示层负责把处理后的数据写进带格式的 Excel包含汇总统计和简单的数据透视。这个分层结构对我来说最大的价值是每一层都能独立验证。采集层跑完看一眼 CSV 有多少行处理层跑完看一眼输出表的字段是否符合预期展示层跑完打开 Excel 直接检查格式。任何一层出了问题都不需要连带到别的代码里去排查。工具选型这一块我也纠结过一段时间。一开始考虑过用别人封装好的爬虫框架但练实战项目的目的不是为了偷懒而是为了理解底层所以网络请求部分直接用 requests。数据处理部分网上有一些人会用纯 Python 的列表和字典来写数据量小的时候确实没问题但这个项目涉及多表合并和分组统计手写太痛苦所以用 pandas。Excel 输出部分openpyxl 是最直接的选择它能控制单元格样式、列宽、字体颜色比 pandas 自带的 to_excel 输出好看很多。| 功能 | 工具 | 选它而不是另一个的理由 | |--------------|------------|------------------------------------------------------------------------------------------| | 数据获取 | requests | 轻量、直观、文档多相比于重框架更容易理解请求头、超时、重试这些底层行为 | | 数据处理 | pandas | DataFrame 的合并、分组、去重、缺失值处理都提供了现成接口学习性价比极高 | | Excel 写入 | openpyxl | 支持单元格级样式控制生成报表好看且能实现冻结首行、自动列宽、条件颜色等 | | 定时调度 | schedule | 纯 Python 实现的轮询调度器语法简单适合练习场景 | | 运行日志 | logging | 标准库自带能同时输出到控制台和文件排查问题比 print 高级太多 |环境搭建方面我建议统一用虚拟环境别图省事全局 pip install。我见过太多次全局环境装了一堆包结果版本互相冲突项目从 A 电脑搬到 B 电脑就跑不起来的案例。创建环境后用 requirements.txt 锁定版本也方便之后复盘。我的要求文件当时是这样写的。requests2.31.0 pandas2.1.4 openpyxl3.1.2 schedule1.2.1整体设计做完我心里其实踏实了一大半。因为此时这个项目不再是脑子里一团模糊的做个报表而是三个清晰的小问题怎么请求数据、怎么处理数据、怎么写 Excel。拆开之后每个问题单独看都不难。3. 核心模块拆解采集、清洗、报表三步走3.1 采集模块别只写爬虫逻辑先管好稳健性采集模块的目标是从某数据平台获取当天的清单列表。数据平台返回的是 JSON 结构里面有列表数据、总数、页码等字段。我第一次写的时候只用了最朴素的 requests.get正则都没加结果跑了不到半小时就报错——对方平台直接返回超时程序就断了。后来我才意识到网络请求这件事最大的坑根本不在于怎么解析数据而在于怎么保证请求的稳健性。我把采集层的代码重构成了三个关键点。超时设置requests.get 里的 timeout 参数必须写否则一个挂死的连接会一直阻塞。重试机制遇到网络波动或服务端 5xx 错误停顿几秒后重新请求最多重试 3 次。随机间隔每两次请求之间加一个随机延时避免请求频率过高被服务端限流。采集核心代码差不多是这个样子。import requests import time import random import logging logger logging.getLogger(__name__) def fetch_page(page_no: int, retries: int 3) - dict: base_url https://api.sample-data-source.local/v1/list params { page: page_no, page_size: 100, } for attempt in range(1, retries 1): try: resp requests.get(base_url, paramsparams, timeout10) resp.raise_for_status() return resp.json() except (requests.Timeout, requests.ConnectionError) as exc: logger.warning(第 %s 次请求第 %s 页失败: %s, attempt, page_no, exc) time.sleep(2 * attempt random.uniform(0, 2)) except requests.HTTPError as exc: logger.error(第 %s 页返回异常状态码: %s, page_no, exc) break return {} def collect_all(): all_items [] page 1 while True: data fetch_page(page) items data.get(items) or [] all_items.extend(items) total data.get(total, 0) if len(all_items) total or not items: break page 1 time.sleep(random.uniform(1, 3)) logger.info(采集完成共获取 %s 条记录, len(all_items)) return all_items这里有一个非常关键的习惯把每一段逻辑都写在单独的函数里并且主流程用清晰的函数名串联起来。这样做的价值在调试的时候体现得淋漓尽致。当第 5 页请求失败时日志能直接告诉我失败原因和重试次数而不用在几十行代码里打断点。3.2 清洗模块数据真正花费时间的地方数据清洗是整个项目里最花时间、也最体现经验的部分。第一次采集成功后我拿到了一万多条 JSON 数据高高兴兴地转成 DataFrame一看结果就头大了数据里同一个商品出现了多次价格字段有的是数字、有的是带货币符号的字符串日期字段有快照时间和更新时间两种有些记录关键字段直接为空。清洗模块我做了一个很实用的小动作把每一步清洗规则拆开写每一步都用单独的 DataFrame 赋值方便检查中间过程。import pandas as pd def clean_data(df: pd.DataFrame) - pd.DataFrame: # 1. 去重按 item_id snap_date 判断是否重复 df df.drop_duplicates(subset[item_id, snap_date], keeplast) # 2. 删除关键字段全为空的行 df df.dropna(subset[item_id, item_name]) # 3. 价格字段清洗去掉货币符号、千分位逗号转为浮点数 df[price] ( df[price] .astype(str) .str.replace(, ) .str.replace(,, ) .astype(float) ) # 4. 日期统一只保留 YYYY-MM-DD 部分 df[snap_date] pd.to_datetime(df[snap_date]).dt.strftime(%Y-%m-%d) # 5. 缺失分类处理没有分类的记录统一标记为“未归类” df[category] df[category].fillna(未归类) return df这段代码看起来简单但每一条规则背后都有实际的坑在等着。比如keeplast是因为同一个 item_id 在一天内可能被更新多次我要保留的是最后一次更新后的版本。比如价格字段要先把整个列转成字符串再 replace否则数值类型会直接报错。再比如时间字段如果不先转成 datetime直接用字符串截取遇到带时区的时间字符串就会出问题。清洗过程中我还专门做了数据量对比。采集层拿到的原始 10080 条记录经过去重剩下 8735 条再经过缺失值处理剩下 8610 条。这个数字对比并不是成本浪费它让我能准确判断每个环节过滤掉了多少数据写周报的时候也能说清楚数据处理的完整过程。3.3 报表模块让非程序员看得懂如果只是把数据重新导出成 CSV那这个练习的成就感会少一半。报表端我做了一份真正的 Excel 文件包含原始明细、筛选后的汇总透视表还有一个简单的统计标签页。openpyxl 在这个环节发挥了大作用。from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils.dataframe import dataframe_to_rows def build_excel_report(df: pd.DataFrame, output_path: str): wb Workbook() header_font Font(boldTrue, colorFFFFFF) header_fill PatternFill(start_color4472C4, end_color4472C4, fill_typesolid) # 明细页 ws1 wb.active ws1.title 明细数据 for row in dataframe_to_rows(df, indexFalse, headerTrue): ws1.append(row) for cell in ws1[1]: cell.font header_font cell.fill header_fill cell.alignment Alignment(horizontalcenter, verticalcenter) ws1.freeze_panes A2 # 汇总页按分类统计条数和平均价格 ws2 wb.create_sheet(分类汇总) summary df.groupby(category).agg(count(item_id, count), avg_price(price, mean)).reset_index() ws2.append([分类, 条数, 平均价格]) for _, row in summary.iterrows(): ws2.append([row[category], row[count], round(row[avg_price], 2)]) ws2[A1].font header_font ws2[B1].font header_font ws2[C1].font header_font # 自动列宽 for ws in wb.worksheets: for col in ws.columns: max_len max(len(str(c.value)) if c.value is not None else 0 for c in col) ws.column_dimensions[col[0].column_letter].width min(max_len 4, 60) wb.save(output_path)这里有一个细节对新手帮助很大dataframe_to_rows这个工具函数能把 DataFrame 一行一行地取出来然后直接用 ws1.append 塞进 Excel 单元格比手动遍历行和列省心很多。自动列宽这段代码我是从踩坑里学来的。第一次生成的 Excel所有列宽默认都是 8.38 个字符中文字符显示成一片乱码一样的窄条看得人头疼。后来我遍历每一列的最大长度动态设置列宽观感瞬间好了很多。还有一点想提醒openpyxl 写入时空值如果直接写进去Excel 里会显示成 None 字符串这非常丑。所以我处理的时候会把 None 统一替换成空字符串或者直接删除这些行否则报表阅读者会以为那是个叫 None 的异常值。4. 关键模块整合与自动化调度让程序每天自己跑一遍模块分开写完之后整合其实是个体力活但也正是在这个环节我体会到了为什么代码要写函数、要写日志。主入口长这样import logging import schedule import time from collector import collect_all from cleaner import clean_data from reporter import build_excel_report def main_job(): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(run.log, encodingutf-8), logging.StreamHandler(), ], ) logger logging.getLogger(main) logger.info(任务开始) raw_data collect_all() df pd.DataFrame(raw_data) df clean_data(df) build_excel_report(df, freport_{time.strftime(%Y%m%d)}.xlsx) logger.info(任务完成报告已生成) if __name__ __main__: schedule.every().day.at(09:30).do(main_job) while True: schedule.run_pending() time.sleep(1)整合完成后的第一件事不是直接等定时任务触发了而是先手动执行 main_job()。这个习惯一定不能省。因为我第一次直接跑代码时调度器在 while True 循环里空转了好几天回头一看日志才发现 main_job 里有一个拼写错误函数名没对上异常又没捕获整个任务静默失败了。从那以后我要求自己写完任何自动化任务必须手动执行一次确认成功再让它进入计划调度。日志这一块我想多花点篇幅说明。很多初学者喜欢用 print 打输出但这在自动化脚本里完全不够。程序挂在后台跑的时候print 的输出你看不到就算看到了也没有时间戳不知道是哪一次运行打出来的。用 logging 可以同时把日志写到控制台和文件并且在任务失败时记录完整的堆栈信息。更重要的是日志文件能帮我做长期复盘——周一的任务为什么比周三晚跑了十分钟翻日志就能找到真相。我在调度逻辑里其实也踩了一个隐形坑。用schedule.every().day.at(09:30)设定好后我一开始没有加异常兜底结果某一天源数据接口返回的是空列表主流程继续往下走清洗后的 DataFrame 是空表Excel 报表照样生成了但打开后里面只有表头没有任何数据。这种半成功的失败最坑人因为程序退出码是 0日志里也没有异常直到有人发现当天报表数据是空的回过头来排查才发现源端数据异常。所以后来我加了数据量阈值的校验如果清洗后的数据条数小于某个最小值就主动发送日志告警并且不生成报表、不覆盖旧文件。这个做法给我的启发是自动化任务不仅要让程序能跑更要让程序在不该跑的时候停下来。5. 常见问题与排查技巧实录我踩过的那几组坑这个项目做完我积累了一整套高频问题排查表这里挑几个最有代表性的整理出来每个问题背后都是一段真实耗时。现象根因解决方法请求时好时坏偶尔抛超时没设置 timeout有的连接挂死了所有 requests 请求统一加 timeout10并做 3 次重试CSV 文件打开全是乱码pandas 默认用 UTF-8 编码写入但 Excel 默认用系统编码读取保存时指定 encodingutf-8-sig或者在打开时手动选编码报表里的数字求和不对有些价格字段是字符串Excel 当成了文本清洗阶段统一 astype(float)并检查是否有空值残留DataFrame 输出 Excel 慢循环逐行 append 一条一条塞改用 openpyxl 的 dataframe_to_rows大幅提升写入速度定时任务到点没跑while True 循环被前面代码阻塞了把耗时长的任务放到子进程或者把 schedule 的循环放到单独线程程序报错却找不到原因只 print 没日志且没有 except 记录 traceback用 logger.exception 在异常分支打印完整堆栈到文件openpyxl 保存的文件模板被破坏直接打开原模板并另存为丢失了原格式改为 locked 或复制模板副本再写入保留原始文件七条问题里耗时最久的是第一个也就是请求超时。那次我调试到怀疑人生一会儿能通一会儿不能通还以为是服务器的问题直到抓包才发现是自己的连接池里堆积了一堆半开连接。加 timeout 和重试机制之后立竿见影。这里我可以给你的建议是所有需要依赖外部网络资源的代码第一版必须带上超时和重试两层保险否则后续所有的调试都会被这种偶发问题干扰。还有一个值得单独拎出来讲的坑就是中文编码。第一次我用 DataFrame.to_csv 保存结果时程序输出的是默认的 UTF-8 编码用 Excel 打开之后中文完全乱掉。当时我以为是 pandas 写坏了折腾了半天才发现是编码标准不统一。utf-8-sig 这个编码里带了一个特殊的 BOM 头Excel 才能准确识别。这个细节到现在我都提醒自己做任何导出操作时先确认目标软件对编码的预期。排查的时候我总结出一个通用套路遇到报错不要急着看倒数第一行先看完整 traceback 的最顶层定位是哪一个函数出了问题然后检查这一轮的输入数据长什么样。大部分数据处理 Bug问题都不在代码逻辑本身而在输入数据不符合你的假设。举个例子我代码里假设每个 item_id 都是字符串结果源端某一个分类下返回的是数字去重时就会因为类型不一致而无法清洗。类似这种情况用df.dtypes和df.head()打印一下五秒钟就能发现。6. 写在最后的几点体会一口气把这个实战项目从头讲到尾最后分享几个让我特别有感的经验。第一练习项目的难度不在会不会用某个库而在能不能把一个模糊目标拆成可执行的小步骤。没有拆解就不会有后面的层模块、独立函数和边界校验。上来就写主逻辑大概率会在中途被各种意外问题打断然后不了了之。第二所有项目里的技巧都可以用踩坑-总结这个模式掌握。我写采集层时第一次超时错误、写报表时第一次列宽显示混乱、写调度时第一次静默失败每一个都不光彩但每一个都是最深刻的教材。建议你也准备一个 bug 记录本把自己项目里每种问题的根因和方法记下来过两三个月回头看那份记录比教程值钱得多。第三也是最重要的实战练习不要贪全贪大。做一个能把数据从源头处理到产出报表的小闭环远远好过同时开三个半成品项目。单点突破做透一件事这种经验才能迁移到更复杂的场景里。这个项目之后我又陆续给它加了并发请求、配置化规则、数据库存储等扩展。每次改动都是一次新的实战练习。如果你现在正处于什么都会一点但又什么都不敢说会的尴尬期别多想找一个真实的小问题从第一行代码写到输出成果再回来读一遍这篇总结你一定会比听十遍课收获更多。