ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Python爬虫采集热搜TOP50并写入Excel

Python爬虫采集热搜TOP50并写入Excel

简介:围绕Python爬虫获取微博TOP50热搜并写入Excel的实战资料包,面向具备Python基础、希望练习网页数据抓取与表格整理的学习者。内容聚焦热搜榜单的请求获取、HTML解析、字段提取及Excel落盘,可用于课堂示例、个人练习或小型数据采集流程的入门参考。压缩包为7z格式,共2个文件,由1个Python脚本和1个xlsx表格组成,整体约7KB,体积轻便,便于直接解压查看与二次修改。脚本承担抓取与数据处理逻辑,表格用于承接整理后的热搜结果。已有478人学习下载。资料覆盖requests发起请求、BeautifulSoup定位榜单结构、pandas导出Excel等关键环节,并涉及时间戳记录、异常处理与重试思路,能帮助读者理解从页面解析到结果保存的完整链路,同时提醒遵守robots.txt及相关规范。

1. 热搜榜 TOP50 抓取:一个 requests 会话加 openpyxl,两三秒落一张表

周五下午被叫去开会,运营想要一份近一周的热搜榜 TOP50 记录,第二天早上就要。人工从网页上一条条复制,50 条乘 7 天就是 350 行,抄到一半眼睛就花了。这种活交给 Python 爬虫最合适:一个 requests 会话打到榜单接口,解析层把嵌套 JSON 摊平成排名、标题、热度、标签四五个字段,openpyxl 直接写进 Excel,50 条数据两三秒落表,顺手补一个采集时间戳,后面做趋势对比就不用再回头翻网页。这套代码适合两类人:刚学完 requests 想找个有真实产出物练手的同学,以及每天要固定留档榜单、不想再手动贴表的内容运营和数据分析岗。技术栈就三样——requests 发请求、json 加正则做清洗、openpyxl 落盘,不上框架、不建数据库,一个文件夹就能跑起来,改造成定时任务也只是加一层调度的事。

2. 项目结构与技术选型:三段式分层,抓取、解析、存储各管一段

2.1 目录结构与模块职责

单文件脚本能跑,但加到第三个功能就开始互相打架:改一下请求头要翻两百行,调 Excel 样式又怕碰坏解析逻辑。所以这个项目按「抓取、解析、存储」三段切开,每个文件只干一件事。检索里不少热搜爬虫示例都是一股脑塞进 main.py,我一开始也这么写,后来加了重试和追加写,整个文件变成黑匣子,出问题只能靠 print 一段段试。

hotspider/ ├── main.py # 编排:抓取 → 解析 → 落盘,异常统一在这里兜 ├── config.py # 全部可调参数:接口、条数、超时、重试、输出路径 ├── fetcher.py # 网络层:Session 复用、超时、重试、退避 ├── parser.py # 解析层:JSON 摊平、热度值归一化、去重 ├── excel_writer.py # 存储层:表头样式、追加写、公式注入防护 ├── requirements.txt ├── .env.example # Cookie 等敏感项的模板,真实值不进仓库 ├── logs/ # 按天滚动,出问题先看这里 └── output/ # 生成的 xlsx,按日期命名

分层的收益在排查时最明显。某天早上发现表是空的,按链路往下走:先看 logs 里有没有失败重试记录,有就说明是网络层;没有就把 resp.text 前 200 个字符打出来,是 HTML 还是 JSON 一眼就分清,前者通常是 Cookie 过期或者接口变了,后者就是解析字段对不上。整条链路三段都能单独跑,写个python -c "from parser import parse_top_list; ..."直接喂本地保存的 JSON 文件,不用每次真去请求一遍接口,调试速度提升非常明显。

main.py保持薄,只做编排,代码里不出现任何requests或openpyxl的直接调用:

