基于Flask+Echarts+爬虫的疫情监控系统
'# 基于Flask+Echarts+爬虫的疫情监控系统
一、背景与问题
随着全球疫情形势的动态变化，实时获取和可视化疫情数据成为公共卫生管理的重要环节。传统疫情监控系统面临三大挑战：
- 数据获取困难：政府网站数据更新滞后，手动收集效率低下
- 数据可视化需求：需要多维度图表展示（折线图/柱状图/热力图）
- 系统可维护性：需要支持定时更新、数据持久化和多终端访问
传统解决方案往往采用静态网页+Excel表格，存在数据延迟、更新繁琐等问题。本文提出的解决方案通过Flask构建后端服务，ECharts实现动态可视化，结合爬虫技术实现数据自动更新，形成完整的疫情监控闭环系统。
二、基本原理
1. 技术架构
系统采用分层架构：
[用户终端] → [Flask API] → [数据存储] → [爬虫服务]
↑ ↓
[ECharts前端] [数据源]- 爬虫层：使用requests+BeautifulSoup解析目标网站，提取疫情数据
- 后端层：Flask提供RESTful API，支持数据查询和图表配置
- 前端层：ECharts动态渲染图表，支持实时数据更新
- 存储层：SQLite存储历史数据，支持快速查询
2. 数据处理流程
- 爬虫定时采集最新数据
- 数据清洗：去除异常值、格式标准化
- 存储到数据库
- 前端请求数据时，Flask返回结构化数据
- ECharts根据配置生成动态图表
三、环境准备
# 安装依赖
pip install flask requests beautifulsoup4 echartsc sqlite3项目结构：
covid_monitor/
├── app.py # Flask主程序
├── data/ # 存储数据文件
│ └── history.db # SQLite数据库
├── crawlers/ # 爬虫模块
│ └── crawler.py # 爬虫核心代码
├── templates/ # 前端模板
│ └── index.html # ECharts页面
└── static/ # 静态资源
└── style.css # 样式文件四、核心实现
1. 爬虫模块实现（crawler.py）
import requests
from bs4 import BeautifulSoup
import sqlite3
import time
import random
# 爬虫配置
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4443.116 Safari/537.36'
}
def fetch_data():
"""爬取疫情数据"""
url = 'https://example-covid-data.com'
try:
response = requests.get(url, headers=HEADERS, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
# 假设数据在表格中
table = soup.find('table', class_='covid-data')
rows = table.find_all('tr')
data = []
for row in rows[1:]: # 跳过表头
cols = row.find_all(['td', 'th'])
if len(cols) == 0:
continue
region = cols[0].text.strip()
confirmed = int(cols[1].text.replace(',', '').strip())
cured = int(cols[2].text.replace(',', '').strip())
dead = int(cols[3].text.replace(',', '').strip())
data.append({
'region': region,
'confirmed': confirmed,
'cured': cured,
'dead': dead,
'timestamp': int(time.time())
})
return data
except Exception as e:
print(f"爬虫错误: {str(e)}")
return []
def save_to_db(data):
"""保存数据到SQLite"""
conn = sqlite3.connect('history.db')
cursor = conn.cursor()
# 创建表（如果不存在）
cursor.execute('''
CREATE TABLE IF NOT EXISTS stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
region TEXT,
confirmed INTEGER,
cured INTEGER,
dead INTEGER,
timestamp INTEGER
)
''')
# 插入新数据
cursor.executemany('''
INSERT INTO stats (region, confirmed, cured, dead, timestamp)
VALUES (?, ?, ?, ?, ?)
''', data)
conn.commit()
conn.close()
def run_crawler():
"""定时爬虫函数"""
while True:
print("开始爬取数据...")
data = fetch_data()
if data:
save_to_db(data)
print(f"成功保存{len(data)}条数据")
else:
print("未获取到新数据")
time.sleep(3600) # 每小时运行一次关键点说明：
- 使用随机User-Agent防止反爬
- 数据清洗处理千分位分隔符
- 使用SQLite保证轻量级存储
- 爬虫服务采用轮询机制
2. Flask接口实现（app.py）
from flask import Flask, jsonify, render_template
import sqlite3
import json
app = Flask(__name__)
# 数据库配置
DB_PATH = 'history.db'
def get_all_data():
"""获取所有历史数据"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT * FROM stats ORDER BY timestamp DESC")
rows = cursor.fetchall()
conn.close()
# 转换为JSON格式
return [{
'id': row[0],
'region': row[1],
'confirmed': row[2],
'cured': row[3],
'dead': row[4],
'timestamp': row[5]
} for row in rows]
@app.route('/api/data', methods=['GET'])
def get_data():
"""获取疫情数据接口"""
data = get_all_data()
return jsonify({
'status': 'success',
'data': data
})
@app.route('/')
def index():
"""首页路由"""
return render_template('index.html')
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)3. ECharts前端实现（index.html）
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>疫情监控系统</title>
<script src="https://cdn.jsdelivr.net/npm/echarts@5.4.0/dist/echarts.min.js"></script>
<style>
body { margin: 0; }
#main { width: 100vw; height: 100vh; }
</style>
</head>
<body>
<div id="main"></div>
<script>
// 获取数据
fetch('/api/data')
.then(response => response.json())
.then(data => {
// 处理数据
const seriesData = data.map(item => ({
name: item.region,
value: [item.confirmed, item.cured, item.dead]
}));
// 创建图表
const chart = echarts.init(document.getElementById('main'));
const option = {
title: {
text: '疫情统计'
},
tooltip: {
trigger: 'axis'
},
xAxis: {
type: 'category',
data: seriesData.map(d => d.name)
},
yAxis: {
type: 'value'
},
series: [{
name: '确诊',
type: 'bar',
data: seriesData.map(d => d.value[0])
}, {
name: '治愈',
type: 'bar',
data: seriesData.map(d => d.value[1])
}, {
name: '死亡',
type: 'bar',
data: seriesData.map(d => d.value[2])
}]
};
chart.setOption(option);
});
</script>
</body>
</html>五、完整案例
1. 项目运行流程
- 启动爬虫服务（可单独运行crawler.py）
- 启动Flask应用（运行app.py）
- 访问 http://localhost:5000 查看实时数据
- ECharts会自动加载最新数据并渲染图表
2. 系统功能演示
- 自动爬取：每小时自动更新数据
- 多维度展示：同时显示确诊/治愈/死亡数据
- 实时更新：页面刷新后自动获取最新数据
- 数据持久化：所有历史数据存储在SQLite中
3. 系统扩展性
- 可添加趋势分析模块（折线图）
- 可增加地图可视化（geo组件）
- 可集成预警系统（阈值检测）
六、源码解析
1. 爬虫代码分析
# 随机User-Agent生成
def get_random_user_agent():
agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4443.116 Safari/537.36",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/91.0.4443.116 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4443.116 Safari/537.36"
]
return random.choice(agents)关键点：使用随机User-Agent防止被封IP
2. 数据库优化
# 查询优化示例
def get_latest_data(limit=10):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute('''
SELECT * FROM stats
ORDER BY timestamp DESC
LIMIT ?
''', (limit,))
rows = cursor.fetchall()
conn.close()
return rows说明：通过LIMIT限制返回数据量，提升查询效率
3. 前端数据处理
// 数据处理逻辑
const seriesData = data.map(item => ({
name: item.region,
value: [item.confirmed, item.cured, item.dead]
}));说明：将原始数据转换为ECharts支持的格式
七、进阶使用
1. 数据持久化改进
使用SQLite的索引优化查询：
# 创建索引
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_timestamp
ON stats (timestamp)
''')2. 增加预警功能
# 预警逻辑示例
def check_alerts(data):
for item in data:
if item['confirmed'] > 10000:
print(f"警报：{item['region']}确诊超1万")3. 前端交互增强
// 添加动态更新
setInterval(() => {
fetch('/api/data')
.then(response => response.json())
.then(data => {
// 更新图表
});
}, 60000); // 每分钟刷新一次八、性能与工程实践
1. 性能优化方案
| 优化项 | 方法 | 效果 |
|---|---|---|
| 爬虫优化 | 使用多线程/异步 | 提升爬取效率 |
| 数据库 | 使用索引/分页查询 | 加快数据检索 |
| 缓存机制 | 使用Redis缓存热点数据 | 降低接口响应时间 |
| 前端优化 | 使用Web Worker | 提升图表渲染性能 |
2. 异常处理机制
# 异常处理示例
def safe_fetch_data():
try:
return fetch_data()
except requests.exceptions.RequestException as e:
print(f"网络错误: {str(e)}")
return []
except Exception as e:
print(f"未知错误: {str(e)}")
return []3. 安全考虑
- 前端接口需添加身份验证（推荐JWT）
- 爬虫需遵守robots.txt规则
- 数据库连接使用环境变量配置
- 对用户输入进行校验和过滤
九、常见问题与踩坑
1. 爬虫被封IP问题
错误现象：爬虫运行后频繁失败
解决方法：
- 使用代理IP池
- 增加请求间隔
- 使用更真实的User-Agent
# 增加请求间隔
time.sleep(random.randint(1, 5))2. 数据解析错误
错误现象：数据字段缺失导致程序崩溃
解决方法：
- 添加字段存在性校验
- 使用default参数处理缺失值
# 字段校验示例
def safe_get_text(tag, default='0'):
return tag.text.strip() if tag else default3. 接口响应慢
错误现象：页面加载卡顿
解决方法：
- 使用缓存机制
- 优化SQL查询
- 前端异步加载
# 缓存示例
from functools import lru_cache
@lru_cache(maxsize=100)
def get_cached_data():
return get_all_data()十、最佳实践
1. 推荐使用场景
- 需要实时监控的中小型项目
- 数据源更新频率较低的系统
- 对可视化要求不高的监控场景
- 需要快速搭建的原型系统
2. 不推荐使用场景
- 需要处理海量数据的系统
- 需要高并发访问的平台
- 数据源需要复杂处理的场景
- 需要严格安全控制的系统
3. 推荐实践方案
- 使用Celery实现异步爬虫
- 使用Gunicorn部署Flask应用
- 使用Redis缓存热点数据
- 使用Docker容器化部署
- 使用Prometheus+Grafana监控系统
十一、总结
基于Flask+ECharts+爬虫的疫情监控系统是一个典型的前后端分离架构案例。通过爬虫获取数据，Flask处理业务逻辑，ECharts实现可视化，形成完整的数据处理闭环。系统在实际应用中需要注意：
- 爬虫策略要符合网站规则，避免被封
- 数据处理要严谨，防止脏数据
- 前端交互要友好，提升用户体验
- 系统要具备扩展性，方便后续升级
虽然该方案适合中小型项目，但在处理复杂业务时需要考虑更专业的技术栈（如Django+DRF+GraphQL）。对于需要处理大规模数据或高并发的场景，建议采用分布式架构和更专业的数据处理方案。
评论已关闭