Python数据工程实战：从爬虫到情感分析的全链路项目拆解

简介：本资源是一套完整的Python数据工程实践项目，面向计算机、数学与电子信息等专业的本科生及初学者，聚焦疫情数据采集、社交媒体舆情分析与可视化呈现全流程。项目涵盖疫情实时爬虫、微博关键词定向抓取（含MySQL数据库存储）、结构化数据清洗、多维度统计可视化（Matplotlib/Seaborn）及基于词典的情感倾向分析（正负样本标注+情感得分计算），可直接用于课程设计、期末大作业或毕业设计参考。压缩包共15个文件，包含7个核心Python脚本（爬虫、入库、预处理、绘图、情感分析）、3个CSV数据集（含标注的nCoV训练样本与情感关键词库）、2个文本词典、1个JSON微博原始数据样例、1张结果图表及1份Markdown项目说明文档，整体大小23.87MB。已有728人学习下载，提供从数据获取到价值提炼的完整链路代码与结构化目录，便于理解模块分工、调试逻辑与扩展功能。

1. 项目全景：一个数据工程师的“练手”与“实战”项目剖析

最近在整理硬盘，翻出来一个前两年做的老项目，一个集成了疫情数据爬取、微博舆情抓取、数据清洗、可视化看板以及情感分析的综合型Python脚本包。当时做这个的初衷，一方面是响应社区里不少朋友对“如何将爬虫、数据库、数据分析串起来做一个完整项目”的需求，另一方面也是想给自己手头的一些技术栈做个压力测试和流程梳理。现在看来，这个项目虽然涉及的领域（疫情、微博）热度有所变化，但其技术栈的整合思路和踩过的坑，对于想从“写单个脚本”进阶到“搭建小型数据管道”的Python开发者来说，依然有很强的参考价值。这不是一个玩具项目，而是一个麻雀虽小五脏俱全的微型数据工程实践，涵盖了从数据采集、存储、加工到分析、呈现的全链路。

简单来说，这个项目帮你解决几个核心问题：第一，面对多个异构数据源（如结构化的公开疫情数据、半结构化的微博网页），如何设计稳定、可维护的爬虫？第二，海量的临时数据如何有效地存入数据库进行管理，而不是散落在无数个CSV文件里？第三，爬下来的“脏数据”如何通过预处理变成“干净数据”，为后续分析扫清障碍？第四，如何将处理好的数据，通过直观的可视化图表讲述故事？第五，如何对文本数据（如微博内容）进行初步的情感倾向判断？如果你对这些问题感兴趣，或者正想找一个综合项目来巩固你的Python、Requests、Pandas、SQL、Matplotlib/Seaborn乃至简单的机器学习库（如SnowNLP或TextBlob）技能，那么这个项目拆解会非常适合你。

2. 双线并行的爬虫架构设计与核心实现

项目的起点是数据，而数据来自两个风格迥异的源头：相对规范的公开疫情数据接口，和动态复杂、反爬机制多样的微博网页。这要求我们的爬虫架构不能是单打独斗的脚本堆砌，而需要有清晰的职责划分和统一的错误处理机制。

2.1 疫情数据爬虫：与公开API的规范对话

对于疫情数据，我们通常可以找到各级卫健委或权威数据平台提供的API或结构化的数据文件。这类爬虫的核心在于“协议遵守”和“数据解析”。

技术选型与核心逻辑：我们首选requests库，因为它简单、高效、社区资源丰富。关键在于构造合法的请求头（User-Agent,Referer等），模拟一个真实浏览器的访问行为。对于返回的数据，通常是JSON格式，我们可以直接用response.json()解析。如果数据是HTML，则可能需要结合BeautifulSoup或lxml进行解析。

import requests import pandas as pd from datetime import datetime def fetch_epidemic_data(api_url): """ 从指定API获取疫情数据 """ headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...', 'Accept': 'application/json', } try: response = requests.get(api_url, headers=headers, timeout=10) response.raise_for_status() # 检查HTTP请求是否成功 data = response.json() # 假设API返回结构为 {'data': [list of records], 'updateTime': '...'} df = pd.DataFrame(data['data']) df['crawl_time'] = datetime.now() # 添加爬取时间戳 return df except requests.exceptions.RequestException as e: print(f"请求疫情API失败: {e}") return pd.DataFrame() # 返回空DataFrame，避免后续流程中断