# main.py import logging from datetime import datetime from pathlib import Path from config import HOT_API, TOP_N, OUTPUT_DIR from fetcher import build_session, fetch_json from parser import parse_top_list from excel_writer import write_sheet def setup_log(): Path("logs").mkdir(exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ logging.FileHandler(f"logs/spider_{datetime.now():%Y%m%d}.log", encoding="utf-8"), logging.StreamHandler(), ], ) def run(): log = logging.getLogger("main") session = build_session() payload = fetch_json(session, HOT_API) rows = parse_top_list(payload, top_n=TOP_N) if not rows: log.error("解析结果为 0 条,接口结构可能已变化,请检查 payload") return path = Path(OUTPUT_DIR) / "hot_top50.xlsx" n = write_sheet(rows, str(path)) log.info("写入 %s 条,文件: %s", n, path) if __name__ == "__main__": setup_log() run()

这里的run()就是一条直线:建会话、取数据、解析、落盘。setup_log把日志同时丢到文件和终端,文件名带日期,跑一个月也不会把单个文件撑爆。rows为空直接 return 而不是继续写,是为了避免生成只有表头的空表——空表交给下游分析会非常难受,会有人来问你为什么今天的热搜全是空。Path比字符串拼路径稳,Windows 上少一堆反斜杠的玄学问题。

2.2 依赖清单与版本约束

依赖只有四个,都算轻量:

requests==2.32.3 openpyxl==3.1.5 python-dotenv==1.0.1 lxml==5.2.2

requests负责 HTTP,openpyxl负责写 xlsx,dotenv用来从.env读 Cookie,lxml是给 HTML 兜底解析路径准备的——接口结构变动时可以从页面直接抠数据,多一个退路。不要装pandas只为了写一个五十行的表,它会把 openpyxl 当写引擎再包一层,多几十兆依赖,样式控制反而不如直接用 openpyxl 顺手。真要后面做分析再装也不迟。

版本号建议锁死。openpyxl在 3.x 各小版本之间对样式对象的复用行为有过调整,同一个PatternFill实例跨工作簿复用偶发报错,锁版本能省掉这类难查的怪问题。装依赖用虚拟环境,别往系统 Python 里堆:

python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -r requirements.txt

2.3 配置项集中管理

所有能调的参数集中到一个文件,改接口、改条数、改输出目录都不动业务代码:

# config.py import os from dotenv import load_dotenv load_dotenv() # 优先从 .env 读取,未配置时回退到默认值 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # 榜单数据接口,换成你目标站点的真实地址即可 HOT_API = os.getenv("HOT_API", "https://example.com/api/hot") TOP_N = int(os.getenv("TOP_N", "50")) # 榜单条数,默认取前 50 TIMEOUT = (5, 10) # (连接超时, 读取超时),单位秒 RETRY_TIMES = 3 # 单次请求最多重试次数 RETRY_BACKOFF = 1.5 # 退避基数,第 n 次等待 backoff ** n 秒 REQUEST_INTERVAL = 0.8 # 两次请求之间的间隔,防触发限流 OUTPUT_DIR = os.path.join(BASE_DIR, "output") COOKIE = os.getenv("HOT_COOKIE", "") # 敏感信息只从环境变量进 HEADERS = { "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"), "Accept": "application/json, text/plain, */*", "Referer": "https://example.com/", "Cookie": COOKIE, }
参数默认值调整场景
TOP_N50想留全量就调大,落表前记得看接口是否分页
TIMEOUT(5, 10)网络差调到 (10, 20),别设成 None,否则会永久挂起
RETRY_TIMES3偶发 5xx 够用,频繁失败说明不是重试能解决的问题
RETRY_BACKOFF1.5遇到 429 调到 2.0 以上,退避更狠一些
REQUEST_INTERVAL0.8只请求一次时无感,改成批量抓多天历史时非常重要

TIMEOUT用元组而不是单值,是因为连接超时和读取超时是两个完全不同的故障:连接超时通常是网络或者域名解析问题,读取超时多半是服务端在处理但慢。区分开之后看日志就知道该等还是该换。Cookie走环境变量这一点别偷懒,硬编码提交到仓库里,第二天就得改密码。

