Web3信息聚合器开发实战:从爬虫到推送的全栈架构设计
1. 项目概述:一个Web3信息聚合器的诞生
最近在GitHub上看到一个挺有意思的项目,叫“arfandi453/web3-daily-digest”。光看名字,你大概就能猜到它的核心功能:一个关于Web3的每日摘要。这听起来像是一个个人兴趣项目,但仔细琢磨,它背后反映了一个非常普遍且强烈的需求——在信息爆炸的Web3世界里,如何高效地获取高质量、结构化的信息。
Web3领域,无论是区块链技术、DeFi、NFT、DAO还是Layer2扩容方案,每天都有海量的新闻、项目更新、技术提案和社区讨论产生。对于开发者、投资者、研究员,甚至是刚入门的好奇者来说,从Twitter、Discord、Telegram、各种博客和新闻网站中手动筛选信息,无异于大海捞针,耗时耗力且容易错过关键内容。这个“web3-daily-digest”项目,本质上就是要解决这个痛点:通过自动化的方式,聚合、筛选、整理并呈现每日最重要的Web3动态。
我自己作为这个领域的持续关注者,深有体会。曾经也尝试过用RSS阅读器订阅几十个源,但信息噪音太大;依赖几个KOL的推文,视角又难免单一。一个理想的“每日摘要”,应该像一个经验丰富的助手,帮你完成了信息的初筛、分类和摘要,让你在早餐的十分钟里就能把握住行业的脉搏。这个项目正是朝着这个方向的一次实践。接下来,我将从项目设计、技术实现、到实际部署和优化,完整拆解如何构建这样一个工具,并分享其中踩过的坑和积累的经验。
2. 核心架构设计与技术选型
构建一个Web3每日摘要系统,远不止是写个爬虫那么简单。它需要一套稳定、可扩展、且能处理非结构化数据的架构。整个系统可以清晰地划分为四个核心层:数据采集层、数据处理层、内容生成层和分发交付层。
2.1 数据源的定义与优先级
数据源的质量直接决定了摘要的价值。我们需要覆盖多种类型的信息源:
- 官方与技术动态源 :这是核心中的核心。包括各大区块链项目的官方博客(如以太坊基金会博客、Solana博客)、核心协议仓库的GitHub动态(特别是
Issues和Pull Requests)、以及像EIP(以太坊改进提案)、BIP(比特币改进提案)这样的技术标准讨论区。这些信息权威性高,但更新频率不固定,需要精准监控。 - 行业新闻与媒体源 :如CoinDesk、The Block、Decrypt等专业媒体,以及一些深度研究机构的报告。它们提供经过初步编辑的行业视角,但需要警惕其可能存在的倾向性或滞后性。
- 社交媒体与社区源 :Twitter/X上重要开发者、研究员、项目方的动态;Reddit上如
r/ethereum、r/CryptoCurrency的热门讨论;Discord/Telegram核心公告频道。这里信息最实时、最原生,但噪音也最大,对筛选算法要求极高。 - 链上数据与指标源 :通过像Dune Analytics、Nansen、Etherscan的API获取的链上数据变化,如巨鲸转账、合约创建激增、Gas费波动等。这些是“用脚投票”的真实信号,能提供新闻之外的洞察。
实操心得:源不在多,在于精。 初期千万不要贪多,先从10-15个最高质量的核心源开始。例如,对于以太坊生态,优先监控
ethereum/ethereum-org-website、ethereum/EIPs仓库,Vitalik Buterin、tayvano等关键人物的推文,以及以太坊基金会博客。确保每个源的信息抓取稳定、解析准确,远比堆砌一百个源但解析得一塌糊涂要有价值。
2.2 技术栈的权衡与决策
基于以上需求,一个典型的技术选型方案如下:
- 后端语言与框架 : Python 是首选。它在数据抓取(Requests, Scrapy)、自然语言处理(NLTK, spaCy)、异步任务(asyncio, Celery)和快速原型方面有巨大生态优势。Web框架可以选择轻量级的 FastAPI ,它异步特性好,自动生成API文档,非常适合构建这类数据服务。
- 数据存储 :需要区分结构化数据和非结构化数据。
- 结构化数据 (如抓取任务状态、源配置、用户订阅关系)使用 PostgreSQL 。它的JSONB类型也能很好地存储一些半结构化数据。
- 非结构化数据 (抓取到的原始HTML、文章全文、处理后的摘要文本)使用 MongoDB 或 Elasticsearch 。Elasticsearch的优势在于其强大的全文检索能力,方便后续做相似内容去重或高级搜索,但维护复杂度稍高。初期从简,可用MongoDB。
- 任务队列与调度 :抓取任务必须是定时且并发的。 Celery + Redis (作为Broker)是经典组合。Celery可以配置定时任务(
celery beat),并发地从多个数据源抓取,并将任务结果回传。Redis同时还可以用作热点数据的缓存。 - 消息推送 :这是交付的关键。国内主流选择是 Server酱 、 PushPlus 等提供微信模板消息推送的服务。如果面向国际用户, Telegram Bot 是极佳选择,API稳定,生态成熟。邮件(通过SMTP或SendGrid等API)作为备选或补充渠道。
- 部署与运维 :使用 Docker 进行容器化封装,确保环境一致。用 Docker Compose 编排PostgreSQL、Redis、MongoDB、Celery Worker和FastAPI服务。生产环境部署到 云服务器 (如阿里云、腾讯云ECS)或 VPS 上,通过 Nginx 做反向代理和SSL管理。
这个技术栈平衡了开发效率、系统性能和可维护性,是经过实践验证的可靠组合。
3. 核心模块实现详解
有了架构设计,我们来深入每个核心模块的实现细节。
3.1 智能爬虫与反爬策略
直接使用 requests 进行简单的GET请求,在当今的Web环境下很快就会碰壁。我们必须构建一个“友好”且“健壮”的爬虫。
import asyncio
import aiohttp
from fake_useragent import UserAgent
import random
import time
class RobustCrawler:
def __init__(self):
self.ua = UserAgent()
# 使用aiohttp的ClientSession管理连接池,提升效率
self.session = None
# 定义一组合理的浏览器头
self.headers_template = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'DNT': '1',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
}
async def fetch(self, url, max_retries=3):
if not self.session:
timeout = aiohttp.ClientTimeout(total=30)
connector = aiohttp.TCPConnector(limit=10, ssl=False) # 适当调整连接数
self.session = aiohttp.ClientSession(timeout=timeout, connector=connector)
for attempt in range(max_retries):
try:
headers = self.headers_template.copy()
headers['User-Agent'] = self.ua.random
# 随机延迟,避免请求过于频繁
await asyncio.sleep(random.uniform(1, 3))
async with self.session.get(url, headers=headers) as response:
if response.status == 200:
html = await response.text()
return html
elif response.status == 429: # Too Many Requests
retry_after = int(response.headers.get('Retry-After', 60))
print(f"Rate limited by {url}, retrying after {retry_after} seconds.")
await asyncio.sleep(retry_after)
continue
else:
print(f"Failed to fetch {url}, status: {response.status}")
return None
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
print(f"Attempt {attempt + 1} failed for {url}: {e}")
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt) # 指数退避
else:
return None
return None
关键点解析:
- 异步与并发 :使用
aiohttp进行异步HTTP请求,可以同时抓取数十个源,极大提升效率。 - 请求头伪装 :使用
fake-useragent随机生成真实的浏览器UA,并设置一套完整的请求头,模拟浏览器行为。 - 速率限制与退避 :正确处理HTTP 429状态码,遵循
Retry-After头。对于其他错误,采用指数退避策略重试。 - 连接管理 :复用
ClientSession和连接池,减少TCP握手开销。
对于JavaScript渲染的页面(如很多现代博客),简单的HTML抓取无效。这时需要引入 Playwright 或 Selenium 进行无头浏览器渲染。但要注意,这类操作资源消耗大,速度慢,应仅作为最后手段,并严格控制并发实例数。
# 示例:使用Playwright抓取动态内容
from playwright.async_api import async_playwright
async def fetch_with_playwright(url):
async with async_playwright() as p:
# 使用Chromium,可配置为无头模式
browser = await p.chromium.launch(headless=True)
context = await browser.new_context(
viewport={'width': 1920, 'height': 1080},
user_agent='Mozilla/5.0...'
)
page = await context.new_page()
try:
await page.goto(url, wait_until='networkidle') # 等待网络空闲
# 等待特定内容加载
await page.wait_for_selector('article.content', timeout=10000)
content = await page.content()
return content
finally:
await browser.close()
3.2 内容解析与信息抽取
抓取到HTML后,如何精准地提取标题、正文、发布时间和作者?规则千变万化,但策略有层次。
- 首选:结构化数据(Schema.org / Open Graph) 。许多新闻和博客网站会嵌入
application/ld+json格式的JSON-LD数据,或使用og:title、og:description等Open Graph标签。这是最准确、最规范的数据源。优先解析这些元数据。 - 次选:基于CSS选择器的启发式规则 。为每个重要的数据源配置独立的解析规则。例如,对于某知名博客,规则可能是
title_selector: 'h1.post-title',content_selector: 'div.post-content',date_selector: 'time.published'。可以建立一个“源配置表”来管理这些规则。 - 保底:通用的正文提取算法 。当以上方法都失效时,使用像
readability、newspaper3k或trafilatura这样的库。它们通过分析DOM树节点密度、链接文本比等启发式方法,猜测正文所在区域,效果通常不错,但并非100%准确。
import json
from bs4 import BeautifulSoup
import trafilatura
def extract_content(html, url, source_config):
"""
根据配置优先级提取内容
:param html: 原始HTML
:param url: 文章URL
:param source_config: 该数据源的配置字典,包含解析规则
:return: 包含标题、正文、发布时间等的字典
"""
soup = BeautifulSoup(html, 'html.parser')
result = {'url': url}
# 1. 尝试提取JSON-LD
json_ld = soup.find('script', type='application/ld+json')
if json_ld:
try:
data = json.loads(json_ld.string)
# 处理可能为列表或字典的情况,提取 headline/articleBody/datePublished
# ... 解析逻辑 ...
if extracted_data:
return {**result, **extracted_data}
except json.JSONDecodeError:
pass
# 2. 尝试使用源特定规则
if source_config.get('selectors'):
title_elem = soup.select_one(source_config['selectors'].get('title'))
content_elem = soup.select_one(source_config['selectors'].get('content'))
date_elem = soup.select_one(source_config['selectors'].get('date'))
# ... 提取和清理逻辑 ...
if title_elem and content_elem:
return {**result, 'title': title_elem.get_text(strip=True), 'content': content_elem.get_text(strip=True), 'date': parsed_date}
# 3. 使用通用提取库
downloaded = trafilatura.extract(html, output_format='json', include_links=False, url=url)
if downloaded and downloaded.get('text'):
return {
**result,
'title': downloaded.get('title'),
'content': downloaded.get('text'),
'date': downloaded.get('date') # trafilatura会尝试提取日期
}
return result # 返回基础结果或None
注意事项: 发布时间(
date)的解析是个大坑。格式五花八门(RFC 3339, ISO 8601, 各种自定义格式),且时区问题频发。务必使用dateutil.parser或python-dateutil库进行灵活解析,并将所有时间统一转换为UTC时间戳存储,在展示时再根据用户时区转换。
3.3 自然语言处理与摘要生成
这是提升摘要价值的关键一步。我们不是简单罗列标题,而是要对正文进行智能摘要。
- 文本预处理 :去除无关的广告文本、导航栏内容、版权声明等噪音。可以使用正则表达式或基于关键词的过滤规则。
- 关键信息抽取 :
- 命名实体识别(NER) :使用spaCy或斯坦福NLP库识别文本中的人名、组织名、项目名、技术术语(如“ZK-Rollup”、“ERC-20”)。这有助于自动打标签和分类。
- 关键词提取 :使用
TF-IDF或TextRank算法从文章中提取核心关键词。
- 自动文摘 :
- 抽取式摘要 :这是最稳定、最常用的方法。从原文中抽取几个最重要的句子组成摘要。可以使用
gensim库的summarize功能,或者基于句子嵌入相似度(如使用Sentence-BERT)的方法。它的优点是忠实于原文,不会产生事实性错误。 - 生成式摘要 :使用像
T5、BART或PEGASUS这样的预训练模型进行文本生成。效果可能更流畅、更像人工,但需要GPU资源,且有产生“幻觉”(生成原文没有的内容)的风险。对于追求稳定性的每日摘要,初期不建议使用。
- 抽取式摘要 :这是最稳定、最常用的方法。从原文中抽取几个最重要的句子组成摘要。可以使用
一个简单的抽取式摘要实现:
from gensim.summarization import summarize
import jieba # 如果是中文,需要分词
# 对于中文,需先进行分词,gensim对中文支持需预处理
def generate_summary(text, word_count=150):
"""
生成抽取式摘要
:param text: 清洗后的正文
:param word_count: 目标摘要字数
:return: 摘要字符串
"""
if not text or len(text.split()) < 100: # 文本太短则不摘要
return text[:200] + '...' if len(text) > 200 else text
try:
# ratio参数控制摘要长度占原文的比例,或使用word_count
summary = summarize(text, word_count=word_count)
# 如果gensim未能生成摘要(例如文本结构问题),则返回前N个字符
if summary:
return summary
else:
return text[:word_count] + '...'
except Exception as e:
print(f"Summarization failed: {e}")
# 降级方案:返回开头部分
return text[:word_count] + '...'
3.4 去重、分类与排序逻辑
每天会有大量内容谈及相似主题,必须去重。同时,需要对文章进行分类(如“Layer2”、“DeFi”、“NFT”、“监管动态”)和排序(决定推送的优先级)。
- 去重 :计算文章内容的 SimHash 或 MinHash 指纹。当两篇文章的指纹距离小于某个阈值时,视为重复。比单纯比较标题或URL更可靠。可以将新文章的指纹与过去几天数据库中的指纹进行比对。
- 分类 :可以采用多标签分类。
- 规则匹配 :建立分类关键词词典。例如,包含“Optimism”、“Arbitrum”、“StarkNet”、“zkSync”的文章打上“Layer2”标签。
- 机器学习分类 :如果数据量足够,可以训练一个简单的文本分类模型(如基于
scikit-learn的TF-IDF + SVM,或使用预训练的BERT进行微调)。规则匹配快速直接,适合初期;机器学习更智能,但需要标注数据。
- 排序(热度/重要性评估) :这是一个综合评分系统,可以考虑以下因素:
- 来源权重 :官方公告权重 > 核心开发者推文 > 一般媒体。
- 社交信号 :抓取文章对应推文/帖子的转发、点赞、评论数(需额外调用社交平台API)。
- 新鲜度 :越新的内容分数越高,但具有里程碑意义的旧公告可能长期置顶。
- 用户交互 :如果系统有用户,可以根据历史点击率进行加权。
最终,根据去重后的文章列表,计算一个综合得分,按分数从高到低排序,选取Top N条作为当日摘要内容。
4. 系统集成、部署与运维
4.1 任务调度与流水线构建
使用 Celery 构建一个清晰的数据处理流水线。每个数据源对应一个定时抓取任务。
# tasks.py
from celery import Celery
from your_crawler import RobustCrawler
from your_parser import extract_content
from your_summarizer import generate_summary
from your_deduplicator import is_duplicate
from your_classifier import classify_article
import asyncio
app = Celery('digest_tasks', broker='redis://localhost:6379/0')
@app.task
def fetch_and_process_source(source_id, source_url, source_config):
"""针对单个数据源的完整处理任务"""
# 1. 抓取
crawler = RobustCrawler()
# 注意:在Celery任务中运行异步函数需要特殊处理,例如使用asyncio.run或在异步worker中
html = asyncio.run(crawler.fetch(source_url))
if not html:
return f"Failed to fetch {source_url}"
# 2. 解析
article_data = extract_content(html, source_url, source_config)
if not article_data.get('title') or not article_data.get('content'):
return f"No valid content extracted from {source_url}"
# 3. 去重检查
if is_duplicate(article_data['content']):
return f"Duplicate article: {article_data['title']}"
# 4. 生成摘要和分类
article_data['summary'] = generate_summary(article_data['content'])
article_data['categories'] = classify_article(article_data['title'] + ' ' + article_data['content'])
# 5. 存储到数据库 (伪代码)
save_to_database(article_data)
return f"Processed: {article_data['title']}"
# 在Celery Beat配置中设置定时任务
# beat_schedule = {
# 'fetch-every-hour': {
# 'task': 'tasks.fetch_all_sources',
# 'schedule': crontab(minute=0, hour='*/2'), # 每2小时执行一次
# },
# }
然后,需要一个主任务( fetch_all_sources )来遍历所有活跃的数据源配置,并发起多个 fetch_and_process_source 子任务。最后,在每天固定时间(如UTC时间23:00)触发一个 generate_daily_digest 任务,执行当日的最终排序、精选,并调用推送接口。
4.2 推送渠道集成
以集成 Telegram Bot 和 邮件 为例。
Telegram Bot推送:
- 通过
@BotFather创建一个新的Bot,获取API Token。 - 使用
python-telegram-bot库,将生成的摘要内容格式化为美观的消息(支持Markdown或HTML格式)。 - 将消息发送到指定的频道(Channel)或群组(Group)。
import telegram
from telegram.constants import ParseMode
async def send_telegram_digest(bot_token, channel_id, digest_content):
bot = telegram.Bot(token=bot_token)
# digest_content是一个包含标题、链接、摘要的列表
message = "**Web3 Daily Digest - {date}**\n\n".format(date=datetime.now().strftime('%Y-%m-%d'))
for idx, item in enumerate(digest_content, 1):
message += f"{idx}. **{item['title']}**\n"
message += f" {item['summary'][:100]}...\n"
message += f" [阅读原文]({item['url']})\n\n"
try:
await bot.send_message(chat_id=channel_id, text=message, parse_mode=ParseMode.MARKDOWN_V2, disable_web_page_preview=False)
print("Telegram digest sent successfully.")
except telegram.error.TelegramError as e:
print(f"Failed to send Telegram message: {e}")
邮件推送: 使用 smtplib 或更友好的 yagmail 库,将摘要内容格式化为HTML邮件,发送给订阅用户列表。
4.3 部署与监控
使用 Docker Compose 定义所有服务。
# docker-compose.yml
version: '3.8'
services:
postgres:
image: postgres:15
environment:
POSTGRES_DB: web3digest
POSTGRES_USER: user
POSTGRES_PASSWORD: strongpassword
volumes:
- pg_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
mongodb:
image: mongo:6
volumes:
- mongo_data:/data/db
backend:
build: ./backend
depends_on:
- postgres
- redis
- mongodb
environment:
- DATABASE_URL=postgresql://user:strongpassword@postgres/web3digest
- REDIS_URL=redis://redis:6379/0
- MONGO_URL=mongodb://mongodb:27017
# 启动FastAPI服务
command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
celery_worker:
build: ./backend
depends_on:
- redis
- postgres
- mongodb
command: celery -A tasks worker --loglevel=info --concurrency=4
# concurrency根据服务器CPU核心数调整
celery_beat:
build: ./backend
depends_on:
- redis
- celery_worker # Beat需要Worker存在
command: celery -A tasks beat --loglevel=info
部署到服务器后,关键的监控点包括:
- 任务执行状态 :监控Celery Worker是否存活,任务是否成功执行。可以通过Celery Flower面板或日志监控。
- 数据抓取成功率 :记录每个源每次抓取的成功/失败状态,定期检查失败率过高的源。
- 内容产出量 :监控每日入库的文章数量,异常减少可能意味着爬虫失效。
- 推送成功率 :监控Telegram消息或邮件是否成功发送。
- 资源使用 :监控CPU、内存、磁盘和网络流量,确保服务稳定。
5. 常见问题、优化与扩展方向
在实际运行中,你一定会遇到各种问题。以下是一些典型问题及解决方案。
5.1 爬虫稳定性与伦理问题
- 问题:IP被封锁。
- 解决 :使用代理IP池。可以考虑付费代理服务,或者使用
scrapy-rotating-proxies这样的中间件。对于非常重要的源,可以考虑使用更慢但更稳定的抓取频率,并严格遵守网站的robots.txt协议。
- 解决 :使用代理IP池。可以考虑付费代理服务,或者使用
- 问题:网站结构频繁变动导致解析失败。
- 解决 :建立解析规则的版本管理和回滚机制。每次解析失败时记录日志,并触发告警。对于核心源,可以编写更健壮的、基于多个备选选择器的解析逻辑。定期(如每周)人工抽查解析结果。
- 问题:动态加载(AJAX)内容抓不到。
- 解决 :如前所述,使用Playwright/Selenium。但应将其作为特定源的“特殊模式”,并限制并发数和执行频率,因为其资源消耗是普通请求的数十倍。
重要提示:网络爬虫伦理。 务必尊重网站资源。设置合理的抓取延迟(如
time.sleep(random.uniform(2,5))),避免对目标服务器造成压力。检查并遵守robots.txt。如果网站提供API(如GitHub API、Twitter API),优先使用API,它更稳定、更友好。
5.2 内容质量与噪音控制
- 问题:摘要生成效果不佳,要么太长,要么抓不住重点。
- 优化 :尝试不同的摘要算法和参数(
ratiovsword_count)。对于技术文章,可以尝试在摘要前先提取章节标题(<h2>,<h3>标签)作为大纲。结合NER提取的关键实体,在摘要中高亮显示。
- 优化 :尝试不同的摘要算法和参数(
- 问题:分类不准,DeFi文章被打上了NFT标签。
- 优化 :优化关键词词典,加入否定词和上下文规则。例如,“NFT”和“借贷”同时出现时,可能属于“NFT-Fi”,而不是单纯的“NFT”。积累一定数据后,可以训练一个简单的分类模型,提升准确率。
- 问题:重要但讨论度不高的技术更新被遗漏。
- 优化 :调整排序算法。为“GitHub仓库Release”、“官方博客”这类来源赋予极高的基础权重,确保它们即使社交信号弱也能被优先收录。
5.3 系统性能与扩展性
- 问题:抓取任务越来越多,执行时间变长,影响当日摘要生成。
- 优化 :
- 横向扩展 :增加Celery Worker的数量,提高并发抓取能力。
- 任务分区 :将数据源按更新频率分区。高频源(如Twitter)每15分钟抓一次,低频源(如官方博客)每天抓一次。
- 异步优化 :确保
aiohttp的客户端会话和连接池配置合理,避免重复创建连接的开销。 - 缓存 :对不常变动的页面(如项目首页)进行缓存,减少不必要的请求。
- 优化 :
5.4 未来可扩展的方向
一个基础的每日摘要系统稳定运行后,可以考虑以下方向深化:
- 个性化推荐 :引入用户系统,让用户选择感兴趣的主题(如“我只关心以太坊和Layer2”)。在每日推送时,除了全局精选,附上个性化推荐列表。
- 多语言支持 :使用翻译API(如DeepL, Google Translate)将英文内容摘要翻译成中文,或反之,服务更广泛的受众。
- 情感分析与趋势预测 :对抓取的内容进行情感分析,识别市场情绪(FOMO/FUD)。聚合一段时间内的主题热度,生成趋势报告。
- 交互式探索 :提供一个简单的Web界面,不仅展示每日摘要,还允许用户按时间、分类、项目检索历史文章,形成一个小型的信息库。
- 与链上数据结合 :当某条新闻提及某个协议或代币时,自动在摘要旁附上其当前的关键链上指标(如TVL、价格、交易量),提供更立体的信息。
构建这样一个系统,是一个典型的“数据流水线”工程,涵盖了爬虫、数据处理、NLP、后端开发和运维等多个领域。它没有高深莫测的算法,但对系统的稳定性、可维护性和细节处理要求极高。从 arfandi453/web3-daily-digest 这个项目标题出发,我们看到的不仅仅是一个脚本,而是一个为解决真实世界信息过载问题而设计的微型系统。每一步的选型和实现,都需要在资源、效率和效果之间做出权衡。希望这份超详细的拆解,能为你实现自己的信息聚合工具提供一份可靠的蓝图。
更多推荐


所有评论(0)