简介本资源是一个基于Python开发的疫情信息发布平台完整源码项目面向计算机类专业学生如计科、大数据、人工智能等及课程设计、毕业设计与期末大作业实践者聚焦疫情数据采集、分析与可视化呈现解决教学场景中缺乏可运行全栈项目案例的痛点。压缩包共72个文件涵盖前端Vue组件7个.vue、JavaScript逻辑18个.js、样式与配置5个.css、2个.html、3个.json等、后端Python核心代码3个.py含app.py主服务与updata.py数据更新脚本、数据库SQL脚本cov.sql及项目说明文档README.md、项目使用说明.md整体大小为7.74MB结构清晰模块划分明确。已有217人学习下载适合从入门到进阶的工程化实践既可直接部署运行查看疫情地图、趋势图表等可视化效果又便于理解前后端分离架构、Flask/Django服务集成、ECharts数据渲染及MySQL数据管理全流程还支持基于country.json等数据源进行二次拓展开发。1. 这不是又一个“疫情看板”而是一套可复用的 Python 全栈数据管道从原始 CSV/Excel 接入、Pandas 清洗聚合、Plotly/Dash 动态渲染到 MySQL 存储与 Flask API 暴露前端 Vue 或 React 可直接消费——适合刚做完课程设计想进真实项目组的应届生也适合需要快速验证数据口径、调整指标逻辑的业务分析师很多人下载这个“疫情信息发布平台”源码后第一反应是界面挺像模像样但跑不起来。不是缺requirements.txt里的包而是根本没搞清它背后的数据流闭环原始疫情数据如国家卫健委每日通报 CSV进来要经过字段标准化省名统一为“北京市”而非“北京”、时间序列对齐补全缺失日期、地理编码把“武汉”转成经纬度、多维聚合按周/省/病种统计确诊/治愈/死亡最后才交给前端画图。中间任何一环断掉可视化就变成静态截图。本项目真正价值不在“展示疫情”而在提供一套可替换数据源、可重定义指标、可插拔前端框架的分析骨架。它用的是最通用的 Python 生态组合Pandas 处理表格、SQLAlchemy 管理数据库连接、Flask 提供 REST 接口、Plotly Express 快速出图——没有冷门库没有私有协议所有依赖都能在 PyPI 直接pip install。如果你正卡在“数据清洗后不知道怎么喂给图表”或“前端调不到后端接口”这篇就是为你写的实操路径。2. 用 Pandas 构建可审计的数据清洗流水线从原始 CSV 解析到结构化宽表生成2.1 原始数据格式识别与字段映射策略疫情原始数据通常以 CSV 或 Excel 形式发布常见字段包括report_date报告日期、province省份、city城市、confirmed确诊数、cured治愈数、dead死亡数。但不同来源命名混乱有的叫date有的叫update_time有的用province_name有的直接写area。Pandas 清洗第一步不是写代码而是建立字段映射表原始字段名标准字段名类型转换规则示例report_date,date,update_timereport_datepd.to_datetime()强制转 datetime64[ns]2022-03-15→2022-03-15 00:00:00province,province_name,areaprovince.str.replace(省市confirmed_count,confirmedconfirmedpd.to_numeric(errorscoerce)转数值异常值设为 NaN1234.5→1234.5-→NaN提示不要在read_csv()里硬编码列名。先用pd.read_csv(file_path, nrows5)读前 5 行用df.columns.tolist()打印原始列名再动态匹配映射表。这样换一份新数据源只需改映射表不用动清洗逻辑。2.2 时间序列补齐与地理维度标准化疫情数据常有缺失某天某省未上报表格里就直接跳过该行。但可视化要求“每天每省都有值”否则折线图会断开。Pandas 的reindex()配合pd.date_range()是标准解法# 假设 df 是清洗后的基础表含 province, report_date, confirmed 列 # 步骤1生成完整时间范围取最小/最大日期 full_date_range pd.date_range( startdf[report_date].min(), enddf[report_date].max(), freqD ) # 步骤2生成所有省×日期的笛卡尔积 provinces df[province].unique() full_index pd.MultiIndex.from_product( [provinces, full_date_range], names[province, report_date] ) # 步骤3reindex 并用前向填充ffill补确诊数 df_full df.set_index([province, report_date]).reindex(full_index).sort_index() df_full[confirmed] df_full.groupby(province)[confirmed].ffill() df_full df_full.reset_index()地理标准化更关键前端地图组件如 Plotly 的choropleth_mapbox需要标准行政区划代码如CN-11代表北京或精确经纬度。不能靠字符串匹配“北京”→“北京市”因为原始数据可能写“京”“BJ”“Beijing”。正确做法是维护一张province_mapping.csvprovince_rawprovince_stdprovince_codelatlng北京北京市CN-1139.9042116.4074广东广东省CN-4423.1291113.2644用df.merge()关联后province_code用于地图着色lat/lng用于散点标注。2.3 多维聚合指标计算从原始记录到分析宽表可视化不直接画原始行数据而是画聚合结果。例如“全国每日新增确诊”需计算confirmed的一阶差分“各省累计治愈率”需cured / confirmed * 100。Pandas 的groupby().agg()是核心# 计算每日全国汇总 daily_national df_full.groupby(report_date).agg({ confirmed: sum, cured: sum, dead: sum }).reset_index() # 计算各省累计值按 report_date 排序后 cumsum df_full df_full.sort_values([province, report_date]) df_full[confirmed_cumsum] df_full.groupby(province)[confirmed].cumsum() df_full[cured_cumsum] df_full.groupby(province)[cured].cumsum() # 计算治愈率避免除零 df_full[cure_rate] np.where( df_full[confirmed_cumsum] 0, (df_full[cured_cumsum] / df_full[confirmed_cumsum] * 100).round(2), 0.0 )关键参数说明agg({})中键是原始列名值是聚合函数sum、mean、max或自定义 lambdacumsum()必须在sort_values()后执行否则顺序错乱np.where()替代df[a]/df[b]直接除防止分母为 0 报inf或NaN。最终输出宽表analysis_wide.csv含列report_date,province,confirmed_cumsum,cure_rate,province_code,lat,lng—— 这才是前端图表真正的数据源。3. Flask SQLAlchemy 构建稳定数据服务层API 设计、数据库建模与跨域处理3.1 MySQL 数据库建模贴合分析宽表的三张核心表源码中models.py定义了 ORM 模型但新手常忽略表结构与清洗逻辑的对应关系。实际部署时必须按宽表字段反向设计数据库# models.py from flask_sqlalchemy import SQLAlchemy db SQLAlchemy() class ProvinceInfo(db.Model): __tablename__ province_info id db.Column(db.Integer, primary_keyTrue) province_std db.Column(db.String(20), nullableFalse) # 北京市 province_code db.Column(db.String(10), uniqueTrue, nullableFalse) # CN-11 lat db.Column(db.Float, nullableFalse) lng db.Column(db.Float, nullableFalse) class DailyReport(db.Model): __tablename__ daily_report id db.Column(db.Integer, primary_keyTrue) report_date db.Column(db.Date, nullableFalse) # 索引字段加速按日期查询 province_id db.Column(db.Integer, db.ForeignKey(province_info.id), nullableFalse) confirmed_cumsum db.Column(db.Integer, default0) cured_cumsum db.Column(db.Integer, default0) dead_cumsum db.Column(db.Integer, default0) cure_rate db.Column(db.Float, default0.0) class DataVersion(db.Model): __tablename__ data_version id db.Column(db.Integer, primary_keyTrue) version db.Column(db.String(20), nullableFalse) # 2023Q4 update_time db.Column(db.DateTime, defaultdb.func.now()) file_hash db.Column(db.String(64)) # 原始CSV的sha256用于校验数据一致性注意DailyReport.province_id是外键关联ProvinceInfo.id而非直接存province_code。这样既保证数据一致性删省份时自动级联又避免字符串重复存储。report_date加db.Index索引否则查“2022年1月数据”会全表扫描。3.2 Flask API 接口设计RESTful 风格与参数校验后端不提供 HTML 页面只暴露 JSON 接口。app.py中的关键路由# app.py from flask import Flask, jsonify, request from models import db, DailyReport, ProvinceInfo app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] mysqlpymysql://user:passlocalhost:3306/epidemic_db app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db.init_app(app) app.route(/api/v1/daily-summary, methods[GET]) def get_daily_summary(): # 参数校验start_date 和 end_date 必须存在且为 YYYY-MM-DD 格式 start_date request.args.get(start_date) end_date request.args.get(end_date) if not start_date or not end_date: return jsonify({error: missing start_date or end_date}), 400 try: pd.to_datetime(start_date) pd.to_datetime(end_date) except ValueError: return jsonify({error: invalid date format}), 400 # 查询JOIN ProvinceInfo 获取 province_std 名称按日期范围过滤 results db.session.query( DailyReport.report_date, ProvinceInfo.province_std, DailyReport.confirmed_cumsum, DailyReport.cure_rate ).join(ProvinceInfo, DailyReport.province_id ProvinceInfo.id).filter( DailyReport.report_date start_date, DailyReport.report_date end_date ).all() # 转字典列表datetime 自动转字符串 data [ { date: r.report_date.isoformat(), province: r.province_std, confirmed: r.confirmed_cumsum, cure_rate: float(r.cure_rate) } for r in results ] return jsonify({data: data})关键点说明methods[GET]明确限定请求方法避免误用 POSTrequest.args.get()获取 URL 参数比request.json更适合简单查询pd.to_datetime()校验而非datetime.strptime()因 Pandas 能自动处理2022-01-01和2022/01/01多种格式db.session.query().join().filter()是 SQLAlchemy 标准写法比原生 SQL 更安全防注入。3.3 解决前后端分离的跨域问题CORS 配置与生产环境代理开发时前端Vue Dev Server运行在http://localhost:8080后端 Flask 在http://localhost:5000浏览器会拦截跨域请求。Flask-CORS 是最简方案pip install flask-cors# app.py 开头添加 from flask_cors import CORS app Flask(__name__) CORS(app) # 允许所有源仅用于开发 # 生产环境应限制CORS(app, origins[https://your-frontend-domain.com])但生产部署时不推荐在 Flask 层处理 CORS。正确做法是 Nginx 反向代理将/api/路径转发到 Flask并由 Nginx 添加响应头# nginx.conf location /api/ { proxy_pass http://127.0.0.1:5000/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; # 关键添加跨域头 add_header Access-Control-Allow-Origin https://your-frontend-domain.com; add_header Access-Control-Allow-Methods GET, POST, OPTIONS; add_header Access-Control-Allow-Headers Content-Type, Authorization; }提示add_header比proxy_set_header更可靠因为它是 Nginx 输出响应时添加不受后端影响。OPTIONS方法必须显式允许否则浏览器预检失败。4. Plotly Dash 与 Vue.js 双前端实现动态图表与响应式布局的落地细节4.1 Dash 服务端渲染用 callback 实现交互式筛选Dash 不是纯前端框架它在服务端生成图表并响应用户操作。dashboard.py的核心是app.callbackimport dash from dash import dcc, html, Input, Output, State import plotly.express as px app dash.Dash(__name__) # 布局下拉框选省份日期选择器图表容器 app.layout html.Div([ html.H1(疫情数据分析看板), html.Div([ dcc.Dropdown( idprovince-selector, options[{label: p, value: p} for p in [全国, 北京市, 广东省]], value全国, clearableFalse ), dcc.DatePickerRange( iddate-range, start_date2022-01-01, end_date2022-12-31, min_date_allowed2020-01-01, max_date_allowed2023-12-31 ) ]), dcc.Graph(idtrend-chart) ]) # 回调当下拉框或日期变化时重新查询数据并更新图表 app.callback( Output(trend-chart, figure), Input(province-selector, value), Input(date-range, start_date), Input(date-range, end_date) ) def update_chart(selected_province, start_date, end_date): # 此处调用 Flask API 或直接查数据库Dash 可直连 SQLAlchemy # 为简化假设已从 DB 获取 df_filtered if selected_province 全国: df_filtered df_daily_national # 全国汇总数据 else: df_filtered df_province[df_province[province] selected_province] # Plotly Express 快速生成折线图 fig px.line( df_filtered, xreport_date, yconfirmed_cumsum, titlef{selected_province} 累计确诊趋势, labels{report_date: 日期, confirmed_cumsum: 累计确诊人} ) fig.update_layout( hovermodex unified, # 鼠标悬停显示所有曲线值 templateplotly_white # 白色背景适配政务风格 ) return fig关键参数说明Input绑定控件的id和属性如valueOutput绑定图表的id和figure属性px.line()的x和y必须是 DataFrame 列名不能是表达式如df[a]/df[b]hovermodex unified让多条线在同一日期悬停时显示所有线的值避免反复移动鼠标。4.2 Vue.js 前端消费 APIAxios 请求与 ECharts 地图渲染若选用 Vue 作为前端src/views/Dashboard.vue需完成三件事请求数据、处理加载状态、渲染 ECharts 地图template div classdashboard h2全国疫情分布/h2 div v-ifloading加载中.../div div v-else refchartRef stylewidth: 100%; height: 600px;/div /div /template script import * as echarts from echarts import axios from axios export default { data() { return { loading: true, chart: null, mapData: [] // 格式[{name: 北京, value: 12345}, ...] } }, async mounted() { await this.fetchData() this.initChart() }, methods: { async fetchData() { try { const res await axios.get(/api/v1/daily-summary, { params: { start_date: 2022-01-01, end_date: 2022-12-31 } }) // 按省份聚合最新一天数据 const latest res.data.data.reduce((acc, item) { const key item.province if (!acc[key] || new Date(item.date) new Date(acc[key].date)) { acc[key] item } return acc }, {}) this.mapData Object.values(latest).map(item ({ name: item.province, value: item.confirmed })) } catch (err) { console.error(API 请求失败:, err) } finally { this.loading false } }, initChart() { this.chart echarts.init(this.$refs.chartRef) // 注册中国地图 JSON需提前下载 china.json echarts.registerMap(china, CHINA_JSON) this.chart.setOption({ tooltip: { trigger: item }, series: [{ type: map, map: china, data: this.mapData, label: { show: true }, emphasis: { label: { show: true } } }] }) } } } /script关键点说明axios.get()的params对象自动生成 URL 查询参数?start_date...end_date...reduce()聚合最新数据因 API 返回的是每日明细需取每个省的report_date最大值echarts.registerMap()必须在setOption()前执行否则报 “map not found”CHINA_JSON是从 ECharts 官网 下载的china.json需放入public/目录并import。5. 生产环境部署与性能调优Nginx Gunicorn MySQL 连接池配置5.1 Gunicorn 启动 Flask 应用进程管理与超时设置Flask 自带服务器app.run()仅用于开发生产必须用 Gunicornpip install gunicorn gunicorn -w 4 -b 127.0.0.1:5000 --timeout 120 --keep-alive 5 app:app参数详解-w 4启动 4 个 worker 进程建议设为 CPU 核心数 × 2如 4 核机器设 8-b 127.0.0.1:5000绑定本地地址由 Nginx 反向代理不对外暴露--timeout 120worker 处理单个请求超时 120 秒避免长查询阻塞如全量导出--keep-alive 5HTTP Keep-Alive 连接保持 5 秒减少 TCP 握手开销。提示创建gunicorn.conf.py配置文件比命令行更易维护# gunicorn.conf.py bind 127.0.0.1:5000 workers 4 timeout 120 keepalive 5 accesslog /var/log/gunicorn_access.log errorlog /var/log/gunicorn_error.log loglevel info启动命令变为gunicorn -c gunicorn.conf.py app:app。5.2 MySQL 连接池优化避免“Too many connections”错误默认 SQLAlchemy 每次请求新建连接高并发时 MySQL 连接数爆满。需配置连接池# app.py app.config[SQLALCHEMY_POOL_SIZE] 10 # 连接池初始大小 app.config[SQLALCHEMY_MAX_OVERFLOW] 20 # 超出池大小时最多新建 20 个连接 app.config[SQLALCHEMY_POOL_RECYCLE] 3600 # 连接空闲 1 小时后回收防 MySQL timeout app.config[SQLALCHEMY_POOL_PRE_PING] True # 每次取连接前执行 SELECT 1 检测有效性验证是否生效登录 MySQL 执行SHOW STATUS LIKE Threads_connected;压力测试时观察该值是否稳定在POOL_SIZE MAX_OVERFLOW范围内而非持续飙升。5.3 Nginx 缓存静态资源与 API 响应前端构建后的dist/目录和 Dash 的静态文件需由 Nginx 直接服务不经过 Flask# nginx.conf server { listen 80; server_name your-domain.com; # 静态文件Vue 构建产物 location / { root /path/to/vue/dist; try_files $uri $uri/ /index.html; } # Dash 静态资源CSS/JS location /_dash-component-suites/ { alias /path/to/venv/lib/python3.x/site-packages/dash/; expires 1h; } # API 接口 location /api/ { proxy_pass http://127.0.0.1:5000/; # 缓存 GET 请求如 /api/v1/daily-summary proxy_cache my_cache; proxy_cache_valid 200 302 10m; # 200/302 响应缓存 10 分钟 proxy_cache_bypass $http_cache_control; # 若请求头含 Cache-Control: no-cache则不走缓存 } } # 缓存区定义 proxy_cache_path /var/cache/nginx levels1:2 keys_zonemy_cache:10m max_size1g inactive60m use_temp_pathoff;关键效果try_files让 Vue Router 的 history 模式正常工作proxy_cache_valid对/api/下的 GET 请求启用缓存减轻后端压力proxy_cache_bypass允许前端加Cache-Control: no-cache强制刷新满足“实时数据”需求。6. 数据溯源与版本控制技巧用 Git LFS 管理原始 CSV用 SQL 注释标记清洗逻辑6.1 原始数据文件的版本化管理Git LFS 替代直接提交大文件疫情原始 CSV 动辄上百 MB直接git add会导致仓库臃肿、克隆缓慢。Git LFSLarge File Storage是标准解法# 安装 Git LFS curl -s https://packagecloud.io/install/repositories/github/git-lfs/script.deb.sh | sudo bash sudo apt-get install git-lfs git lfs install # 跟踪所有 CSV 文件 git lfs track *.csv git lfs track *.xlsx # 提交 .gitattributesLFS 规则文件 git add .gitattributes git commit -m track csv files with lfs此后git add data/raw/2022-12-01.csv会上传文件到 LFS 服务器Git 仓库只存指针。协作时队友git clone后需git lfs pull才能获取真实文件。6.2 清洗脚本中的 SQL 注释让后续维护者一眼看懂指标含义Pandas 清洗逻辑常被写成黑盒函数。更好的做法是在.py文件顶部用 SQL 风格注释说明每列计算规则 -- 数据来源国家卫健委官网每日通报 CSV -- 清洗目标生成 analysis_wide.csv供前端图表消费 -- 字段说明 -- report_date: 报告日期datetime64[ns]已标准化 -- province: 省份标准名称去省/市/自治区后缀 -- confirmed_cumsum: 该省截至 report_date 的累计确诊数按 province 分组 cumsum -- cure_rate: (cured_cumsum / confirmed_cumsum * 100).round(2)分母为 0 时设为 0.0 -- province_code: 行政区划代码CN-11用于地图着色 import pandas as pd import numpy as np def clean_epidemic_data(raw_df): # ... 清洗代码 ... return wide_df提示这种注释可被文档生成工具如 Sphinx提取形成团队内部数据字典。比写 Wiki 更可靠因为随代码同步更新。6.3 验证清洗结果一致性的自动化检查用 pytest 断言关键指标每次更新清洗逻辑必须验证输出不变。在tests/test_cleaning.py中写断言import pytest import pandas as pd from src.cleaning import clean_epidemic_data def test_confirmed_cumsum_consistency(): # 给定固定输入数据 raw_df pd.DataFrame({ report_date: [2022-01-01, 2022-01-02, 2022-01-03], province: [北京市, 北京市, 北京市], confirmed: [100, 50, 30] }) raw_df[report_date] pd.to_datetime(raw_df[report_date]) result clean_epidemic_data(raw_df) # 断言累计值应为 [100, 150, 180] assert result.loc[0, confirmed_cumsum] 100 assert result.loc[1, confirmed_cumsum] 150 assert result.loc[2, confirmed_cumsum] 180 if __name__ __main__: pytest.main([__file__])运行pytest tests/test_cleaning.py即可验证。CI/CD 流水线中加入此步骤能防止清洗逻辑修改引发指标漂移。本文还有配套的精品资源点击获取