“抓了个寂寞”：一次实时信息采集的意外和修复

原创

jackcode

发布于 2025-07-24 10:33:40

7500

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

1. 那天下午，舆情系统“迟到”了

那天下午，公司运营那边突然在群里喊：“XX事件都快上热搜榜前十了，咱们系统咋没反应？”

我愣了几秒，立马翻后台日志、爬虫执行记录，结果一查，还真有点“抓瞎”的感觉。

明明脚本一直在跑，但偏偏就错过了那一小段爆发期。这不是掉链子是什么？

再仔细看时间戳，我们是每10分钟爬一次热榜。嗯……那确实，如果热点刚好冒头又迅速被其他话题压下，就容易被我们“跳过去了”。

说白了，我们只是“定时拍照”，而不是“持续录像”。

2. 问题在哪？我们原来只会定时“扫一眼”

回头捋逻辑，我发现旧的方式过于粗糙：

# 简化逻辑：10分钟跑一轮
while True:
    fetch_hotlist()
    time.sleep(600)

这和“公交车每小时来一趟”差不多，偏偏你想搭车的时候它刚走，你就只能干等下次。

更麻烦的是，微博这种平台热度变化太快，有些关键词一两个爆料就能冲上去再掉下来。如果这期间没被我们爬到，就等于白忙活。

3. 得改！用“最近几分钟都回看一遍”的思路试试？

我们没打算加快频率（怕被封），于是想到个折中方案：

与其加速，不如每次多看一点时间段，用“最近10分钟的数据”来弥补单点采集盲区。

说白了就是——

不光盯现在，还得回头看看“刚刚”是不是有啥动静。

4. 实战起来：加上代理，用滑动窗口 + 去重逻辑跑一次

我们继续用微博热搜举例，思路是这样：

每分钟爬一次（频率提高，但控制节奏）；
每次都拿完整榜单，分析有没有“刚冒出来”的词；
通过关键词生成MD5做“去重”，避免重复入库；
用亿牛云的动态代理，防止高频被限制IP。

核心代码片段如下：

import requests
import hashlib
import time
from datetime import datetime

# 动态代理设置（来自亿牛云示例）
proxy_host = "proxy.16yun.cn"
proxy_port = "3100"
proxy_user = "16YUN"
proxy_pass = "16IP"

proxies = {
    "http": f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}",
    "https": f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
}

# 简单“去重表”，防止刷屏式重复入库
seen = set()

def fetch_weibo_hot():
    try:
        headers = {"User-Agent": "Mozilla/5.0"}
        url = "https://s.weibo.com/top/summary"
        resp = requests.get(url, headers=headers, proxies=proxies, timeout=8)
        if resp.status_code == 200:
            return resp.text
    except Exception as e:
        print("抓取失败：", e)
    return ""

def extract_titles(html):
    import re
    pattern = r'<td class="td-02">.*?<a.*?>(.*?)</a>'
    return re.findall(pattern, html)

def is_new(topic):
    h = hashlib.md5(topic.encode()).hexdigest()
    if h in seen:
        return False
    seen.add(h)
    return True

def run_monitor():
    print("启动微博热搜监控（含回看机制）...")
    while True:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{now}] 正在检查更新...")

        html = fetch_weibo_hot()
        if not html:
            time.sleep(60)
            continue

        titles = extract_titles(html)
        for t in titles:
            if is_new(t):
                print("发现新词：", t)
                # 这里你可以换成数据库写入或告警逻辑

        time.sleep(60)  # 下一轮