关键细节与避坑：

1. 频率限制与礼貌爬取：公开API也可能有调用频率限制。务必在代码中加入time.sleep(random.uniform(1, 3))这样的随机延时，避免对服务器造成压力，这也是基本的网络礼仪。
2. 数据更新逻辑：这是一个典型的增量型爬虫场景。我们不应该每次都全量爬取所有历史数据。理想的做法是，在数据库中记录上次爬取的最新数据日期，本次只请求该日期之后的新数据。这极大地节省了网络和存储资源。
3. 错误处理与重试：网络请求充满不确定性。必须用try...except包裹核心请求代码，并设计重试机制（如使用tenacity库）。记录失败的URL和原因，便于后续排查。
4. 数据验证：解析后的数据，在入库前应进行基本的验证，如检查必要字段是否存在、数据类型是否正确、是否有异常值（如负的确诊数）。

2.2 微博关键词爬虫：在动态网页与反爬机制间周旋

微博爬虫的复杂度提升了一个数量级。它面临登录态、动态加载、反爬算法（如滑块验证、请求签名）等问题。这里我们探讨一种相对可行（但需严格遵守微博Robots协议和使用限制）的思路。

技术选型：对于动态内容，Selenium或Playwright这类浏览器自动化工具可以完美渲染页面，但速度慢、资源消耗大。更高效的方法是分析微博的XHR/Fetch请求，找到直接返回数据的接口，然后用requests模拟调用。这需要用到浏览器的开发者工具（F12）进行网络抓包分析。

核心实现步骤：

1. 关键词搜索接口分析：在微博网页或手机端进行关键词搜索，观察网络请求，找到一个返回JSON格式搜索结果的请求。复制其cURL命令，并导入到requests代码中。
2. 参数逆向：该接口的URL和请求头（Headers）中通常包含加密参数，如_s、sign等。这些参数可能由前端JavaScript生成。一种方法是使用execjs库执行相关的JS代码来生成；另一种更简单（但可能不稳定）的方法是，这些参数有时在页面源码的某个全局变量里可以找到。
3. 请求模拟与数据提取：构造包含关键词、页码、时间范围等参数的请求。解析返回的JSON，提取博文ID、内容、发布时间、发布人、转发评论点赞数等信息。

import requests import execjs import json def fetch_weibo_by_keyword(keyword, page=1): """ 模拟微博关键词搜索请求（示例，实际参数需动态生成） """ # 1. 加载生成签名参数的JS文件 with open('weibo_sign.js', 'r', encoding='utf-8') as f: js_code = f.read() ctx = execjs.compile(js_code) # 2. 调用JS函数生成动态参数 dynamic_params = ctx.call('get_sign_params', keyword, page) url = 'https://m.weibo.cn/api/container/getIndex' headers = { 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) ...', 'Referer': f'https://m.weibo.cn/search?keyword={keyword}', 'X-Requested-With': 'XMLHttpRequest' } params = { 'containerid': f'100103type=1&q={keyword}', 'page_type': 'searchall', 'page': page, **dynamic_params # 合并动态生成的签名参数 } try: resp = requests.get(url, headers=headers, params=params, timeout=15) data = resp.json() # 解析data['data']['cards'] 结构，提取博文信息 weibo_list = [] for card in data.get('data', {}).get('cards', []): mblog = card.get('mblog') if mblog: weibo_list.append({ 'id': mblog.get('id'), 'text': mblog.get('text'), 'created_at': mblog.get('created_at'), 'user_name': mblog.get('user', {}).get('screen_name'), 'reposts_count': mblog.get('reposts_count'), 'comments_count': mblog.get('comments_count'), 'attitudes_count': mblog.get('attitudes_count') }) return weibo_list except Exception as e: print(f"爬取微博关键词[{keyword}]失败: {e}") return []

重要注意事项与伦理边界：

警告：爬取微博等社交媒体数据必须严格遵守其robots.txt协议，并尊重用户隐私和数据版权。本示例仅用于技术学习交流。在实际应用中，务必：

1. 控制爬取频率，避免对目标服务器造成干扰。
2. 不要爬取非公开信息。
3. 对爬取的数据进行脱敏处理，避免泄露用户个人身份信息。
4. 明确数据用途，不用于非法或不道德的目的。
5. 考虑到微博反爬机制的持续更新，此方法可能需要定期维护。