3. 抓取层拆解:一个热搜请求从构造到拿到 JSON 的完整链路

3.1 Session 复用与请求头对齐

单次抓取其实用requests.get就够,但一旦要做批量、要抓多天、要定时跑,Session的优势就出来了:TCP 连接复用、Cookie 自动保持、默认头统一注入。热搜接口对Referer和User-Agent的校验很常见,头不对会返回一个结构完全不同的小 JSON 或者直接 403,所以头统一在 Session 层塞进去最省心。

# fetcher.py(上半部分) import logging import time import requests from config import HEADERS, TIMEOUT, RETRY_TIMES, RETRY_BACKOFF, REQUEST_INTERVAL log = logging.getLogger(__name__) def build_session(cookie: str = "") -> requests.Session: """建一个复用的会话,把公共请求头一次性注入""" s = requests.Session() s.headers.update(HEADERS) if cookie: s.headers["Cookie"] = cookie # 关闭自动重定向会暴露真实跳转,这里保持默认开启即可 s.max_redirects = 5 return s def polite_sleep(): """单次抓取感受不到,批量抓历史时这一行能救你的 IP""" time.sleep(REQUEST_INTERVAL)

build_session里max_redirects = 5是防止某些场景下的重定向循环把程序卡死;默认是 30,其实也能用,但明确写出来后续看代码的人不用查文档。polite_sleep单独抽成函数,是为了以后把间隔改成随机抖动时只动一处——固定间隔的请求节奏太规律,加个random.uniform(0.8, 1.6)会更自然,属于常见做法。

请求头里Accept写成 JSON 优先很关键。有些站点会根据Accept决定返回 JSON 还是 HTML 页面,声明要 JSON 能省一次解析分支。Referer填站点首页就行,不用精确到具体路径,填太细反而容易过期。

3.2 重试、超时与状态码判断

网络请求失败是常态,不重试的脚本跑一周总会断一次。但也别见错就重试——4xx 类错误重试一百次也是同样的结果,只会拖慢整体节奏。

# fetcher.py(下半部分) def fetch_json(session, url, params=None, retry=RETRY_TIMES): """带指数退避的 GET,返回解析后的 JSON;失败抛 RuntimeError""" last_err = None for i in range(retry): try: resp = session.get(url, params=params, timeout=TIMEOUT) # 限流和服务端错误值得重试,其余 4xx 直接失败 if resp.status_code == 429 or resp.status_code >= 500: raise requests.HTTPError(f"status={resp.status_code}") resp.raise_for_status() # 接口返回非 JSON 时 .json() 会抛 ValueError,被下面的 except 接住 return resp.json() except (requests.RequestException, ValueError) as e: last_err = e wait = RETRY_BACKOFF ** (i + 1) # 1.5s → 2.25s → 3.375s log.warning("第 %s/%s 次请求失败: %s,%.2fs 后重试", i + 1, retry, e, wait) time.sleep(wait) raise RuntimeError(f"重试 {retry} 次仍失败: {last_err}")

逻辑上有三个判断值得说。第一,429 和 5xx 手动抛成HTTPError走重试分支,其余 4xx(比如 403、404)在raise_for_status()处抛出后其实也会被except接住,这里没有做区分——如果你想让 403 立刻失败不浪费时间,可以在except里加一句if isinstance(e, requests.HTTPError) and "40" in str(e): break,属于可选优化。第二,resp.json()包的ValueError一定要接,接口在 Cookie 失效时经常返回一段 HTML 登录页,.json()直接就炸了,不接住的话日志里只有一句Expecting value,看不出是登录页。第三,退避用的是backoff ** (i+1),三次失败分别等 1.5、2.25、3.375 秒,加起来不到 8 秒,既给了服务端缓冲,又不会让定时任务卡太久。

