1. 腾讯疫情API接口解析与数据获取腾讯疫情数据API曾经是获取国内疫情动态的重要渠道虽然现在已经停止更新但其中的历史数据仍然具有很高的研究价值。我们先来看看如何找到并解析这些API接口。我花了些时间研究腾讯疫情数据的接口结构发现主要有两类关键接口省级历史数据接口https://api.inews.qq.com/newsqa/v1/query/pubished/daily/list?province省份名称市级历史数据接口https://api.inews.qq.com/newsqa/v1/query/pubished/daily/list?province省名称city市名称这些接口返回的是标准的JSON格式数据包含日期、确诊数、治愈数等关键字段。实测发现数据时间跨度从2020年1月持续到接口停更前覆盖了国内所有省级行政区。import requests import json # 获取广东省疫情数据示例 url https://api.inews.qq.com/newsqa/v1/query/pubished/daily/list?province广东 response requests.get(url) data json.loads(response.text)[data] print(data[0]) # 打印第一条数据返回的数据结构大致如下{ date: 2020.01.21, confirm: 6, suspect: 0, dead: 0, heal: 0, province: 广东, city: }2. 构建自动化爬虫采集系统2.1 获取完整的行政区划列表要爬取全国数据首先需要知道有哪些省份和城市。通过分析腾讯疫情首页的接口我发现这个接口可以获取完整的行政区树def get_area_tree(): url https://view.inews.qq.com/g2/getOnsInfo?namedisease_h5 response requests.get(url).json() data json.loads(response[data]) return data[areaTree][0][children] # 中国数据在第一个节点 area_tree get_area_tree() provinces [item[name] for item in area_tree] print(f共获取到{len(provinces)}个省级行政区)2.2 实现多级数据爬取有了行政区列表就可以编写爬虫逐级获取数据。这里我设计了一个三级爬取策略遍历所有省份对每个省份获取其下辖城市列表分别获取省级汇总数据和各城市数据def crawl_all_data(): all_data {} headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) } for province in provinces: # 获取省级数据 province_url fhttps://api.inews.qq.com/newsqa/v1/query/pubished/daily/list?province{province} province_data requests.get(province_url, headersheaders).json()[data] all_data[province] {province_data: province_data} # 获取市级数据 cities [city[name] for city in area_tree if city[name] province][0][children] for city in cities: city_url fhttps://api.inews.qq.com/newsqa/v1/query/pubished/daily/list?province{province}city{city} city_data requests.get(city_url, headersheaders).json()[data] all_data[province][city] city_data time.sleep(1) # 礼貌性延迟 return all_data3. 数据清洗与标准化处理原始数据往往存在各种问题需要进行清洗才能使用。我总结了几个常见的处理步骤3.1 处理缺失值与异常值检查数据时发现有些日期的字段值为null还有些早期数据存在录入错误。我的处理方法是def clean_data(raw_data): # 填充缺失值 for record in raw_data: for key in [confirm, suspect, heal, dead]: if record.get(key) is None: record[key] 0 # 修正早期数据错误示例 if raw_data[0][province] 湖北 and raw_data[0][date] 2020.01.23: raw_data[0][confirm] 444 # 修正确诊数 return raw_data3.2 日期格式标准化原始数据中的日期格式不统一需要转换为标准格式from datetime import datetime def standardize_dates(data): for item in data: # 统一转换为YYYY-MM-DD格式 if . in item[date]: dt datetime.strptime(item[date], %Y.%m.%d) else: dt datetime.strptime(item[date], %Y-%m-%d) item[date] dt.strftime(%Y-%m-%d) return data3.3 数据验证与去重def validate_data(data): valid_data [] dates_seen set() for item in data: # 检查必填字段 if not all(key in item for key in [date, confirm, heal, dead]): continue # 去重 if item[date] in dates_seen: continue dates_seen.add(item[date]) valid_data.append(item) return valid_data4. 数据库设计与实现4.1 MySQL数据库表设计经过多次迭代我最终采用了这样的表结构设计CREATE TABLE province_daily ( id int(11) NOT NULL AUTO_INCREMENT, date date NOT NULL, province varchar(50) NOT NULL, city varchar(50) DEFAULT NULL, confirm int(11) DEFAULT 0, confirm_add int(11) DEFAULT 0, suspect int(11) DEFAULT 0, suspect_add int(11) DEFAULT 0, heal int(11) DEFAULT 0, heal_add int(11) DEFAULT 0, dead int(11) DEFAULT 0, dead_add int(11) DEFAULT 0, update_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY idx_date_province_city (date,province,city) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这个设计有以下几个特点使用复合唯一索引防止重复数据同时记录累计值和当日新增值支持省、市两级数据存储自动记录数据更新时间4.2 使用SQLite实现轻量级存储如果不想部署MySQLSQLite是个不错的替代方案。Python内置支持单文件即可运行import sqlite3 def init_sqlite_db(): conn sqlite3.connect(tencent_covid.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS province_daily ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, province TEXT NOT NULL, city TEXT, confirm INTEGER DEFAULT 0, confirm_add INTEGER DEFAULT 0, suspect INTEGER DEFAULT 0, suspect_add INTEGER DEFAULT 0, heal INTEGER DEFAULT 0, heal_add INTEGER DEFAULT 0, dead INTEGER DEFAULT 0, dead_add INTEGER DEFAULT 0, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(date, province, city) ) ) conn.commit() return conn4.3 批量数据入库优化当处理大量数据时逐条插入效率极低。我推荐使用批量插入方式def batch_insert(conn, data): sql INSERT OR IGNORE INTO province_daily (date, province, city, confirm, confirm_add, suspect, suspect_add, heal, heal_add, dead, dead_add) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) batch [] for item in data: batch.append(( item[date], item[province], item.get(city, ), item[confirm], item.get(confirm_add, 0), item.get(suspect, 0), item.get(suspect_add, 0), item[heal], item.get(heal_add, 0), item[dead], item.get(dead_add, 0) )) conn.executemany(sql, batch) conn.commit()5. 完整代码实现与部署5.1 项目结构tencent-covid-archiver/ ├── config.py # 配置文件 ├── crawler.py # 爬虫主逻辑 ├── database.py # 数据库操作 ├── main.py # 主程序入口 ├── requirements.txt # 依赖库 └── utils.py # 工具函数5.2 核心爬虫代码# crawler.py import requests import json import time from datetime import datetime from typing import Dict, List class TencentCovidCrawler: def __init__(self): self.base_url https://api.inews.qq.com/newsqa/v1/query/pubished/daily/list self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) } def get_provinces(self) - List[str]: 获取所有省份列表 url https://view.inews.qq.com/g2/getOnsInfo?namedisease_h5 response requests.get(url).json() data json.loads(response[data]) return [item[name] for item in data[areaTree][0][children]] def get_cities(self, province: str) - List[str]: 获取指定省份下的城市列表 url https://view.inews.qq.com/g2/getOnsInfo?namedisease_h5 response requests.get(url).json() data json.loads(response[data]) for item in data[areaTree][0][children]: if item[name] province: return [city[name] for city in item[children]] return [] def crawl_province_data(self, province: str) - List[Dict]: 爬取省级历史数据 url f{self.base_url}?province{province} try: response requests.get(url, headersself.headers) response.raise_for_status() return json.loads(response.text)[data] except Exception as e: print(f获取{province}数据失败: {str(e)}) return [] def crawl_city_data(self, province: str, city: str) - List[Dict]: 爬取市级历史数据 url f{self.base_url}?province{province}city{city} try: response requests.get(url, headersself.headers) response.raise_for_status() return json.loads(response.text)[data] except Exception as e: print(f获取{province}-{city}数据失败: {str(e)}) return [] def crawl_all_data(self, delay: float 1.0) - Dict[str, Dict]: 爬取全国所有数据 provinces self.get_provinces() all_data {} for province in provinces: print(f正在处理 {province}...) province_data self.crawl_province_data(province) all_data[province] {province: province_data} cities self.get_cities(province) for city in cities: city_data self.crawl_city_data(province, city) all_data[province][city] city_data time.sleep(delay) time.sleep(delay) return all_data5.3 数据处理器# processor.py import json from datetime import datetime from typing import Dict, List class DataProcessor: staticmethod def clean_raw_data(raw_data: List[Dict]) - List[Dict]: 清洗原始数据 cleaned [] for item in raw_data: # 处理缺失值 item.setdefault(confirm, 0) item.setdefault(suspect, 0) item.setdefault(heal, 0) item.setdefault(dead, 0) item.setdefault(confirm_add, 0) item.setdefault(suspect_add, 0) item.setdefault(heal_add, 0) item.setdefault(dead_add, 0) # 标准化日期 if date in item: if . in item[date]: dt datetime.strptime(item[date], %Y.%m.%d) else: dt datetime.strptime(item[date], %Y-%m-%d) item[date] dt.strftime(%Y-%m-%d) cleaned.append(item) return cleaned staticmethod def transform_to_db_format(data: Dict[str, Dict]) - List[Dict]: 转换为数据库存储格式 db_records [] for province, cities in data.items(): # 处理省级数据 if province in cities: for record in cities[province]: db_records.append({ date: record[date], province: province, city: , confirm: record[confirm], confirm_add: record.get(confirm_add, 0), suspect: record[suspect], suspect_add: record.get(suspect_add, 0), heal: record[heal], heal_add: record.get(heal_add, 0), dead: record[dead], dead_add: record.get(dead_add, 0) }) # 处理市级数据 for city, records in cities.items(): if city province: continue for record in records: db_records.append({ date: record[date], province: province, city: city, confirm: record[confirm], confirm_add: record.get(confirm_add, 0), suspect: record[suspect], suspect_add: record.get(suspect_add, 0), heal: record[heal], heal_add: record.get(heal_add, 0), dead: record[dead], dead_add: record.get(dead_add, 0) }) return db_records6. 实际应用与数据分析建议有了本地疫情数据库后可以进行各种分析。这里分享几个实际应用场景6.1 疫情发展趋势分析import pandas as pd import matplotlib.pyplot as plt def plot_province_trend(conn, province): 绘制省份疫情趋势图 sql f SELECT date, confirm, heal, dead FROM province_daily WHERE province{province} AND city ORDER BY date df pd.read_sql(sql, conn) df[date] pd.to_datetime(df[date]) plt.figure(figsize(12, 6)) plt.plot(df[date], df[confirm], label确诊) plt.plot(df[date], df[heal], label治愈) plt.plot(df[date], df[dead], label死亡) plt.title(f{province}疫情发展趋势) plt.xlabel(日期) plt.ylabel(人数) plt.legend() plt.grid() plt.show()6.2 区域对比分析def compare_provinces(conn, provinces): 多省份对比分析 dfs [] for province in provinces: sql f SELECT date, confirm FROM province_daily WHERE province{province} AND city ORDER BY date df pd.read_sql(sql, conn) df[province] province dfs.append(df) combined pd.concat(dfs) combined[date] pd.to_datetime(combined[date]) plt.figure(figsize(12, 6)) for province, group in combined.groupby(province): plt.plot(group[date], group[confirm], labelprovince) plt.title(各省份确诊人数对比) plt.xlabel(日期) plt.ylabel(确诊人数) plt.legend() plt.grid() plt.show()6.3 数据导出与共享为了方便其他研究者使用可以将数据导出为通用格式def export_to_csv(conn, output_dir): 导出数据为CSV文件 # 导出省级数据 province_sql SELECT date, province, confirm, heal, dead FROM province_daily WHERE city ORDER BY province, date province_df pd.read_sql(province_sql, conn) province_df.to_csv(f{output_dir}/province_level.csv, indexFalse) # 导出市级数据 city_sql SELECT date, province, city, confirm, heal, dead FROM province_daily WHERE city! ORDER BY province, city, date city_df pd.read_sql(city_sql, conn) city_df.to_csv(f{output_dir}/city_level.csv, indexFalse)7. 项目部署与维护建议7.1 定时自动更新虽然腾讯疫情数据已经停止更新但这个框架可以用于其他类似场景。使用crontab设置定时任务# 每天凌晨3点执行一次爬虫 0 3 * * * /usr/bin/python3 /path/to/main.py /var/log/covid_crawler.log 217.2 异常处理与监控建议添加以下监控措施记录每次爬取的数据量监控API响应时间设置失败告警机制# 在爬虫类中添加监控方法 def monitor_crawl(self): start_time time.time() data self.crawl_all_data() end_time time.time() stats { timestamp: datetime.now().isoformat(), duration: end_time - start_time, province_count: len(data), total_records: sum(len(v) for v in data.values()) } with open(crawl_stats.json, a) as f: f.write(json.dumps(stats) \n) return data7.3 数据备份策略建议定期备份数据库MySQL可以使用mysqldumpSQLite直接复制数据库文件# MySQL备份示例 mysqldump -u username -p covid_db backup_$(date %Y%m%d).sql # SQLite备份示例 cp tencent_covid.db backup_$(date %Y%m%d).db