3. 数据存储层：SQLite与MySQL的选型及表结构设计

爬取到的数据是“流沙”，必须存入“容器”才能进行有效的管理和分析。在这个项目中，数据库的选择和设计是承上启下的关键一环。

3.1 数据库选型：轻量级SQLite vs 生产级MySQL

对于个人学习、小型项目或原型验证，SQLite是绝佳选择。它是一个无服务器、零配置、事务性的SQL数据库引擎，整个数据库就是一个文件，用Python标准库sqlite3即可操作，无需安装和运行独立的数据库服务。

import sqlite3 import pandas as pd # 连接数据库（如果不存在则创建） conn = sqlite3.connect('epidemic_weibo.db') # 将Pandas DataFrame直接写入表 df_epidemic.to_sql('epidemic_data', conn, if_exists='append', index=False) conn.close()

对于数据量更大、需要并发访问或考虑未来扩展的项目，MySQL或PostgreSQL更合适。它们提供了更完善的用户权限管理、存储过程、触发器和更强的并发性能。可以使用pymysql或sqlalchemy库进行连接。

选型建议：本项目作为学习整合，初期强烈推荐使用SQLite。它让你聚焦于SQL操作和表结构设计本身，而无需分心于数据库服务的安装、配置和维护。当数据量增长到数十万条以上，或需要多人协作时，再考虑迁移到MySQL。

3.2 表结构设计：为分析与关联做准备

良好的表结构是高效查询和分析的基础。我们需要为疫情数据和微博数据分别设计表，并考虑它们之间潜在的关联点（例如，通过“日期”和“地区”）。

疫情数据表 (epidemic_data)：

CREATE TABLE IF NOT EXISTS epidemic_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 自增主键 date DATE NOT NULL, -- 统计日期 province VARCHAR(50), -- 省份 city VARCHAR(50), -- 城市 confirmed_new INTEGER DEFAULT 0, -- 新增确诊 confirmed_total INTEGER DEFAULT 0, -- 累计确诊 cured_new INTEGER DEFAULT 0, -- 新增治愈 cured_total INTEGER DEFAULT 0, -- 累计治愈 dead_new INTEGER DEFAULT 0, -- 新增死亡 dead_total INTEGER DEFAULT 0, -- 累计死亡 crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP -- 爬取时间 ); -- 创建复合索引，加速按日期和地区的查询 CREATE INDEX idx_date_province ON epidemic_data (date, province);

微博数据表 (weibo_data)：

CREATE TABLE IF NOT EXISTS weibo_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, weibo_id BIGINT UNIQUE, -- 微博唯一ID，防止重复插入 keyword VARCHAR(100), -- 搜索使用的关键词 text TEXT, -- 微博正文 clean_text TEXT, -- 预处理后的干净文本（后续填充） sentiment_score FLOAT, -- 情感分析得分（后续填充） created_at DATETIME, -- 微博发布时间 user_name VARCHAR(100), reposts_count INTEGER DEFAULT 0, comments_count INTEGER DEFAULT 0, attitudes_count INTEGER DEFAULT 0, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 可以添加地理位置字段，用于和疫情数据关联 location_province VARCHAR(50), location_city VARCHAR(50) ); CREATE INDEX idx_keyword_time ON weibo_data (keyword, created_at); CREATE INDEX idx_sentiment ON weibo_data (sentiment_score);

设计心得：

1. 唯一约束与去重：为微博ID (weibo_id) 设置UNIQUE约束，这样在使用INSERT OR IGNORE或ON DUPLICATE KEY UPDATE（MySQL）语句时，可以自动跳过重复数据，这是实现增量爬虫的关键。
2. 预留加工字段：表中直接设计了clean_text和sentiment_score字段。这样，数据预处理的流水线就可以直接更新原表，保持数据的一致性，也方便后续查询。
3. 索引是双刃剑：索引能极大加速查询（如按关键词和时间筛选微博），但会降低数据插入的速度。对于爬虫这种写多读少的初期阶段，可以在数据爬取完成后再统一创建索引。

4. 数据预处理流水线：从“脏数据”到“干净数据”

