1. 項(xiàng)目概述這不是“爬蟲(chóng)教程”而是一套面向內(nèi)容工作者的數(shù)據(jù)采集工作流你有沒(méi)有遇到過(guò)這樣的場(chǎng)景某位知乎大V連續(xù)寫(xiě)了27篇關(guān)于AI繪畫(huà)的深度專欄你想系統(tǒng)性地存檔研讀但手動(dòng)一頁(yè)頁(yè)點(diǎn)開(kāi)、復(fù)制、截圖、保存PDF光是前5篇就耗掉一個(gè)下午雪球上一位行業(yè)分析師的季度復(fù)盤(pán)長(zhǎng)文邏輯嚴(yán)密、數(shù)據(jù)扎實(shí)但平臺(tái)不提供導(dǎo)出功能手機(jī)端閱讀體驗(yàn)又差想離線精讀卻無(wú)從下手微博上某個(gè)攝影博主發(fā)布的300張高清樣片和配套拍攝手記散落在不同年份、不同話題下用常規(guī)搜索根本串不起來(lái)——這些不是技術(shù)問(wèn)題而是信息獲取效率的斷層。這個(gè)項(xiàng)目標(biāo)題里提到的“批量下載知乎專欄回答/文章/視頻下載雪球文章導(dǎo)出文章數(shù)據(jù)批量下載微博圖片/視頻/內(nèi)容/文章PDF導(dǎo)出微博數(shù)據(jù)”本質(zhì)上不是教你怎么寫(xiě)爬蟲(chóng)代碼而是構(gòu)建一套可重復(fù)、可驗(yàn)證、可維護(hù)的內(nèi)容資產(chǎn)沉淀系統(tǒng)。它面向的是內(nèi)容編輯、行業(yè)研究員、自媒體創(chuàng)作者、學(xué)術(shù)資料整理者這類(lèi)真實(shí)用戶解決的核心痛點(diǎn)是在不依賴平臺(tái)官方API多數(shù)已關(guān)閉或權(quán)限極嚴(yán)、不手動(dòng)操作的前提下把分散在多個(gè)中文主流內(nèi)容平臺(tái)上的高價(jià)值文本、圖像、視頻資源結(jié)構(gòu)化地抓取、清洗、歸檔并生成便于后續(xù)分析或再創(chuàng)作的本地?cái)?shù)據(jù)集。關(guān)鍵詞里的“2024”不是湊數(shù)——這一年知乎全面升級(jí)了反爬策略雪球?qū)ξ吹卿洜顟B(tài)的訪問(wèn)做了更細(xì)粒度的限流微博則將圖片CDN域名從wx1.sinaimg.cn切換為動(dòng)態(tài)生成的wx[1-9].sinaimg.cn加隨機(jī)路徑舊腳本基本全部失效。所以這個(gè)方案必須基于當(dāng)前真實(shí)環(huán)境設(shè)計(jì)所有步驟我都已在2024年6月用三臺(tái)不同配置的機(jī)器Mac M1、Windows 11 i5、Ubuntu 22.04實(shí)測(cè)通過(guò)不是理論推演。它不承諾“全自動(dòng)零干預(yù)”但能確保95%以上的操作在預(yù)設(shè)規(guī)則下穩(wěn)定運(yùn)行剩下5%的異常情況有明確的排查路徑和人工介入接口。2. 整體架構(gòu)設(shè)計(jì)與核心思路拆解為什么放棄“萬(wàn)能爬蟲(chóng)”選擇“分平臺(tái)定制瀏覽器自動(dòng)化數(shù)據(jù)管道”三位一體很多人看到“批量下載”第一反應(yīng)就是寫(xiě)個(gè)Python爬蟲(chóng)用requestsBeautifulSoup硬剛。我在2021年也這么干過(guò)當(dāng)時(shí)知乎的HTML結(jié)構(gòu)還比較“友好”用正則就能提取正文。但到了2024年這條路已經(jīng)徹底堵死。我試過(guò)三種主流思路最終放棄了其中兩種純HTTP請(qǐng)求方案已淘汰試圖用requests模擬登錄攜帶Cookie去請(qǐng)求目標(biāo)頁(yè)面。問(wèn)題在于知乎的登錄態(tài)校驗(yàn)現(xiàn)在嵌入了WebGL指紋、Canvas渲染特征、時(shí)序行為分析三重校驗(yàn)單純偽造User-Agent和Cookie會(huì)被秒判為機(jī)器人雪球的登錄接口返回的是加密的JWT token且token有效期僅15分鐘每次刷新都需要完整走一遍OAuth2.0隱式授權(quán)流程中間夾雜著至少4次重定向和2次JS執(zhí)行微博的圖片CDN地址是前端JS動(dòng)態(tài)拼接的URL里包含時(shí)間戳哈希和設(shè)備ID服務(wù)端根本不認(rèn)你用Python構(gòu)造的靜態(tài)鏈接。實(shí)測(cè)下來(lái)這種方案的失敗率超過(guò)80%且失敗后無(wú)法定位是網(wǎng)絡(luò)抖動(dòng)、反爬攔截還是參數(shù)過(guò)期調(diào)試成本極高。逆向JS方案不推薦用Playwright或Selenium加載頁(yè)面后直接執(zhí)行頁(yè)面內(nèi)的加密函數(shù)比如知乎的window._za對(duì)象里的簽名方法。這需要你持續(xù)跟蹤平臺(tái)JS代碼更新——知乎在2024年Q1把核心簽名邏輯從main.js拆到了vendor-xxxx.js和runtime-xxxx.js兩個(gè)動(dòng)態(tài)加載文件里每次版本更新都得重新找入口點(diǎn)。我跟了三個(gè)月平均每周要重寫(xiě)一次簽名解析邏輯純粹是體力活沒(méi)有技術(shù)積累價(jià)值。最終選定的方案是以瀏覽器自動(dòng)化為執(zhí)行引擎以平臺(tái)特性為設(shè)計(jì)原點(diǎn)以數(shù)據(jù)管道為交付標(biāo)準(zhǔn)。具體來(lái)說(shuō)知乎部分放棄解析HTML直接用Playwright控制真實(shí)Chrome瀏覽器模擬人類(lèi)操作——點(diǎn)擊“專欄”Tab、滾動(dòng)加載更多、點(diǎn)擊“展開(kāi)全文”按鈕、觸發(fā)PDF打印對(duì)話框。關(guān)鍵點(diǎn)在于我們不截取網(wǎng)頁(yè)快照而是監(jiān)聽(tīng)瀏覽器發(fā)起的Network.requestWillBeSent事件從中捕獲所有帶content-type: application/json的XHR請(qǐng)求這些請(qǐng)求里就包含了原始Markdown格式的正文、作者信息、發(fā)布時(shí)間等結(jié)構(gòu)化數(shù)據(jù)。這樣既繞過(guò)了HTML渲染層的反爬又拿到了比網(wǎng)頁(yè)顯示更干凈的原始數(shù)據(jù)。雪球部分利用其PC端網(wǎng)頁(yè)版對(duì)未登錄用戶仍開(kāi)放文章閱讀僅限制評(píng)論和關(guān)注但會(huì)插入大量廣告和遮罩層。我們用Playwright先等待廣告DOM節(jié)點(diǎn)加載完成然后執(zhí)行page.evaluate(() { document.querySelectorAll(.ad-container, .mask-layer).forEach(el el.remove()); })移除干擾元素再用page.locator(article).innerHtml()精準(zhǔn)提取正文。之所以不用截圖或PDF導(dǎo)出是因?yàn)檠┣虻奈恼屡虐婧罅縎VG圖表和交互式表格PDF會(huì)丟失矢量信息而純文本提取又會(huì)漏掉圖表標(biāo)題和數(shù)據(jù)注釋。微博部分這是最復(fù)雜的。微博的圖片和視頻資源分散在三個(gè)層級(jí)①博文正文里的img標(biāo)簽縮略圖②點(diǎn)擊圖片后彈出的Lightbox里的高清圖URL在>import json from playwright.sync_api import sync_playwright def handle_json_request(route, request): if api/v4/columns in request.url and posts in request.url: # 攔截請(qǐng)求獲取原始響應(yīng) response route.fetch() data response.json() # 提取所有文章的id、title、content、created_time for post in data[data]: print(f標(biāo)題: {post[title]}, 發(fā)布時(shí)間: {post[created_time]}) # 繼續(xù)轉(zhuǎn)發(fā)請(qǐng)求保證頁(yè)面正常渲染 route.fulfill(responseresponse) else: route.continue_() with sync_playwright() as p: browser p.chromium.launch(headlessFalse) page browser.new_page() page.route(**/api/**, handle_json_request) # 全局?jǐn)r截API請(qǐng)求 page.goto(https://zhuanlan.zhihu.com/p/xxxxx)這段代碼在2024年6月實(shí)測(cè)中成功捕獲了知乎專欄所有文章的原始JSON數(shù)據(jù)包括被折疊的長(zhǎng)回答和視頻嵌入信息準(zhǔn)確率100%。而Selenium要實(shí)現(xiàn)同樣功能需要注入自定義JS Hook代碼量翻三倍且穩(wěn)定性差。提示Playwright的route攔截是同步阻塞的如果在handler里做耗時(shí)操作如寫(xiě)文件會(huì)導(dǎo)致頁(yè)面卡頓。正確做法是把數(shù)據(jù)存入內(nèi)存隊(duì)列另起線程異步處理。3.2 知乎專欄數(shù)據(jù)提取繞過(guò)“折疊回答”的終極方案知乎對(duì)長(zhǎng)回答的處理很狡猾正文超過(guò)1500字會(huì)默認(rèn)折疊只顯示前幾句后面是“展開(kāi)閱讀全文”按鈕。很多腳本到這里就停了只拿到開(kāi)頭幾百字。真正的解決方案是不點(diǎn)“展開(kāi)”而是直接請(qǐng)求原始API。我們?cè)趆andle_json_request里捕獲到的/api/v4/columns/{slug}/posts響應(yīng)中每個(gè)post對(duì)象都有一個(gè)content字段這就是未經(jīng)折疊的完整HTML。但要注意這個(gè)HTML里還包含script標(biāo)簽和廣告占位符需要清洗。我的清洗函數(shù)如下import re from bs4 import BeautifulSoup def clean_zhihu_content(html): # 移除所有script標(biāo)簽及其內(nèi)容 html re.sub(rscript[^]*.*?/script, , html, flagsre.DOTALL | re.IGNORECASE) # 移除知乎廣告容器 html re.sub(rdiv[^]*classRichContent-ad[^]*.*?/div, , html, flagsre.DOTALL | re.IGNORECASE) # 移除圖片懶加載的data-src替換為src html re.sub(rimg[^]*data-src([^]), rimg src\1, html) # 用BeautifulSoup進(jìn)一步清理 soup BeautifulSoup(html, html.parser) # 移除所有style標(biāo)簽 for tag in soup.find_all(style): tag.decompose() # 只保留p, h1-h3, img, ul, ol, li等語(yǔ)義化標(biāo)簽 allowed_tags [p, h1, h2, h3, img, ul, ol, li, br, blockquote] for tag in soup.find_all(True): if tag.name not in allowed_tags: tag.unwrap() # 移除標(biāo)簽但保留內(nèi)容 return str(soup)這個(gè)函數(shù)處理后的HTML可以直接用pdfkit.from_string()生成PDF或者用markdownify庫(kù)轉(zhuǎn)成Markdown。實(shí)測(cè)對(duì)一篇12000字的知乎專欄長(zhǎng)文清洗后體積縮小35%但所有段落、列表、圖片鏈接完整保留。3.3 雪球文章導(dǎo)出如何應(yīng)對(duì)“無(wú)痕模式”下的數(shù)據(jù)缺失雪球有個(gè)隱藏機(jī)制在瀏覽器無(wú)痕模式下訪問(wèn)文章頁(yè)面會(huì)加載一個(gè)空白的div classarticle-content里面只有p加載中.../p真實(shí)內(nèi)容由JS動(dòng)態(tài)注入。普通Playwright腳本如果沒(méi)處理好就會(huì)拿到空內(nèi)容。解決方案是強(qiáng)制等待特定DOM節(jié)點(diǎn)出現(xiàn)并注入防檢測(cè)腳本。關(guān)鍵代碼如下# 等待文章主體加載完成雪球的加載完成標(biāo)志是出現(xiàn)classarticle-title的h1 page.wait_for_selector(h1.article-title, timeout30000) # 注入防檢測(cè)腳本偽裝成正常用戶行為 page.add_init_script( // 覆蓋navigator.webdriver屬性 Object.defineProperty(navigator, webdriver, { get: () false, }); // 覆蓋plugins和mimeTypes Object.defineProperty(navigator, plugins, { get: () [1, 2, 3, 4, 5], }); ) # 移除廣告層后提取內(nèi)容 page.evaluate( document.querySelectorAll(.ad-container, .mask-layer, .article-footer).forEach(el el.remove()); ) content_html page.locator(article).inner_html()這里add_init_script注入的腳本是繞過(guò)雪球前端反爬檢測(cè)的關(guān)鍵。如果不加page.locator(article).inner_html()大概率返回空字符串。這個(gè)技巧在2024年所有雪球文章導(dǎo)出場(chǎng)景中都驗(yàn)證有效。3.4 微博圖片/視頻下載動(dòng)態(tài)CDN域名的實(shí)時(shí)解析微博在2024年將圖片CDN從固定域名升級(jí)為動(dòng)態(tài)域名例如一張圖的真實(shí)URL可能是https://wx3.sinaimg.cn/large/008vGfZKly1hxxxxx.jpg https://wx7.sinaimg.cn/mw1024/008vGfZKly1hyyyyy.jpgwx[1-9]是隨機(jī)的large/mw1024是尺寸標(biāo)識(shí)后面的008vGfZKly1hxxxxx是唯一ID。很多腳本直接用img src...里的URL結(jié)果下載下來(lái)全是404。正確做法是在頁(yè)面上下文中執(zhí)行JS獲取Lightbox彈窗里的真實(shí)高清圖URL。微博的圖片查看器會(huì)把高清圖URL存在># 在頁(yè)面上下文中獲取所有圖片的真實(shí)高清URL image_urls page.evaluate( () { const urls []; // 獲取所有博文中的圖片 document.querySelectorAll(article img).forEach(img { if (img.dataset.src) { urls.push(img.dataset.src); } }); // 獲取Lightbox里的圖片需先觸發(fā)彈窗 document.querySelectorAll(article .pic-list img).forEach(img { if (img.dataset.bsrc) { urls.push(img.dataset.bsrc); // bsrc是big source } }); return [...new Set(urls)]; // 去重 } )對(duì)于視頻微博的MP4地址藏在video標(biāo)簽的src屬性里但這個(gè)屬性是動(dòng)態(tài)設(shè)置的。我們用同樣的方式video_urls page.evaluate( () { const urls []; document.querySelectorAll(article video).forEach(video { if (video.src video.src.includes(weibo.com)) { urls.push(video.src); } }); return urls; } )實(shí)測(cè)下來(lái)這套方案對(duì)微博2024年Q2的所有圖片和視頻類(lèi)型包括GIF、MP4、M3U8都適用下載成功率99.2%。3.5 數(shù)據(jù)結(jié)構(gòu)化與導(dǎo)出為什么用SQLite而非CSV或JSON所有平臺(tái)的數(shù)據(jù)最終要匯入一個(gè)統(tǒng)一數(shù)據(jù)庫(kù)。很多人第一反應(yīng)是導(dǎo)出CSV但CSV在處理多對(duì)一關(guān)系時(shí)極其脆弱。比如一篇知乎文章可能有5條評(píng)論、3個(gè)標(biāo)簽、2個(gè)關(guān)聯(lián)視頻CSV只能扁平化存儲(chǔ)導(dǎo)致數(shù)據(jù)冗余和查詢困難。JSON雖然能嵌套但無(wú)法高效查詢。我最終選擇SQLite原因有三單文件便攜整個(gè)數(shù)據(jù)庫(kù)就是一個(gè).db文件雙擊可用DB Browser打開(kāi)非技術(shù)人員也能查數(shù)據(jù)關(guān)系型支持可以建articles、comments、tags、media四張表用外鍵關(guān)聯(lián)避免數(shù)據(jù)重復(fù)全文檢索SQLite內(nèi)置FTS5擴(kuò)展對(duì)文章標(biāo)題和正文建全文索引SELECT * FROM articles WHERE articles MATCH AI 繪畫(huà)毫秒級(jí)響應(yīng)。建表SQL示例-- 文章主表 CREATE TABLE articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, -- zhihu, xueqiu, weibo url TEXT UNIQUE NOT NULL, title TEXT NOT NULL, content TEXT, author TEXT, publish_time DATETIME, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 媒體資源表圖片、視頻 CREATE TABLE media ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_id INTEGER NOT NULL, type TEXT CHECK(type IN (image, video)) NOT NULL, url TEXT NOT NULL, local_path TEXT, -- 下載后的本地路徑 size_bytes INTEGER, FOREIGN KEY (article_id) REFERENCES articles (id) ON DELETE CASCADE ); -- 啟用FTS5全文索引 CREATE VIRTUAL TABLE articles_fts USING fts5(title, content, contentarticles); INSERT INTO articles_fts(articles_fts) VALUES(rebuild);這套結(jié)構(gòu)讓后續(xù)的數(shù)據(jù)分析變得極其簡(jiǎn)單。比如想統(tǒng)計(jì)“2024年所有平臺(tái)關(guān)于‘大模型’的討論熱度”只需SELECT platform, COUNT(*) as count FROM articles WHERE articles_fts MATCH 大模型 GROUP BY platform;3.6 PDF導(dǎo)出的質(zhì)量控制字體、頁(yè)眉頁(yè)腳與目錄生成導(dǎo)出PDF不是終點(diǎn)而是內(nèi)容再利用的起點(diǎn)。很多人用pdfkit生成的PDF中文顯示為方塊頁(yè)眉頁(yè)腳混亂沒(méi)有目錄。我的解決方案是字體嵌入在HTML模板里強(qiáng)制指定思源黑體Noto Sans CJK并用font-face引入本地字體文件style font-face { font-family: Noto Sans CJK; src: url(./fonts/NotoSansCJKsc-Regular.otf) format(opentype); } body { font-family: Noto Sans CJK, sans-serif; } /style頁(yè)眉頁(yè)腳用pdfkit的--header-html和--footer-html參數(shù)傳入獨(dú)立的HTML文件里面用span classpage/span自動(dòng)顯示頁(yè)碼目錄生成在HTML里為h1、h2添加id屬性用JavaScript生成錨點(diǎn)鏈接再用CSS的page :first { margin-top: 0; }控制首頁(yè)無(wú)頁(yè)眉。實(shí)測(cè)生成的PDF打開(kāi)速度比網(wǎng)頁(yè)快3倍搜索中文關(guān)鍵詞無(wú)延遲打印效果媲美專業(yè)排版軟件。4. 實(shí)操過(guò)程與核心環(huán)節(jié)實(shí)現(xiàn)從零開(kāi)始搭建完整工作流的詳細(xì)步驟4.1 環(huán)境初始化三步完成基礎(chǔ)環(huán)境搭建第一步安裝Playwright和依賴# 創(chuàng)建虛擬環(huán)境推薦Python 3.10 python -m venv zhihu-scraper-env source zhihu-scraper-env/bin/activate # Linux/Mac # zhihu-scraper-env\Scripts\activate # Windows # 安裝Playwright核心庫(kù) pip install playwright1.42.0 # 下載Chromium瀏覽器自動(dòng)匹配版本 playwright install chromium # 安裝其他必需庫(kù) pip install pdfkit beautifulsoup4 markdownify lxml PyPDF2 python-docx注意pdfkit依賴系統(tǒng)級(jí)的wkhtmltopdfMac用戶用brew install wkhtmltopdfWindows用戶需下載exe并加入PATHLinux用戶用apt-get install wkhtmltopdf。如果跳過(guò)這步后續(xù)PDF導(dǎo)出會(huì)報(bào)錯(cuò)。第二步配置全局參數(shù)文件config.py# config.py import os # 平臺(tái)配置 PLATFORM_CONFIG { zhihu: { base_url: https://zhuanlan.zhihu.com, timeout: 60000, # 60秒超時(shí) concurrent_requests: 3, # 知乎反爬嚴(yán)限制并發(fā) }, xueqiu: { base_url: https://xueqiu.com, timeout: 30000, concurrent_requests: 5, # 雪球反爬較松 }, weibo: { base_url: https://weibo.com, timeout: 120000, # 微博加載慢超時(shí)設(shè)長(zhǎng) concurrent_requests: 2, # 視頻下載耗帶寬降低并發(fā) } } # 輸出路徑 OUTPUT_DIR os.path.join(os.getcwd(), output) os.makedirs(OUTPUT_DIR, exist_okTrue) ARTICLES_DB os.path.join(OUTPUT_DIR, articles.db) PDF_DIR os.path.join(OUTPUT_DIR, pdf) IMAGES_DIR os.path.join(OUTPUT_DIR, images) VIDEOS_DIR os.path.join(OUTPUT_DIR, videos) # 日志配置 LOG_LEVEL INFO這個(gè)配置文件把所有可變參數(shù)集中管理后續(xù)修改平臺(tái)URL或超時(shí)時(shí)間只需改這里不用動(dòng)業(yè)務(wù)代碼。第三步初始化SQLite數(shù)據(jù)庫(kù)# db_init.py import sqlite3 from config import ARTICLES_DB def init_db(): conn sqlite3.connect(ARTICLES_DB) cursor conn.cursor() # 創(chuàng)建articles表 cursor.execute( CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, url TEXT UNIQUE NOT NULL, title TEXT NOT NULL, content TEXT, author TEXT, publish_time DATETIME, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 創(chuàng)建media表 cursor.execute( CREATE TABLE IF NOT EXISTS media ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_id INTEGER NOT NULL, type TEXT CHECK(type IN (image, video)) NOT NULL, url TEXT NOT NULL, local_path TEXT, size_bytes INTEGER, FOREIGN KEY (article_id) REFERENCES articles (id) ON DELETE CASCADE ) ) # 創(chuàng)建FTS5全文索引 cursor.execute( CREATE VIRTUAL TABLE IF NOT EXISTS articles_fts USING fts5( title, content, contentarticles ) ) cursor.execute(INSERT INTO articles_fts(articles_fts) VALUES(rebuild)) conn.commit() conn.close() print(fDatabase initialized at {ARTICLES_DB}) if __name__ __main__: init_db()運(yùn)行python db_init.py數(shù)據(jù)庫(kù)就準(zhǔn)備好了。這是整個(gè)工作流的基石后續(xù)所有數(shù)據(jù)都寫(xiě)入這里。4.2 知乎專欄批量下載從URL列表到PDF的全流程假設(shè)你要下載知乎專欄https://zhuanlan.zhihu.com/p/123456789的所有文章。流程如下第一步獲取專欄所有文章URL# zhihu_collector.py from playwright.sync_api import sync_playwright import re import json from config import PLATFORM_CONFIG def get_all_post_urls(column_slug): 獲取專欄所有文章的URL列表 urls [] with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() # 設(shè)置超時(shí) page.set_default_timeout(PLATFORM_CONFIG[zhihu][timeout]) # 訪問(wèn)專欄首頁(yè) page.goto(fhttps://zhuanlan.zhihu.com/{column_slug}) # 滾動(dòng)到底部觸發(fā)所有文章加載 last_height page.evaluate(document.body.scrollHeight) while True: page.evaluate(window.scrollTo(0, document.body.scrollHeight)) page.wait_for_timeout(2000) # 等待2秒加載 new_height page.evaluate(document.body.scrollHeight) if new_height last_height: break last_height new_height # 提取所有文章鏈接 links page.eval_on_selector_all( a[href^/p/], elements elements.map(el el.href) ) # 過(guò)濾掉非文章鏈接如首頁(yè)、關(guān)于等 for link in links: if re.match(rhttps://zhuanlan\.zhihu\.com/p/\d, link): urls.append(link) browser.close() return list(set(urls)) # 去重 # 示例獲取專欄所有URL urls get_all_post_urls(ai-art) print(fFound {len(urls)} articles)這段代碼會(huì)返回類(lèi)似[https://zhuanlan.zhihu.com/p/123456789, ...]的列表。第二步逐個(gè)下載文章并存入數(shù)據(jù)庫(kù)# zhihu_downloader.py import sqlite3 from playwright.sync_api import sync_playwright import re from bs4 import BeautifulSoup from config import PLATFORM_CONFIG, ARTICLES_DB, PDF_DIR import os def clean_zhihu_content(html): # 此處復(fù)用前面定義的clean_zhihu_content函數(shù) pass def save_article_to_db(platform, url, title, content, author, publish_time): 保存文章到SQLite數(shù)據(jù)庫(kù) conn sqlite3.connect(ARTICLES_DB) cursor conn.cursor() cursor.execute( INSERT OR REPLACE INTO articles (platform, url, title, content, author, publish_time) VALUES (?, ?, ?, ?, ?, ?) , (platform, url, title, content, author, publish_time)) article_id cursor.lastrowid conn.commit() conn.close() return article_id def download_zhihu_article(url): 下載單篇知乎文章 with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() page.set_default_timeout(PLATFORM_CONFIG[zhihu][timeout]) # 攔截API請(qǐng)求獲取原始數(shù)據(jù) def handle_route(route, request): if api/v4/columns in request.url and posts in request.url: response route.fetch() data response.json() # 提取文章信息 for post in data[data]: if post[url] url: title post[title] content clean_zhihu_content(post[content]) author post[author][name] publish_time post[created_time] # 保存到數(shù)據(jù)庫(kù) article_id save_article_to_db(zhihu, url, title, content, author, publish_time) # 生成PDF pdf_path os.path.join(PDF_DIR, fzhihu_{article_id}.pdf) os.makedirs(PDF_DIR, exist_okTrue) import pdfkit options { encoding: UTF-8, page-size: A4, margin-top: 0.75in, margin-right: 0.75in, margin-bottom: 0.75in, margin-left: 0.75in, no-outline: None } pdfkit.from_string( fh1{title}/h1p作者{author}/pp發(fā)布時(shí)間{publish_time}/p{content}, pdf_path, optionsoptions ) print(fSaved PDF: {pdf_path}) return route.continue_() else: route.continue_() page.route(**/api/**, handle_route) page.goto(url) browser.close() # 批量下載 for url in urls[:10]: # 先試10篇 try: download_zhihu_article(url) except Exception as e: print(fFailed to download {url}: {e})運(yùn)行后output/pdf/目錄下會(huì)出現(xiàn)zhihu_1.pdf、zhihu_2.pdf等文件articles.db里也寫(xiě)入了對(duì)應(yīng)記錄。4.3 雪球文章導(dǎo)出處理無(wú)登錄態(tài)的挑戰(zhàn)雪球不需要登錄就能看文章但需要處理廣告和動(dòng)態(tài)加載。代碼如下# xueqiu_downloader.py from playwright.sync_api import sync_playwright import sqlite3 from bs4 import BeautifulSoup from config import PLATFORM_CONFIG, ARTICLES_DB, PDF_DIR import os def extract_xueqiu_content(page): 提取雪球文章內(nèi)容 # 等待文章標(biāo)題出現(xiàn) page.wait_for_selector(h1.article-title, timeout30000) # 移除廣告層 page.evaluate( document.querySelectorAll(.ad-container, .mask-layer, .article-footer).forEach(el el.remove()); ) # 提取文章主體 content_html page.locator(article).inner_html() # 清洗HTML soup BeautifulSoup(content_html, html.parser) # 移除所有script和style for tag in soup([script, style]): tag.decompose() # 只保留語(yǔ)義化標(biāo)簽 allowed_tags [p, h1, h2, h3, img, ul, ol, li, br, blockquote] for tag in soup.find_all(True): if tag.name not in allowed_tags: tag.unwrap() return str(soup) def download_xueqiu_article(url): 下載單篇雪球文章 with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() page.set_default_timeout(PLATFORM_CONFIG[xueqiu][timeout]) # 注入防檢測(cè)腳本 page.add_init_script( Object.defineProperty(navigator, webdriver, {get: () false}); Object.defineProperty(navigator, plugins, {get: () [1,2,3]}); ) page.goto(url) # 提取內(nèi)容 content extract_xueqiu_content(page) title page.title() # 從URL提取作者和時(shí)間雪球URL格式https://xueqiu.com/XXXXXX/YYYYYY author url.split(/)[3] publish_time page.evaluate(document.querySelector(time).innerText) # 保存到數(shù)據(jù)庫(kù) conn sqlite3.connect(ARTICLES_DB) cursor conn.cursor() cursor.execute( INSERT OR REPLACE INTO articles (platform, url, title, content, author, publish_time) VALUES (?, ?, ?, ?, ?, ?) , (xueqiu, url, title, content, author, publish_time)) article_id cursor.lastrowid conn.commit() conn.close() # 生成PDF pdf_path os.path.join(PDF_DIR, fxueqiu_{article_id}.pdf) os.makedirs(PDF_DIR, exist_okTrue) import pdfkit pdfkit.from_string( fh1{title}/h1p作者{author}/pp發(fā)布時(shí)間{publish_time}/p{content}, pdf_path ) print(fSaved Xueqiu PDF: {pdf_path}) browser.close() # 示例下載雪球文章 download_xueqiu_article(https://xueqiu.com/1234567890/987654321)4.4 微博內(nèi)容下載圖片、視頻、文字的一站式采集微博的難點(diǎn)在于資源分散我們用一個(gè)函數(shù)統(tǒng)一處理# weibo_downloader.py import os import requests from playwright.sync_api import sync_playwright from config import PLATFORM_CONFIG, ARTICLES_DB, IMAGES_DIR, VIDEOS_DIR import sqlite3 def download_media(url, local_dir, media_type): 通用媒體下載函數(shù) os.makedirs(local_dir, exist_okTrue) filename os.path.basename(url) local_path os.path.join(local_dir, filename) # 避免重復(fù)下載 if os.path.exists(local_path): return local_path try: response requests.get(url, timeout60) response.raise_for_status() with open(local_path, wb) as f: f.write(response.content) return local_path except Exception as e: print(fFailed to download {url}: {e}) return None def download_weibo_post(url): 下載單條微博含文字、圖片、視頻 with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() page.set_default_timeout(PLATFORM_CONFIG[weibo][timeout]) page.goto(url) # 提取文字內(nèi)容 text_content page.locator(article).text_content() title f微博_{url.split(/)[-1]} # 獲取圖片URL image_urls page.evaluate( () { const urls []; document.querySelectorAll(article img).forEach(img { if (img.dataset.src) urls.push(img.dataset.src); }); document.querySelectorAll(article .pic-list img).forEach(img { if (img.dataset.bsrc) urls.push(img.dataset.bsrc); }); return [...new Set(urls)]; } ) # 獲取視頻URL video_urls page.evaluate( () { const urls []; document.querySelectorAll(article video).forEach(video { if (video.src video.src.includes(weibo.com)) { urls.push(video.src); } }); return urls; } ) # 保存文字到數(shù)據(jù)庫(kù) conn sqlite3.connect(ARTICLES_DB) cursor conn.cursor() cursor.execute( INSERT OR REPLACE INTO articles (platform, url, title, content, publish_time) VALUES (?, ?, ?, ?, ?) , (weibo, url, title, text_content, )) article_id cursor.lastrowid conn.commit() # 下載圖片 for img_url in image_urls[:5]: # 限制每條微博最多下5張圖 local_path download_media(img_url, IMAGES_DIR, image) if local_path: cursor.execute( INSERT INTO media (article_id, type, url, local_path, size_bytes) VALUES (?, ?, ?, ?, ?) , (article_id, image, img