爬虫笔记（10/2）——用 TaoToken 统一 Key 跑通 Scrapy 爬虫框架

1. 从零搭建 Scrapy 爬虫框架为什么要把模型调用统一走 TaoTokenScrapy 是 Python 生态里最成熟的爬虫框架之一它把请求调度、去重、下载、解析、数据管道这几件事拆成了独立组件你只需要写少量代码就能跑起一个结构清晰的抓取任务。很多人第一次接触爬虫框架时会纠结选 Scrapy 还是 Crawley、Portia、newspaper 这些方案但真正落到工程里Scrapy 的中间件机制和管道机制是最容易扩展的——尤其是当你的爬虫不只是抓静态页面还要在解析阶段调用大模型做内容抽取、分类、摘要时Scrapy 的下载中间件和 Spider 中间件就成了天然的接入点。问题也正好出在这里。爬虫项目里一旦引入模型调用Key 管理就会变得很乱有的写在 settings.py有的塞进环境变量有的直接在 Spider 里硬编码。多个 Spider 共用同一个 Key 时你根本不知道哪个任务消耗了多少换一个模型供应商又要满项目改 Base URL。我试过在一个十几个 Spider 的项目里逐个替换调用地址改到后面自己都记不清哪个文件漏了。所以这篇笔记的思路是把 Scrapy 项目里所有模型调用类的请求统一收敛到 TaoToken 的 API 通道https://taotoken.net/api。TaoToken 在这里扮演的是一个统一入口——你只需要维护一份 Base URL 和一个 KeySpider、中间件、管道都从同一处读取配置。这样做的直接好处是换模型只改一个 Model ID换通道只改一个 Base URL排查问题时日志里能清楚看到请求是从哪个组件发出的。这篇文章适合谁如果你已经会写一点 Python想系统入门 Scrapy 爬虫框架同时又希望项目里预留模型调用的扩展位那这篇的配置可以直接抄。如果你只是单纯想跑一个静态页面抓取也可以跳过模型部分前面的项目创建、Spider 编写、管道落库照样能用。下面我会从项目创建开始一步步给出可复制的 settings.py、items.py、pipelines.py 片段最后用一次本地抓取任务验证请求发出、数据落库和日志输出是否正常。2. TaoToken 前置准备Key、Base URL 与 Scrapy 项目的对接位置在动手改 Scrapy 配置之前先把 TaoToken 这边的三样东西准备好API Key、Base URL、以及你要用的 Model ID。这三样是后面所有配置的基础缺一个请求都发不出去。先说 Base URL。TaoToken 的 API 通道地址是 https://taotoken.net/api注意这里结尾没有多余的斜杠写配置的时候也别自己加。很多请求报 404 就是因为 Base URL 多拼了一段路径比如写成 https://taotoken.net/api/v1 这种具体路径应该由你调用的 SDK 或请求库去拼接。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册和查看文档都从这里进。再说 API Key。登录之后进控制台在 API Keys 页面创建一个新的 Key。创建时建议按项目命名比如 scrapy-demo这样后面在日志里看到消耗记录时能对上号。Key 只在创建时完整显示一次复制下来存到安全的地方别直接提交到 Git 仓库。控制台地址是 https://taotoken.net/consoleAPI Keys 页面是 https://taotoken.net/api-keys。Model ID 这块你要根据自己实际要调用的模型来填。TaoToken 支持多种模型具体可用的 Model ID 在文档里能查到文档入口是 https://taotoken.net/doc。如果你只是想让爬虫在解析阶段做点文本处理选一个通用对话模型就够了如果要做代码相关的抽取可以选 coding 方向的模型。想先手动验证模型能不能通可以用模型对话页面 https://taotoken.net/models 发一条测试消息确认返回正常再写进 Scrapy 配置。现在把这三样东西落到 Scrapy 项目里。Scrapy 的配置入口是 settings.py但我不建议把 Key 直接写死在 settings.py 里而是通过环境变量注入settings.py 只负责读取。这样本地开发和部署到服务器时可以用不同的 Key也不会因为误提交泄露。具体做法是在项目根目录建一个 .env 文件记得加进 .gitignore里面写TAOTOKEN_API_KEY你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_ID你的ModelID然后在 settings.py 里用 os.environ 读取。如果你不想引入 python-dotenv也可以直接在启动爬虫前 export 这些变量。下面第三节会给出完整的 settings.py 片段包括怎么把这些值传给中间件和管道。这里有个容易踩的坑Scrapy 的 settings 是全局的但不同 Spider 可能想用不同的 Model ID。我的处理方式是在 settings.py 里放一个默认 Model ID然后在 Spider 的 custom_settings 里按需覆盖。这样既保留了统一入口又给了单个 Spider 灵活度。下一节就进入具体的可复制配置。3. 可复制配置settings.py、items.py 与 pipelines.py 完整片段这一节是整篇的核心所有片段都可以直接复制到你的 Scrapy 项目里改掉 Key 和 Model ID 就能跑。先看项目结构假设你用scrapy startproject taotoken_spider创建了项目目录大概是这样taotoken_spider/ ├── scrapy.cfg └── taotoken_spider/ ├── __init__.py ├── items.py ├── middlewares.py ├── pipelines.py ├── settings.py └── spiders/ └── __init__.py先配 settings.py。下面这段是精简后的关键部分保留了和 TaoToken 对接相关的配置其他 Scrapy 默认项按需保留# settings.py import os BOT_NAME taotoken_spider SPIDER_MODULES [taotoken_spider.spiders] NEWSPIDER_MODULE taotoken_spider.spiders # 遵守 robots.txt按需调整 ROBOTSTXT_OBEY True # 并发与延迟本地调试建议调低 CONCURRENT_REQUESTS 8 DOWNLOAD_DELAY 0.5 # TaoToken 统一配置 TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) TAOTOKEN_BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) TAOTOKEN_MODEL_ID os.environ.get(TAOTOKEN_MODEL_ID, 你的默认ModelID) # 开启管道 ITEM_PIPELINES { taotoken_spider.pipelines.TaoTokenValidationPipeline: 300, taotoken_spider.pipelines.SqlitePipeline: 400, } # 日志级别调试时用 DEBUG LOG_LEVEL INFO REQUEST_FINGERPRINTER_IMPLEMENTATION 2.7 TWISTED_REACTOR twisted.internet.asyncioreactor.AsyncioSelectorReactor FEED_EXPORT_ENCODING utf-8注意TAOTOKEN_BASE_URL默认值写的是 https://taotoken.net/api没有结尾斜杠。ITEM_PIPELINES里的数字是优先级越小越先执行这里先做校验再落库。接着是 items.py定义你要抓的字段。假设我们抓的是一批文章列表字段包括标题、链接、正文摘要以及一个留给模型处理的字段# items.py import scrapy class ArticleItem(scrapy.Item): title scrapy.Field() url scrapy.Field() summary scrapy.Field() model_tag scrapy.Field() crawled_at scrapy.Field()model_tag就是留给模型调用结果的比如让模型给文章打一个分类标签。crawled_at记录抓取时间方便后面排查。然后是 pipelines.py这里放两个管道一个负责调用 TaoToken 做校验或打标一个负责落库到 SQLite。先看调用部分用 requests 直接发请求保持依赖简单# pipelines.py import sqlite3 import os import requests from datetime import datetime from itemadapter import ItemAdapter class TaoTokenValidationPipeline: def open_spider(self, spider): self.api_key spider.settings.get(TAOTOKEN_API_KEY) self.base_url spider.settings.get(TAOTOKEN_BASE_URL) self.model_id spider.settings.get(TAOTOKEN_MODEL_ID) if not self.api_key: spider.logger.warning(TAOTOKEN_API_KEY 未设置模型调用将被跳过) def process_item(self, item, spider): adapter ItemAdapter(item) if not self.api_key: adapter[model_tag] skipped return item prompt f请用不超过10个字给下面内容打一个分类标签{adapter.get(title, )} try: resp requests.post( f{self.base_url}/v1/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, json{ model: self.model_id, messages: [{role: user, content: prompt}], max_tokens: 32, }, timeout30, ) resp.raise_for_status() data resp.json() adapter[model_tag] data[choices][0][message][content].strip() except Exception as e: spider.logger.error(fTaoToken 调用失败: {e}) adapter[model_tag] error return item这里请求路径拼的是/v1/chat/completionsBase URL 是 https://taotoken.net/api拼起来就是完整的调用地址。如果你的模型接口路径不同按文档调整这一段即可。注意异常处理里把错误记进日志同时给 item 打上 error 标记这样不会因为单条失败中断整个抓取。落库管道用 SQLite简单直接class SqlitePipeline: def open_spider(self, spider): db_path os.path.join(os.getcwd(), articles.db) self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, url TEXT UNIQUE, summary TEXT, model_tag TEXT, crawled_at TEXT ) ) self.conn.commit() def process_item(self, item, spider): adapter ItemAdapter(item) adapter[crawled_at] datetime.now().isoformat() try: self.conn.execute( INSERT OR IGNORE INTO articles (title, url, summary, model_tag, crawled_at) VALUES (?, ?, ?, ?, ?), ( adapter.get(title), adapter.get(url), adapter.get(summary), adapter.get(model_tag), adapter.get(crawled_at), ), ) self.conn.commit() except Exception as e: spider.logger.error(f落库失败: {e}) return item def close_spider(self, spider): self.conn.close()url字段加了 UNIQUE 约束配合INSERT OR IGNORE重复抓取时不会产生重复记录。到这里三个文件就配好了下一节写 Spider 并跑一次验证。4. 验证请求与落库写一个 Spider 跑通完整链路配置写完之后得有一个真实的 Spider 来验证整条链路。在 spiders 目录下新建demo_spider.py写一个抓取本地或公开测试页面的 Spider。为了不依赖外部站点稳定性这里用 Scrapy 自带的测试思路抓一个结构简单的页面重点验证请求发出、管道执行、日志输出。# spiders/demo_spider.py import scrapy from taotoken_spider.items import ArticleItem class DemoSpider(scrapy.Spider): name demo start_urls [https://quotes.toscrape.com/] custom_settings { TAOTOKEN_MODEL_ID: 你的ModelID, } def parse(self, response): for quote in response.css(div.quote): item ArticleItem() item[title] quote.css(span.text::text).get() item[url] response.urljoin(quote.css(a::attr(href)).get()) item[summary] quote.css(small.author::text).get() yield item这个 Spider 抓的是 quotes.toscrape.com一个专门给爬虫练习用的站点结构稳定。custom_settings里覆盖了 Model ID演示单个 Spider 独立配置的用法。运行命令export TAOTOKEN_API_KEY你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL_ID你的ModelID scrapy crawl demo -L INFO跑起来之后日志里应该能看到类似这样的输出2024-10-02 10:12:33 [scrapy.core.engine] INFO: Spider opened 2024-10-02 10:12:34 [scrapy.core.scraper] DEBUG: Scraped from 200 https://quotes.toscrape.com/ 2024-10-02 10:12:35 [scrapy.extensions.logstats] INFO: Crawled 1 pages (at 1 pages/min), scraped 10 items 2024-10-02 10:12:36 [scrapy.core.engine] INFO: Closing spider (finished)如果模型调用成功model_tag字段会有值如果 Key 没设日志里会出现TAOTOKEN_API_KEY 未设置模型调用将被跳过的警告item 的model_tag是 skipped。落库验证用 sqlite3 命令行sqlite3 articles.db SELECT title, model_tag, crawled_at FROM articles LIMIT 5;正常的话能看到五行记录model_tag有分类结果crawled_at是 ISO 格式时间。这一步跑通说明请求发出、模型调用、数据落库、日志输出四个环节都正常。这里补充一个调试技巧如果模型调用慢可以把DOWNLOAD_DELAY调大或者在管道里加一个简单的重试。Scrapy 的日志级别用-L DEBUG能看到更细的请求信息但生产环境建议保持 INFO避免日志膨胀。下一节专门讲常见的报错和排查方法。5. 常见报错排查401、local proxy failed、reading choices 与 OAuth跑 Scrapy 加模型调用的组合时报错基本集中在几个固定位置。这一节按真实报错信息来对照排查每条都给出定位思路。401 Unauthorized。这是最常见的日志里通常长这样taotoken_spider.pipelines.TaoTokenValidationPipeline: TaoToken 调用失败: 401 Client Error: Unauthorized for url: https://taotoken.net/api/v1/chat/completions原因就三类Key 没读到、Key 写错、Key 被禁用。先在管道里打印一下self.api_key的前几位确认读到了没有注意别把完整 Key 打进日志。如果环境变量没生效检查是不是在启动爬虫的同一个 shell 里 export 的或者 .env 文件没被加载。Key 本身的问题就去控制台 https://taotoken.net/api-keys 重新生成一个。local proxy failed。这个报错通常出现在请求库层面日志里会带ProxyError或local proxy failed字样。Scrapy 默认会读取系统的代理环境变量如果你的环境里残留了 HTTP_PROXY 或 HTTPS_PROXY请求就会走那个地址然后失败。排查方式是先清掉这些变量unset HTTP_PROXY HTTPS_PROXY http_proxy https_proxy然后在 settings.py 里显式关掉 Scrapy 的代理中间件或者确认没有配置DOWNLOADER_MIDDLEWARES里的代理相关项。清掉之后重跑请求应该能直连到 https://taotoken.net/api。reading choices 报错。这个一般出现在解析响应的时候日志类似KeyError: choices说明返回的 JSON 里没有 choices 字段通常是请求本身失败了但状态码是 200比如返回了错误信息体。处理方式是在管道里先判断resp.status_code再判断data里有没有choices没有就把完整响应体打到日志里看。常见原因是 Model ID 写错或者请求体格式不对。对照文档 https://taotoken.net/doc 检查 messages 结构。OAuth 相关报错。如果你用的是需要 OAuth 的客户端工具可能会看到 token 过期或 scope 不足的提示。Scrapy 项目里直接用 API Key 的 Bearer 方式一般不会碰到 OAuth但如果你在中间件里集成了某个 CLI 工具就要确认它的认证方式。遇到 OAuth 报错时先确认你用的是 API Key 而不是 OAuth token两者不能混用。排查顺序建议固定下来先看 Key 有没有读到再看 Base URL 拼得对不对然后看请求体格式最后看返回体内容。把这四步走一遍大部分问题都能定位。如果还是不通用模型对话页面 https://taotoken.net/models 手动发一条同样的请求对比返回结果能快速区分是配置问题还是代码问题。6. 长期跑爬虫任务把 Key 管理和模型调用收进 Coding Plan单次抓取验证通过之后接下来要考虑的是长期运行。爬虫任务往往是定时跑的可能一天几次也可能持续几小时。这时候 Key 的管理和调用成本就需要有个稳定的方案。我自己的做法是把 Scrapy 项目里的模型调用统一收口到 TaoToken然后根据任务类型选择不同的使用方式。如果是短期的、一次性的抓取验证直接用 API Key 按量调用就够了配置就是前面那套。如果是长期跑的编码类或 Agent 类任务比如爬虫要持续做内容抽取、分类、甚至自动生成摘要那更适合用 Coding Plan入口在 https://taotoken.net/coding-plan。它的好处是额度可预期不用每次盯着按量消耗。具体到 Scrapy 项目里长期运行要注意几点。第一Key 不要写死在代码里用环境变量或配置中心注入前面已经演示过。第二管道里的模型调用要加超时和重试避免单条卡住拖垮整个任务。第三日志里记录每次调用的 Model ID 和耗时方便后面分析哪个模型更适合你的抓取场景。第四如果 Spider 数量多建议在 settings.py 里维护一份默认配置单个 Spider 用 custom_settings 覆盖保持统一入口。如果你还想在爬虫之外做更多模型相关的实验比如对比不同模型对同一批数据的抽取效果可以用模型对话页面 https://taotoken.net/models 手动试。需要看调用文档和参数说明就去 https://taotoken.net/doc。Key 的创建和管理在 https://taotoken.net/api-keys。这几个入口配合起来基本能覆盖从调试到长期运行的全流程。最后说一个实际经验爬虫项目里引入模型调用之后最容易出问题的不是模型本身而是配置分散。把 Base URL、Key、Model ID 这三样收敛到一处后面换模型、换通道、排查问题都会轻松很多。Scrapy 的中间件和管道机制正好给了这种收敛的天然位置用好了比在每个 Spider 里各写一套要省心得多。