直接从网上爬下来的数据，我们称之为“原始数据”或“脏数据”。它可能包含HTML标签、特殊字符、无关信息、缺失值、重复项等。不经过清洗，直接进行分析或可视化，结果很可能是错误的。

4.1 文本数据清洗：以微博内容为例

微博文本清洗是一个典型的多步骤流水线作业，使用pandas和正则表达式 (re) 是主力。

import re import pandas as pd import jieba from sqlalchemy import create_engine def clean_weibo_text(text): """清洗单条微博文本""" if not isinstance(text, str): return '' # 1. 移除HTML标签和URL text = re.sub(r'<[^>]+>', '', text) text = re.sub(r'http[s]?://\S+', '', text) # 2. 移除@用户和话题# text = re.sub(r'@[\w\u4e00-\u9fa5\-]+', '', text) text = re.sub(r'#([^#]+)#', r'\1', text) # 保留话题内容，去掉#号 # 3. 移除表情符号（如[笑cry]）和特殊字符 text = re.sub(r'\[.*?\]', '', text) text = re.sub(r'[^\w\u4e00-\u9fa5，。！？、；：“”‘’\s]', '', text) # 4. 合并多余空白字符 text = re.sub(r'\s+', ' ', text).strip() return text def batch_text_preprocessing(db_path='epidemic_weibo.db'): """批量预处理微博文本并更新数据库""" engine = create_engine(f'sqlite:///{db_path}') # 读取未清洗的原始文本 query = "SELECT id, text FROM weibo_data WHERE clean_text IS NULL OR clean_text = ''" df = pd.read_sql_query(query, engine) if df.empty: print("没有需要清洗的新微博数据。") return # 应用清洗函数 df['clean_text'] = df['text'].apply(clean_weibo_text) # 可选：进行分词，便于后续深度分析 df['segmented'] = df['clean_text'].apply(lambda x: ' '.join(jieba.lcut(x))) # 将清洗结果更新回数据库 # 这里采用逐条更新或批量更新策略。对于SQLite，批量更新效率更高。 with engine.begin() as conn: # 使用事务 for _, row in df.iterrows(): update_sql = """ UPDATE weibo_data SET clean_text = :clean_text, segmented = :segmented WHERE id = :id """ conn.execute(update_sql, { 'clean_text': row['clean_text'], 'segmented': row['segmented'], 'id': row['id'] }) print(f"已批量清洗并更新 {len(df)} 条微博数据。")

4.2 结构化数据清洗：以疫情数据为例

疫情数据的清洗更侧重于数据完整性和逻辑一致性。

def clean_epidemic_data(df): """ 清洗疫情DataFrame """ # 1. 处理缺失值：对于数值列，用0填充；对于文本列，用‘未知’填充 numeric_cols = ['confirmed_new', 'confirmed_total', 'cured_new', 'cured_total', 'dead_new', 'dead_total'] text_cols = ['province', 'city'] df[numeric_cols] = df[numeric_cols].fillna(0) df[text_cols] = df[text_cols].fillna('未知') # 2. 数据类型转换 df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors='coerce').fillna(0).astype('int') df['date'] = pd.to_datetime(df['date'], errors='coerce') # 3. 逻辑校验与修正：例如，累计数不应小于新增数 # 这里可以添加更复杂的业务规则校验 mask = df['confirmed_total'] < df['confirmed_new'] df.loc[mask, 'confirmed_total'] = df.loc[mask, 'confirmed_new'] # 简单处理，实际需根据业务判断 # 4. 去除完全重复的行（基于所有字段） df = df.drop_duplicates() # 5. 去除关键字段为空的无效行 df = df.dropna(subset=['date', 'province']) return df

预处理阶段的核心经验：

1. 流水线化：将清洗步骤封装成函数，形成可重复执行的流水线。新的数据来了，跑一遍流水线即可。
2. 保留原始数据：永远不要在原始数据上直接修改。我们的策略是：原始数据存入数据库后，通过新增字段（如clean_text）或新表来存储清洗后的结果。
3. 日志记录：在清洗函数中加入日志，记录处理了多少条数据、遇到了多少异常、修正了哪些问题。这对于监控数据质量和调试至关重要。

5. 数据可视化：用图表讲述疫情与舆情的“故事”

