如果你也是那种每天不刷一遍 Product Hunt 就浑身不自在的人一定懂这种感觉首页翻到第四五屏全是差不多的 AI 工具等晚上再回来看真正涨势凶猛的产品早被淹没在海量更新里了。手动盯能盯出感觉但效率太低还容易漏。我做的这个 Python 爬虫就是奔着“自动化”三个字去的——每天早上自动把 Product Hunt 的产品数据抓回来存好再用增量算法跑出一份“今日爆款”名单。你不用熬夜蹲首页每天早上花三分钟看结果就行。整个项目用到的都是 Python 爬虫里最基础、但恰恰也最容易被忽略的能力请求公开接口、解析结构化数据、用 SQLAlchemy 落库、用 Crontab 定时调度、用 SQL 做增量分析。难度曲线比较平缓适合刚学完 requests 和 SQL 基础的人当第二个练手项目对已经在做产品、做市场、做独立开发的人同样有用因为每天一份 Product Hunt 爆款榜单本质上就是一份免费的市场调研报告。在正式开始之前有一句话我必须放在最前面爬虫技术本身没有问题有问题的永远是使用方式。现在的网站普遍有使用条款、robots 协议和数据保护要求抓公开数据做个人学习、做非商业分析和抓取登录后数据再付费转卖完全是两码事。后面所有代码都只基于 Product Hunt 公开 API 暴露的数据也请读者自己把握好边界。1. 选对练手对象这个项目真正让你掌握的能力与合规红线1.1 这个项目到底让你练会什么很多人练爬虫喜欢挑一些比较“重”的网站比如电商、社交平台、视频网站一上来就要处理登录态、验证码、字体反爬、JS 逆向结果写了三天还在跟反爬机制搏斗项目热情直接清零。Product Hunt 完全不一样它有结构清晰的后端接口返回的数据字段完整、含义明确非常适合把精力放在“爬虫该有的完整链路”上。这个链路就是数据源分析 → 请求采集 → 解析清洗 → 入库存储 → 定时调度 → 数据分析。每一步都能迁移到其他项目里。比如这套代码把 API 地址换成你内部系统的接口稍微改下字段映射就是一个内部数据采集工具把 SQLAlchemy 模型换成 JSON 文件又变成了一个轻量爬虫脚本。真正值钱的不是“会 requests”这个动作而是你脑子里有没有这条链路。我建议的练手路线是先能跑通单次采集再手动跑几天攒数据最后再上定时任务和增量分析。别一上来就追求分布式、可视化管理平台那些东西等你有真实数据量了自然会发现需求没有数据量的时候搭出来全是空架子。1.2 合规红线爬虫第一课不是技术而是边界“因爬虫入狱”这个词频繁上热搜不是没有原因的。这个领域每年都有人因为越界把自己搭进去最常见的翻车方式就几种抓了需要登录才能看的数据、绕过反爬机制抓源码层的内容、把抓来的数据传输给第三方牟利。任何一个只要踩中性质就完全不同了这不是用“学习目的”能解释过去的。所以我在这个项目里给自己定的规则很简单只采集公开接口、公开页面能直接看到的数据不碰任何需要登录态的内容请求频率控制在“人肉浏览”级别宁可慢一点绝不当炮轰请求头里写清用途和联系方式至少让对方知道我是一个可追溯的自动化脚本数据只用于个人学习、非商业分析不转售、不批量复制、不对外公开原始数据集。这套规则同样适用于你拿这套代码去套别的网站。判断标准就一句话如果网站所有者看到你的爬虫行为会感到不舒服那这件事大概率越界了。技术是工具边界是人定的别让工具替你决定风险。2. 别急着写代码先把 Product Hunt 的数据源结构摸清楚2.1 从页面反推接口别直接去解析 HTML刚入门的时候很多人拿到一个网站第一反应就是requests.get然后交给 BeautifulSoup 解析 HTML。这套路对付静态页面没问题但 Product Hunt 首页是前端渲染的你直接拿 HTML 会发现内容很少产品卡片全是动态渲染进去的。与其绕一大圈不如先打开浏览器开发者工具刷新页面在 Network 面板里找 XHR 请求。操作路径很简单刷新首页 → Network 面板 → 过滤 XHR 或 Fetch → 看哪个请求返回了产品数据。Product Hunt 走的是 GraphQL接口地址是https://api.producthunt.com/v2/api/graphql请求方式是 POST返回的是 JSON。拿到这个请求以后你的爬虫目标就非常明确了模拟它而不是跟 HTML 纠缠。我把两种方案放在一起对比过差异很直观对比维度直接解析 HTML请求公开 API稳定性前端改版就挂字段相对稳定字段完整度页面不一定全展示结构化字段齐全维护成本选择器要反复改只需要维护查询语句合规风险相对较高使用官方公开接口更透明学习价值练了 BeautifulSoup接触 GraphQL更贴近现代后端2.2 核心接口长什么样需要哪些字段关于 GraphQL你不需要把它当成一个很玄的东西。你可以理解成后端给你一个“数据菜单”你想点什么菜就传什么字段它按你的要求原样返回。相比 REST 接口动辄返回一整坨无关字段GraphQL 对爬虫来说反而更友好。当时我用的查询结构大致是下面这样。GraphQL 的 schema 会不时调整如果你跑的时候报了字段错误按响应里的提示修改字段名即可query { posts(order: {createdAt: DESC}, first: 50) { edges { node { id name tagline description url votesCount commentsCount createdAt topics { edges { node { name } } } } } } }这个查询里我重点拿这几个字段id产品唯一标识去重时会用到name、tagline、description产品名称、一句话简介和完整介绍url产品官网或 PH 详情页链接votesCount、commentsCount累计点赞数和评论数这是后面算爆款的核心数据createdAt产品发布日期用于区分“今日新品”和“老产品”topics产品打的话题标签可以用来观察它属于什么赛道。3. 核心采集代码请求 GraphQL、解析 JSON 与容错兜底3.1 工程结构和配置我不建议把所有代码塞进一个main.py虽然小项目能跑但后面加定时、加分析、加存储会越来越乱。我用的目录结构长这样product_hunt_spider/ ├── config.py # 配置API地址、Token等 ├── spider.py # 请求和解析逻辑 ├── models.py # SQLAlchemy 模型 ├── pipeline.py # 入口抓取 → 解析 → 入库 ├── analysis.py # 增量排行分析 └── requirements.txt # requests、sqlalchemy依赖很简单就两个核心库pip install requests sqlalchemyconfig.py里我特意用了环境变量来保护 Token而不是把它写死在代码里import os PH_API_URL https://api.producthunt.com/v2/api/graphql PH_TOKEN os.getenv(PH_TOKEN, ) def build_headers() - dict: return { Authorization: fBearer {PH_TOKEN}, Content-Type: application/json, Accept: application/json, User-Agent: PH-Daily-Spider/1.0 (个人学习用途; 联系方式: youremail.com) }3.2 请求部分怎么写才稳我习惯用requests.Session()它能复用底层连接多次请求时效率更高也能统一设置 headers。请求体里最好把query和variables分开写后面做分页时改参数会方便很多import time import requests from config import PH_API_URL, build_headers def fetch_posts(session: requests.Session, after: str | None None, limit: int 50) - dict: query query($limit: Int!, $after: String) { posts(order: {createdAt: DESC}, first: $limit, after: $after) { edges { node { id name tagline description url votesCount commentsCount createdAt topics { edges { node { name } } } } } pageInfo { endCursor hasNextPage } } } payload { query: query, variables: {limit: limit, after: after} } resp session.post(PH_API_URL, jsonpayload, headersbuild_headers(), timeout15) resp.raise_for_status() return resp.json()timeout15很重要不加这个参数一旦网络波动脚本可能挂在那里几分钟不动。稍微解释一下resp.raise_for_status()它会在 HTTP 状态码不是 2xx 时主动抛出异常帮你把错误拦截在源头不会带病继续往下跑。3.3 解析数据能容错的解析才有资格上线GraphQL 返回的 JSON 是有固定壳子的数据在data.posts.edges每个edge里有一个nodenode才是产品对象。解析时最忌讳直接写node[votesCount]因为一旦哪个产品缺字段整个脚本就 KeyError 崩了。我写了个normalize_post函数统一做字段清洗和默认值兜底from datetime import datetime def normalize_post(node: dict) - dict: raw_created_at node.get(createdAt) created_at None if raw_created_at: try: created_at datetime.fromisoformat(raw_created_at.replace(Z, 00:00)) except ValueError: created_at None topics [] for item in node.get(topics, {}).get(edges, []): topic_name item.get(node, {}).get(name) if topic_name: topics.append(topic_name) return { ph_id: node.get(id), name: node.get(name), tagline: node.get(tagline, ), description: node.get(description, ), url: node.get(url, ), votes_count: node.get(votesCount, 0), comments_count: node.get(commentsCount, 0), created_at: created_at, topics: |.join(topics), }话题标签我用|拼成一个字符串是为了在 SQLite 里好存真要做分析时再按|拆开就行。时间字段的处理我比较谨慎createdAt是 ISO8601 格式带Z结尾先替换成00:00再用fromisoformat解析解析失败就置空绝不让一条坏数据拖垮整个流程。3.4 分页需要全量数据时怎么循环对“每日爆款”来说前 50 个热门产品基本已经覆盖了值得关注的对象所以刚开始可以只用first50。但如果你想把当天发布的产品全量抓下来那就要处理分页。GraphQL 的分页逻辑不复杂响应里会给你pageInfo.hasNextPage和pageInfo.endCursor把endCursor作为下一次请求的after参数传进去就行def fetch_all_posts(session: requests.Session, max_pages: int 5) - list[dict]: all_posts [] cursor None for _ in range(max_pages): data fetch_posts(session, aftercursor) if errors in data: raise RuntimeError(fGraphQL errors: {data[errors]}) posts data.get(data, {}).get(posts, {}) edges posts.get(edges, []) for edge in edges: node edge.get(node, {}) if node: all_posts.append(node) page_info posts.get(pageInfo, {}) if not page_info.get(hasNextPage): break cursor page_info.get(endCursor) time.sleep(1) # 保持礼貌的频率 return all_postsmax_pages是一个保险丝防止某个异常场景下脚本无限翻页刷爆接口。实测中这个循环跑得很稳关键就在于每页之间sleep(1)既保证速度又不会撞限流。4. 数据落库不是随便存SQLAlchemy 模型与每日去重4.1 为什么用 SQLAlchemy而不是直接写 SQL如果你是第一次做落库可能会觉得直接写 SQLite 的execute也很简单。但一旦字段数量多起来你就会发现代码里到处都是INSERT INTO、UPDATE改一个字段要改好几处特别容易遗漏。SQLAlchemy 的 ORM 模型把表和 Python 对象映射起来改模型就能同步改表结构对爬虫这种字段变化频繁的项目非常友好。还有一个很实际的好处开发阶段用 SQLite文件型数据库零配置等数据量大到需要多人协作、需要远程访问时把连接串从sqlite:///products.db换成postgresql://user:passhost/dbname模型代码几乎不用动。这种平滑迁移能力直接裸写 SQL 是享受不到的。4.2 表结构设计这张表怎么定义“一天一份快照”我的核心设计思路是每天给每个产品存一条快照。这样你既能查今天的总点赞数也能拿昨天和今天的数据做差值算出真实的增长量。表结构用 SQLAlchemy 2.0 的声明式写法from datetime import datetime from sqlalchemy import create_engine, String, Integer, Text, DateTime, UniqueConstraint from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker engine create_engine(sqlite:///products.db, echoFalse) SessionLocal sessionmaker(bindengine) class Base(DeclarativeBase): pass class Product(Base): __tablename__ products __table_args__ ( UniqueConstraint(ph_id, collected_date, nameuq_ph_id_collected_date), ) id: Mapped[int] mapped_column(primary_keyTrue) ph_id: Mapped[str] mapped_column(String(32), indexTrue) name: Mapped[str] mapped_column(String(200)) tagline: Mapped[str] mapped_column(String(500), default) description: Mapped[str] mapped_column(Text, default) url: Mapped[str] mapped_column(String(500)) topics: Mapped[str] mapped_column(String(500), default) votes_count: Mapped[int] mapped_column(Integer, default0) comments_count: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, nullableTrue) collected_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.now) collected_date: Mapped[str] mapped_column(String(10), indexTrue) Base.metadata.create_all(engine)重点看这行唯一约束UniqueConstraint(ph_id, collected_date)。它保证“同一个产品在同一天只能有一条记录”。这个设计直接解决了重复抓取的问题——就算你今天把这个脚本手动跑了三遍数据库里每个产品也只会保留一条快照不会出现十行一模一样的数据。那collected_date用哪一天的我直接取 UTC 日期。下面会专讲时区问题这里先记住一个原则所有“日期”都用 UTC 算展示时再转本地时区。4.3 入库逻辑主键冲突时什么都不做入库我用的是 SQLite 方言的插入语句加上on_conflict_do_nothing。它的语义很直白“能插进去就插遇到唯一键冲突就跳过”。这样脚本天然具备幂等性重复执行不会造成脏数据from datetime import datetime, timezone from sqlalchemy.dialects.sqlite import insert as sqlite_insert from models import Product, SessionLocal def save_products(products: list[dict]) - int: collected_date datetime.now(timezone.utc).strftime(%Y-%m-%d) rows [{**product, collected_date: collected_date} for product in products] session SessionLocal() try: stmt sqlite_insert(Product).values(rows) stmt stmt.on_conflict_do_nothing(index_elements[ph_id, collected_date]) session.execute(stmt) session.commit() finally: session.close() return len(rows)如果你以后换到 PostgreSQL只需要把from sqlalchemy.dialects.sqlite import insert as sqlite_insert换成from sqlalchemy.dialects.postgresql import insert as pg_insert后面调.on_conflict_do_nothing()的写法基本一致。这就是选 ORM 的底气底层存储换了业务代码纹丝不动。5. 每天早上八点自动跑Crontab 调度与日志监控5.1 定时任务我为什么选 Crontab实现“每日自动跑”有好几种方式比如用 Python 的schedule库、用 Celery 配 beat或者直接上系统级 Crontab。我的选择是无脑 Crontab理由很简单它不需要常驻进程没有额外依赖系统自带对“每天跑一次”这种低频任务来说是最轻量的方案。schedule库虽然写起来好看但你的 Python 脚本必须一直挂着不退出资源浪费不说服务器一重启就失效。Celery 是给大规模分布式任务用的这个场景去上 Celery 属于杀鸡用牛刀。Crontab 的缺点是人机交互不太友好配置写错了没有直观提示但只要你掌握了五段式格式它就是最可靠的定时器。5.2 Crontab 配置怎么写先运行crontab -e打开当前用户的定时任务表然后加一行0 8 * * * cd /home/ubuntu/product_hunt_spider /usr/bin/python3 pipeline.py /var/log/product_hunt/ph_spider.log 21五段式从左到右分别是分钟、小时、日、月、周。这行的意思就是每天早上 8 点整切到项目目录用系统 Python 执行pipeline.py标准输出和错误输出都追加到日志文件里。命令里cd那一步很多人会忘。Crontab 执行命令时默认工作目录是用户目录不cd进去的话脚本里所有相对路径都会指错地方比如sqlite:///products.db这个相对路径就会建到别的位置去。我吃过这个亏所以后来配置文件全用绝对路径Crontab 里强制先cd到项目目录。日志路径如果不存在记得先建好mkdir -p /var/log/product_hunt有了日志文件脚本真的挂了你也知道去哪里看现场。我一般还会加一句date到日志里方便确认每次任务真实执行时间。5.3 本机跑还是服务器跑如果你只是自己用在本机跑也行但前提是电脑一直开着、而且不能休眠。这很不现实所以我后来直接放到一台低配轻量云服务器上一个月成本很低资源占用也小。整个项目跑起来的内存占用不到 200MB基本不挤占其他服务的资源。Token 在服务器上不要写进代码仓库用环境变量注入export PH_TOKEN你的_token然后手动执行一遍脚本确认日志正常再配置 Crontab。千万别一上来就配定时任务否则你根本分不清是定时配置写错还是脚本本身有 bug。6. “爆款”不是拍脑袋用 SQL 算增量、做排行6.1 增量比绝对量更说明问题很多人拿到votesCount就直接排序这有一个坑它叫“累计点赞数”不是“今天新增点赞数”。一个上周上线、已经攒了三千赞的老产品和一个今天刚上线、两小时涨了两百赞的新品放在一起排序老产品永远靠前但论“此刻的势头”新品明显更值得关注。所以我的分析逻辑是用今天的数据减去昨天的数据得到 24 小时增量。这个差值才是“爆款潜力”最直接的指标。也正是因为要算差值前面才会设计成“每天存一条快照”数据积累两三天后增量分析才能跑起来。6.2 两个榜单新品榜和增速榜我会把它拆成两个维度看这样信息更干净。第一个是“今日新品榜”判断标准是createdAt在最近 24 小时内按累计赞数和评论数排序第二个是“24 小时增速榜”用今天和昨天两张快照 join 起来算每个产品的点赞增量和评论增量。增速榜的 SQL 大概是这样的from datetime import datetime, timedelta, timezone import pandas as pd from sqlalchemy import text from models import engine today datetime.now(timezone.utc).strftime(%Y-%m-%d) yesterday (datetime.now(timezone.utc) - timedelta(days1)).strftime(%Y-%m-%d) sql text( WITH today AS ( SELECT ph_id, name, url, votes_count, comments_count FROM products WHERE collected_date :today ), yesterday AS ( SELECT ph_id, votes_count, comments_count FROM products WHERE collected_date :yesterday ) SELECT t.name, t.url, t.votes_count - COALESCE(y.votes_count, 0) AS votes_delta, t.comments_count - COALESCE(y.comments_count, 0) AS comments_delta FROM today t LEFT JOIN yesterday y ON t.ph_id y.ph_id WHERE t.votes_count - COALESCE(y.votes_count, 0) 0 ORDER BY votes_delta DESC, comments_delta DESC LIMIT 20; ) df pd.read_sql(sql, engine, params{today: today, yesterday: yesterday})这段 SQL 里LEFT JOIN的写法值得多看两眼如果某个产品今天上榜但昨天不在榜上yesterday表里没有它的记录COALESCE会把昨天的赞数补成 0。这种产品通常是“今天刚发布”的新品它的votes_delta实际等于累计赞数出现在增速榜顶部是正常的不影响判断。6.3 一个可调的“爆款系数”公式单纯按点赞增速排容易忽视讨论热度高的产品。一个产品赞数中等但评论区吵翻了天往往说明用户参与度极高这种产品后劲往往更足。我后来加了一个经验公式把评论的权重提上去hot_score votes_delta * 1 comments_delta * 5 topic_bonus解释一下两个参数思路comments_delta * 5写评论的成本比点一个赞高得多用户愿意花这个成本说明产品引发了深度讨论我给 5 倍权重topic_bonus话题标签多于 3 个的产品说明运营在不同赛道做了分发我给 5 到 10 分的额外加成。这不是官方指标只是我自己的经验公式权重完全按你的需求调。比如你做 B 端产品调研可能更看重 tagline 和描述里的关键词你做增长分析可能想拿评论增量和点赞增量对比判断讨论转化效率。数据已经存进库里了怎么算都行。6.4 把榜单变成每天早上能看的东西我每天早上会收到一份精简版报表其实就是把上面这个 DataFrame 输出成 Markdown 文件再配合一个简单的邮件发送脚本。如果你还需要可视化加一个 pandas matplotlib 的折线图把“每日爆款”的连续增长趋势画出来每天对比着看感觉非常直观。这一步不是必须的但对数据敏感度培养有很大帮助。7. 实测里最常踩的四个坑限流、Token 失效、字段缺失与时区错位7.1 请求频率一高就被限流第一次把分页循环跑全量时我翻页翻得特别快结果到第三页直接返回 429 Too Many Requests。这个状态码的意思是“请求太频繁请歇一会儿”。后来我在响应头里看到了 PH 的限流信息X-RateLimit-Remaining会告诉你还剩几次额度。处理方案是加一个带重试退避的请求封装def request_with_retry(session, payload, retries3): for attempt in range(retries): resp session.post(PH_API_URL, jsonpayload, headersbuild_headers(), timeout15) if resp.status_code 429: wait int(resp.headers.get(Retry-After, 5 * (attempt 1))) time.sleep(wait) continue resp.raise_for_status() return resp.json() raise RuntimeError(request failed after retries)另外就是给自己定死规矩分页循环每页之间至少sleep(1)如果只想抓前 50 条就绝不贪多。爬虫慢一点不会死但被封了数据就拿不到了。7.2 Token 失效401 的坑比想象中频繁Product Hunt 的 Developer Token 不是永久有效的。我第一次写了挺久隔了一周再跑直接 401 Unauthorized。这个坑不踩一次很难意识到因为错误信息非常模糊。处理方式就三步第一步在请求封装里先判断status_code 401不要再盲目重试否则会把有限的重试次数浪费在无效 token 上第二步去开发者后台检查 token 状态刷新后更新环境变量第三步把 token 放在config.py里从环境变量读取而不是硬编码。这样 token 失效时你只需要在服务器上改环境变量不需要去翻代码。7.3 字段缺失与数据结构变化容错不是可选项GraphQL 的好处是按需取字段但坏处是后端加字段、改字段名非常频繁。我的第一版只跑了三天就遇到了description字段为空的产品以及某个产品的topics边缘列表返回了空数组。如果你用的是node[description]遇到空值直接 KeyError整批数据入库失败。所以我在解析层做了一层完整容错所有字段全部走.get()并且给出默认值。与此同时我在normalize_post里会先检查最外层的errors字段如果 GraphQL 本身报了错误宁可抛出异常让脚本停下来也不要带着残缺数据入库。7.4 时区错位为什么你的“每日”会差一天这个坑最隐蔽。Product Hunt 的createdAt是 ISO8601 格式带时区信息的你的服务器可能在东八区可能在欧洲如果你直接用datetime.now()去算“今天”那么同一个 UTC 时间在不同时区的人看来可能是两个不同日期。我的统一做法是所有入库的created_at和collected_date全部以 UTC 为准展示给用户看时才转成本地时区。具体到代码里就是from datetime import datetime, timezone utc_today datetime.now(timezone.utc).strftime(%Y-%m-%d)不要用datetime.now().strftime(...)。这个细节如果不对你在东八区早上 8 点跑任务抓到的“今天”的数据可能还是美国那边的昨天增量算出来全是负数看起来就像产品集体掉赞实际上只是日期边界错位了。这个项目我断断续续跑了两个多月最大的体会不是写代码有多难而是“自动化之后你才有时间思考”。前一个星期几乎都在调接口、跟字段、改存储后面每天只需要扫一眼报表爆款产品背后的规律慢慢就在脑子里形成直觉了。我的建议是先把今天这套流程跑通别追求花哨坚持收两周数据你再看那些产品的共同点一定比刷一个月首页更有体感。如果你把它当成学习项目一定自己从零写一遍尤其是 SQL 增量那段——手写一遍和看一遍完全不是一回事。 用Python爬虫打造Product Hunt每日爆款榜单
如果你也是那种每天不刷一遍 Product Hunt 就浑身不自在的人一定懂这种感觉首页翻到第四五屏全是差不多的 AI 工具等晚上再回来看真正涨势凶猛的产品早被淹没在海量更新里了。手动盯能盯出感觉但效率太低还容易漏。我做的这个 Python 爬虫就是奔着“自动化”三个字去的——每天早上自动把 Product Hunt 的产品数据抓回来存好再用增量算法跑出一份“今日爆款”名单。你不用熬夜蹲首页每天早上花三分钟看结果就行。整个项目用到的都是 Python 爬虫里最基础、但恰恰也最容易被忽略的能力请求公开接口、解析结构化数据、用 SQLAlchemy 落库、用 Crontab 定时调度、用 SQL 做增量分析。难度曲线比较平缓适合刚学完 requests 和 SQL 基础的人当第二个练手项目对已经在做产品、做市场、做独立开发的人同样有用因为每天一份 Product Hunt 爆款榜单本质上就是一份免费的市场调研报告。在正式开始之前有一句话我必须放在最前面爬虫技术本身没有问题有问题的永远是使用方式。现在的网站普遍有使用条款、robots 协议和数据保护要求抓公开数据做个人学习、做非商业分析和抓取登录后数据再付费转卖完全是两码事。后面所有代码都只基于 Product Hunt 公开 API 暴露的数据也请读者自己把握好边界。1. 选对练手对象这个项目真正让你掌握的能力与合规红线1.1 这个项目到底让你练会什么很多人练爬虫喜欢挑一些比较“重”的网站比如电商、社交平台、视频网站一上来就要处理登录态、验证码、字体反爬、JS 逆向结果写了三天还在跟反爬机制搏斗项目热情直接清零。Product Hunt 完全不一样它有结构清晰的后端接口返回的数据字段完整、含义明确非常适合把精力放在“爬虫该有的完整链路”上。这个链路就是数据源分析 → 请求采集 → 解析清洗 → 入库存储 → 定时调度 → 数据分析。每一步都能迁移到其他项目里。比如这套代码把 API 地址换成你内部系统的接口稍微改下字段映射就是一个内部数据采集工具把 SQLAlchemy 模型换成 JSON 文件又变成了一个轻量爬虫脚本。真正值钱的不是“会 requests”这个动作而是你脑子里有没有这条链路。我建议的练手路线是先能跑通单次采集再手动跑几天攒数据最后再上定时任务和增量分析。别一上来就追求分布式、可视化管理平台那些东西等你有真实数据量了自然会发现需求没有数据量的时候搭出来全是空架子。1.2 合规红线爬虫第一课不是技术而是边界“因爬虫入狱”这个词频繁上热搜不是没有原因的。这个领域每年都有人因为越界把自己搭进去最常见的翻车方式就几种抓了需要登录才能看的数据、绕过反爬机制抓源码层的内容、把抓来的数据传输给第三方牟利。任何一个只要踩中性质就完全不同了这不是用“学习目的”能解释过去的。所以我在这个项目里给自己定的规则很简单只采集公开接口、公开页面能直接看到的数据不碰任何需要登录态的内容请求频率控制在“人肉浏览”级别宁可慢一点绝不当炮轰请求头里写清用途和联系方式至少让对方知道我是一个可追溯的自动化脚本数据只用于个人学习、非商业分析不转售、不批量复制、不对外公开原始数据集。这套规则同样适用于你拿这套代码去套别的网站。判断标准就一句话如果网站所有者看到你的爬虫行为会感到不舒服那这件事大概率越界了。技术是工具边界是人定的别让工具替你决定风险。2. 别急着写代码先把 Product Hunt 的数据源结构摸清楚2.1 从页面反推接口别直接去解析 HTML刚入门的时候很多人拿到一个网站第一反应就是requests.get然后交给 BeautifulSoup 解析 HTML。这套路对付静态页面没问题但 Product Hunt 首页是前端渲染的你直接拿 HTML 会发现内容很少产品卡片全是动态渲染进去的。与其绕一大圈不如先打开浏览器开发者工具刷新页面在 Network 面板里找 XHR 请求。操作路径很简单刷新首页 → Network 面板 → 过滤 XHR 或 Fetch → 看哪个请求返回了产品数据。Product Hunt 走的是 GraphQL接口地址是https://api.producthunt.com/v2/api/graphql请求方式是 POST返回的是 JSON。拿到这个请求以后你的爬虫目标就非常明确了模拟它而不是跟 HTML 纠缠。我把两种方案放在一起对比过差异很直观对比维度直接解析 HTML请求公开 API稳定性前端改版就挂字段相对稳定字段完整度页面不一定全展示结构化字段齐全维护成本选择器要反复改只需要维护查询语句合规风险相对较高使用官方公开接口更透明学习价值练了 BeautifulSoup接触 GraphQL更贴近现代后端2.2 核心接口长什么样需要哪些字段关于 GraphQL你不需要把它当成一个很玄的东西。你可以理解成后端给你一个“数据菜单”你想点什么菜就传什么字段它按你的要求原样返回。相比 REST 接口动辄返回一整坨无关字段GraphQL 对爬虫来说反而更友好。当时我用的查询结构大致是下面这样。GraphQL 的 schema 会不时调整如果你跑的时候报了字段错误按响应里的提示修改字段名即可query { posts(order: {createdAt: DESC}, first: 50) { edges { node { id name tagline description url votesCount commentsCount createdAt topics { edges { node { name } } } } } } }这个查询里我重点拿这几个字段id产品唯一标识去重时会用到name、tagline、description产品名称、一句话简介和完整介绍url产品官网或 PH 详情页链接votesCount、commentsCount累计点赞数和评论数这是后面算爆款的核心数据createdAt产品发布日期用于区分“今日新品”和“老产品”topics产品打的话题标签可以用来观察它属于什么赛道。3. 核心采集代码请求 GraphQL、解析 JSON 与容错兜底3.1 工程结构和配置我不建议把所有代码塞进一个main.py虽然小项目能跑但后面加定时、加分析、加存储会越来越乱。我用的目录结构长这样product_hunt_spider/ ├── config.py # 配置API地址、Token等 ├── spider.py # 请求和解析逻辑 ├── models.py # SQLAlchemy 模型 ├── pipeline.py # 入口抓取 → 解析 → 入库 ├── analysis.py # 增量排行分析 └── requirements.txt # requests、sqlalchemy依赖很简单就两个核心库pip install requests sqlalchemyconfig.py里我特意用了环境变量来保护 Token而不是把它写死在代码里import os PH_API_URL https://api.producthunt.com/v2/api/graphql PH_TOKEN os.getenv(PH_TOKEN, ) def build_headers() - dict: return { Authorization: fBearer {PH_TOKEN}, Content-Type: application/json, Accept: application/json, User-Agent: PH-Daily-Spider/1.0 (个人学习用途; 联系方式: youremail.com) }3.2 请求部分怎么写才稳我习惯用requests.Session()它能复用底层连接多次请求时效率更高也能统一设置 headers。请求体里最好把query和variables分开写后面做分页时改参数会方便很多import time import requests from config import PH_API_URL, build_headers def fetch_posts(session: requests.Session, after: str | None None, limit: int 50) - dict: query query($limit: Int!, $after: String) { posts(order: {createdAt: DESC}, first: $limit, after: $after) { edges { node { id name tagline description url votesCount commentsCount createdAt topics { edges { node { name } } } } } pageInfo { endCursor hasNextPage } } } payload { query: query, variables: {limit: limit, after: after} } resp session.post(PH_API_URL, jsonpayload, headersbuild_headers(), timeout15) resp.raise_for_status() return resp.json()timeout15很重要不加这个参数一旦网络波动脚本可能挂在那里几分钟不动。稍微解释一下resp.raise_for_status()它会在 HTTP 状态码不是 2xx 时主动抛出异常帮你把错误拦截在源头不会带病继续往下跑。3.3 解析数据能容错的解析才有资格上线GraphQL 返回的 JSON 是有固定壳子的数据在data.posts.edges每个edge里有一个nodenode才是产品对象。解析时最忌讳直接写node[votesCount]因为一旦哪个产品缺字段整个脚本就 KeyError 崩了。我写了个normalize_post函数统一做字段清洗和默认值兜底from datetime import datetime def normalize_post(node: dict) - dict: raw_created_at node.get(createdAt) created_at None if raw_created_at: try: created_at datetime.fromisoformat(raw_created_at.replace(Z, 00:00)) except ValueError: created_at None topics [] for item in node.get(topics, {}).get(edges, []): topic_name item.get(node, {}).get(name) if topic_name: topics.append(topic_name) return { ph_id: node.get(id), name: node.get(name), tagline: node.get(tagline, ), description: node.get(description, ), url: node.get(url, ), votes_count: node.get(votesCount, 0), comments_count: node.get(commentsCount, 0), created_at: created_at, topics: |.join(topics), }话题标签我用|拼成一个字符串是为了在 SQLite 里好存真要做分析时再按|拆开就行。时间字段的处理我比较谨慎createdAt是 ISO8601 格式带Z结尾先替换成00:00再用fromisoformat解析解析失败就置空绝不让一条坏数据拖垮整个流程。3.4 分页需要全量数据时怎么循环对“每日爆款”来说前 50 个热门产品基本已经覆盖了值得关注的对象所以刚开始可以只用first50。但如果你想把当天发布的产品全量抓下来那就要处理分页。GraphQL 的分页逻辑不复杂响应里会给你pageInfo.hasNextPage和pageInfo.endCursor把endCursor作为下一次请求的after参数传进去就行def fetch_all_posts(session: requests.Session, max_pages: int 5) - list[dict]: all_posts [] cursor None for _ in range(max_pages): data fetch_posts(session, aftercursor) if errors in data: raise RuntimeError(fGraphQL errors: {data[errors]}) posts data.get(data, {}).get(posts, {}) edges posts.get(edges, []) for edge in edges: node edge.get(node, {}) if node: all_posts.append(node) page_info posts.get(pageInfo, {}) if not page_info.get(hasNextPage): break cursor page_info.get(endCursor) time.sleep(1) # 保持礼貌的频率 return all_postsmax_pages是一个保险丝防止某个异常场景下脚本无限翻页刷爆接口。实测中这个循环跑得很稳关键就在于每页之间sleep(1)既保证速度又不会撞限流。4. 数据落库不是随便存SQLAlchemy 模型与每日去重4.1 为什么用 SQLAlchemy而不是直接写 SQL如果你是第一次做落库可能会觉得直接写 SQLite 的execute也很简单。但一旦字段数量多起来你就会发现代码里到处都是INSERT INTO、UPDATE改一个字段要改好几处特别容易遗漏。SQLAlchemy 的 ORM 模型把表和 Python 对象映射起来改模型就能同步改表结构对爬虫这种字段变化频繁的项目非常友好。还有一个很实际的好处开发阶段用 SQLite文件型数据库零配置等数据量大到需要多人协作、需要远程访问时把连接串从sqlite:///products.db换成postgresql://user:passhost/dbname模型代码几乎不用动。这种平滑迁移能力直接裸写 SQL 是享受不到的。4.2 表结构设计这张表怎么定义“一天一份快照”我的核心设计思路是每天给每个产品存一条快照。这样你既能查今天的总点赞数也能拿昨天和今天的数据做差值算出真实的增长量。表结构用 SQLAlchemy 2.0 的声明式写法from datetime import datetime from sqlalchemy import create_engine, String, Integer, Text, DateTime, UniqueConstraint from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker engine create_engine(sqlite:///products.db, echoFalse) SessionLocal sessionmaker(bindengine) class Base(DeclarativeBase): pass class Product(Base): __tablename__ products __table_args__ ( UniqueConstraint(ph_id, collected_date, nameuq_ph_id_collected_date), ) id: Mapped[int] mapped_column(primary_keyTrue) ph_id: Mapped[str] mapped_column(String(32), indexTrue) name: Mapped[str] mapped_column(String(200)) tagline: Mapped[str] mapped_column(String(500), default) description: Mapped[str] mapped_column(Text, default) url: Mapped[str] mapped_column(String(500)) topics: Mapped[str] mapped_column(String(500), default) votes_count: Mapped[int] mapped_column(Integer, default0) comments_count: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, nullableTrue) collected_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.now) collected_date: Mapped[str] mapped_column(String(10), indexTrue) Base.metadata.create_all(engine)重点看这行唯一约束UniqueConstraint(ph_id, collected_date)。它保证“同一个产品在同一天只能有一条记录”。这个设计直接解决了重复抓取的问题——就算你今天把这个脚本手动跑了三遍数据库里每个产品也只会保留一条快照不会出现十行一模一样的数据。那collected_date用哪一天的我直接取 UTC 日期。下面会专讲时区问题这里先记住一个原则所有“日期”都用 UTC 算展示时再转本地时区。4.3 入库逻辑主键冲突时什么都不做入库我用的是 SQLite 方言的插入语句加上on_conflict_do_nothing。它的语义很直白“能插进去就插遇到唯一键冲突就跳过”。这样脚本天然具备幂等性重复执行不会造成脏数据from datetime import datetime, timezone from sqlalchemy.dialects.sqlite import insert as sqlite_insert from models import Product, SessionLocal def save_products(products: list[dict]) - int: collected_date datetime.now(timezone.utc).strftime(%Y-%m-%d) rows [{**product, collected_date: collected_date} for product in products] session SessionLocal() try: stmt sqlite_insert(Product).values(rows) stmt stmt.on_conflict_do_nothing(index_elements[ph_id, collected_date]) session.execute(stmt) session.commit() finally: session.close() return len(rows)如果你以后换到 PostgreSQL只需要把from sqlalchemy.dialects.sqlite import insert as sqlite_insert换成from sqlalchemy.dialects.postgresql import insert as pg_insert后面调.on_conflict_do_nothing()的写法基本一致。这就是选 ORM 的底气底层存储换了业务代码纹丝不动。5. 每天早上八点自动跑Crontab 调度与日志监控5.1 定时任务我为什么选 Crontab实现“每日自动跑”有好几种方式比如用 Python 的schedule库、用 Celery 配 beat或者直接上系统级 Crontab。我的选择是无脑 Crontab理由很简单它不需要常驻进程没有额外依赖系统自带对“每天跑一次”这种低频任务来说是最轻量的方案。schedule库虽然写起来好看但你的 Python 脚本必须一直挂着不退出资源浪费不说服务器一重启就失效。Celery 是给大规模分布式任务用的这个场景去上 Celery 属于杀鸡用牛刀。Crontab 的缺点是人机交互不太友好配置写错了没有直观提示但只要你掌握了五段式格式它就是最可靠的定时器。5.2 Crontab 配置怎么写先运行crontab -e打开当前用户的定时任务表然后加一行0 8 * * * cd /home/ubuntu/product_hunt_spider /usr/bin/python3 pipeline.py /var/log/product_hunt/ph_spider.log 21五段式从左到右分别是分钟、小时、日、月、周。这行的意思就是每天早上 8 点整切到项目目录用系统 Python 执行pipeline.py标准输出和错误输出都追加到日志文件里。命令里cd那一步很多人会忘。Crontab 执行命令时默认工作目录是用户目录不cd进去的话脚本里所有相对路径都会指错地方比如sqlite:///products.db这个相对路径就会建到别的位置去。我吃过这个亏所以后来配置文件全用绝对路径Crontab 里强制先cd到项目目录。日志路径如果不存在记得先建好mkdir -p /var/log/product_hunt有了日志文件脚本真的挂了你也知道去哪里看现场。我一般还会加一句date到日志里方便确认每次任务真实执行时间。5.3 本机跑还是服务器跑如果你只是自己用在本机跑也行但前提是电脑一直开着、而且不能休眠。这很不现实所以我后来直接放到一台低配轻量云服务器上一个月成本很低资源占用也小。整个项目跑起来的内存占用不到 200MB基本不挤占其他服务的资源。Token 在服务器上不要写进代码仓库用环境变量注入export PH_TOKEN你的_token然后手动执行一遍脚本确认日志正常再配置 Crontab。千万别一上来就配定时任务否则你根本分不清是定时配置写错还是脚本本身有 bug。6. “爆款”不是拍脑袋用 SQL 算增量、做排行6.1 增量比绝对量更说明问题很多人拿到votesCount就直接排序这有一个坑它叫“累计点赞数”不是“今天新增点赞数”。一个上周上线、已经攒了三千赞的老产品和一个今天刚上线、两小时涨了两百赞的新品放在一起排序老产品永远靠前但论“此刻的势头”新品明显更值得关注。所以我的分析逻辑是用今天的数据减去昨天的数据得到 24 小时增量。这个差值才是“爆款潜力”最直接的指标。也正是因为要算差值前面才会设计成“每天存一条快照”数据积累两三天后增量分析才能跑起来。6.2 两个榜单新品榜和增速榜我会把它拆成两个维度看这样信息更干净。第一个是“今日新品榜”判断标准是createdAt在最近 24 小时内按累计赞数和评论数排序第二个是“24 小时增速榜”用今天和昨天两张快照 join 起来算每个产品的点赞增量和评论增量。增速榜的 SQL 大概是这样的from datetime import datetime, timedelta, timezone import pandas as pd from sqlalchemy import text from models import engine today datetime.now(timezone.utc).strftime(%Y-%m-%d) yesterday (datetime.now(timezone.utc) - timedelta(days1)).strftime(%Y-%m-%d) sql text( WITH today AS ( SELECT ph_id, name, url, votes_count, comments_count FROM products WHERE collected_date :today ), yesterday AS ( SELECT ph_id, votes_count, comments_count FROM products WHERE collected_date :yesterday ) SELECT t.name, t.url, t.votes_count - COALESCE(y.votes_count, 0) AS votes_delta, t.comments_count - COALESCE(y.comments_count, 0) AS comments_delta FROM today t LEFT JOIN yesterday y ON t.ph_id y.ph_id WHERE t.votes_count - COALESCE(y.votes_count, 0) 0 ORDER BY votes_delta DESC, comments_delta DESC LIMIT 20; ) df pd.read_sql(sql, engine, params{today: today, yesterday: yesterday})这段 SQL 里LEFT JOIN的写法值得多看两眼如果某个产品今天上榜但昨天不在榜上yesterday表里没有它的记录COALESCE会把昨天的赞数补成 0。这种产品通常是“今天刚发布”的新品它的votes_delta实际等于累计赞数出现在增速榜顶部是正常的不影响判断。6.3 一个可调的“爆款系数”公式单纯按点赞增速排容易忽视讨论热度高的产品。一个产品赞数中等但评论区吵翻了天往往说明用户参与度极高这种产品后劲往往更足。我后来加了一个经验公式把评论的权重提上去hot_score votes_delta * 1 comments_delta * 5 topic_bonus解释一下两个参数思路comments_delta * 5写评论的成本比点一个赞高得多用户愿意花这个成本说明产品引发了深度讨论我给 5 倍权重topic_bonus话题标签多于 3 个的产品说明运营在不同赛道做了分发我给 5 到 10 分的额外加成。这不是官方指标只是我自己的经验公式权重完全按你的需求调。比如你做 B 端产品调研可能更看重 tagline 和描述里的关键词你做增长分析可能想拿评论增量和点赞增量对比判断讨论转化效率。数据已经存进库里了怎么算都行。6.4 把榜单变成每天早上能看的东西我每天早上会收到一份精简版报表其实就是把上面这个 DataFrame 输出成 Markdown 文件再配合一个简单的邮件发送脚本。如果你还需要可视化加一个 pandas matplotlib 的折线图把“每日爆款”的连续增长趋势画出来每天对比着看感觉非常直观。这一步不是必须的但对数据敏感度培养有很大帮助。7. 实测里最常踩的四个坑限流、Token 失效、字段缺失与时区错位7.1 请求频率一高就被限流第一次把分页循环跑全量时我翻页翻得特别快结果到第三页直接返回 429 Too Many Requests。这个状态码的意思是“请求太频繁请歇一会儿”。后来我在响应头里看到了 PH 的限流信息X-RateLimit-Remaining会告诉你还剩几次额度。处理方案是加一个带重试退避的请求封装def request_with_retry(session, payload, retries3): for attempt in range(retries): resp session.post(PH_API_URL, jsonpayload, headersbuild_headers(), timeout15) if resp.status_code 429: wait int(resp.headers.get(Retry-After, 5 * (attempt 1))) time.sleep(wait) continue resp.raise_for_status() return resp.json() raise RuntimeError(request failed after retries)另外就是给自己定死规矩分页循环每页之间至少sleep(1)如果只想抓前 50 条就绝不贪多。爬虫慢一点不会死但被封了数据就拿不到了。7.2 Token 失效401 的坑比想象中频繁Product Hunt 的 Developer Token 不是永久有效的。我第一次写了挺久隔了一周再跑直接 401 Unauthorized。这个坑不踩一次很难意识到因为错误信息非常模糊。处理方式就三步第一步在请求封装里先判断status_code 401不要再盲目重试否则会把有限的重试次数浪费在无效 token 上第二步去开发者后台检查 token 状态刷新后更新环境变量第三步把 token 放在config.py里从环境变量读取而不是硬编码。这样 token 失效时你只需要在服务器上改环境变量不需要去翻代码。7.3 字段缺失与数据结构变化容错不是可选项GraphQL 的好处是按需取字段但坏处是后端加字段、改字段名非常频繁。我的第一版只跑了三天就遇到了description字段为空的产品以及某个产品的topics边缘列表返回了空数组。如果你用的是node[description]遇到空值直接 KeyError整批数据入库失败。所以我在解析层做了一层完整容错所有字段全部走.get()并且给出默认值。与此同时我在normalize_post里会先检查最外层的errors字段如果 GraphQL 本身报了错误宁可抛出异常让脚本停下来也不要带着残缺数据入库。7.4 时区错位为什么你的“每日”会差一天这个坑最隐蔽。Product Hunt 的createdAt是 ISO8601 格式带时区信息的你的服务器可能在东八区可能在欧洲如果你直接用datetime.now()去算“今天”那么同一个 UTC 时间在不同时区的人看来可能是两个不同日期。我的统一做法是所有入库的created_at和collected_date全部以 UTC 为准展示给用户看时才转成本地时区。具体到代码里就是from datetime import datetime, timezone utc_today datetime.now(timezone.utc).strftime(%Y-%m-%d)不要用datetime.now().strftime(...)。这个细节如果不对你在东八区早上 8 点跑任务抓到的“今天”的数据可能还是美国那边的昨天增量算出来全是负数看起来就像产品集体掉赞实际上只是日期边界错位了。这个项目我断断续续跑了两个多月最大的体会不是写代码有多难而是“自动化之后你才有时间思考”。前一个星期几乎都在调接口、跟字段、改存储后面每天只需要扫一眼报表爆款产品背后的规律慢慢就在脑子里形成直觉了。我的建议是先把今天这套流程跑通别追求花哨坚持收两周数据你再看那些产品的共同点一定比刷一个月首页更有体感。如果你把它当成学习项目一定自己从零写一遍尤其是 SQL 增量那段——手写一遍和看一遍完全不是一回事。