params参数留着是为了处理分页接口,比如{"page": 1, "size": 50},不用改函数签名,直接在 main 里传就行。

3.3 双解析路径:优先 JSON,退回 HTML

榜单接口返回的结构通常长这样:外层一个data,里面挂realtime、list或者items,每个元素用word、num、label_name这类字段名。不同站点字段名差异很大,写死一个键名等于把代码和环境绑死,所以解析层做三层兜底。

# parser.py import re _UNIT = {"万": 10_000, "亿": 100_000_000} def parse_hot(raw) -> int: """把 '1234567' / '1.2万' / '3.5亿' 统一成整数,方便排序和做图""" if raw is None: return 0 if isinstance(raw, (int, float)): return int(raw) s = str(raw).strip().replace(",", "") m = re.match(r"^([\d.]+)\s*([万亿]?)$", s) if not m: return 0 num, unit = float(m.group(1)), m.group(2) return int(num * _UNIT.get(unit, 1)) def parse_top_list(payload: dict, top_n: int = 50) -> list[dict]: """把嵌套 JSON 摊平成统一字段,顺带按标题去重""" data = payload.get("data") or {} items = data.get("realtime") or data.get("list") or data.get("items") or [] rows, seen = [], set() for it in items: title = (it.get("word") or it.get("title") or "").strip() if not title or title in seen: # 去重:同一话题有时会挂在两个位置 continue seen.add(title) rows.append({ "rank": len(rows) + 1, # 排名按去重后的顺序重编,不信任原始序号 "title": title, "hot": parse_hot(it.get("num") or it.get("hot") or it.get("hotValue")), "label": (it.get("label_name") or it.get("category") or "").strip(), "url": it.get("url") or it.get("scheme") or "", }) if len(rows) >= top_n: break return rows

parse_hot是整个解析层最容易被忽略的一段。接口返回的热度有时是纯数字,有时是「123.4万」这种带单位的字符串,直接拿去排序会得到「9万 > 100万」这种离谱结果,因为字符串比较按字符走。转成整数之后再排、再做图、再算环比都不会翻车。正则^([\d.]+)\s*([万亿]?)$只认这两种单位,遇到「1.2w」这种写法会返回 0——看到 0 就说明字段格式变了,比悄悄写错好。

parse_top_list里的排名重编是有意的。原始数据里偶尔会出现重复话题或者置顶位,序号不连续,直接照抄会让 Excel 里出现两个第 3 名。按去重后的顺序重编,表就是规整的 1 到 50。seen集合用标题去重而不是 id,是因为热搜话题的 id 在跨天之后会复用,按 id 去重反而会误杀。

如果接口彻底变了、只能从页面抓,就走 HTML 兜底路径,用 lxml 的 xpath 定位列表项:

# parser.py 追加:HTML 兜底解析 from lxml import html as lxml_html def parse_from_html(text: str, top_n: int = 50) -> list[dict]: """页面版兜底:结构变了就改 xpath,不要动上层调用""" doc = lxml_html.fromstring(text) nodes = doc.xpath('//div[contains(@class,"hot-list")]//li') rows = [] for i, li in enumerate(nodes[:top_n], start=1): title = "".join(li.xpath('.//span[@class="title"]/text()')).strip() hot = "".join(li.xpath('.//span[@class="num"]/text()')).strip() if title: rows.append({"rank": i, "title": title, "hot": parse_hot(hot), "label": "", "url": ""}) return rows

xpath 用contains(@class,...)而不是精确匹配,是因为前端加个响应式类名hot-list active就会让精确匹配全部落空,这类血泪经验踩一次就记住了。两条路径的输出结构完全一致,上层main.py里加个if "data" not in payload的分支就能无缝切换,存储层一行都不用改。

4. 数据清洗与 Excel 落盘:openpyxl 写出一张能直接看的 TOP50 表

4.1 字段模型与热度归一化