数据清洗后，就到了最直观的环节——可视化。我们使用matplotlib和seaborn库，目标是生成能清晰反映趋势、分布和关联的图表。

5.1 疫情数据可视化：趋势与分布

全国疫情趋势折线图：展示每日全国新增确诊、治愈、死亡的变化趋势。

import matplotlib.pyplot as plt import seaborn as sns from matplotlib.font_manager import FontProperties # 解决中文显示问题 plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False def plot_national_trend(engine): """绘制全国疫情趋势图""" query = """ SELECT date, SUM(confirmed_new) as new_confirmed, SUM(cured_new) as new_cured, SUM(dead_new) as new_dead FROM epidemic_data WHERE province != '未知' GROUP BY date ORDER BY date """ df_trend = pd.read_sql_query(query, engine) fig, ax = plt.subplots(figsize=(14, 7)) ax.plot(df_trend['date'], df_trend['new_confirmed'], label='新增确诊', linewidth=2, marker='o') ax.plot(df_trend['date'], df_trend['new_cured'], label='新增治愈', linewidth=2, marker='s') ax.plot(df_trend['date'], df_trend['new_dead'], label='新增死亡', linewidth=2, marker='^') ax.set_xlabel('日期') ax.set_ylabel('人数') ax.set_title('全国疫情每日新增趋势') ax.legend() ax.grid(True, linestyle='--', alpha=0.6) plt.xticks(rotation=45) plt.tight_layout() plt.savefig('national_trend.png', dpi=300) plt.show()

各省累计确诊分布热力图（需地理坐标数据）或条形图：如果没有地理坐标，可以用条形图展示Top 10省份。

def plot_top_provinces(engine, top_n=10): """绘制累计确诊最多的前N个省份""" query = """ SELECT province, SUM(confirmed_total) as total_confirmed FROM epidemic_data WHERE date = (SELECT MAX(date) FROM epidemic_data) -- 取最新一天的数据 AND province != '未知' AND province != '中国' GROUP BY province ORDER BY total_confirmed DESC LIMIT ? """ df_province = pd.read_sql_query(query, engine, params=(top_n,)) fig, ax = plt.subplots(figsize=(12, 6)) bars = ax.barh(df_province['province'], df_province['total_confirmed'], color=sns.color_palette("Reds_r", top_n)) ax.set_xlabel('累计确诊人数') ax.set_title(f'累计确诊人数最多的前{top_n}个省份（截至最新数据）') # 在条形末端添加数据标签 for bar in bars: width = bar.get_width() ax.text(width + width*0.01, bar.get_y() + bar.get_height()/2, f'{int(width):,}', va='center') plt.tight_layout() plt.savefig('top_provinces.png', dpi=300) plt.show()

5.2 微博舆情可视化：情感与热词

微博情感得分分布直方图：查看公众情绪的整体倾向。

def plot_sentiment_distribution(engine, keyword='疫情'): """绘制特定关键词下微博情感得分分布""" query = """ SELECT sentiment_score FROM weibo_data WHERE keyword = ? AND sentiment_score IS NOT NULL """ df_sentiment = pd.read_sql_query(query, engine, params=(keyword,)) if df_sentiment.empty: print(f"关键词 '{keyword}' 下没有情感分析数据。") return fig, ax = plt.subplots(figsize=(10, 6)) # 情感得分范围通常在0-1之间，0.5为中性 ax.hist(df_sentiment['sentiment_score'], bins=30, edgecolor='black', alpha=0.7, color='skyblue') ax.axvline(x=0.5, color='red', linestyle='--', label='中性线 (0.5)') ax.set_xlabel('情感得分 (0: 负面, 1: 正面)') ax.set_ylabel('微博数量') ax.set_title(f'关键词“{keyword}”相关微博情感分布') ax.legend() ax.grid(True, linestyle='--', alpha=0.3) plt.tight_layout() plt.savefig(f'sentiment_dist_{keyword}.png', dpi=300) plt.show()

微博发布数量时间序列图：观察舆情热度随时间的变化。

