Python爬虫+SQLAlchemy：GPTs市场增量数据监控与需求分析实战

每天早上八点服务器上的定时任务会把昨天新上线的 GPTs 记录一次性抓到数据库里跑完大约需要两分钟。这件事我坚持做了一个月每天能稳定捡到 80 到 150 条新需求。很多人看到这个标题以为重点在爬虫但其实项目的核心是另一个词——需求。我用 Python 写这个 GPTs 市场爬虫真正的目的不是堆数据而是想知道每天都有哪些人在用哪些方向的新东西、解决什么问题。如果你是做 AI 产品、独立开发或者内容创业的这个思路值得参考。爬虫本身不难难的是你怎么从每天一百条新数据里找到真正值得投入的缝隙。这篇文章会把整套方案拆开讲清楚数据接口怎么找、SQLAlchemy 怎么存、定时任务怎么设计、上线后踩了哪些坑以及最后怎么把数据变成需求判断。1. 为什么我会去爬 GPTs 市场先聊清楚这 100 个“需求”到底值多少1.1 GPTs 市场是一个难得的“需求样本池”先说一个很反直觉的观察GPTs 市场里大量上架的作品作者根本不是专业开发者。很多人只是把一段写得还算完整的系统提示词包装一下加上一两个 Actions 接口就成了一个新 GPTs。技术门槛低带来的直接结果就是——所有被生活和工作卡住的人都有能力把自己的需求直接实物化。这就有意思了。如果一个普通用户愿意花半小时把一个想法做成 GPTs 丢到公网上说明这个需求不是他凭空想出来的而是真实存在、并且让他困扰了很久的。一个人做合同审查机器人背后是大量法务、创业者、中小企业主在处理合同上的重复劳动十个人做小红书文案助手背后是整个内容生态的焦虑。GPTs 市场因此成为一面镜子照出的是普通人在 AI 时代最急着被解决的麻烦。我从这个市场里盯的也不是某个具体的爆款 GPTs而是“每天新增的东西都在往哪些方向聚集”。标题里说的“100 个新需求”其实就是每天新上线的 80 到 150 个 GPTs 背后所代表的用户痛点集合。1.2 爬虫的设计目标每天增量而不是拉全量刚开始我的想法很简单一次性把 GPTs 市场全部数据拉下来存好慢慢分析。试了一轮之后立刻放弃了。全量数据是一个静态快照存下来确实很厚实但它回答不了最核心的问题——什么方向正在变热。某个方向的 GPTs 数量从 20 变成 80这中间的过程比 80 这个结果重要得多。所以我把目标改成了增量监控每天跑一次抓取过去 24 小时新上线或明显更新的记录。这个“每天自动捡”的机制才是整个项目真正的核心价值。增量数据是一条连续的需求流拉到足够长的时间之后你能看到某个需求从出现、增长到拥挤的全过程。整个爬虫的设计目标因此浓缩成一句话稳定、自动、可增量、可统计。我不追求抓取速度有多快也不追求把每个字段都完整拿到只追求每天早上准点拿到一批干净、可复用、能支撑后续分析的数据。1.3 必须说清楚的边界只抓公开数据控制频率跑这个项目之前我给自己定了三条规矩。第一只抓公开可见的列表信息包括名称、描述、作者、时间、分类这些展示在公网页面上的字段不碰登录态不碰个人私密数据。第二请求频率严格控制单个接口访问间隔保持在秒级不给目标服务器造成压力。第三解析逻辑对字段变化保持宽容字段缺失时直接跳过而不是崩溃。这三条规矩不是道德说教而是让这个爬虫能长期跑下去的前提。抓取频率太高结果就是 IP 被限制项目中断访问带鉴权的数据合规风险陡增而且技术上也不可持续。我把这个原则写进了项目 README 的第一行这是一个公开数据的观察工具不是一个抢数据的轰炸机。2. 数据从哪来先找到那个返回 JSON 的接口再谈爬虫2.1 列表页是动态渲染的直接 requests 拿不到内容第一版脚本我用 requests 直接请求 GPTs 市场的首页心想最多带个分页参数就能搞定。结果拿回来的 HTML 里几乎没有有用的内容只有一堆空的 div 和压缩过的 JavaScript 代码。这是典型的单页应用SPA结构页面框架先返回数据由浏览器里的 JS 在渲染阶段再向后端接口请求。不懂这个机制的人在这里就会卡住容易往“上无头浏览器模拟渲染”的方向走。无头浏览器比如 Playwright、Selenium确实能解决动态渲染的问题但它在这个场景下是过度设计启动浏览器实例非常吃内存一页一页翻又慢又容易被识别。我当时的判断是先打开浏览器开发者工具切到 Network 面板然后手动翻几页列表。翻页过程中能看到很多网络请求绝大多数是图片、静态资源和埋点日志。真正有价值的请求特征是返回 JSON 而不是 HTMLURL 里带着分页和排序参数响应速度快。找到这个接口之后用 requests 直接请求它基本就能拿到结构清晰的数据。这一步是整个爬虫的地基花上一个小时把它确认清楚后面会省很多事。2.2 接口返回结构与字段识别这个返回 JSON 的接口响应的结构大致长这样{ data: { total: 12345, items: [ { id: g-abc123, name: 小红书文案小助手, description: 帮小红书博主生成种草文案和标题, author: { name: user_001, id: u_123 }, created_at: 2025-06-10T08:30:00.000Z, updated_at: 2025-06-10T09:15:00.000Z, categories: [content, marketing] } ] } }我用黑名单的思路来看这些字段名称和描述是分析的核心所有需求判断都从这里提取created_at 用来识别新上线updated_at 用来捕捉老作品的改动categories 是官方分类后面做聚合时可以直接用id 是整个表的自然主键去重靠它。description 这个字段包含的信息量最大。很多 GPTs 作者会直接在描述里写“帮我做某件事”这段话就是用户需求最真实的表达。后面做词频分析时最大的词源就在这里。2.3 请求参数与请求头让请求看起来像一个正常访客接口有了接下来就是构造请求。参数层面通常离不开三个分页游标offset 或 page、单页数量limit、排序规则sort。这里的 sort 很关键增量爬虫必须按时间倒序拉取才能保证游标逻辑不会乱。请求头是我踩过的一个小坑。第一版脚本只带了 User-Agent结果请求少数几次之后就开始收到异常状态码最后发现是没有带 Accept 头。很多 JSON 接口对请求头有隐性要求我现在的固定配置是这几项import requests session requests.Session() session.headers.update({ 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, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, }) params { limit: 50, offset: 0, sort: newest, } resp session.get(https://example.com/api/gpts_public, paramsparams, timeout15) data resp.json()用 requests.Session 而不是每次新建请求对象是为了复用底层连接效率和稳定性都会好一些。这个细节在小数据量时感觉不到分页抓几百条之后就能明显感知到连接建立的时间省下来了。2.4 一个小细节时间戳是 ISO 字符串不是 Unix 时间戳观察接口返回时我还注意到一个容易忽略的细节created_at 和 updated_at 返回的不是常见的 Unix 时间戳而是“2025-06-10T08:30:00.000Z”这种带时区信息的 ISO 8601 字符串。这个细节一开始没当回事想着反正存到数据库里之前可以转成 Python 的 datetime。结果后面做每日统计的时候被它坑了一次——如果你直接把这个字符串塞进 DateTime 字段数据库里存的是“无时区”的本地时间概念和程序里跑的本地时间混在一起统计当天新增时就会差出 8 个小时。正确的做法是解析时显式保留时区信息统一存成 UTC显示的时候再转本地时间这个后面存储章节会专门展开。3. 存储层设计SQLAlchemy 把那一天的数据沉淀下来3.1 表结构设计一列都不能浪费数据拉到内存里只是第一步不落到存储层就无法做时间维度的分析和增量判断。我选 SQLite SQLAlchemy 的组合零配置就能跑单机项目完全够用如果哪一天数据量大到 SQLite 扛不住SQLAlchemy 的连接层可以相对平滑地切到 PostgreSQL代码基本不用大改。建表时我用了下面这个模型from datetime import datetime, timezone from sqlalchemy import create_engine, Column, String, Text, DateTime, JSON from sqlalchemy.orm import declarative_base, sessionmaker Base declarative_base() class Gpts(Base): __tablename__ gpts_snapshot id Column(String(64), primary_keyTrue) name Column(String(255), nullableFalse) description Column(Text, default) author_name Column(String(255), default) author_id Column(String(64), default) categories Column(JSON, defaultlist) created_at Column(DateTime, nullableFalse) updated_at Column(DateTime, nullableFalse) first_seen_at Column(DateTime, defaultlambda: datetime.now(timezone.utc)) batch_no Column(String(32), default20250610) engine create_engine(sqlite:///gpts.db, echoFalse) Base.metadata.create_all(engine) SessionLocal sessionmaker(bindengine)逐列说下设计意图。id 是自然主键直接用接口返回的 GPT 标识一张表只能有一条同 id 记录去重靠它不需要额外自增主键。name、description、author 这些是分析字段描述必须用 Text 类型不能设成 String(255)因为有些 GPTs 的描述能写很长。created_at 和 updated_at 存的是对象在平台上的生命周期时间first_seen_at 是“我第一次抓到它”的时间这个字段是“每天新增”统计的基础很多人会忽略它。batch_no 表示本次抓取批次相当于给每条数据盖一个当天的戳方便回溯某个批次的数据质量。3.2 写入策略有则更新没有则插入爬虫的写入模式和普通应用不太一样——大部分记录第二次抓取时已经存在了只有少量是真正的新条目。如果每次全量删了重插等于把所有数据的 created_at 历史都搞丢了增量分析就成了笑话。SQLAlchemy 的 merge 可以自动根据主键判断该插入还是更新但我更习惯手动写这个分支因为可以把“哪些字段允许被更新、哪些字段不允许”控制得更精细。比如 created_at 是平台给的属于一次性数据更新时不应该被覆盖而 name、description、updated_at 这些随时可能变抓到新值就应该覆盖。def upsert_items(items, batch_no): session SessionLocal() try: for item in items: gpt session.get(Gpts, item[id]) if gpt is None: gpt Gpts( iditem[id], nameitem[name], descriptionitem.get(description, ), author_nameitem.get(author, {}).get(name, ), author_iditem.get(author, {}).get(id, ), categoriesitem.get(categories, []), created_atparse_iso(item[created_at]), updated_atparse_iso(item[updated_at]), batch_nobatch_no, ) session.add(gpt) else: # 只允许更新的字段 gpt.name item[name] gpt.description item.get(description, ) gpt.categories item.get(categories, []) gpt.updated_at parse_iso(item[updated_at]) gpt.batch_no batch_no session.commit() except Exception: session.rollback() raise finally: session.close()解析时间的函数也一起放这里def parse_iso(s): return datetime.fromisoformat(s.replace(Z, 00:00))batch_no 我固定用当天日期字符串生成每天跑批时传进去。这样哪怕某天抓的数据有问题也能通过批次号快速定位、回滚或重跑。3.3 容易被忽略的时区问题所有时间统一存 UTC第三章开头提到的时区问题具体故事是这样的上线第二天我看数据库里 created_at发现早上 8 点抓的任务统计“今日新增”时数量少得可疑。排查后发现接口返回的是 UTC 时间“2025-06-10T08:30:00.000Z”而我解析时用了本地时间语境存进去之后再用本地时间的当天边界去过滤等于每天少了 8 小时的数据窗口。我的处理方案是程序里所有时间戳一律用 datetime.now(timezone.utc) 生成接口回来的时间统一转成带 UTC 时区的 aware datetime 再存储需要做“当天统计”的时候先把“今天凌晨 00:00:00”换算成 UTC 对应的时间点再拿这个边界去数据库里比较。这样无论服务器在哪个时区统计口径都不会乱。4. 定时与增量让爬虫“每天自动”跑起来的三个关键点4.1 增量判定的底层逻辑游标 主键双保险定时任务的核心问题只有一个怎么知道哪些是“新的”我用了两层机制。第一层是请求游标。每次跑批开始前从数据库里查一个 max_created_at把它当作这次抓取的起点。接口按时间倒序返回数据我逐页往下翻直到某一条记录的 created_at 小于或等于上次游标为止就不再继续翻页。这个机制让每天的请求量稳定在几页以内不会越跑越重。第二层是主键去重。游标依赖 created_at 字段但这个字段可能有误差——比如某作者把自己的旧作品改了改名重新上架created_at 可能看起来是新的但 id 可能还是同一个。所以存储层必须用数据库主键做二次校验重复 id 直接走更新逻辑。游标控制“请求哪些数据”主键控制“入库哪些数据”两者配合才完整。from sqlalchemy import func def get_cursor(session): last session.query(func.max(Gpts.created_at)).scalar() return last or datetime(2020, 1, 1, tzinfotimezone.utc)实测下来这套逻辑非常稳每天新增量基本都能对上平台实际的上新节奏。如果某天数量异常少先查游标是不是被一条“旧记录后补”的数据污染了再查接口是否改了返回排序。4.2 定时方案选型cron、APScheduler 还是云函数定时任务我给了自己三个选项。第一是 Linux 自带的 cron最轻量一条配置就能跑但缺点是任务一旦挂掉只能靠日志发现没有重试机制。第二是 Python 的 APScheduler可以在代码里管理任务、设置时区、控制异常适合想统一管理所有定时逻辑的人。第三是 GitHub Actions 这类云端定时方案不需要自己的服务器但前提是仓库权限允许而且每次跑批要重新拉代码装依赖灵活性差一些。我最后选了 cron Python 脚本的组合脚本内部自己做异常捕获和重试cron 负责每天 8 点把它叫醒。0 8 * * * cd /home/user/gpts-tracker /usr/bin/python3 daily_track.py logs/daily.log 21如果你更习惯在代码里管理全部逻辑APScheduler 的最小配置也很简单from apscheduler.schedulers.blocking import BlockingScheduler def daily_run(): print(start tracking...) # 爬取 入库 print(done) scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job( daily_run, triggercron, hour8, minute0, iddaily_gpts_tracker, ) scheduler.start()选 cron 的主要原因是这个项目只有一个任务没必要为了一个任务引入常驻进程。如果你有多套爬虫任务需要统一管理调度和失败重试APScheduler 更合适。4.3 异常兜底让任务“反脆弱”定时任务最怕的其实不是抓不到数据而是“静默失败”——脚本崩了日志没记录第二天早上你看到的还是昨天那批数据还以为一切正常。我的做法是在脚本里做几层保护。第一层每个阶段都有日志。请求接口前记一条“batch started”入库结束后记一条“batch finished, inserted X, updated Y”。第二天早上只需看结束日志里的数字就能判断昨晚跑没跑正常。第二层主流程包 try/except异常信息写入单独的 error 表而不是让堆栈直接打到终端就消失。第三层给每次请求设置超时和重试超时时间 15 秒连续失败 3 次就放弃本轮并留下告警。告警通道我用的是一个很轻的方案把失败信息发到一个群机器人 Webhook。这样做的好处是半夜任务挂了也能第一时间知道不用等早上手动看日志。5. 上线一个月我踩过的四个坑5.1 第一轮就抓了 5000 条增量判断全是废的项目启动第一天我想着先把市场现有的数据全部拉下来存好于是写了一个单独的历史数据 dump 脚本翻开所有分页一口气抓了 5000 多条入库。抓完确实挺爽但第二天做增量分析时就发现问题了按“新增时间在今天”过滤捞出来的不到 20 条。原因很简单历史 dump 会把所有老条目都打上 batch_no而我在设计表结构时没有区分“这是历史存量”和“这是今天增量”。历史数据的 created_at 天然分散在过去几个月统计当天新增时它们当然不会被算进去但因为数据量太大人为干扰了游标判断。解决办法是把两个任务彻底拆开历史 dump 是一个只跑一次的一次性脚本输出到单独的表日常增量是另一个每天跑的脚本两者互不干扰。你也别想着偷懒合并混在一起排查时非常痛苦。5.2 请求频率太高第三天碰壁第一版爬虫为了快点抓完基本是 0.1 秒一个请求地翻页。跑了三天某个时段开始接连返回异常状态码请求直接被挡了。我当时第一反应是代码坏了排查一圈发现完全没问题单纯是请求太密集触发了限流。处理方案也很简单把请求间隔提升到每页 3 到 5 秒并加一个随机抖动避免请求节奏过于规律。用 time.sleep(random.uniform(3, 5)) 替换掉固定 sleep。当天晚上调整完第二天数据恢复正常。用 Requests Session 复用连接对限流也有帮助因为 TCP 连接握手次数少了行为更像一个真实浏览器。5.3 字段结构会变解析时不要硬编码路径跑了大概两周某个早上起来发现入库的新增数量突然降为 0。去查日志请求成功、返回也有数据但 Python 脚本在解析时崩了。原因是一个条目的 description 字段变成了 null而我的代码里写的是 item[description]取不到键就直接 KeyError。我以前写爬虫也习惯直接 item[字段名] 这种硬编码写法因为快。但在一个要长期跑的增量项目里这种方式太脆弱了。我改成用一个安全取值的工具函数def get_field(item, path, defaultNone): cur item try: for key in path: if not isinstance(cur, dict): return default cur cur[key] return cur except (KeyError, TypeError, IndexError): return default解析描述时用 get_field(item, [description], ) 代替 item[description]哪怕字段缺失也只是拿到空字符串不会让整个任务崩溃。后来 categories 结构调整过一次这个工具函数让我免于又一次紧急修复。5.4 跨越日期边界时区让“今天的新增”统计错乱这是上一个坑的延续。某天早上看统计报表发现“今日新增”的数量比预期的少一截。查了二十分钟最后定位到是时区判断的问题接口返回的 created_at 是 UTC 时间和我数据库里存的 aware datetime 比对时我用了本地时区“今天零点”作为边界结果凌晨 0 点到 8 点之间平台新增的数据被我排除在了“昨天”的区间里。修复方式是统一口径。写一个函数把本地时区的任意时间点转成 UTC 再查库from datetime import datetime, timezone, timedelta def local_midnight_utc(offset_hours8): local_now datetime.now(timezone(timedelta(hoursoffset_hours))) local_midnight local_now.replace(hour0, minute0, second0, microsecond0) return local_midnight.astimezone(timezone.utc)每天统计时取这个函数返回的 UTC 零点作为过滤条件问题就消失了。这个坑非常经典所有涉及跨时区定时统计的项目都会遇到建议你从第一天就把时区策略定死。6. 从一百条新数据到“需求判断”爬虫价值的最后一公里6.1 把标题和描述变成词频信号数据入库后如果只是每天看一眼数量那爬虫的价值基本为零。我每天固定会跑一个小分析脚本把当天新增 GPTs 的 name 和 description 拼成一段文本用 jieba 分词过滤掉“助手”“机器人”“工具”“GPT”“智能”“自动”这类没有区分度的词统计 Top 30 词频。import jieba from collections import Counter stopwords set(助手 机器人 工具 GPT AI 智能 自动 生成 一个 可以 帮您 支持.split()) words Counter() for gpt in today_gpts: text f{gpt.name} {gpt.description} words.update(w for w in jieba.cut(text) if len(w) 2 and w not in stopwords) for word, count in words.most_common(30): print(word, count)这一步的输出就是那天“大家集中在做的事情”。比如某天排名靠前的是“小红书”“文案”“视频”“合同”说明这波新增需求集中在这几个场景。词频本身不完美但它是最快把一百条零散标题压缩成几个方向的方法。6.2 双维度聚合官方分类和自建分类交叉验证光看词频还不够我会再按分类做两次聚合。第一次用接口返回的 categories 字段统计官方网站自己划分的类别下每天有多少新增能看到内容、营销、编程、效率这些大类的新增趋势。第二次自建一套规则分类比如标题或描述里出现“小红书/抖音/快手/公众号”就归为“社媒运营”出现“合同/条款/协议/法律”就归为“法务文本”出现“Excel/表格/公式”就归为“表格处理”。rules { 社媒运营: [小红书, 抖音, 快手, 视频号, 公众号, 微博], 法务文本: [合同, 条款, 协议, 法律, 合规], 表格处理: [excel, 表格, 公式, wps, 电子表格], 销售增长: [销售, 获客, 报价, 成交, 客户, crm], } def classify(gpt): text (gpt.name gpt.description).lower() for category, keys in rules.items(): if any(k in text for k in keys): return category return 其他两组维度交叉看能发现官方分类比较抽象自建分类更贴近真实业务。比如官方分类叫“Marketing”你只看这个看不出大家在营销的哪个环节卡住自建规则拆开之后你就会发现“小红书文案”和“报价单”是两类完全不同的需求。6.3 找“需求密度高但供给少”的方向爬虫数据的价值不是让你看到哪里有热闹而是让你看到哪里热闹但没被满足。判断方法是把高频词和实际 GPTs 列表并列着看词频排名前 30 的方向如果对应的 GPTs 数量才几个这就是一个潜在机会。打个比方如果某天新增里有 15 个都是“Excel 公式生成”但仔细看这 15 个的 description几乎全部停留在“帮你写 VLOOKUP 和 IF 公式”这个层面。这时候你就能判断出表格处理本身是高需求场景但发言大多很浅。如果你能做出一个“直接上传 CSV 返回清洗方案”的细分 GPTs就是在高需求低供给的空隙里插进去了。这个筛选逻辑不能全自动化需要人工过一遍。我会每周把所有高频词拉出来挑三个方向逐一打开对应的 GPTs 看描述记录它们的共通点和空白点。周复一周“需求密度高但供给少”的方向会越来越清晰。6.4 一个实操案例一周数据找到的细分方向举一个真实发生的小例子。跑了一周后我注意到一个有意思的现象销售相关的新增 GPTs集中在“报价单生成”和“客户跟进提醒”两个点但描述里的高频词却是“打电话”“逼单”“CRM 记录”。当时的存量 GPTs 里几乎没有一个专门做“客户跟进日报自动生成”的——大家都在给单点做工具没人把“跟进过程管理”串起来。基于这个观察我拿当天的新增数据做了一次验证把近 7 天销售方向的所有 GPTs 描述拉出来确认“跟进”“日报”“记录”这些词出现的频率确实在上升。于是花了两个晚上做了一个原型功能很简单——根据聊天记录自动生成客户跟进摘要和下一步建议。这个原型后面有没有成为产品是另一回事但关键是如果没有这个每天的增量爬虫我根本不会在第一时间看到这个需求聚集的过程。所以我一直觉得爬虫本身只是一种数据触角真正的价值在于它每天给你递上一百条来自真实世界的需求信号。你需要的不是更多数据而是从数据里提炼出判断的能力。最后分享一个我自己的体会跑了整整一个月之后我越来越确定增量数据的独特价值——每天的记录单独看都只是一些零散的新产品但攒上一个月再回头看你能看到一个市场需求从萌芽、蔓延到拥挤的完整过程。建议你从第一天就把每天的增量数据好好存着不要只留最近几天的结果时间会赋予这些数据最大的意义。