落盘前先把字段定死,六个列足够覆盖日常需求:排名、标题、热度、标签、链接、采集时间。多一列原文摘要会让表格变宽,五十行看着还行,攒一个月就没人愿意打开了。采集时间一定要有,它的作用不是给人看的,后面做「同一话题第几天上榜、热度涨了多少」的对比全靠它。

热度归一化在 3.3 已经做了,这里补一个统计口径的问题:热度数字在一天内会波动,同一条话题早上 80 万、晚上 120 万很正常,做趋势图的时候不要拿绝对值直接比,先按当天最大值归一化成百分比,曲线才有可比性。这个处理可以放在 Excel 里用公式做,也可以落盘前就多算一列,我一般选后者,下游拿到表就能直接用。

写入前还有一次防御性检查:标题里如果混进换行符或者制表符,Excel 单元格会撑得很高,看起来像格式错乱,用replace("\n", " ").replace("\t", " ")统一拍平。链接列如果为空就写空字符串而不是None,None写进单元格是空值,某些版本的 Excel 打开会提示修复。

4.2 openpyxl 写表与样式控制

# excel_writer.py import os from datetime import datetime from openpyxl import Workbook, load_workbook from openpyxl.styles import Font, Alignment, PatternFill from openpyxl.utils import get_column_letter COLUMNS = ["排名", "标题", "热度", "标签", "链接", "采集时间"] # 以这些字符开头的标题会被 Excel 当公式解析,前面补单引号强制成文本 FORMULA_PREFIX = ("=", "+", "-", "@") def _safe_text(v: str) -> str: v = (v or "").replace("\n", " ").replace("\t", " ").strip() return "'" + v if v.startswith(FORMULA_PREFIX) else v def _style_sheet(ws): fill = PatternFill("solid", fgColor="1F4E79") for cell in ws[1]: cell.font = Font(bold=True, color="FFFFFF") cell.fill = fill cell.alignment = Alignment(horizontal="center", vertical="center") for i, w in enumerate([6, 52, 12, 10, 58, 20], start=1): ws.column_dimensions[get_column_letter(i)].width = w ws.freeze_panes = "A2" # 冻结表头,向下滚动时列名始终可见 def write_sheet(rows, path, sheet_name="热搜TOP50"): """已存在则追加,不存在则新建;返回本次写入行数""" if os.path.exists(path): wb = load_workbook(path) ws = wb[sheet_name] if sheet_name in wb.sheetnames else wb.create_sheet(sheet_name) if ws.max_row == 1: # 只有表头说明还没写过数据 ws.append(COLUMNS) else: os.makedirs(os.path.dirname(path) or ".", exist_ok=True) wb = Workbook() ws = wb.active ws.title = sheet_name ws.append(COLUMNS) now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for r in rows: ws.append([r["rank"], _safe_text(r["title"]), r["hot"], r["label"], r["url"], now]) _style_sheet(ws) wb.save(path) return len(rows)

参数说明几个关键点。sheet_name留给以后按周分 sheet 用,默认一张表从头写到尾。追加模式的判断条件是ws.max_row == 1,意思是只剩表头那一行,此时补一次表头;如果max_row大于 1 说明已经有数据,直接往下 append,不会出现重复表头。这个判断比「文件存在就追加」更可靠,因为文件可能被手动删了内容只留表头。

_safe_text处理公式注入:如果某条热搜标题恰好以=或+开头,openpyxl 写进去之后 Excel 打开会尝试当公式计算,轻则显示#NAME?,重则触发外部引用警告弹窗。前面加一个单引号强制成文本,这是 Excel 自身的约定写法,看着有点丑但能保证打开不报错。

freeze_panes和列宽这类样式,重跑时会重复应用一次,openpyxl对重复赋值是幂等的,不会把表头越涂越深,可以放心每次调用。样式对象PatternFill建议在函数内部创建(如上),跨工作簿复用同一个实例在部分版本上会报Style相关的错。

4.3 追加模式与按日期分表