def plot_weibo_volume_trend(engine, keyword='疫情'): """绘制特定关键词下微博发布数量随时间变化趋势""" query = """ SELECT DATE(created_at) as post_date, COUNT(*) as weibo_count FROM weibo_data WHERE keyword = ? GROUP BY DATE(created_at) ORDER BY post_date """ df_volume = pd.read_sql_query(query, engine, params=(keyword,)) fig, ax = plt.subplots(figsize=(14, 6)) ax.plot(df_volume['post_date'], df_volume['weibo_count'], marker='o', linewidth=2, color='orange') ax.set_xlabel('日期') ax.set_ylabel('微博发布数量') ax.set_title(f'关键词“{keyword}”每日微博讨论量趋势') ax.fill_between(df_volume['post_date'], df_volume['weibo_count'], alpha=0.3, color='orange') plt.xticks(rotation=45) plt.grid(True, linestyle='--', alpha=0.6) plt.tight_layout() plt.savefig(f'weibo_volume_trend_{keyword}.png', dpi=300) plt.show()

可视化经验谈：

1. 图表服务于故事：不要为了炫技而作图。先想清楚你想通过图表表达什么信息（例如：疫情是否得到控制？舆情是乐观还是悲观？），再选择合适的图表类型。
2. 细节决定专业度：添加清晰的标题、轴标签、图例。合理设置颜色、线宽、标记点。使用plt.tight_layout()避免标签重叠。保存图片时指定高dpi(如300) 以保证印刷或展示质量。
3. 自动化报告：可以将上述绘图函数封装起来，并编写一个主函数，在数据更新后自动运行，生成一套最新的图表，甚至可以用Jinja2模板生成HTML报告。

6. 微博文本情感分析：SnowNLP实战与结果解读

情感分析是自然语言处理(NLP)的入门应用，旨在判断一段文本所表达的情感倾向是正面、负面还是中性。对于中文微博，SnowNLP库是一个简单易用的选择。

6.1 使用SnowNLP进行批量情感分析

SnowNLP基于朴素贝叶斯算法训练，可以直接对中文文本进行情感打分（0到1之间，越接近1表示越正面）。

from snownlp import SnowNLP import numpy as np def analyze_sentiment_batch(text_list): """ 批量分析情感，返回情感得分列表。 注意：SnowNLP的情感分析基于商品评论训练，对社交媒体文本可能有一定偏差。 """ scores = [] for text in text_list: if not text or not isinstance(text, str) or len(text.strip()) < 2: scores.append(None) # 对空文本或过短文本返回None continue try: s = SnowNLP(text) scores.append(s.sentiments) # 情感得分 except Exception as e: print(f"分析文本情感时出错: {e}, 文本: {text[:50]}...") scores.append(None) return scores def update_sentiment_to_db(engine): """为未分析情感的微博计算情感得分并更新数据库""" query = "SELECT id, clean_text FROM weibo_data WHERE sentiment_score IS NULL AND clean_text IS NOT NULL AND LENGTH(clean_text) > 1" df_to_analyze = pd.read_sql_query(query, engine) if df_to_analyze.empty: print("没有需要分析情感的新微博。") return print(f"开始为 {len(df_to_analyze)} 条微博进行情感分析...") sentiment_scores = analyze_sentiment_batch(df_to_analyze['clean_text'].tolist()) df_to_analyze['sentiment_score'] = sentiment_scores # 过滤掉分析失败的记录（得分为None） df_to_update = df_to_analyze.dropna(subset=['sentiment_score']) # 批量更新数据库 with engine.begin() as conn: for _, row in df_to_update.iterrows(): update_sql = "UPDATE weibo_data SET sentiment_score = :score WHERE id = :id" conn.execute(update_sql, {'score': row['sentiment_score'], 'id': row['id']}) print(f"情感分析完成，成功更新 {len(df_to_update)} 条记录。") # 打印一些统计信息 if not df_to_update.empty: avg_score = df_to_update['sentiment_score'].mean() print(f"平均情感得分: {avg_score:.3f}") print(f"正面微博占比 (>0.6): {(df_to_update['sentiment_score'] > 0.6).mean()*100:.1f}%") print(f"负面微博占比 (<0.4): {(df_to_update['sentiment_score'] < 0.4).mean()*100:.1f}%")

6.2 情感分析结果的局限性及优化方向

SnowNLP开箱即用，但必须清醒认识其局限性：

