★ FEATURED ARTICLE 1. 项目缘起与整体设计思路1.1 这个项目到底在做什么先把话说在前头这个项目的核心任务是把一批公开的、历史性的公共卫生统计数据从原始来源采集下来、清洗干净、结构化入库最终对外提供一份可查询、可下载、可追溯的数据集。标题里写的“紧急上线”意味着整个项目从立项到交付的时间窗口被压得极短通常只有常规周期的三分之一甚至更少。这种节奏下技术选型的第一原则不是“最优解”而是“最快能跑通且不出错的解”。我接手这个需求的时候第一反应是数据本身不复杂复杂的是时间。历史数据往往分散在多个页面、多种格式里有的是表格有的是文本段落有的还夹在图片里。如果按常规做法先做需求评审、再设计架构、再开发测试黄花菜都凉了。所以整个项目的设计思路就八个字先通后优边跑边修。具体来说我把项目拆成了四个阶段数据源摸底、采集脚本快速实现、数据清洗与校验、接口与页面交付。每个阶段都设了一个“最小可用版本”的硬性标准只要达到这个标准就立刻推进下一阶段绝不恋战。比如采集阶段我不追求一次抓全所有字段先把最核心的几个字段跑通剩下的字段后续迭代补上。这个思路的好处是任何时刻你手里都有一个能演示、能交付的东西而不是一堆半成品。坏处也很明显技术债会堆积。所以我在每个阶段都留了“还债窗口”比如采集脚本里预留了字段扩展位清洗逻辑里预留了规则插槽方便后续快速修补。1.2 为什么选择这套技术方案时间紧任务重技术选型就得务实。我最终确定的方案是Python Requests Pandas SQLite Flask。这套组合看起来平平无奇但每一个选择都有明确的理由。选 Python 是因为它的生态最全遇到任何格式的数据几乎都能找到现成的库来处理。Requests 负责网络请求简单直接不需要额外配置。Pandas 负责数据清洗和转换它的 DataFrame 结构处理表格数据简直是降维打击。SQLite 作为存储层零配置、单文件、支持 SQL 查询对于这种中小规模的历史数据集来说完全够用而且部署时直接拷贝文件就行省去了数据库安装和权限配置的麻烦。Flask 负责对外提供接口和简单页面轻量到极致几行代码就能跑起来。这里我要特别说一下为什么没用更“高级”的方案。有人可能会问为什么不用 Scrapy 做采集为什么不用 PostgreSQL 做存储为什么不用 FastAPI 做接口答案很简单在这个时间窗口下每多一个依赖就多一个出问题的环节。Scrapy 的异步模型确实快但它的学习曲线和调试成本在紧急项目里是负资产。PostgreSQL 功能强大但安装配置、用户权限、连接池调优每一项都要时间。FastAPI 性能好但 Flask 我闭着眼睛都能写出错概率更低。紧急项目的技术选型铁律选你最熟的那套而不是最强的那套。熟悉意味着你能预判问题意味着出错了你知道去哪找原因。另外我把整个项目做成了单文件脚本加一个轻量 Web 服务的结构。采集和清洗是一个独立的 Python 脚本可以单独运行、单独调试。Web 服务是另一个独立文件只负责读数据库和返回结果。两者通过 SQLite 文件解耦互不干扰。这样即使 Web 服务挂了采集脚本照样能跑采集脚本出问题了也不影响已经入库的数据对外服务。1.3 数据源摸底的关键发现动手写代码之前我花了大概两个小时做数据源摸底。这一步绝对不能省否则后面全是返工。摸底的目标是搞清楚三件事数据在哪、数据长什么样、数据有多少。数据在哪历史数据通常有几个常见的存放位置。一是专门的公开数据页面以表格形式呈现二是新闻稿或公告正文里以文字描述形式出现三是附件下载比如 PDF 或 Excel 文件。我这次遇到的情况是三种都有所以采集策略必须分开处理。数据长什么样我随机抽取了几个时间点的数据发现格式并不统一。早期的数据字段少、命名不规范后期的数据字段多、结构清晰。这意味着清洗规则不能一刀切得按时间段做兼容处理。数据有多少我粗略估算了一下总记录数在几千条左右字段数在十个上下。这个规模用 SQLite 绰绰有余用 Pandas 在内存里处理也毫无压力。摸底完成后我画了一张简单的数据流图在脑子里没来得及画出来原始页面 - 采集脚本 - 原始数据暂存 - 清洗脚本 - 结构化数据 - SQLite - Flask 接口 - 前端页面。每个环节之间都有明确的输入输出约定方便并行开发和调试。2. 核心细节解析与实操要点2.1 采集环节的难点与应对采集是第一个硬骨头。历史数据页面往往有几个特点分页多、加载慢、结构乱。我遇到的分页问题最典型有的页面用 URL 参数分页有的用 JavaScript 动态加载有的干脆把所有数据塞在一个超长页面里。对于 URL 参数分页处理起来最简单改一下参数循环请求就行。对于 JavaScript 动态加载就得分析网络请求找到真正的数据接口。这里有个小技巧打开浏览器的开发者工具切换到 Network 面板筛选 XHR 或 Fetch 请求通常能直接看到返回 JSON 数据的接口。拿到接口地址后用 Requests 直接请求比解析 HTML 快得多也稳得多。对于超长页面我的做法是先请求整个页面然后用 Pandas 的read_html一次性解析所有表格。这个函数会自动把页面里的 table 标签转成 DataFrame 列表省去了手写解析逻辑的麻烦。不过它也有坑如果页面里有多个表格返回的列表顺序不一定和页面视觉顺序一致需要根据列名或内容特征来判断哪个是目标表格。采集环节的注意事项一定要加请求间隔和重试机制。历史数据页面通常不会做严格的反爬限制但如果你请求太快对方服务器可能直接拒绝服务。我的做法是每次请求间隔 1 到 2 秒失败后重试三次每次重试间隔翻倍。还有一个容易被忽略的点编码问题。中文页面的编码可能是 UTF-8、GBK、GB2312 等多种Requests 自动识别的编码有时候会出错导致中文变成乱码。我的做法是手动指定编码或者用chardet库先检测再解码。这个细节看似小但一旦出错后面所有数据都是废的。2.2 数据清洗的规则设计采集下来的原始数据就像刚从地里拔出来的萝卜带着泥。清洗的目标就是去泥、削皮、切块让它变成能直接下锅的食材。我设计的清洗规则分四层格式统一、缺失处理、异常值修正、字段标准化。格式统一日期字段是最典型的。原始数据里可能有“2020年1月23日”、“2020-01-23”、“01/23/2020”等多种写法。我的做法是全部转成 ISO 标准格式YYYY-MM-DD用 Pandas 的to_datetime函数配合format参数逐个尝试解析。数值字段也要统一比如有的带千分位逗号有的带单位后缀全部去掉后转成数值类型。缺失处理历史数据早期字段缺失是常态。我的策略是关键字段缺失的记录直接标记为“不完整”但不删除保留在数据库里供参考非关键字段缺失的用空值或默认值填充。这里要特别注意绝对不能随意填充关键字段的缺失值比如用前一天的数值填充今天的这在数据伦理上是严重问题。异常值修正有些数据明显不合逻辑比如数值为负数、日期超出合理范围、同一记录重复出现等。我的做法是写一个校验函数逐条检查发现异常就记录到日志里人工确认后再决定是修正还是剔除。字段标准化原始数据的列名可能五花八门比如“确诊人数”、“确诊病例数”、“累计确诊”其实是一个意思。我建了一个映射表把所有可能的列名统一映射到标准字段名上。这个映射表是清洗脚本的核心配置后续如果发现新的列名写法只需要往映射表里加一行就行。2.3 存储结构的设计考量SQLite 虽然简单但表结构设计不能马虎。我设计了两张表一张是原始数据表一张是清洗后的标准数据表。原始数据表的结构很宽松基本上就是采集到什么存什么字段类型全部用 TEXT方便兼容各种格式。这张表的作用是留底万一清洗规则出了问题还能从原始数据重新跑一遍。标准数据表的结构就严格多了。我定义了明确的字段类型和约束日期字段用 DATE 类型数值字段用 INTEGER 或 REAL关键字段加 NOT NULL 约束。另外还加了几个辅助字段data_source记录数据来源created_at记录入库时间updated_at记录最后更新时间。这些字段在后续排查问题时非常有用。索引方面我在日期字段上建了索引因为查询大多是按时间范围来的。这个索引让查询速度从秒级降到了毫秒级效果立竿见影。存储设计的一个经验永远保留原始数据。清洗规则可能会变业务需求可能会变但原始数据不会变。保留原始数据就保留了重新处理的可能。2.4 接口与页面的快速实现Flask 的轻量在这里体现得淋漓尽致。我写了三个接口一个按日期范围查询一个按关键词搜索一个返回数据总量和日期范围等元信息。每个接口的代码都不超过二十行核心逻辑就是拼 SQL、查数据库、返回 JSON。前端页面更简单一个 HTML 文件加一点原生 JavaScript。页面顶部是搜索框和日期选择器中间是数据表格底部是分页控件。没有用任何前端框架因为在这个时间窗口下引入 React 或 Vue 的构建工具链纯属自找麻烦。原生 JavaScript 虽然写起来啰嗦一点但零依赖、零构建、直接打开就能用。这里有个小技巧数据表格的渲染用了一个简单的模板字符串拼接而不是 DOM 操作。因为数据量不大一次性渲染几百行完全没问题。如果数据量再大一些可以考虑用虚拟滚动但那是后续优化的事了。3. 实操过程与核心环节实现3.1 环境准备与依赖安装环境准备这一步我建议用虚拟环境避免污染全局 Python 环境。命令很简单python -m venv venv source venv/bin/activate # Linux/Mac # 或者 venv\Scripts\activate # Windows然后安装依赖pip install requests pandas flask chardet lxml这里解释一下每个依赖的作用。requests负责网络请求pandas负责数据处理flask负责 Web 服务chardet负责编码检测lxml是 Pandas 解析 HTML 时的底层依赖不装的话read_html会报错。注意lxml在某些系统上安装可能会遇到编译问题如果报错可以换成html5lib或BeautifulSoup作为解析器。不过lxml的速度最快优先尝试。3.2 采集脚本的完整实现采集脚本的核心逻辑是遍历所有分页请求每一页解析表格追加到列表最后合并成一个 DataFrame。import requests import pandas as pd import time import chardet def fetch_page(url, paramsNone): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } for attempt in range(3): try: resp requests.get(url, paramsparams, headersheaders, timeout15) resp.raise_for_status() # 编码检测 detected chardet.detect(resp.content) if detected[encoding]: resp.encoding detected[encoding] return resp.text except Exception as e: print(f请求失败第 {attempt 1} 次重试: {e}) time.sleep(2 ** attempt) return None def parse_tables(html): try: tables pd.read_html(html) # 根据列名特征筛选目标表格 for table in tables: if any(日期 in str(col) or date in str(col).lower() for col in table.columns): return table except Exception as e: print(f解析表格失败: {e}) return None def collect_all(base_url, total_pages): all_data [] for page in range(1, total_pages 1): print(f正在采集第 {page} 页...) html fetch_page(base_url, params{page: page}) if html: table parse_tables(html) if table is not None: all_data.append(table) time.sleep(1.5) # 请求间隔 if all_data: return pd.concat(all_data, ignore_indexTrue) return pd.DataFrame()这段代码有几个关键点。第一fetch_page里做了编码检测避免中文乱码。第二重试机制用了指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒给服务器喘息时间。第三parse_tables里根据列名特征筛选目标表格而不是盲目取第一个因为页面里可能有导航表格、布局表格等干扰项。3.3 清洗脚本的完整实现清洗脚本的核心逻辑是重命名列、转换类型、处理缺失、修正异常、输出标准数据。import pandas as pd import numpy as np # 列名映射表 COLUMN_MAP { 确诊人数: confirmed, 确诊病例数: confirmed, 累计确诊: confirmed, 治愈人数: recovered, 治愈病例数: recovered, 累计治愈: recovered, 死亡人数: deaths, 死亡病例数: deaths, 累计死亡: deaths, 日期: date, 时间: date, } def clean_data(df): # 重命名列 df df.rename(columnsCOLUMN_MAP) # 只保留标准字段 standard_cols [date, confirmed, recovered, deaths] existing_cols [c for c in standard_cols if c in df.columns] df df[existing_cols] # 日期转换 if date in df.columns: df[date] pd.to_datetime(df[date], errorscoerce) df df.dropna(subset[date]) df[date] df[date].dt.strftime(%Y-%m-%d) # 数值转换 for col in [confirmed, recovered, deaths]: if col in df.columns: df[col] df[col].astype(str).str.replace(,, ).str.replace(例, ) df[col] pd.to_numeric(df[col], errorscoerce) df[col] df[col].fillna(0).astype(int) # 异常值修正负数置零 for col in [confirmed, recovered, deaths]: if col in df.columns: df.loc[df[col] 0, col] 0 # 去重 df df.drop_duplicates(subset[date], keeplast) # 按日期排序 df df.sort_values(date).reset_index(dropTrue) return df这里有几个细节值得展开。第一列名映射表是清洗脚本的核心配置后续发现新的列名写法只需要往字典里加一行。第二日期转换用了errorscoerce无法解析的日期会变成 NaT然后被dropna剔除避免脏数据进入数据库。第三数值转换先转字符串再去掉逗号和单位然后转数值无法转换的变成 NaN最后填充为 0。第四去重时保留最后一条因为后面的数据通常更准确。3.4 入库与接口实现入库逻辑很简单用 Pandas 的to_sql方法直接写入 SQLiteimport sqlite3 def save_to_db(df, db_pathdata.db, table_namerecords): conn sqlite3.connect(db_path) df.to_sql(table_name, conn, if_existsreplace, indexFalse) # 创建索引 conn.execute(fCREATE INDEX IF NOT EXISTS idx_date ON {table_name}(date)) conn.commit() conn.close()Flask 接口的实现from flask import Flask, request, jsonify import sqlite3 app Flask(__name__) def query_db(sql, params()): conn sqlite3.connect(data.db) conn.row_factory sqlite3.Row cursor conn.execute(sql, params) rows [dict(row) for row in cursor.fetchall()] conn.close() return rows app.route(/api/records) def get_records(): start request.args.get(start, 1900-01-01) end request.args.get(end, 2100-01-01) sql SELECT * FROM records WHERE date BETWEEN ? AND ? ORDER BY date return jsonify(query_db(sql, (start, end))) app.route(/api/summary) def get_summary(): sql SELECT MIN(date) as min_date, MAX(date) as max_date, COUNT(*) as total FROM records return jsonify(query_db(sql)) if __name__ __main__: app.run(host0.0.0.0, port5000)这两个接口覆盖了最核心的查询需求。/api/records支持日期范围过滤/api/summary返回元信息。代码量极少但功能完整。3.5 前端页面的快速搭建前端页面用一个 HTML 文件搞定核心是 fetch 请求和表格渲染!DOCTYPE html html head meta charsetUTF-8 title历史数据查询/title style body { font-family: sans-serif; margin: 20px; } table { border-collapse: collapse; width: 100%; } th, td { border: 1px solid #ddd; padding: 8px; text-align: left; } th { background: #f5f5f5; } .controls { margin-bottom: 20px; } /style /head body h1历史数据查询/h1 div classcontrols label开始日期: input typedate idstart/label label结束日期: input typedate idend/label button onclickloadData()查询/button /div table idresult thead trth日期/thth确诊/thth治愈/thth死亡/th/tr /thead tbody/tbody /table script async function loadData() { const start document.getElementById(start).value || 1900-01-01; const end document.getElementById(end).value || 2100-01-01; const resp await fetch(/api/records?start${start}end${end}); const data await resp.json(); const tbody document.querySelector(#result tbody); tbody.innerHTML data.map(row trtd${row.date}/tdtd${row.confirmed}/tdtd${row.recovered}/tdtd${row.deaths}/td/tr ).join(); } loadData(); /script /body /html这个页面没有任何依赖打开就能用。表格渲染用了模板字符串简单直接。日期选择器用了浏览器原生的input typedate省去了引入日期库的麻烦。4. 常见问题与排查技巧实录4.1 采集环节的典型问题问题一请求返回 403 或 429这是最常见的反爬响应。403 表示服务器拒绝访问429 表示请求过于频繁。我的排查思路是先检查 User-Agent 是否被识别为爬虫换成常见浏览器的 UA 试试如果还不行降低请求频率把间隔从 1 秒调到 3 秒甚至 5 秒再不行就检查是否有其他请求头缺失比如 Referer、Accept-Language 等。问题二页面返回的内容和浏览器看到的不一样这种情况通常是 JavaScript 动态渲染导致的。浏览器看到的是 JS 执行后的结果而 Requests 拿到的是原始 HTML。解决办法是分析网络请求找到真正的数据接口。如果找不到接口就得考虑用 Selenium 或 Playwright 这类浏览器自动化工具但那会显著增加复杂度和时间成本。问题三中文乱码前面提过编码问题是中文采集的常见坑。排查方法是打印resp.encoding和resp.apparent_encoding对比一下。如果两者不一致手动设置成正确的那个。chardet库的检测结果也不是百分之百准确遇到乱码时可以尝试常见的几种编码UTF-8、GBK、GB2312、GB18030。4.2 清洗环节的典型问题问题一日期解析失败不同格式的日期混在一起to_datetime可能解析不了。我的做法是先用errorscoerce把无法解析的变成 NaT然后打印出这些 NaT 对应的原始值看看是什么格式再往解析规则里加对应的 format。Pandas 的to_datetime支持传入 format 列表会逐个尝试。问题二数值字段混入了非数值字符比如“1,234例”、“约 500 人”、“暂无”等。我的做法是先用正则表达式提取数字部分再转数值。正则可以用re.search(r\d, str(value))提取不到的就置为 0 或 NaN。问题三重复记录同一日期出现多条记录可能是采集时重复抓取也可能是原始数据本身就有重复。我的做法是先按日期分组看看重复的情况。如果是完全相同的记录直接去重如果是同一天不同数值保留最后一条因为后面的数据通常经过了修正。4.3 部署与运行环节的典型问题问题一SQLite 并发写入报错SQLite 默认的锁机制在并发写入时会报 “database is locked”。解决办法是写入时用conn.execute(BEGIN IMMEDIATE)显式开启事务或者干脆把写入操作串行化避免并发。对于这个项目的数据量串行写入完全够用。问题二Flask 开发服务器性能不足Flask 自带的开发服务器是单线程的并发请求多了会阻塞。解决办法是用 Gunicorn 或 uWSGI 部署或者至少开启多线程模式app.run(threadedTrue)。对于内部使用的小规模查询多线程模式已经足够。问题三数据更新后页面不刷新浏览器缓存了旧的接口响应。解决办法是在接口响应头里加上Cache-Control: no-cache或者在前端 fetch 请求里加上时间戳参数强制绕过缓存。4.4 常见问题速查表问题现象可能原因排查方法解决方案请求返回 403UA 被识别为爬虫打印请求头对比浏览器更换 UA降低频率页面内容与浏览器不一致JS 动态渲染查看 Network 面板找数据接口或用自动化工具中文乱码编码识别错误打印 encoding 对比手动指定编码日期解析失败格式不统一打印 NaT 原始值增加 format 规则数值字段含非数字单位或描述混入抽样检查正则提取数字数据库锁定并发写入查看报错堆栈串行写入或显式事务页面不刷新浏览器缓存查看响应头加 no-cache 头排查问题的通用思路先定位问题发生在哪个环节再缩小范围到具体函数最后打印关键变量确认。不要一上来就改代码先搞清楚问题是什么。4.5 几个独家避坑技巧第一个技巧采集前先手动保存一份原始页面。用浏览器的“另存为”功能把页面存下来然后用这份本地文件调试解析逻辑。这样调试时不需要反复请求网络速度快也不会给对方服务器造成压力。解析逻辑调通后再换成在线请求。第二个技巧清洗脚本要能重复运行。也就是说同一份原始数据跑两次清洗结果应该完全一样。这要求清洗逻辑是幂等的不能有随机性不能依赖外部状态。做到这一点后续发现清洗规则有问题时直接改规则重跑就行不用担心数据被搞乱。第三个技巧数据库文件要定期备份。SQLite 是单文件数据库备份就是拷贝文件。我在每次批量更新数据之前都会先拷贝一份data.db到data.db.bak。万一更新出错直接回滚文件就行比写回滚 SQL 快得多。第四个技巧接口返回的数据量要控制。如果查询结果有几千条一次性返回 JSON 会导致响应体过大前端渲染也会卡顿。我的做法是默认只返回最近 100 条如果需要更多通过分页参数控制。这个细节在紧急上线时容易被忽略但用户体验差别很大。5. 上线后的观察与后续优化方向5.1 上线初期的监控要点紧急上线不等于上线完就不管了。上线后的头几个小时是最关键的我重点盯三个指标接口响应时间、错误日志、数据完整性。接口响应时间用 Flask 的日志就能看到如果某个查询超过 1 秒就得检查是不是索引没生效或者 SQL 写得太差。错误日志主要看有没有 500 错误有的话立刻排查。数据完整性是抽查几条记录和原始数据对比确认没有在清洗或入库环节丢失。上线初期的一个经验准备一个手动回滚方案。如果发现数据有严重问题能在一分钟内把服务切回旧版本或旧数据。这个方案不需要多复杂拷贝文件加重启服务就够了但必须提前准备好不能等出事了再想。5.2 数据质量的持续校验历史数据的一次性采集和清洗只是开始后续还需要持续校验。我的做法是写一个校验脚本定期跑一遍检查几个关键点日期是否连续、数值是否单调递增对于累计数据、字段是否有异常空值。发现异常就发通知人工确认后再决定是否修正。这个校验脚本本身不复杂核心就是几条 SQL 查询加一些 Pandas 检查逻辑。但它的价值很大能在问题扩大之前及时发现。5.3 后续可以扩展的方向如果时间允许这个项目还有几个明显的优化方向。一是采集环节可以做成增量更新只抓取新增的数据而不是每次全量重跑。二是清洗环节可以引入更完善的规则引擎把清洗规则配置化而不是硬编码在脚本里。三是接口层可以加缓存对于不常变的数据缓存几分钟能显著降低数据库压力。四是前端可以加图表展示用 ECharts 或 Chart.js 把数据可视化比纯表格直观得多。不过这些都是后话了。在紧急上线的场景下最重要的是先把核心功能跑通让数据能查、能用。优化的事情等系统稳定了再慢慢做。我在这个项目里最大的体会就是紧急项目拼的不是技术深度而是判断力。知道什么该做、什么不该做、什么先做、什么后做比会写多复杂的代码重要得多。
阅读完成 · 觉得有帮助？