两种落盘策略按使用场景选。单文件追加适合攒长期历史,一个 xlsx 里几百上千行,用采集时间列做筛选,配数据透视表就能看出话题生命周期。按日期分文件适合每天归档给不同人,文件名带日期,谁要哪天的直接拿走。

# excel_writer.py 追加:按日期分文件 from datetime import datetime import os def daily_path(output_dir: str) -> str: """一天一个文件,跑多次写同一个文件,避免覆盖丢失""" day = datetime.now().strftime("%Y%m%d") return os.path.join(output_dir, f"hot_top50_{day}.xlsx")

daily_path和write_sheet组合起来就是幂等的:同一天跑十次,数据都追加到同一个文件,不会出现output_1.xlsx、output_2.xlsx这种越攒越乱的目录。如果希望同一天只保留最新一版,可以在写入前判断文件是否存在并删掉重写,但从留档角度看,追加更安全——重复总比丢数据好,重复的可以去重,丢了的找不回来。

另外提醒一句,xlsx 单表上限是 1048576 行,按每天 50 条算能写五十多年,不用担心。倒是文件体积会慢慢涨,load_workbook加载几万行明显变慢,此时换成按天分文件或者切到 sqlite,是对的方向。

5. 避坑与排查:Cookie 失效、空表、公式注入这几个坑都踩过

5.1 请求返回 200 但解析出 0 条

现象:日志显示请求成功,rows却是空列表,Excel 里只有表头,或者干脆连文件都没生成。

原因:绝大多数情况是 Cookie 过期。榜单接口在未登录时可能返回一个结构完全不同的降级 JSON,键名还是data,但里面没有realtime;也可能直接返回一段 HTML 登录页,状态码依然是 200。两种情况下fetch_json都不会报错,问题被推迟到解析层才暴露。

解决:在fetch_json里加一行受控的响应快照,失败时把前 200 个字符写进日志,一眼就能区分是 JSON 还是 HTML。同时把parse_top_list里items为空的情况显式打日志,别让它静默返回空列表。

raw = resp.text[:200] if len(resp.text) < 2000 else resp.text[:200] log.debug("响应快照: %s", raw.replace("\n", " "))

Cookie 每天刷新一次是最省事的做法,从浏览器里复制最新的值写进.env,别硬编码。如果频繁过期,说明该站点的会话时效很短,考虑在请求前先访问一次首页拿基础凭证,属于常见做法。

5.2 表头重复出现或者数据写到第二个 sheet

现象:同一份 Excel 打开,表头出现了两次、三次,中间还夹着一批数据;或者数据莫名其妙跑到了Sheet2里。

原因:表头重复通常是判断条件写成了「文件存在就 append 表头」。openpyxl的Workbook()默认建一个名为Sheet的活动表,如果新建文件时改过ws.title = "热搜TOP50",之后用wb["热搜TOP50"]是找不到的,代码走create_sheet又建了一张同名新表,数据就分流了。

解决:追加前统一用ws.max_row == 1判断,而不是文件是否存在;取 sheet 时用if name in wb.sheetnames兜底,找不到再创建。读文件之前先print(wb.sheetnames)看清楚里面到底有几张表,这个习惯能省掉一半的排查时间。

5.3 Excel 里标题变成 #NAME? 或者一串异常字符

现象:明明抓到了中文标题,打开 Excel 却显示#NAME?,或者单元格内容变成了日期、科学计数法。

原因:openpyxl写入时不会主动转义,以=、+、-、@开头的字符串会被 Excel 当作公式解析;纯数字组成的标题(比如某个年份话题)会被自动识别成数值并截断前导零。

解决:用 4.2 里的_safe_text,以公式前缀开头的标题前面加单引号;纯数字标题同样加单引号强制文本。写完一定要用 Excel 真实打开看一遍,不要只靠「文件生成了」来判断成功,文件生成了但内容是错的是最隐蔽的失败。

5.4 短时间重复运行触发 429

现象:本地调试时连着跑几次,突然全部请求返回 429,或者返回一个明显被裁剪过的列表。