1. 领域适应性：其模型主要在电商评论上训练，对微博上的网络用语、反讽、缩写等可能识别不准。例如，“这操作真是绝了！”在电商可能是好评，在微博可能是负面吐槽。
2. 中性区域模糊：得分在0.4-0.6之间的文本，情感倾向非常模糊，简单二分法（正面/负面）会丢失大量信息。
3. 缺乏强度区分：“不错”和“太好了！”都是正面，但强度不同，得分可能接近。

优化建议：

• 自定义训练：SnowNLP支持用自己的标注数据重新训练情感分析模型。可以手动标注一批微博数据（正面、负面、中性），训练一个更贴合社交媒体领域的模型。
• 结合词典与规则：针对特定领域（如疫情），可以构建一个情感词典，包含该领域特有的正向词（如“加油”、“致敬”）和负向词（如“恐慌”、“失望”），结合规则进行加权判断。
• 尝试其他工具：对于更复杂的分析，可以探索BosonNLP、百度NLP、腾讯NLP等商业API，或者使用Transformers库加载预训练的中文情感分析模型（如bert-base-chinese微调后的模型），这些通常效果更好，但复杂度也更高。

7. 项目整合、调度与展望

至此，我们完成了数据采集、存储、清洗、分析和可视化的全流程。但一个完整的项目还需要一个“大脑”来调度这一切。

7.1 使用Apache Airflow或简单脚本进行任务调度

对于生产环境，通常会使用Apache Airflow这样的工作流调度平台，以DAG（有向无环图）的形式定义任务依赖关系（如：先爬虫，再清洗，再分析，最后可视化）。

对于个人项目或学习，一个简单的Python调度脚本足矣。我们可以使用schedule库或操作系统的crontab(Linux/macOS) /任务计划程序(Windows)。

# main_scheduler.py import time import schedule from datetime import datetime import logging # 导入各个模块的功能函数 from epidemic_spider import fetch_and_save_epidemic_data from weibo_spider import fetch_and_save_weibo_by_keywords from data_cleaner import batch_text_preprocessing, clean_and_update_epidemic_data from sentiment_analyzer import update_sentiment_to_db from visualizer import generate_all_reports logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def daily_job(): """每日执行的任务流水线""" logging.info("开始执行每日数据管道任务...") try: # 1. 爬取数据 fetch_and_save_epidemic_data() fetch_and_save_weibo_by_keywords(['疫情', '疫苗', '隔离']) # 2. 清洗数据 clean_and_update_epidemic_data() batch_text_preprocessing() # 3. 情感分析 update_sentiment_to_db() # 4. 生成可视化报告 generate_all_reports() logging.info("每日数据管道任务执行完毕！") except Exception as e: logging.error(f"任务执行失败: {e}", exc_info=True) if __name__ == '__main__': # 每天凌晨2点执行 schedule.every().day.at("02:00").do(daily_job) logging.info("调度器已启动，等待执行时间...") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次

7.2 项目总结与扩展思考

回顾这个项目，它本质上构建了一个微型的、闭环的数据流水线。在这个过程中，我们不仅练习了Python多个库的使用，更重要的是学习了如何将分散的技术点串联起来解决一个实际问题。

可以进一步扩展的方向：

1. 前端展示：使用Flask或Streamlit快速搭建一个Web仪表盘，将可视化图表集成到网页中，实现交互式查询。
2. 更复杂的分析：除了情感分析，还可以做主题模型（如LDA）来发现微博讨论的热点话题；或者将疫情数据与微博情感数据在时间序列上进行相关性分析。
3. 容器化与部署：使用Docker将整个项目（Python环境、代码、SQLite数据库）容器化，方便在任何地方一键运行。
4. 数据源扩展：接入更多数据源，如新闻网站、论坛帖子，进行多源信息对比分析。
5. 告警机制：当某地疫情数据突变，或微博负面情感激增时，自动发送邮件或短信告警。

最后的实操心得：数据项目中最耗时、最棘手的部分往往不是编写核心算法，而是数据采集的稳定性和数据清洗的琐碎性。在开始任何分析之前，请务必投入足够的时间确保你的数据管道是可靠和干净的。这个项目提供了一个完整的框架和许多踩坑经验，希望能帮助你更顺畅地开启自己的数据之旅。记住，从能跑通的简单版本开始，逐步迭代和优化，是学习这类综合项目的最佳路径。

本文还有配套的精品资源，点击获取