據(jù)工程實戰(zhàn)：從爬蟲到情感分析的全鏈路項目拆解)
簡介本資源是一套完整的Python數(shù)據(jù)工程實踐項目面向計算機、數(shù)學(xué)與電子信息等專業(yè)的本科生及初學(xué)者聚焦疫情數(shù)據(jù)采集、社交媒體輿情分析與可視化呈現(xiàn)全流程。項目涵蓋疫情實時爬蟲、微博關(guān)鍵詞定向抓取含MySQL數(shù)據(jù)庫存儲、結(jié)構(gòu)化數(shù)據(jù)清洗、多維度統(tǒng)計可視化Matplotlib/Seaborn及基于詞典的情感傾向分析正負樣本標注情感得分計算可直接用于課程設(shè)計、期末大作業(yè)或畢業(yè)設(shè)計參考。壓縮包共15個文件包含7個核心Python腳本爬蟲、入庫、預(yù)處理、繪圖、情感分析、3個CSV數(shù)據(jù)集含標注的nCoV訓(xùn)練樣本與情感關(guān)鍵詞庫、2個文本詞典、1個JSON微博原始數(shù)據(jù)樣例、1張結(jié)果圖表及1份Markdown項目說明文檔整體大小23.87MB。已有728人學(xué)習(xí)下載提供從數(shù)據(jù)獲取到價值提煉的完整鏈路代碼與結(jié)構(gòu)化目錄便于理解模塊分工、調(diào)試邏輯與擴展功能。1. 項目全景一個數(shù)據(jù)工程師的“練手”與“實戰(zhàn)”項目剖析最近在整理硬盤翻出來一個前兩年做的老項目一個集成了疫情數(shù)據(jù)爬取、微博輿情抓取、數(shù)據(jù)清洗、可視化看板以及情感分析的綜合型Python腳本包。當時做這個的初衷一方面是響應(yīng)社區(qū)里不少朋友對“如何將爬蟲、數(shù)據(jù)庫、數(shù)據(jù)分析串起來做一個完整項目”的需求另一方面也是想給自己手頭的一些技術(shù)棧做個壓力測試和流程梳理?，F(xiàn)在看來這個項目雖然涉及的領(lǐng)域疫情、微博熱度有所變化但其技術(shù)棧的整合思路和踩過的坑對于想從“寫單個腳本”進階到“搭建小型數(shù)據(jù)管道”的Python開發(fā)者來說依然有很強的參考價值。這不是一個玩具項目而是一個麻雀雖小五臟俱全的微型數(shù)據(jù)工程實踐涵蓋了從數(shù)據(jù)采集、存儲、加工到分析、呈現(xiàn)的全鏈路。簡單來說這個項目幫你解決幾個核心問題第一面對多個異構(gòu)數(shù)據(jù)源如結(jié)構(gòu)化的公開疫情數(shù)據(jù)、半結(jié)構(gòu)化的微博網(wǎng)頁如何設(shè)計穩(wěn)定、可維護的爬蟲第二海量的臨時數(shù)據(jù)如何有效地存入數(shù)據(jù)庫進行管理而不是散落在無數(shù)個CSV文件里第三爬下來的“臟數(shù)據(jù)”如何通過預(yù)處理變成“干凈數(shù)據(jù)”為后續(xù)分析掃清障礙第四如何將處理好的數(shù)據(jù)通過直觀的可視化圖表講述故事第五如何對文本數(shù)據(jù)如微博內(nèi)容進行初步的情感傾向判斷如果你對這些問題感興趣或者正想找一個綜合項目來鞏固你的Python、Requests、Pandas、SQL、Matplotlib/Seaborn乃至簡單的機器學(xué)習(xí)庫如SnowNLP或TextBlob技能那么這個項目拆解會非常適合你。2. 雙線并行的爬蟲架構(gòu)設(shè)計與核心實現(xiàn)項目的起點是數(shù)據(jù)而數(shù)據(jù)來自兩個風(fēng)格迥異的源頭相對規(guī)范的公開疫情數(shù)據(jù)接口和動態(tài)復(fù)雜、反爬機制多樣的微博網(wǎng)頁。這要求我們的爬蟲架構(gòu)不能是單打獨斗的腳本堆砌而需要有清晰的職責(zé)劃分和統(tǒng)一的錯誤處理機制。2.1 疫情數(shù)據(jù)爬蟲與公開API的規(guī)范對話對于疫情數(shù)據(jù)我們通?？梢哉业礁骷壭l(wèi)健委或權(quán)威數(shù)據(jù)平臺提供的API或結(jié)構(gòu)化的數(shù)據(jù)文件。這類爬蟲的核心在于“協(xié)議遵守”和“數(shù)據(jù)解析”。技術(shù)選型與核心邏輯我們首選requests庫因為它簡單、高效、社區(qū)資源豐富。關(guān)鍵在于構(gòu)造合法的請求頭User-Agent,Referer等模擬一個真實瀏覽器的訪問行為。對于返回的數(shù)據(jù)通常是JSON格式我們可以直接用response.json()解析。如果數(shù)據(jù)是HTML則可能需要結(jié)合BeautifulSoup或lxml進行解析。import requests import pandas as pd from datetime import datetime def fetch_epidemic_data(api_url): 從指定API獲取疫情數(shù)據(jù) headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., Accept: application/json, } try: response requests.get(api_url, headersheaders, timeout10) response.raise_for_status() # 檢查HTTP請求是否成功 data response.json() # 假設(shè)API返回結(jié)構(gòu)為 {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避免后續(xù)流程中斷關(guān)鍵細節(jié)與避坑頻率限制與禮貌爬取公開API也可能有調(diào)用頻率限制。務(wù)必在代碼中加入time.sleep(random.uniform(1, 3))這樣的隨機延時避免對服務(wù)器造成壓力這也是基本的網(wǎng)絡(luò)禮儀。數(shù)據(jù)更新邏輯這是一個典型的增量型爬蟲場景。我們不應(yīng)該每次都全量爬取所有歷史數(shù)據(jù)。理想的做法是在數(shù)據(jù)庫中記錄上次爬取的最新數(shù)據(jù)日期本次只請求該日期之后的新數(shù)據(jù)。這極大地節(jié)省了網(wǎng)絡(luò)和存儲資源。錯誤處理與重試網(wǎng)絡(luò)請求充滿不確定性。必須用try...except包裹核心請求代碼并設(shè)計重試機制如使用tenacity庫。記錄失敗的URL和原因便于后續(xù)排查。數(shù)據(jù)驗證解析后的數(shù)據(jù)在入庫前應(yīng)進行基本的驗證如檢查必要字段是否存在、數(shù)據(jù)類型是否正確、是否有異常值如負的確診數(shù)。2.2 微博關(guān)鍵詞爬蟲在動態(tài)網(wǎng)頁與反爬機制間周旋微博爬蟲的復(fù)雜度提升了一個數(shù)量級。它面臨登錄態(tài)、動態(tài)加載、反爬算法如滑塊驗證、請求簽名等問題。這里我們探討一種相對可行但需嚴格遵守微博Robots協(xié)議和使用限制的思路。技術(shù)選型對于動態(tài)內(nèi)容Selenium或Playwright這類瀏覽器自動化工具可以完美渲染頁面但速度慢、資源消耗大。更高效的方法是分析微博的XHR/Fetch請求找到直接返回數(shù)據(jù)的接口然后用requests模擬調(diào)用。這需要用到瀏覽器的開發(fā)者工具F12進行網(wǎng)絡(luò)抓包分析。核心實現(xiàn)步驟關(guān)鍵詞搜索接口分析在微博網(wǎng)頁或手機端進行關(guān)鍵詞搜索觀察網(wǎng)絡(luò)請求找到一個返回JSON格式搜索結(jié)果的請求。復(fù)制其cURL命令并導(dǎo)入到requests代碼中。參數(shù)逆向該接口的URL和請求頭Headers中通常包含加密參數(shù)如_s、sign等。這些參數(shù)可能由前端JavaScript生成。一種方法是使用execjs庫執(zhí)行相關(guān)的JS代碼來生成另一種更簡單但可能不穩(wěn)定的方法是這些參數(shù)有時在頁面源碼的某個全局變量里可以找到。請求模擬與數(shù)據(jù)提取構(gòu)造包含關(guān)鍵詞、頁碼、時間范圍等參數(shù)的請求。解析返回的JSON提取博文ID、內(nèi)容、發(fā)布時間、發(fā)布人、轉(zhuǎn)發(fā)評論點贊數(shù)等信息。import requests import execjs import json def fetch_weibo_by_keyword(keyword, page1): 模擬微博關(guān)鍵詞搜索請求示例實際參數(shù)需動態(tài)生成 # 1. 加載生成簽名參數(shù)的JS文件 with open(weibo_sign.js, r, encodingutf-8) as f: js_code f.read() ctx execjs.compile(js_code) # 2. 調(diào)用JS函數(shù)生成動態(tài)參數(shù) 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: fhttps://m.weibo.cn/search?keyword{keyword}, X-Requested-With: XMLHttpRequest } params { containerid: f100103type1q{keyword}, page_type: searchall, page: page, **dynamic_params # 合并動態(tài)生成的簽名參數(shù) } try: resp requests.get(url, headersheaders, paramsparams, timeout15) data resp.json() # 解析data[data][cards] 結(jié)構(gòu)提取博文信息 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爬取微博關(guān)鍵詞[{keyword}]失敗: {e}) return []重要注意事項與倫理邊界警告爬取微博等社交媒體數(shù)據(jù)必須嚴格遵守其robots.txt協(xié)議并尊重用戶隱私和數(shù)據(jù)版權(quán)。本示例僅用于技術(shù)學(xué)習(xí)交流。在實際應(yīng)用中務(wù)必控制爬取頻率避免對目標服務(wù)器造成干擾。不要爬取非公開信息。對爬取的數(shù)據(jù)進行脫敏處理避免泄露用戶個人身份信息。明確數(shù)據(jù)用途不用于非法或不道德的目的?？紤]到微博反爬機制的持續(xù)更新此方法可能需要定期維護。3. 數(shù)據(jù)存儲層SQLite與MySQL的選型及表結(jié)構(gòu)設(shè)計爬取到的數(shù)據(jù)是“流沙”必須存入“容器”才能進行有效的管理和分析。在這個項目中數(shù)據(jù)庫的選擇和設(shè)計是承上啟下的關(guān)鍵一環(huán)。3.1 數(shù)據(jù)庫選型輕量級SQLite vs 生產(chǎn)級MySQL對于個人學(xué)習(xí)、小型項目或原型驗證SQLite是絕佳選擇。它是一個無服務(wù)器、零配置、事務(wù)性的SQL數(shù)據(jù)庫引擎整個數(shù)據(jù)庫就是一個文件用Python標準庫sqlite3即可操作無需安裝和運行獨立的數(shù)據(jù)庫服務(wù)。import sqlite3 import pandas as pd # 連接數(shù)據(jù)庫如果不存在則創(chuàng)建 conn sqlite3.connect(epidemic_weibo.db) # 將Pandas DataFrame直接寫入表 df_epidemic.to_sql(epidemic_data, conn, if_existsappend, indexFalse) conn.close()對于數(shù)據(jù)量更大、需要并發(fā)訪問或考慮未來擴展的項目MySQL或PostgreSQL更合適。它們提供了更完善的用戶權(quán)限管理、存儲過程、觸發(fā)器和更強的并發(fā)性能?？梢允褂胮ymysql或sqlalchemy庫進行連接。選型建議本項目作為學(xué)習(xí)整合初期強烈推薦使用SQLite。它讓你聚焦于SQL操作和表結(jié)構(gòu)設(shè)計本身而無需分心于數(shù)據(jù)庫服務(wù)的安裝、配置和維護。當數(shù)據(jù)量增長到數(shù)十萬條以上或需要多人協(xié)作時再考慮遷移到MySQL。3.2 表結(jié)構(gòu)設(shè)計為分析與關(guān)聯(lián)做準備良好的表結(jié)構(gòu)是高效查詢和分析的基礎(chǔ)。我們需要為疫情數(shù)據(jù)和微博數(shù)據(jù)分別設(shè)計表并考慮它們之間潛在的關(guān)聯(lián)點例如通過“日期”和“地區(qū)”。疫情數(shù)據(jù)表 (epidemic_data)CREATE TABLE IF NOT EXISTS epidemic_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 自增主鍵 date DATE NOT NULL, -- 統(tǒng)計日期 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 -- 爬取時間 ); -- 創(chuàng)建復(fù)合索引加速按日期和地區(qū)的查詢 CREATE INDEX idx_date_province ON epidemic_data (date, province);微博數(shù)據(jù)表 (weibo_data)CREATE TABLE IF NOT EXISTS weibo_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, weibo_id BIGINT UNIQUE, -- 微博唯一ID防止重復(fù)插入 keyword VARCHAR(100), -- 搜索使用的關(guān)鍵詞 text TEXT, -- 微博正文 clean_text TEXT, -- 預(yù)處理后的干凈文本后續(xù)填充 sentiment_score FLOAT, -- 情感分析得分后續(xù)填充 created_at DATETIME, -- 微博發(fā)布時間 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, -- 可以添加地理位置字段用于和疫情數(shù)據(jù)關(guān)聯(lián) 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);設(shè)計心得唯一約束與去重為微博ID (weibo_id) 設(shè)置UNIQUE約束這樣在使用INSERT OR IGNORE或ON DUPLICATE KEY UPDATEMySQL語句時可以自動跳過重復(fù)數(shù)據(jù)這是實現(xiàn)增量爬蟲的關(guān)鍵。預(yù)留加工字段表中直接設(shè)計了clean_text和sentiment_score字段。這樣數(shù)據(jù)預(yù)處理的流水線就可以直接更新原表保持數(shù)據(jù)的一致性也方便后續(xù)查詢。索引是雙刃劍索引能極大加速查詢?nèi)绨搓P(guān)鍵詞和時間篩選微博但會降低數(shù)據(jù)插入的速度。對于爬蟲這種寫多讀少的初期階段可以在數(shù)據(jù)爬取完成后再統(tǒng)一創(chuàng)建索引。4. 數(shù)據(jù)預(yù)處理流水線從“臟數(shù)據(jù)”到“干凈數(shù)據(jù)”直接從網(wǎng)上爬下來的數(shù)據(jù)我們稱之為“原始數(shù)據(jù)”或“臟數(shù)據(jù)”。它可能包含HTML標簽、特殊字符、無關(guān)信息、缺失值、重復(fù)項等。不經(jīng)過清洗直接進行分析或可視化結(jié)果很可能是錯誤的。4.1 文本數(shù)據(jù)清洗以微博內(nèi)容為例微博文本清洗是一個典型的多步驟流水線作業(yè)使用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(rhttp[s]?://\S, , text) # 2. 移除用戶和話題# text re.sub(r[\w\u4e00-\u9fa5\-], , text) text re.sub(r#([^#])#, r\1, text) # 保留話題內(nèi)容去掉#號 # 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_pathepidemic_weibo.db): 批量預(yù)處理微博文本并更新數(shù)據(jù)庫 engine create_engine(fsqlite:///{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(沒有需要清洗的新微博數(shù)據(jù)。) return # 應(yīng)用清洗函數(shù) df[clean_text] df[text].apply(clean_weibo_text) # 可選進行分詞便于后續(xù)深度分析 df[segmented] df[clean_text].apply(lambda x: .join(jieba.lcut(x))) # 將清洗結(jié)果更新回數(shù)據(jù)庫 # 這里采用逐條更新或批量更新策略。對于SQLite批量更新效率更高。 with engine.begin() as conn: # 使用事務(wù) 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)} 條微博數(shù)據(jù)。)4.2 結(jié)構(gòu)化數(shù)據(jù)清洗以疫情數(shù)據(jù)為例疫情數(shù)據(jù)的清洗更側(cè)重于數(shù)據(jù)完整性和邏輯一致性。def clean_epidemic_data(df): 清洗疫情DataFrame # 1. 處理缺失值對于數(shù)值列用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. 數(shù)據(jù)類型轉(zhuǎn)換 df[numeric_cols] df[numeric_cols].apply(pd.to_numeric, errorscoerce).fillna(0).astype(int) df[date] pd.to_datetime(df[date], errorscoerce) # 3. 邏輯校驗與修正例如累計數(shù)不應(yīng)小于新增數(shù) # 這里可以添加更復(fù)雜的業(yè)務(wù)規(guī)則校驗 mask df[confirmed_total] df[confirmed_new] df.loc[mask, confirmed_total] df.loc[mask, confirmed_new] # 簡單處理實際需根據(jù)業(yè)務(wù)判斷 # 4. 去除完全重復(fù)的行基于所有字段 df df.drop_duplicates() # 5. 去除關(guān)鍵字段為空的無效行 df df.dropna(subset[date, province]) return df預(yù)處理階段的核心經(jīng)驗流水線化將清洗步驟封裝成函數(shù)形成可重復(fù)執(zhí)行的流水線。新的數(shù)據(jù)來了跑一遍流水線即可。保留原始數(shù)據(jù)永遠不要在原始數(shù)據(jù)上直接修改。我們的策略是原始數(shù)據(jù)存入數(shù)據(jù)庫后通過新增字段如clean_text或新表來存儲清洗后的結(jié)果。日志記錄在清洗函數(shù)中加入日志記錄處理了多少條數(shù)據(jù)、遇到了多少異常、修正了哪些問題。這對于監(jiān)控數(shù)據(jù)質(zhì)量和調(diào)試至關(guān)重要。5. 數(shù)據(jù)可視化用圖表講述疫情與輿情的“故事”數(shù)據(jù)清洗后就到了最直觀的環(huán)節(jié)——可視化。我們使用matplotlib和seaborn庫目標是生成能清晰反映趨勢、分布和關(guān)聯(lián)的圖表。5.1 疫情數(shù)據(jù)可視化趨勢與分布全國疫情趨勢折線圖展示每日全國新增確診、治愈、死亡的變化趨勢。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新增確診, linewidth2, markero) ax.plot(df_trend[date], df_trend[new_cured], label新增治愈, linewidth2, markers) ax.plot(df_trend[date], df_trend[new_dead], label新增死亡, linewidth2, marker^) ax.set_xlabel(日期) ax.set_ylabel(人數(shù)) ax.set_title(全國疫情每日新增趨勢) ax.legend() ax.grid(True, linestyle--, alpha0.6) plt.xticks(rotation45) plt.tight_layout() plt.savefig(national_trend.png, dpi300) plt.show()各省累計確診分布熱力圖需地理坐標數(shù)據(jù)或條形圖如果沒有地理坐標可以用條形圖展示Top 10省份。def plot_top_provinces(engine, top_n10): 繪制累計確診最多的前N個省份 query SELECT province, SUM(confirmed_total) as total_confirmed FROM epidemic_data WHERE date (SELECT MAX(date) FROM epidemic_data) -- 取最新一天的數(shù)據(jù) 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], colorsns.color_palette(Reds_r, top_n)) ax.set_xlabel(累計確診人數(shù)) ax.set_title(f累計確診人數(shù)最多的前{top_n}個省份截至最新數(shù)據(jù)) # 在條形末端添加數(shù)據(jù)標簽 for bar in bars: width bar.get_width() ax.text(width width*0.01, bar.get_y() bar.get_height()/2, f{int(width):,}, vacenter) plt.tight_layout() plt.savefig(top_provinces.png, dpi300) plt.show()5.2 微博輿情可視化情感與熱詞微博情感得分分布直方圖查看公眾情緒的整體傾向。def plot_sentiment_distribution(engine, keyword疫情): 繪制特定關(guān)鍵詞下微博情感得分分布 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關(guān)鍵詞 {keyword} 下沒有情感分析數(shù)據(jù)。) return fig, ax plt.subplots(figsize(10, 6)) # 情感得分范圍通常在0-1之間0.5為中性 ax.hist(df_sentiment[sentiment_score], bins30, edgecolorblack, alpha0.7, colorskyblue) ax.axvline(x0.5, colorred, linestyle--, label中性線 (0.5)) ax.set_xlabel(情感得分 (0: 負面, 1: 正面)) ax.set_ylabel(微博數(shù)量) ax.set_title(f關(guān)鍵詞“{keyword}”相關(guān)微博情感分布) ax.legend() ax.grid(True, linestyle--, alpha0.3) plt.tight_layout() plt.savefig(fsentiment_dist_{keyword}.png, dpi300) plt.show()微博發(fā)布數(shù)量時間序列圖觀察輿情熱度隨時間的變化。def plot_weibo_volume_trend(engine, keyword疫情): 繪制特定關(guān)鍵詞下微博發(fā)布數(shù)量隨時間變化趨勢 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], markero, linewidth2, colororange) ax.set_xlabel(日期) ax.set_ylabel(微博發(fā)布數(shù)量) ax.set_title(f關(guān)鍵詞“{keyword}”每日微博討論量趨勢) ax.fill_between(df_volume[post_date], df_volume[weibo_count], alpha0.3, colororange) plt.xticks(rotation45) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.savefig(fweibo_volume_trend_{keyword}.png, dpi300) plt.show()可視化經(jīng)驗談圖表服務(wù)于故事不要為了炫技而作圖。先想清楚你想通過圖表表達什么信息例如疫情是否得到控制輿情是樂觀還是悲觀再選擇合適的圖表類型。細節(jié)決定專業(yè)度添加清晰的標題、軸標簽、圖例。合理設(shè)置顏色、線寬、標記點。使用plt.tight_layout()避免標簽重疊。保存圖片時指定高dpi(如300) 以保證印刷或展示質(zhì)量。自動化報告可以將上述繪圖函數(shù)封裝起來并編寫一個主函數(shù)在數(shù)據(jù)更新后自動運行生成一套最新的圖表甚至可以用Jinja2模板生成HTML報告。6. 微博文本情感分析SnowNLP實戰(zhàn)與結(jié)果解讀情感分析是自然語言處理(NLP)的入門應(yīng)用旨在判斷一段文本所表達的情感傾向是正面、負面還是中性。對于中文微博SnowNLP庫是一個簡單易用的選擇。6.1 使用SnowNLP進行批量情感分析SnowNLP基于樸素貝葉斯算法訓(xùn)練可以直接對中文文本進行情感打分0到1之間越接近1表示越正面。from snownlp import SnowNLP import numpy as np def analyze_sentiment_batch(text_list): 批量分析情感返回情感得分列表。 注意SnowNLP的情感分析基于商品評論訓(xùn)練對社交媒體文本可能有一定偏差。 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): 為未分析情感的微博計算情感得分并更新數(shù)據(jù)庫 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]) # 批量更新數(shù)據(jù)庫 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)} 條記錄。) # 打印一些統(tǒng)計信息 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 情感分析結(jié)果的局限性及優(yōu)化方向SnowNLP開箱即用但必須清醒認識其局限性領(lǐng)域適應(yīng)性其模型主要在電商評論上訓(xùn)練對微博上的網(wǎng)絡(luò)用語、反諷、縮寫等可能識別不準。例如“這操作真是絕了”在電商可能是好評在微博可能是負面吐槽。中性區(qū)域模糊得分在0.4-0.6之間的文本情感傾向非常模糊簡單二分法正面/負面會丟失大量信息。缺乏強度區(qū)分“不錯”和“太好了”都是正面但強度不同得分可能接近。優(yōu)化建議自定義訓(xùn)練SnowNLP支持用自己的標注數(shù)據(jù)重新訓(xùn)練情感分析模型。可以手動標注一批微博數(shù)據(jù)正面、負面、中性訓(xùn)練一個更貼合社交媒體領(lǐng)域的模型。結(jié)合詞典與規(guī)則針對特定領(lǐng)域如疫情可以構(gòu)建一個情感詞典包含該領(lǐng)域特有的正向詞如“加油”、“致敬”和負向詞如“恐慌”、“失望”結(jié)合規(guī)則進行加權(quán)判斷。嘗試其他工具對于更復(fù)雜的分析可以探索BosonNLP、百度NLP、騰訊NLP等商業(yè)API或者使用Transformers庫加載預(yù)訓(xùn)練的中文情感分析模型如bert-base-chinese微調(diào)后的模型這些通常效果更好但復(fù)雜度也更高。7. 項目整合、調(diào)度與展望至此我們完成了數(shù)據(jù)采集、存儲、清洗、分析和可視化的全流程。但一個完整的項目還需要一個“大腦”來調(diào)度這一切。7.1 使用Apache Airflow或簡單腳本進行任務(wù)調(diào)度對于生產(chǎn)環(huán)境通常會使用Apache Airflow這樣的工作流調(diào)度平臺以DAG有向無環(huán)圖的形式定義任務(wù)依賴關(guān)系如先爬蟲再清洗再分析最后可視化。對于個人項目或?qū)W習(xí)一個簡單的Python調(diào)度腳本足矣。我們可以使用schedule庫或操作系統(tǒng)的crontab(Linux/macOS) /任務(wù)計劃程序(Windows)。# main_scheduler.py import time import schedule from datetime import datetime import logging # 導(dǎo)入各個模塊的功能函數(shù) 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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def daily_job(): 每日執(zhí)行的任務(wù)流水線 logging.info(開始執(zhí)行每日數(shù)據(jù)管道任務(wù)...) try: # 1. 爬取數(shù)據(jù) fetch_and_save_epidemic_data() fetch_and_save_weibo_by_keywords([疫情, 疫苗, 隔離]) # 2. 清洗數(shù)據(jù) clean_and_update_epidemic_data() batch_text_preprocessing() # 3. 情感分析 update_sentiment_to_db() # 4. 生成可視化報告 generate_all_reports() logging.info(每日數(shù)據(jù)管道任務(wù)執(zhí)行完畢) except Exception as e: logging.error(f任務(wù)執(zhí)行失敗: {e}, exc_infoTrue) if __name__ __main__: # 每天凌晨2點執(zhí)行 schedule.every().day.at(02:00).do(daily_job) logging.info(調(diào)度器已啟動等待執(zhí)行時間...) while True: schedule.run_pending() time.sleep(60) # 每分鐘檢查一次7.2 項目總結(jié)與擴展思考回顧這個項目它本質(zhì)上構(gòu)建了一個微型的、閉環(huán)的數(shù)據(jù)流水線。在這個過程中我們不僅練習(xí)了Python多個庫的使用更重要的是學(xué)習(xí)了如何將分散的技術(shù)點串聯(lián)起來解決一個實際問題?？梢赃M一步擴展的方向前端展示使用Flask或Streamlit快速搭建一個Web儀表盤將可視化圖表集成到網(wǎng)頁中實現(xiàn)交互式查詢。更復(fù)雜的分析除了情感分析還可以做主題模型如LDA來發(fā)現(xiàn)微博討論的熱點話題或者將疫情數(shù)據(jù)與微博情感數(shù)據(jù)在時間序列上進行相關(guān)性分析。容器化與部署使用Docker將整個項目Python環(huán)境、代碼、SQLite數(shù)據(jù)庫容器化方便在任何地方一鍵運行。數(shù)據(jù)源擴展接入更多數(shù)據(jù)源如新聞網(wǎng)站、論壇帖子進行多源信息對比分析。告警機制當某地疫情數(shù)據(jù)突變或微博負面情感激增時自動發(fā)送郵件或短信告警。最后的實操心得數(shù)據(jù)項目中最耗時、最棘手的部分往往不是編寫核心算法而是數(shù)據(jù)采集的穩(wěn)定性和數(shù)據(jù)清洗的瑣碎性。在開始任何分析之前請務(wù)必投入足夠的時間確保你的數(shù)據(jù)管道是可靠和干凈的。這個項目提供了一個完整的框架和許多踩坑經(jīng)驗希望能幫助你更順暢地開啟自己的數(shù)據(jù)之旅。記住從能跑通的簡單版本開始逐步迭代和優(yōu)化是學(xué)習(xí)這類綜合項目的最佳路徑。本文還有配套的精品資源點擊獲取 Python數(shù)據(jù)工程實戰(zhàn)：從爬蟲到情感分析的全鏈路項目拆解
簡介本資源是一套完整的Python數(shù)據(jù)工程實踐項目面向計算機、數(shù)學(xué)與電子信息等專業(yè)的本科生及初學(xué)者聚焦疫情數(shù)據(jù)采集、社交媒體輿情分析與可視化呈現(xiàn)全流程。項目涵蓋疫情實時爬蟲、微博關(guān)鍵詞定向抓取含MySQL數(shù)據(jù)庫存儲、結(jié)構(gòu)化數(shù)據(jù)清洗、多維度統(tǒng)計可視化Matplotlib/Seaborn及基于詞典的情感傾向分析正負樣本標注情感得分計算可直接用于課程設(shè)計、期末大作業(yè)或畢業(yè)設(shè)計參考。壓縮包共15個文件包含7個核心Python腳本爬蟲、入庫、預(yù)處理、繪圖、情感分析、3個CSV數(shù)據(jù)集含標注的nCoV訓(xùn)練樣本與情感關(guān)鍵詞庫、2個文本詞典、1個JSON微博原始數(shù)據(jù)樣例、1張結(jié)果圖表及1份Markdown項目說明文檔整體大小23.87MB。已有728人學(xué)習(xí)下載提供從數(shù)據(jù)獲取到價值提煉的完整鏈路代碼與結(jié)構(gòu)化目錄便于理解模塊分工、調(diào)試邏輯與擴展功能。1. 項目全景一個數(shù)據(jù)工程師的“練手”與“實戰(zhàn)”項目剖析最近在整理硬盤翻出來一個前兩年做的老項目一個集成了疫情數(shù)據(jù)爬取、微博輿情抓取、數(shù)據(jù)清洗、可視化看板以及情感分析的綜合型Python腳本包。當時做這個的初衷一方面是響應(yīng)社區(qū)里不少朋友對“如何將爬蟲、數(shù)據(jù)庫、數(shù)據(jù)分析串起來做一個完整項目”的需求另一方面也是想給自己手頭的一些技術(shù)棧做個壓力測試和流程梳理?，F(xiàn)在看來這個項目雖然涉及的領(lǐng)域疫情、微博熱度有所變化但其技術(shù)棧的整合思路和踩過的坑對于想從“寫單個腳本”進階到“搭建小型數(shù)據(jù)管道”的Python開發(fā)者來說依然有很強的參考價值。這不是一個玩具項目而是一個麻雀雖小五臟俱全的微型數(shù)據(jù)工程實踐涵蓋了從數(shù)據(jù)采集、存儲、加工到分析、呈現(xiàn)的全鏈路。簡單來說這個項目幫你解決幾個核心問題第一面對多個異構(gòu)數(shù)據(jù)源如結(jié)構(gòu)化的公開疫情數(shù)據(jù)、半結(jié)構(gòu)化的微博網(wǎng)頁如何設(shè)計穩(wěn)定、可維護的爬蟲第二海量的臨時數(shù)據(jù)如何有效地存入數(shù)據(jù)庫進行管理而不是散落在無數(shù)個CSV文件里第三爬下來的“臟數(shù)據(jù)”如何通過預(yù)處理變成“干凈數(shù)據(jù)”為后續(xù)分析掃清障礙第四如何將處理好的數(shù)據(jù)通過直觀的可視化圖表講述故事第五如何對文本數(shù)據(jù)如微博內(nèi)容進行初步的情感傾向判斷如果你對這些問題感興趣或者正想找一個綜合項目來鞏固你的Python、Requests、Pandas、SQL、Matplotlib/Seaborn乃至簡單的機器學(xué)習(xí)庫如SnowNLP或TextBlob技能那么這個項目拆解會非常適合你。2. 雙線并行的爬蟲架構(gòu)設(shè)計與核心實現(xiàn)項目的起點是數(shù)據(jù)而數(shù)據(jù)來自兩個風(fēng)格迥異的源頭相對規(guī)范的公開疫情數(shù)據(jù)接口和動態(tài)復(fù)雜、反爬機制多樣的微博網(wǎng)頁。這要求我們的爬蟲架構(gòu)不能是單打獨斗的腳本堆砌而需要有清晰的職責(zé)劃分和統(tǒng)一的錯誤處理機制。2.1 疫情數(shù)據(jù)爬蟲與公開API的規(guī)范對話對于疫情數(shù)據(jù)我們通?？梢哉业礁骷壭l(wèi)健委或權(quán)威數(shù)據(jù)平臺提供的API或結(jié)構(gòu)化的數(shù)據(jù)文件。這類爬蟲的核心在于“協(xié)議遵守”和“數(shù)據(jù)解析”。技術(shù)選型與核心邏輯我們首選requests庫因為它簡單、高效、社區(qū)資源豐富。關(guān)鍵在于構(gòu)造合法的請求頭User-Agent,Referer等模擬一個真實瀏覽器的訪問行為。對于返回的數(shù)據(jù)通常是JSON格式我們可以直接用response.json()解析。如果數(shù)據(jù)是HTML則可能需要結(jié)合BeautifulSoup或lxml進行解析。import requests import pandas as pd from datetime import datetime def fetch_epidemic_data(api_url): 從指定API獲取疫情數(shù)據(jù) headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., Accept: application/json, } try: response requests.get(api_url, headersheaders, timeout10) response.raise_for_status() # 檢查HTTP請求是否成功 data response.json() # 假設(shè)API返回結(jié)構(gòu)為 {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避免后續(xù)流程中斷關(guān)鍵細節(jié)與避坑頻率限制與禮貌爬取公開API也可能有調(diào)用頻率限制。務(wù)必在代碼中加入time.sleep(random.uniform(1, 3))這樣的隨機延時避免對服務(wù)器造成壓力這也是基本的網(wǎng)絡(luò)禮儀。數(shù)據(jù)更新邏輯這是一個典型的增量型爬蟲場景。我們不應(yīng)該每次都全量爬取所有歷史數(shù)據(jù)。理想的做法是在數(shù)據(jù)庫中記錄上次爬取的最新數(shù)據(jù)日期本次只請求該日期之后的新數(shù)據(jù)。這極大地節(jié)省了網(wǎng)絡(luò)和存儲資源。錯誤處理與重試網(wǎng)絡(luò)請求充滿不確定性。必須用try...except包裹核心請求代碼并設(shè)計重試機制如使用tenacity庫。記錄失敗的URL和原因便于后續(xù)排查。數(shù)據(jù)驗證解析后的數(shù)據(jù)在入庫前應(yīng)進行基本的驗證如檢查必要字段是否存在、數(shù)據(jù)類型是否正確、是否有異常值如負的確診數(shù)。2.2 微博關(guān)鍵詞爬蟲在動態(tài)網(wǎng)頁與反爬機制間周旋微博爬蟲的復(fù)雜度提升了一個數(shù)量級。它面臨登錄態(tài)、動態(tài)加載、反爬算法如滑塊驗證、請求簽名等問題。這里我們探討一種相對可行但需嚴格遵守微博Robots協(xié)議和使用限制的思路。技術(shù)選型對于動態(tài)內(nèi)容Selenium或Playwright這類瀏覽器自動化工具可以完美渲染頁面但速度慢、資源消耗大。更高效的方法是分析微博的XHR/Fetch請求找到直接返回數(shù)據(jù)的接口然后用requests模擬調(diào)用。這需要用到瀏覽器的開發(fā)者工具F12進行網(wǎng)絡(luò)抓包分析。核心實現(xiàn)步驟關(guān)鍵詞搜索接口分析在微博網(wǎng)頁或手機端進行關(guān)鍵詞搜索觀察網(wǎng)絡(luò)請求找到一個返回JSON格式搜索結(jié)果的請求。復(fù)制其cURL命令并導(dǎo)入到requests代碼中。參數(shù)逆向該接口的URL和請求頭Headers中通常包含加密參數(shù)如_s、sign等。這些參數(shù)可能由前端JavaScript生成。一種方法是使用execjs庫執(zhí)行相關(guān)的JS代碼來生成另一種更簡單但可能不穩(wěn)定的方法是這些參數(shù)有時在頁面源碼的某個全局變量里可以找到。請求模擬與數(shù)據(jù)提取構(gòu)造包含關(guān)鍵詞、頁碼、時間范圍等參數(shù)的請求。解析返回的JSON提取博文ID、內(nèi)容、發(fā)布時間、發(fā)布人、轉(zhuǎn)發(fā)評論點贊數(shù)等信息。import requests import execjs import json def fetch_weibo_by_keyword(keyword, page1): 模擬微博關(guān)鍵詞搜索請求示例實際參數(shù)需動態(tài)生成 # 1. 加載生成簽名參數(shù)的JS文件 with open(weibo_sign.js, r, encodingutf-8) as f: js_code f.read() ctx execjs.compile(js_code) # 2. 調(diào)用JS函數(shù)生成動態(tài)參數(shù) 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: fhttps://m.weibo.cn/search?keyword{keyword}, X-Requested-With: XMLHttpRequest } params { containerid: f100103type1q{keyword}, page_type: searchall, page: page, **dynamic_params # 合并動態(tài)生成的簽名參數(shù) } try: resp requests.get(url, headersheaders, paramsparams, timeout15) data resp.json() # 解析data[data][cards] 結(jié)構(gòu)提取博文信息 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爬取微博關(guān)鍵詞[{keyword}]失敗: {e}) return []重要注意事項與倫理邊界警告爬取微博等社交媒體數(shù)據(jù)必須嚴格遵守其robots.txt協(xié)議并尊重用戶隱私和數(shù)據(jù)版權(quán)。本示例僅用于技術(shù)學(xué)習(xí)交流。在實際應(yīng)用中務(wù)必控制爬取頻率避免對目標服務(wù)器造成干擾。不要爬取非公開信息。對爬取的數(shù)據(jù)進行脫敏處理避免泄露用戶個人身份信息。明確數(shù)據(jù)用途不用于非法或不道德的目的?？紤]到微博反爬機制的持續(xù)更新此方法可能需要定期維護。3. 數(shù)據(jù)存儲層SQLite與MySQL的選型及表結(jié)構(gòu)設(shè)計爬取到的數(shù)據(jù)是“流沙”必須存入“容器”才能進行有效的管理和分析。在這個項目中數(shù)據(jù)庫的選擇和設(shè)計是承上啟下的關(guān)鍵一環(huán)。3.1 數(shù)據(jù)庫選型輕量級SQLite vs 生產(chǎn)級MySQL對于個人學(xué)習(xí)、小型項目或原型驗證SQLite是絕佳選擇。它是一個無服務(wù)器、零配置、事務(wù)性的SQL數(shù)據(jù)庫引擎整個數(shù)據(jù)庫就是一個文件用Python標準庫sqlite3即可操作無需安裝和運行獨立的數(shù)據(jù)庫服務(wù)。import sqlite3 import pandas as pd # 連接數(shù)據(jù)庫如果不存在則創(chuàng)建 conn sqlite3.connect(epidemic_weibo.db) # 將Pandas DataFrame直接寫入表 df_epidemic.to_sql(epidemic_data, conn, if_existsappend, indexFalse) conn.close()對于數(shù)據(jù)量更大、需要并發(fā)訪問或考慮未來擴展的項目MySQL或PostgreSQL更合適。它們提供了更完善的用戶權(quán)限管理、存儲過程、觸發(fā)器和更強的并發(fā)性能?？梢允褂胮ymysql或sqlalchemy庫進行連接。選型建議本項目作為學(xué)習(xí)整合初期強烈推薦使用SQLite。它讓你聚焦于SQL操作和表結(jié)構(gòu)設(shè)計本身而無需分心于數(shù)據(jù)庫服務(wù)的安裝、配置和維護。當數(shù)據(jù)量增長到數(shù)十萬條以上或需要多人協(xié)作時再考慮遷移到MySQL。3.2 表結(jié)構(gòu)設(shè)計為分析與關(guān)聯(lián)做準備良好的表結(jié)構(gòu)是高效查詢和分析的基礎(chǔ)。我們需要為疫情數(shù)據(jù)和微博數(shù)據(jù)分別設(shè)計表并考慮它們之間潛在的關(guān)聯(lián)點例如通過“日期”和“地區(qū)”。疫情數(shù)據(jù)表 (epidemic_data)CREATE TABLE IF NOT EXISTS epidemic_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 自增主鍵 date DATE NOT NULL, -- 統(tǒng)計日期 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 -- 爬取時間 ); -- 創(chuàng)建復(fù)合索引加速按日期和地區(qū)的查詢 CREATE INDEX idx_date_province ON epidemic_data (date, province);微博數(shù)據(jù)表 (weibo_data)CREATE TABLE IF NOT EXISTS weibo_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, weibo_id BIGINT UNIQUE, -- 微博唯一ID防止重復(fù)插入 keyword VARCHAR(100), -- 搜索使用的關(guān)鍵詞 text TEXT, -- 微博正文 clean_text TEXT, -- 預(yù)處理后的干凈文本后續(xù)填充 sentiment_score FLOAT, -- 情感分析得分后續(xù)填充 created_at DATETIME, -- 微博發(fā)布時間 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, -- 可以添加地理位置字段用于和疫情數(shù)據(jù)關(guān)聯(lián) 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);設(shè)計心得唯一約束與去重為微博ID (weibo_id) 設(shè)置UNIQUE約束這樣在使用INSERT OR IGNORE或ON DUPLICATE KEY UPDATEMySQL語句時可以自動跳過重復(fù)數(shù)據(jù)這是實現(xiàn)增量爬蟲的關(guān)鍵。預(yù)留加工字段表中直接設(shè)計了clean_text和sentiment_score字段。這樣數(shù)據(jù)預(yù)處理的流水線就可以直接更新原表保持數(shù)據(jù)的一致性也方便后續(xù)查詢。索引是雙刃劍索引能極大加速查詢?nèi)绨搓P(guān)鍵詞和時間篩選微博但會降低數(shù)據(jù)插入的速度。對于爬蟲這種寫多讀少的初期階段可以在數(shù)據(jù)爬取完成后再統(tǒng)一創(chuàng)建索引。4. 數(shù)據(jù)預(yù)處理流水線從“臟數(shù)據(jù)”到“干凈數(shù)據(jù)”直接從網(wǎng)上爬下來的數(shù)據(jù)我們稱之為“原始數(shù)據(jù)”或“臟數(shù)據(jù)”。它可能包含HTML標簽、特殊字符、無關(guān)信息、缺失值、重復(fù)項等。不經(jīng)過清洗直接進行分析或可視化結(jié)果很可能是錯誤的。4.1 文本數(shù)據(jù)清洗以微博內(nèi)容為例微博文本清洗是一個典型的多步驟流水線作業(yè)使用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(rhttp[s]?://\S, , text) # 2. 移除用戶和話題# text re.sub(r[\w\u4e00-\u9fa5\-], , text) text re.sub(r#([^#])#, r\1, text) # 保留話題內(nèi)容去掉#號 # 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_pathepidemic_weibo.db): 批量預(yù)處理微博文本并更新數(shù)據(jù)庫 engine create_engine(fsqlite:///{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(沒有需要清洗的新微博數(shù)據(jù)。) return # 應(yīng)用清洗函數(shù) df[clean_text] df[text].apply(clean_weibo_text) # 可選進行分詞便于后續(xù)深度分析 df[segmented] df[clean_text].apply(lambda x: .join(jieba.lcut(x))) # 將清洗結(jié)果更新回數(shù)據(jù)庫 # 這里采用逐條更新或批量更新策略。對于SQLite批量更新效率更高。 with engine.begin() as conn: # 使用事務(wù) 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)} 條微博數(shù)據(jù)。)4.2 結(jié)構(gòu)化數(shù)據(jù)清洗以疫情數(shù)據(jù)為例疫情數(shù)據(jù)的清洗更側(cè)重于數(shù)據(jù)完整性和邏輯一致性。def clean_epidemic_data(df): 清洗疫情DataFrame # 1. 處理缺失值對于數(shù)值列用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. 數(shù)據(jù)類型轉(zhuǎn)換 df[numeric_cols] df[numeric_cols].apply(pd.to_numeric, errorscoerce).fillna(0).astype(int) df[date] pd.to_datetime(df[date], errorscoerce) # 3. 邏輯校驗與修正例如累計數(shù)不應(yīng)小于新增數(shù) # 這里可以添加更復(fù)雜的業(yè)務(wù)規(guī)則校驗 mask df[confirmed_total] df[confirmed_new] df.loc[mask, confirmed_total] df.loc[mask, confirmed_new] # 簡單處理實際需根據(jù)業(yè)務(wù)判斷 # 4. 去除完全重復(fù)的行基于所有字段 df df.drop_duplicates() # 5. 去除關(guān)鍵字段為空的無效行 df df.dropna(subset[date, province]) return df預(yù)處理階段的核心經(jīng)驗流水線化將清洗步驟封裝成函數(shù)形成可重復(fù)執(zhí)行的流水線。新的數(shù)據(jù)來了跑一遍流水線即可。保留原始數(shù)據(jù)永遠不要在原始數(shù)據(jù)上直接修改。我們的策略是原始數(shù)據(jù)存入數(shù)據(jù)庫后通過新增字段如clean_text或新表來存儲清洗后的結(jié)果。日志記錄在清洗函數(shù)中加入日志記錄處理了多少條數(shù)據(jù)、遇到了多少異常、修正了哪些問題。這對于監(jiān)控數(shù)據(jù)質(zhì)量和調(diào)試至關(guān)重要。5. 數(shù)據(jù)可視化用圖表講述疫情與輿情的“故事”數(shù)據(jù)清洗后就到了最直觀的環(huán)節(jié)——可視化。我們使用matplotlib和seaborn庫目標是生成能清晰反映趨勢、分布和關(guān)聯(lián)的圖表。5.1 疫情數(shù)據(jù)可視化趨勢與分布全國疫情趨勢折線圖展示每日全國新增確診、治愈、死亡的變化趨勢。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新增確診, linewidth2, markero) ax.plot(df_trend[date], df_trend[new_cured], label新增治愈, linewidth2, markers) ax.plot(df_trend[date], df_trend[new_dead], label新增死亡, linewidth2, marker^) ax.set_xlabel(日期) ax.set_ylabel(人數(shù)) ax.set_title(全國疫情每日新增趨勢) ax.legend() ax.grid(True, linestyle--, alpha0.6) plt.xticks(rotation45) plt.tight_layout() plt.savefig(national_trend.png, dpi300) plt.show()各省累計確診分布熱力圖需地理坐標數(shù)據(jù)或條形圖如果沒有地理坐標可以用條形圖展示Top 10省份。def plot_top_provinces(engine, top_n10): 繪制累計確診最多的前N個省份 query SELECT province, SUM(confirmed_total) as total_confirmed FROM epidemic_data WHERE date (SELECT MAX(date) FROM epidemic_data) -- 取最新一天的數(shù)據(jù) 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], colorsns.color_palette(Reds_r, top_n)) ax.set_xlabel(累計確診人數(shù)) ax.set_title(f累計確診人數(shù)最多的前{top_n}個省份截至最新數(shù)據(jù)) # 在條形末端添加數(shù)據(jù)標簽 for bar in bars: width bar.get_width() ax.text(width width*0.01, bar.get_y() bar.get_height()/2, f{int(width):,}, vacenter) plt.tight_layout() plt.savefig(top_provinces.png, dpi300) plt.show()5.2 微博輿情可視化情感與熱詞微博情感得分分布直方圖查看公眾情緒的整體傾向。def plot_sentiment_distribution(engine, keyword疫情): 繪制特定關(guān)鍵詞下微博情感得分分布 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關(guān)鍵詞 {keyword} 下沒有情感分析數(shù)據(jù)。) return fig, ax plt.subplots(figsize(10, 6)) # 情感得分范圍通常在0-1之間0.5為中性 ax.hist(df_sentiment[sentiment_score], bins30, edgecolorblack, alpha0.7, colorskyblue) ax.axvline(x0.5, colorred, linestyle--, label中性線 (0.5)) ax.set_xlabel(情感得分 (0: 負面, 1: 正面)) ax.set_ylabel(微博數(shù)量) ax.set_title(f關(guān)鍵詞“{keyword}”相關(guān)微博情感分布) ax.legend() ax.grid(True, linestyle--, alpha0.3) plt.tight_layout() plt.savefig(fsentiment_dist_{keyword}.png, dpi300) plt.show()微博發(fā)布數(shù)量時間序列圖觀察輿情熱度隨時間的變化。def plot_weibo_volume_trend(engine, keyword疫情): 繪制特定關(guān)鍵詞下微博發(fā)布數(shù)量隨時間變化趨勢 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], markero, linewidth2, colororange) ax.set_xlabel(日期) ax.set_ylabel(微博發(fā)布數(shù)量) ax.set_title(f關(guān)鍵詞“{keyword}”每日微博討論量趨勢) ax.fill_between(df_volume[post_date], df_volume[weibo_count], alpha0.3, colororange) plt.xticks(rotation45) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.savefig(fweibo_volume_trend_{keyword}.png, dpi300) plt.show()可視化經(jīng)驗談圖表服務(wù)于故事不要為了炫技而作圖。先想清楚你想通過圖表表達什么信息例如疫情是否得到控制輿情是樂觀還是悲觀再選擇合適的圖表類型。細節(jié)決定專業(yè)度添加清晰的標題、軸標簽、圖例。合理設(shè)置顏色、線寬、標記點。使用plt.tight_layout()避免標簽重疊。保存圖片時指定高dpi(如300) 以保證印刷或展示質(zhì)量。自動化報告可以將上述繪圖函數(shù)封裝起來并編寫一個主函數(shù)在數(shù)據(jù)更新后自動運行生成一套最新的圖表甚至可以用Jinja2模板生成HTML報告。6. 微博文本情感分析SnowNLP實戰(zhàn)與結(jié)果解讀情感分析是自然語言處理(NLP)的入門應(yīng)用旨在判斷一段文本所表達的情感傾向是正面、負面還是中性。對于中文微博SnowNLP庫是一個簡單易用的選擇。6.1 使用SnowNLP進行批量情感分析SnowNLP基于樸素貝葉斯算法訓(xùn)練可以直接對中文文本進行情感打分0到1之間越接近1表示越正面。from snownlp import SnowNLP import numpy as np def analyze_sentiment_batch(text_list): 批量分析情感返回情感得分列表。 注意SnowNLP的情感分析基于商品評論訓(xùn)練對社交媒體文本可能有一定偏差。 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): 為未分析情感的微博計算情感得分并更新數(shù)據(jù)庫 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]) # 批量更新數(shù)據(jù)庫 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)} 條記錄。) # 打印一些統(tǒng)計信息 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 情感分析結(jié)果的局限性及優(yōu)化方向SnowNLP開箱即用但必須清醒認識其局限性領(lǐng)域適應(yīng)性其模型主要在電商評論上訓(xùn)練對微博上的網(wǎng)絡(luò)用語、反諷、縮寫等可能識別不準。例如“這操作真是絕了”在電商可能是好評在微博可能是負面吐槽。中性區(qū)域模糊得分在0.4-0.6之間的文本情感傾向非常模糊簡單二分法正面/負面會丟失大量信息。缺乏強度區(qū)分“不錯”和“太好了”都是正面但強度不同得分可能接近。優(yōu)化建議自定義訓(xùn)練SnowNLP支持用自己的標注數(shù)據(jù)重新訓(xùn)練情感分析模型。可以手動標注一批微博數(shù)據(jù)正面、負面、中性訓(xùn)練一個更貼合社交媒體領(lǐng)域的模型。結(jié)合詞典與規(guī)則針對特定領(lǐng)域如疫情可以構(gòu)建一個情感詞典包含該領(lǐng)域特有的正向詞如“加油”、“致敬”和負向詞如“恐慌”、“失望”結(jié)合規(guī)則進行加權(quán)判斷。嘗試其他工具對于更復(fù)雜的分析可以探索BosonNLP、百度NLP、騰訊NLP等商業(yè)API或者使用Transformers庫加載預(yù)訓(xùn)練的中文情感分析模型如bert-base-chinese微調(diào)后的模型這些通常效果更好但復(fù)雜度也更高。7. 項目整合、調(diào)度與展望至此我們完成了數(shù)據(jù)采集、存儲、清洗、分析和可視化的全流程。但一個完整的項目還需要一個“大腦”來調(diào)度這一切。7.1 使用Apache Airflow或簡單腳本進行任務(wù)調(diào)度對于生產(chǎn)環(huán)境通常會使用Apache Airflow這樣的工作流調(diào)度平臺以DAG有向無環(huán)圖的形式定義任務(wù)依賴關(guān)系如先爬蟲再清洗再分析最后可視化。對于個人項目或?qū)W習(xí)一個簡單的Python調(diào)度腳本足矣。我們可以使用schedule庫或操作系統(tǒng)的crontab(Linux/macOS) /任務(wù)計劃程序(Windows)。# main_scheduler.py import time import schedule from datetime import datetime import logging # 導(dǎo)入各個模塊的功能函數(shù) 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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def daily_job(): 每日執(zhí)行的任務(wù)流水線 logging.info(開始執(zhí)行每日數(shù)據(jù)管道任務(wù)...) try: # 1. 爬取數(shù)據(jù) fetch_and_save_epidemic_data() fetch_and_save_weibo_by_keywords([疫情, 疫苗, 隔離]) # 2. 清洗數(shù)據(jù) clean_and_update_epidemic_data() batch_text_preprocessing() # 3. 情感分析 update_sentiment_to_db() # 4. 生成可視化報告 generate_all_reports() logging.info(每日數(shù)據(jù)管道任務(wù)執(zhí)行完畢) except Exception as e: logging.error(f任務(wù)執(zhí)行失敗: {e}, exc_infoTrue) if __name__ __main__: # 每天凌晨2點執(zhí)行 schedule.every().day.at(02:00).do(daily_job) logging.info(調(diào)度器已啟動等待執(zhí)行時間...) while True: schedule.run_pending() time.sleep(60) # 每分鐘檢查一次7.2 項目總結(jié)與擴展思考回顧這個項目它本質(zhì)上構(gòu)建了一個微型的、閉環(huán)的數(shù)據(jù)流水線。在這個過程中我們不僅練習(xí)了Python多個庫的使用更重要的是學(xué)習(xí)了如何將分散的技術(shù)點串聯(lián)起來解決一個實際問題?？梢赃M一步擴展的方向前端展示使用Flask或Streamlit快速搭建一個Web儀表盤將可視化圖表集成到網(wǎng)頁中實現(xiàn)交互式查詢。更復(fù)雜的分析除了情感分析還可以做主題模型如LDA來發(fā)現(xiàn)微博討論的熱點話題或者將疫情數(shù)據(jù)與微博情感數(shù)據(jù)在時間序列上進行相關(guān)性分析。容器化與部署使用Docker將整個項目Python環(huán)境、代碼、SQLite數(shù)據(jù)庫容器化方便在任何地方一鍵運行。數(shù)據(jù)源擴展接入更多數(shù)據(jù)源如新聞網(wǎng)站、論壇帖子進行多源信息對比分析。告警機制當某地疫情數(shù)據(jù)突變或微博負面情感激增時自動發(fā)送郵件或短信告警。最后的實操心得數(shù)據(jù)項目中最耗時、最棘手的部分往往不是編寫核心算法而是數(shù)據(jù)采集的穩(wěn)定性和數(shù)據(jù)清洗的瑣碎性。在開始任何分析之前請務(wù)必投入足夠的時間確保你的數(shù)據(jù)管道是可靠和干凈的。這個項目提供了一個完整的框架和許多踩坑經(jīng)驗希望能幫助你更順暢地開啟自己的數(shù)據(jù)之旅。記住從能跑通的簡單版本開始逐步迭代和優(yōu)化是學(xué)習(xí)這類綜合項目的最佳路徑。本文還有配套的精品資源點擊獲取