原因:固定间隔加固定 UA,请求特征太规律。榜单接口对无 Cookie 的低频访问通常宽容,但连续触发会被判定为异常。

解决:REQUEST_INTERVAL改成随机抖动,单次抓取场景影响为零,批量场景能把命中率降下来。真遇到 429 不要硬刚,退避时间翻倍,等一两分钟再试。调试阶段把响应存成本地 JSON 反复回放,别每次都真去请求,这条经验值得专门记住——本地有快照,解析逻辑随便改,一天能省下几百次请求。

6. 二次开发:把单次脚本改成定时采集加趋势对比

单次跑通只是起点,真正有价值的用法是每天定时采一份,攒出时间序列,然后回答「哪个话题连续三天在榜」「热度涨得最快的是哪条」。改造分三步走,都不复杂。

第一步,落库代替直接落 Excel。用标准库 sqlite3 建一张最简单的事实表,字段和 Excel 对齐即可,采集时间存 ISO 格式字符串。好处是查询灵活,做环比不用把几百个 xlsx 打开再拼。

# storage.py import sqlite3 from contextlib import closing DDL = """ CREATE TABLE IF NOT EXISTS hot_snapshot ( id INTEGER PRIMARY KEY AUTOINCREMENT, rank INTEGER, title TEXT NOT NULL, hot INTEGER, label TEXT, url TEXT, collected_at TEXT NOT NULL, UNIQUE(title, collected_at) -- 同一次采集内不重复 ) """ def save(rows, db_path="data/hot.db"): with closing(sqlite3.connect(db_path)) as conn: conn.execute(DDL) conn.executemany( "INSERT OR IGNORE INTO hot_snapshot(rank,title,hot,label,url,collected_at) " "VALUES(?,?,?,?,?,datetime('now','localtime'))", [(r["rank"], r["title"], r["hot"], r["label"], r["url"]) for r in rows], ) conn.commit()

UNIQUE(title, collected_at)配上INSERT OR IGNORE让重复执行变得安全,同一天多跑几次不会产生重复行。datetime('now','localtime')用数据库本地时间,和日志时间对齐,排查时不用做时区换算。

第二步,定时调度。Linux 上直接 cron,一行搞定,比装 APScheduler 省一个依赖;Windows 上用任务计划程序。cron 表达式按每两小时采一次写:

# 0 */2 * * * cd /path/to/hotspider && .venv/bin/python main.py >> logs/cron.log 2>&1

注意必须写绝对路径,cron 的工作目录不是项目目录,相对路径会找不到config.py;输出重定向也不能省,否则报错信息直接进邮件黑洞。

第三步,做趋势对比。有了历史数据,一条 SQL 就能找出连续在榜的话题:

SELECT title, COUNT(DISTINCT date(collected_at)) AS days, MAX(hot) - MIN(hot) AS delta FROM hot_snapshot WHERE collected_at >= datetime('now', '-7 day', 'localtime') GROUP BY title HAVING days >= 3 ORDER BY delta DESC LIMIT 20;

days >= 3过滤出持续在榜的长尾话题,delta是七天内热度的最大涨幅,按涨幅排序就能看出哪条在持续发酵。这套查询的结果导出成第二张 sheet,和当天的 TOP50 放在一个文件里,运营拿到的就是「今日榜单 + 七日趋势」的组合,比单纯的快照有用得多。

最后说个习惯。以前我改这类脚本总喜欢直接动 main.py,改完顺手跑一次,看到文件生成了就以为成了。结果有次_safe_text写漏了,标题全变成#NAME?,文件照样生成,日志照样报成功,第二天才发现一整天的数据废了。从那以后我每次改完解析或写入逻辑,都会先拿本地保存的响应快照做一次离线回放,再用 Excel 真实打开检查表头和前三行内容,最后才接上真实请求。多花两分钟,能省掉一整天的返工。希望帮到你。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进