第一章微信公众号文章爬取技术概述1.1 微信公众号的价值与爬取难点微信公众号作为中国最大的内容创作平台之一每天产生数百万篇优质原创内容。对于数据分析师、市场研究人员和内容创业者而言批量获取公众号文章具有重要价值。然而微信平台采取了严格的反爬措施动态Token验证每次请求需携带最新TokenIP访问频率限制单个IP每秒最多3-5次请求内容加密传输文章正文通过JS动态加载账号行为监控异常操作触发滑块验证文章链接时效性72小时内有效1.2 爬虫技术选型对比方案优点缺点适用场景模拟浏览器(Selenium)无需处理JS渲染速度慢、资源消耗大小规模采集API接口模拟数据返回结构化需逆向破解加密参数大规模采集中间人代理可直接捕获数据包配置复杂专业爬虫开发微信读书通道数据获取稳定需要微信读书账号长期稳定采集本文将重点讲解基于API接口模拟的最新解决方案使用Python 3.10和异步技术实现高效采集。第二章环境搭建与依赖安装2.1 基础环境配置bash# 创建虚拟环境 python -m venv wechat_spider source wechat_spider/bin/activate # Linux/Mac wechat_spider\Scripts\activate # Windows # 安装核心依赖 pip install aiohttp3.8.4 pip install asyncio3.4.3 pip install beautifulsoup44.12.2 pip install lxml4.9.2 pip install redis4.5.5 pip install pymongo4.3.3 pip install elasticsearch8.8.2 pip install fake-useragent1.4.0 pip install tenacity8.2.2 pip install loguru0.7.22.2 数据库与中间件安装bash# 使用Docker快速部署 docker run -d -p 6379:6379 --name redis redis:7-alpine docker run -d -p 27017:27017 --name mongo mongo:6 docker run -d -p 9200:9200 -e discovery.typesingle-node --name es elasticsearch:8.8.2第三章核心爬虫代码实现3.1 公众号文章列表爬取python# wechat_spider.py import asyncio import aiohttp from typing import List, Dict from fake_useragent import UserAgent from loguru import logger from tenacity import retry, stop_after_attempt, wait_exponential import hashlib import json from datetime import datetime class WeChatSpider: 微信公众号异步爬虫 def __init__(self, max_concurrency10): self.semaphore asyncio.Semaphore(max_concurrency) self.ua UserAgent() self.session None self.token_cache {} async def __aenter__(self): self.session aiohttp.ClientSession( headers{User-Agent: self.ua.random}, timeoutaiohttp.ClientTimeout(total30) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.session.close() retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) async def get_article_list(self, biz: str, uin: str, key: str, offset: int 0) - Dict: 获取公众号文章列表 :param biz: 公众号唯一标识 :param uin: 用户uin :param key: 动态key :param offset: 分页偏移量 url https://mp.weixin.qq.com/mp/profile_ext params { action: getmsg, __biz: biz, f: json, offset: offset, count: 10, uin: uin, key: key, wxtoken: , x5: 0, f: json } # 生成请求签名 timestamp str(int(datetime.now().timestamp())) params[appmsg_token] await self._get_appmsg_token(biz) params[sign] self._generate_sign(params, timestamp) async with self.semaphore: async with self.session.get(url, paramsparams) as resp: if resp.status ! 200: logger.error(f请求失败: {resp.status}) return None data await resp.json() if data.get(ret) ! 0: logger.warning(f获取失败: {data.get(errmsg)}) return None return data def _generate_sign(self, params: Dict, timestamp: str) - str: 生成请求签名逆向微信加密算法 # 简化版签名算法实际需逆向真实逻辑 sign_str f{params[__biz]}{params[offset]}{timestamp} return hashlib.md5(sign_str.encode()).hexdigest() async def _get_appmsg_token(self, biz: str) - str: 获取动态appmsg_token需定期更新 # 实际开发中需维护token池 if biz in self.token_cache: return self.token_cache[biz] # 模拟从Redis获取token token your_dynamic_token self.token_cache[biz] token return token async def parse_article_list(self, raw_data: Dict) - List[Dict]: 解析文章列表数据 articles [] general_msg_list json.loads(raw_data.get(general_msg_list, {})) for item in general_msg_list.get(list, []): app_msg_ext_info item.get(app_msg_ext_info, {}) article { title: app_msg_ext_info.get(title), digest: app_msg_ext_info.get(digest), cover: app_msg_ext_info.get(cover), author: app_msg_ext_info.get(author), content_url: app_msg_ext_info.get(content_url, ).replace(amp;, ), source_url: app_msg_ext_info.get(source_url), publish_time: item.get(comm_msg_info, {}).get(datetime), msgid: item.get(comm_msg_info, {}).get(id), biz: raw_data.get(biz) } # 处理多图文 if multi_app_msg_item_list in app_msg_ext_info: for sub_item in app_msg_ext_info[multi_app_msg_item_list]: sub_article article.copy() sub_article.update({ title: sub_item.get(title), digest: sub_item.get(digest), content_url: sub_item.get(content_url, ).replace(amp;, ) }) articles.append(sub_article) else: articles.append(article) return articles3.2 文章详情页内容提取python# article_parser.py import re from bs4 import BeautifulSoup from typing import Dict import html class ArticleParser: 微信公众号文章解析器 staticmethod def parse_content(html_content: str) - Dict: 解析文章正文内容 :return: 包含文本、图片、视频的结构化数据 soup BeautifulSoup(html_content, lxml) # 定位文章主体微信文章通常包含在div classrich_media_content中 content_div soup.find(div, class_rich_media_content) if not content_div: return None # 提取纯文本 text_content content_div.get_text(stripTrue) # 提取所有图片 images [] for img in content_div.find_all(img): img_url img.get(data-src) or img.get(src) if img_url and mmbiz.qpic.cn in img_url: images.append({ url: img_url, alt: img.get(data-copyright) or img.get(alt, ) }) # 提取所有视频 videos [] for iframe in content_div.find_all(iframe, class_video_iframe): video_url iframe.get(data-src) if video_url: videos.append({ url: video_url, type: iframe }) # 提取标题 title_tag soup.find(h1, class_rich_media_title) or \ soup.find(h2, class_rich_media_title) title title_tag.get_text(stripTrue) if title_tag else None # 提取作者信息 author_tag soup.find(a, idjs_name) or \ soup.find(span, class_profile_nickname) author author_tag.get_text(stripTrue) if author_tag else None # 提取发布时间 time_tag soup.find(em, idpublish_time) publish_time time_tag.get_text(stripTrue) if time_tag else None return { title: title, author: author, publish_time: publish_time, text: text_content, images: images, videos: videos, word_count: len(text_content), html: str(content_div) } staticmethod def clean_content(raw_text: str) - str: 清洗文章正文 # 去除多余空白字符 text re.sub(r\s, , raw_text) # 去除特殊字符 text re.sub(r[^\u4e00-\u9fa5a-zA-Z0-9。、【】], , text) return text[:5000] # 限制长度用于ES索引3.3 分布式任务队列实现python# task_queue.py import asyncio import redis.asyncio as redis from typing import Optional, Dict import pickle from loguru import logger class RedisTaskQueue: 基于Redis的分布式任务队列 def __init__(self, redis_urlredis://localhost:6379/0): self.redis None self.redis_url redis_url self.task_key wechat:tasks self.processing_key wechat:processing async def connect(self): 建立Redis连接 self.redis await redis.from_url(self.redis_url, decode_responsesFalse) logger.info(Redis连接成功) async def push_task(self, task: Dict): 推送任务到队列 task_data pickle.dumps(task) await self.redis.lpush(self.task_key, task_data) logger.debug(f任务已推送: {task.get(biz)} offset{task.get(offset)}) async def pop_task(self, timeout5) - Optional[Dict]: 从队列获取任务阻塞式 result await self.redis.brpop(self.task_key, timeouttimeout) if result: _, task_data result task pickle.loads(task_data) # 标记为处理中 await self.redis.sadd(self.processing_key, task_data) return task return None async def complete_task(self, task: Dict): 完成任务从处理集合移除 task_data pickle.dumps(task) await self.redis.srem(self.processing_key, task_data) async def retry_failed_tasks(self, timeout300): 重新加入超时未完成的任务 now asyncio.get_event_loop().time() async for task_data in self.redis.sscan_iter(self.processing_key): task pickle.loads(task_data) if now - task.get(start_time, 0) timeout: await self.redis.srem(self.processing_key, task_data) await self.redis.lpush(self.task_key, task_data) logger.warning(f任务超时重试: {task.get(biz)})第四章数据存储与检索系统4.1 MongoDB原始数据存储python# mongo_storage.py from motor.motor_asyncio import AsyncIOMotorClient from typing import Dict from loguru import logger class MongoStorage: 异步MongoDB存储 def __init__(self, urimongodb://localhost:27017): self.client AsyncIOMotorClient(uri) self.db self.client[wechat_articles] self.articles self.db[articles] self.accounts self.db[accounts] # 创建索引 asyncio.create_task(self._init_indexes()) async def _init_indexes(self): 初始化数据库索引 await self.articles.create_index(msgid, uniqueTrue) await self.articles.create_index(publish_time) await self.articles.create_index(biz) await self.accounts.create_index(biz, uniqueTrue) logger.info(MongoDB索引创建完成) async def save_article(self, article: Dict): 保存文章存在则更新 filter {msgid: article[msgid]} update {$set: article} result await self.articles.update_one(filter, update, upsertTrue) if result.upserted_id: logger.info(f新增文章: {article[title]}) else: logger.debug(f更新文章: {article[title]}) async def save_account(self, account: Dict): 保存公众号信息 filter {biz: account[biz]} update {$set: account} await self.accounts.update_one(filter, update, upsertTrue) async def get_articles_by_date(self, start_date, end_date): 按时间范围查询文章 cursor self.articles.find({ publish_time: {$gte: start_date, $lte: end_date} }).sort(publish_time, -1) return await cursor.to_list(length1000)4.2 Elasticsearch全文检索实现python# es_search.py from elasticsearch import AsyncElasticsearch from typing import List, Dict from loguru import logger class ElasticSearchEngine: 基于ES的全文检索引擎 def __init__(self, hosts[http://localhost:9200]): self.es AsyncElasticsearch(hostshosts) self.index_name wechat_articles async def init_index(self): 创建索引和映射 mapping { mappings: { properties: { title: { type: text, analyzer: ik_max_word, search_analyzer: ik_smart }, content: { type: text, analyzer: ik_max_word, search_analyzer: ik_smart }, author: {type: keyword}, biz: {type: keyword}, publish_time: {type: date}, word_count: {type: integer}, msgid: {type: keyword}, cover: {type: keyword}, images: {type: nested}, summary: {type: text, analyzer: ik_smart} } }, settings: { number_of_shards: 3, number_of_replicas: 1, analysis: { analyzer: { ik_analyzer: { type: custom, tokenizer: ik_max_word } } } } } if not await self.es.indices.exists(indexself.index_name): await self.es.indices.create(indexself.index_name, bodymapping) logger.info(ES索引创建成功) async def index_article(self, article: Dict): 索引单篇文章 # 生成文档ID doc_id article[msgid] # 构建ES文档 doc { title: article[title], content: article.get(text, )[:10000], # 限制长度 author: article.get(author), biz: article.get(biz), publish_time: article.get(publish_time), word_count: article.get(word_count, 0), msgid: article[msgid], cover: article.get(cover), summary: article.get(digest, article[title]), images: [img[url] for img in article.get(images, [])] } await self.es.index(indexself.index_name, iddoc_id, bodydoc) logger.debug(f文档索引成功: {doc_id}) async def search(self, query: str, page: int 1, size: int 10) - Dict: 全文检索 :param query: 搜索关键词 :param page: 页码 :param size: 每页数量 body { query: { bool: { should: [ {match: {title: {query: query, boost: 3}}}, {match: {content: {query: query, boost: 1}}}, {match: {summary: {query: query, boost: 2}}} ] } }, highlight: { fields: { title: {number_of_fragments: 0}, content: {fragment_size: 150, number_of_fragments: 3} }, pre_tags: [em classhighlight], post_tags: [/em] }, sort: [{publish_time: {order: desc}}], from: (page - 1) * size, size: size } response await self.es.search(indexself.index_name, bodybody) # 格式化结果 results [] for hit in response[hits][hits]: result hit[_source] result[id] hit[_id] result[score] hit[_score] result[highlight] hit.get(highlight, {}) results.append(result) return { total: response[hits][total][value], page: page, size: size, results: results } async def close(self): 关闭连接 await self.es.close()4.3 数据管道整合python# pipeline.py import asyncio from typing import Dict from loguru import logger class DataPipeline: 数据管道处理清洗、存储、索引 def __init__(self, mongo_storage, es_engine): self.mongo mongo_storage self.es es_engine self.buffer [] self.buffer_size 50 async def process_article(self, article: Dict): 处理单篇文章 try: # 1. 数据清洗 cleaned_article self._clean_article(article) # 2. 存储到MongoDB await self.mongo.save_article(cleaned_article) # 3. 添加到ES缓冲区 self.buffer.append(cleaned_article) # 4. 缓冲区满则批量索引 if len(self.buffer) self.buffer_size: await self._flush_buffer() except Exception as e: logger.error(f文章处理失败: {article.get(msgid)} - {str(e)}) def _clean_article(self, article: Dict) - Dict: 数据清洗 # 去除HTML标签 if text in article: article[text] re.sub(r[^], , article[text]) # 标准化时间格式 if publish_time in article: # 转换为时间戳 article[publish_time] int(article[publish_time]) # 去除空值 return {k: v for k, v in article.items() if v is not None} async def _flush_buffer(self): 批量写入ES if not self.buffer: return tasks [self.es.index_article(article) for article in self.buffer] await asyncio.gather(*tasks) logger.info(f批量索引完成: {len(self.buffer)}篇) self.buffer.clear() async def close(self): 关闭管道 await self._flush_buffer() await self.es.close()第五章高级功能实现5.1 代理IP池管理python# proxy_pool.py import aiohttp import asyncio from typing import Optional, List from loguru import logger import random class ProxyPool: 动态代理IP池 def __init__(self, min_proxies10, max_proxies50): self.proxies [] self.min_proxies min_proxies self.max_proxies max_proxies self.lock asyncio.Lock() self.failed_count {} async def fetch_proxies_from_api(self): 从代理API获取新IP api_url http://api.proxyprovider.com/get?count10 async with aiohttp.ClientSession() as session: try: async with session.get(api_url, timeout10) as resp: if resp.status 200: data await resp.json() new_proxies [fhttp://{p[ip]}:{p[port]} for p in data] async with self.lock: self.proxies.extend(new_proxies) # 限制最大数量 if len(self.proxies) self.max_proxies: self.proxies self.proxies[-self.max_proxies:] logger.info(f获取到{len(new_proxies)}个新代理) except Exception as e: logger.error(f获取代理失败: {e}) async def get_proxy(self) - Optional[str]: 获取一个可用代理 async with self.lock: if len(self.proxies) self.min_proxies: asyncio.create_task(self.fetch_proxies_from_api()) if not self.proxies: return None # 随机选择并考虑失败次数 proxy random.choice(self.proxies) return proxy async def report_failure(self, proxy: str): 报告代理失败 async with self.lock: self.failed_count[proxy] self.failed_count.get(proxy, 0) 1 if self.failed_count[proxy] 3: self.proxies.remove(proxy) del self.failed_count[proxy] logger.warning(f移除失效代理: {proxy})5.2 验证码识别与处理python# captcha_solver.py import asyncio from typing import Optional from loguru import logger import base64 import aiohttp class CaptchaSolver: 验证码识别服务 def __init__(self, api_key): self.api_key api_key self.solving_tasks {} async def solve_slider(self, image_url: str, bg_url: str) - Optional[int]: 解决滑块验证码 :return: 滑动距离 # 方案1使用第三方打码平台 try: async with aiohttp.ClientSession() as session: # 下载图片 img_task asyncio.gather( self._download_image(image_url), self._download_image(bg_url) ) img_data, bg_data await img_task # 调用第三方API payload { key: self.api_key, method: slide_captcha, image: base64.b64encode(img_data).decode(), background: base64.b64encode(bg_data).decode() } async with session.post(http://api.captcha.solver, jsonpayload) as resp: result await resp.json() if result[success]: return result[distance] except Exception as e: logger.error(f验证码识别失败: {e}) # 方案2本地OpenCV识别备选 return await self._local_solve(image_url, bg_url) async def _local_solve(self, img_url, bg_url): 本地OpenCV识别 # 实现图像匹配算法 import cv2 import numpy as np # 下载图片... # 边缘检测 # 模板匹配 # 返回距离 return 120 # 示例返回值 async def _download_image(self, url: str) - bytes: 下载图片 async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.read()5.3 定时调度与监控python# scheduler.py import asyncio from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from datetime import datetime from loguru import logger import psutil import os class SpiderScheduler: 爬虫调度器 def __init__(self, spider, task_queue, pipeline): self.scheduler AsyncIOScheduler() self.spider spider self.queue task_queue self.pipeline pipeline self.stats { start_time: datetime.now(), articles_crawled: 0, failed_tasks: 0 } def start(self): 启动调度器 # 添加定时任务 self.scheduler.add_job( self.collect_tasks, triggerIntervalTrigger(minutes30), idcollect_tasks ) self.scheduler.add_job( self.monitor_system, triggerIntervalTrigger(minutes5), idmonitor ) self.scheduler.add_job( self.rotate_proxies, triggerIntervalTrigger(hours1), idrotate_proxies ) self.scheduler.start() logger.info(调度器已启动) async def collect_tasks(self): 收集新的采集任务 # 从数据库获取待采集的公众号列表 accounts await self.pipeline.mongo.accounts.find().to_list(100) for account in accounts: task { biz: account[biz], uin: account[uin], key: account[key], offset: 0, max_offset: account.get(max_offset, 100) } await self.queue.push_task(task) logger.info(f已生成{len(accounts)}个采集任务) async def monitor_system(self): 系统监控 cpu_percent psutil.cpu_percent(interval1) memory psutil.virtual_memory() logger.info(f系统状态 - CPU: {cpu_percent}%, 内存: {memory.percent}%) # 检查队列长度 queue_len await self.queue.redis.llen(self.queue.task_key) if queue_len 1000: logger.warning(f任务队列过长: {queue_len}) # 记录统计信息 runtime (datetime.now() - self.stats[start_time]).seconds speed self.stats[articles_crawled] / (runtime / 3600) if runtime 0 else 0 logger.info(f爬取统计 - 总数: {self.stats[articles_crawled]}, f失败: {self.stats[failed_tasks]}, 速度: {speed:.2f}篇/小时) async def rotate_proxies(self): 轮换代理 # 清理失效代理 # 获取新代理 await self.spider.proxy_pool.fetch_proxies_from_api()第六章Web API服务构建6.1 FastAPI搜索接口python# api_server.py from fastapi import FastAPI, Query, HTTPException from fastapi.responses import HTMLResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional, List import uvicorn app FastAPI(title微信公众号搜索引擎, version1.0.0) # 跨域支持 app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) # 依赖注入 es_engine ElasticSearchEngine() class SearchResponse(BaseModel): total: int page: int size: int results: List[dict] app.get(/, response_classHTMLResponse) async def home(): 首页 return !DOCTYPE html html head title微信公众号搜索引擎/title style body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; } .search-box { width: 100%; padding: 10px; font-size: 16px; } .result-item { border-bottom: 1px solid #eee; padding: 15px 0; } .highlight { background-color: #ffeb3b; } /style /head body h1 微信公众号搜索引擎/h1 input typetext classsearch-box idquery placeholder输入搜索关键词... button onclicksearch()搜索/button div idresults/div script async function search() { const q document.getElementById(query).value; const res await fetch(/search?q${encodeURIComponent(q)}); const data await res.json(); let html p找到 ${data.total} 条结果/p; data.results.forEach(article { html div classresult-item h3${article.title}/h3 p${article.summary || article.content.substring(0,200)}.../p small${article.author} | ${new Date(article.publish_time*1000).toLocaleString()}/small /div ; }); document.getElementById(results).innerHTML html; } /script /body /html app.get(/search, response_modelSearchResponse) async def search( q: str Query(..., min_length1, description搜索关键词), page: int Query(1, ge1, description页码), size: int Query(10, ge1, le100, description每页数量) ): 全文搜索接口 try: result await es_engine.search(q, page, size) return result except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/article/{msgid}) async def get_article(msgid: str): 获取文章详情 # 从MongoDB获取原始内容 pass app.on_event(shutdown) async def shutdown(): 关闭时清理资源 await es_engine.close() if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)第七章部署与运维7.1 Docker容器化部署dockerfile# Dockerfile FROM python:3.10-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ g \ libffi-dev \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制项目代码 COPY . . # 设置环境变量 ENV PYTHONPATH/app ENV REDIS_URLredis://redis:6379/0 ENV MONGO_URLmongodb://mongo:27017 ENV ES_HOSTShttp://es:9200 # 启动命令 CMD [python, main.py]7.2 Docker Compose编排yaml# docker-compose.yml version: 3.8 services: redis: image: redis:7-alpine ports: - 6379:6379 volumes: - redis_data:/data restart: always mongo: image: mongo:6 ports: - 27017:27017 volumes: - mongo_data:/data/db restart: always es: image: elasticsearch:8.8.2 ports: - 9200:9200 environment: - discovery.typesingle-node - ES_JAVA_OPTS-Xms512m -Xmx512m volumes: - es_data:/usr/share/elasticsearch/data restart: always spider: build: . depends_on: - redis - mongo - es environment: - REDIS_URLredis://redis:6379/0 - MONGO_URLmongodb://mongo:27017 - ES_HOSTShttp://es:9200 volumes: - ./logs:/app/logs restart: always api: build: . command: uvicorn api_server:app --host 0.0.0.0 --port 8000 ports: - 8000:8000 depends_on: - es - spider restart: always volumes: redis_data: mongo_data: es_data:7.3 Kubernetes部署配置yaml# k8s-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: wechat-spider spec: replicas: 3 selector: matchLabels: app: spider template: metadata: labels: app: spider spec: containers: - name: spider image: wechat-spider:latest env: - name: REDIS_URL value: redis://redis-service:6379/0 - name: MONGO_URL value: mongodb://mongo-service:27017 resources: requests: memory: 512Mi cpu: 250m limits: memory: 1Gi cpu: 500m livenessProbe: exec: command: - python - -c - import redis; redis.Redis.from_url($REDIS_URL).ping() initialDelaySeconds: 30 periodSeconds: 10 --- apiVersion: v1 kind: Service metadata: name: spider-service spec: selector: app: spider ports: - port: 8000 targetPort: 8000