简介面向计算机专业学生和爬虫开发者的基于Python和Scrapy框架的农业数据爬虫项目针对农业科研、政策制定与市场分析等场景的数据采集需求完整覆盖爬虫结构设计、Spider编写、Item定义、Pipeline数据清洗与存储、中间件异常处理及系统部署等环节适用于课程设计、毕业设计或企业数据采集方案演示。整个压缩包共34个文件大小仅35KB包含10个Python源文件、9个编译后的pyc文件、6个HTML页面模板、cfg配置及Markdown说明文档等目录层次清楚便于按模块对照阅读。目前已有67人学习下载该项目源自个人高分答辩作品经导师认可代码经过完整测试运行稳定。资源内附部署文档与设计思路详细说明了不同环境下的服务器搭建、所需配置与安全注意事项即使基础薄弱也能按说明完成部署有经验的用户还可针对现有Spider、Pipeline等组件进行扩展融入更多数据源与定时调度策略快速构建属于自己的农业数据采集系统兼备学术参考与工程实用价值。1. 农业数据爬虫的难点不在“爬”而在数据的时效与清洗把“PythonScrapy实现的农业数据爬虫设计与部署”拆开看前半段是技术选型后半段是数据属性。多数农业数据页面并不复杂——行情价格、气象墒情、品种产地大部分是列表页加详情页的经典结构Scrapy 完全可以覆盖。真正的难点在于农业数据的时间属性一条玉米价格上午 10 点发布和下午 3 点发布的记录含义完全不同晚抓一小时拿到的可能是隔天的行情快照去重逻辑、存储粒度、调度周期全都会跟着变。这个标题要解决的不只是“怎么把网页抓下来”而是“怎么在三个月内持续稳定地抓到可入库、可分析、可追溯的新鲜数据”。适合的人群分两类一类是农业信息化项目里负责数据采集的工程师另一类是把公开农业数据当作竞品分析素材的数据从业者。这套设计更偏向于把采集做成一个低运维成本的生产任务而不是一次性的脚本。2. Scrapy Item 设计把农业数据的时间属性写进字段模型2.1 字段建模发布日期、行情类型、地区编码一个都不能少农业数据源的字段差异比想象中大。同一张批发价格表有的站点叫“品名”有的叫“品种”有的把产地写在品名里比如“山东大葱”有的单独有“产地”列价格单位更是混乱元/斤、元/公斤、元/100g 都见过。Item 设计的目标不是对齐字段名而是对齐字段语义否则进了数据库再进行清洗成本会翻倍。一个通用的农业行情 Item 模板可以做成下面这样import scrapy class AgriPriceItem(scrapy.Item): # 数据源标识方便追溯某条数据来自哪个站点 source scrapy.Field() # 详情页 URL作为溯源入口 url scrapy.Field() # 页面上的“信息发布/更新”时间不是爬虫抓取时间 publish_date scrapy.Field() publish_time scrapy.Field() # 行情类型批发价 / 零售价 / 产地收购价 price_type scrapy.Field() # 品种名称保持站点原始写法清洗动作放到 pipeline commodity scrapy.Field() # 地区字段只存站点给的最细粒度 region scrapy.Field() # 统一转成 float单位统一转成 元/公斤 price scrapy.Field() unit scrapy.Field() # 规格描述如“中等、混等”不参与计算但值得保留 spec scrapy.Field() # 抓取时间戳用于判断数据新鲜度 crawl_ts scrapy.Field()这里最容易犯的错误是拿抓取时间当发布时间用。爬虫跑在服务器上请求到页面的那一刻是crawl_ts页面正文里写的“统计日期 2025-06-03”才是publish_date。排错的时候如果只看抓取时间去判断数据是否过期会被页面本身的更新节奏误导。另外建议把unit单独留出来因为后续清洗时不同单位要换算单位一旦在源头上合并掉换算关系就查不回来了。2.2 去重指纹用“页面内容摘要 发布日期”替代简单 URL 哈希Scrapy 默认的request_fingerprint机制针对的是请求去重同一个 URL 不会请求两遍。但农业数据网页有个特点同样一个 URL页面内容每天在变。比如一个批发市场每日报价页URL 恒定今天请求和明天请求会拿到不同日期的价格数据。所以请求去了重不等于数据去了重数据层面的去重必须自己动手。我一般会在 Item Pipeline 里做指纹计算并把指纹直接写进 Redis 集合用SISMEMBER查重import hashlib import json import redis class DuplicatePipeline: def __init__(self, redis_host, redis_port, redis_db): self.r redis.Redis(hostredis_host, portredis_port, dbredis_db) # 指纹集合的 key按站点区分避免不同站点数据互相碰撞 self.fp_key agri:price:fingerprint classmethod def from_crawler(cls, crawler): return cls( redis_hostcrawler.settings.get(REDIS_HOST, localhost), redis_portcrawler.settings.get(REDIS_PORT, 6379), redis_dbcrawler.settings.get(REDIS_PIPE_DB, 1), ) def process_item(self, item, spider): # 指纹组合站点 品种 地区 发布日期天然适配每日更新的场景 raw json.dumps({ source: item.get(source), commodity: item.get(commodity), region: item.get(region), publish_date: item.get(publish_date), }, ensure_asciiFalse, sort_keysTrue) fp hashlib.sha1(raw.encode(utf-8)).hexdigest() # set 添加成功说明是首次出现添加失败说明已存在直接丢弃 if self.r.sadd(self.fp_key, fp): item[_fingerprint] fp return item raise DropItem(fduplicate item: {item.get(commodity)})这段逻辑里有几个细节值得说明。指纹用sha1而不是md5主要考虑是在分布式部署场景下避免极端碰撞性能差距可以忽略。sadd的返回值是关键Redis 的集合操作天然支持原子判重新增返回 1重复返回 0省掉了“先查再写”的竞态窗口。publish_date放进指纹意味着同一天重复抓到的同一条数据会被丢弃而第二天再次爬取时因为日期变化指纹变成新的数据能正常更新入库。这个方案核心解决了“URL 不变但内容变了”的页面更新跟踪问题。2.3 清洗 Pipeline单位、价格、规格乱成这样也能归一化清洗动作放在 Pipeline 的好处是后置且统一爬虫里不用改动解析代码。农业数据的清洗主要集中在三件事单位换算、价格类型归一化、产地解析。UNIT_MAP { 元/斤: 2.0, # 1 斤 0.5 公斤乘以 2 得到 元/公斤 元/公斤: 1.0, 元/kg: 1.0, 元/500g: 2.0, # 500g 即 1 市斤 元/千克: 1.0, } def normalize_price(item: dict): unit item.get(unit) price float(item.get(price)) if unit in UNIT_MAP: item[price] round(price * UNIT_MAP[unit], 2) item[unit] 元/公斤 return item TYPE_ALIAS { 批发: wholesale, 批发价: wholesale, 市场价: market, 零售: retail, 收购价: purchase, } def normalize_price_type(item: dict): raw_type item.get(price_type) if raw_type in TYPE_ALIAS: item[price_type] TYPE_ALIAS[raw_type] return itemUNIT_MAP里的换算系数需要人工确认特别是“元/500g”这种写法很多南方信息站习惯用这个单位报价不换算直接入库后续做同比分析时价格会整体差一倍。normalize_price_type把中文描述映射为英文枚举统一price_type之后才能用 SQL 做GROUP BY汇总。清洗规则不要写死在爬虫里Pipeline 阶段统一处理爬虫只保留原始值这样同样一套 Item 可以复用到不同站点只需替换 Pipeline 配置不用动解析代码。3. 用 scrapy-playwright 和中间件处理动态页面与访问节奏3.1 静态页面交给 Scrapyiframe 里的报价交给 Playwright农业站点里有一批老旧的政府信息平台页面结构还停留在 iframe 嵌套加表格。Scrapy 原生解析 iframe 比较麻烦需要先把这个页面发请求拿到 iframe 子页面的 URL再对子页面发起二次请求。如果 iframe 里再用 JavaScript 动态拼接表格纯 Scrapy 处理起来成本很高。这时用scrapy-playwright把动态页面交给浏览器引擎渲染。配置上做一次中间件启用即可DOWNLOAD_HANDLERS { http: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, https: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, } TWISTED_REACTOR twisted.internet.asyncioreactor.AsyncioSelectorReactor PLAYWRIGHT_LAUNCH_OPTIONS { headless: True, }请求时通过meta参数控制哪些页面走渲染yield scrapy.Request( urliframe_page_url, meta{ playwright: True, # 等待市场报价表格渲染完成 playwright_page_methods: [ {method: wait_for_selector, selector: table#priceTable}, ], }, callbackself.parse_price_table, )需要特别说的是路由策略要分流而不是全站 Playwright。价格列表页如果本身就是静态 HTML直接走默认下载器速度快且不占内存。只有遇到 iframe、动态填充表格、点击“更多”加载这类交互时才把请求交给 Playwright。全站无脑开 Playwright 的后果是爬虫并发数降到原来的五分之一而且服务器内存经常报警。3.2 Retry 中间件与 UA 轮换从日志判断该降速还是该换伪装反爬处理的原则是先看日志再动手不要一遇 403 就上重武器。农业信息站的反爬强度普遍不高很多 403 只是因为没有携带常规请求头或者请求频率超过了站点容忍度。在settings.py里做基础配置DOWNLOADER_MIDDLEWARES { scrapy.downloadermiddlewares.useragent.UserAgentMiddleware: None, scrapy.downloadermiddlewares.retry.RetryMiddleware: 550, my_project.middlewares.RotateUserAgentMiddleware: 500, } RETRY_ENABLED True RETRY_TIMES 2 # 最多重试 2 次重试太多次会放大请求压力 RETRY_HTTP_CODES [403, 500, 502, 503, 504] CONCURRENT_REQUESTS 8 DOWNLOAD_DELAY 1.0UA 轮换中间件本身不复杂从配置一个 UA 列表每次请求分配一个class RotateUserAgentMiddleware: def __init__(self, ua_list): self.ua_list ua_list classmethod def from_crawler(cls, crawler): return cls(crawler.settings.getlist(USER_AGENT_LIST)) def process_request(self, request, spider): request.headers[User-Agent] random.choice(self.ua_list) return None USER_AGENT_LIST [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15, ]排错的关键看日志。如果 403 出现在爬虫刚启动的前几个请求里大概率是 UA 被拦如果跑了十几分钟后才密集出现说明是频率问题这时候调低并发和增加DOWNLOAD_DELAY比换 UA 更有效。我处理的农业站点里一半以上的 403 调整完频率后就不再出现说明根本没有高级风控在拦纯粹是请求太密集。3.3 并发与限速参数按域名区分配置而不是全局一刀切一个爬虫往往同时采集多个农业站点各站点对访问频率的容忍度完全不同。政府公开数据接口一般能接受每秒几次请求而小型的民营行情站可能一两秒一次就触发限制。Scrapy 提供了一个容易被忽略的参数——每个域名的独立配置per-domain配置。在 Spider 里覆写custom_settingsclass MultiSitePriceSpider(scrapy.Spider): name agri_prices custom_settings { CONCURRENT_REQUESTS_PER_DOMAIN: { data.gov-example.cn: 4, vegmarket.example.com: 1, }, DOWNLOAD_DELAY: { data.gov-example.cn: 0.5, vegmarket.example.com: 3.0, }, AUTOTHROTTLE_ENABLED: True, AUTOTHROTTLE_TARGET_CONCURRENCY: 4.0, }AUTOTHROTTLE_ENABLED开启后Scrapy 会依据响应耗时动态调整延迟目标并发适中站点响应慢就自动降速。但要注意AUTOTHROTTLE和手动设置的DOWNLOAD_DELAY是叠加关系如果两个都设了实际请求间隔会比预期长。简单配置方式是用DOWNLOAD_DELAY做基础限速AUTOTHROTTLE只在目标站响应波动大时开启。4. 增量更新与调度按发布窗口而不是固定频率去跑4.1 先做站点时间窗分析再做调度计划农业数据的发布节奏高度依赖自然日和交易时段。批发市场价格一般在清晨 5 点到 8 点更新产地收购价集中在下午 3 点到 6 点而生鲜电商的实时价格则可能每隔 10 分钟刷新一次。固定频率每小时扫一遍最短的发布窗口往往扫到的是没有变化的数据还容易因为请求频繁触发限制。因此在写调度之前先做一轮时间窗分析方法很简单用带时间戳的下载器抓取目标站两到三天统计页面里publish_date分布。得到一个典型的更新规律后按规律决定跑批时间。数据类型典型更新时间建议调度策略批发市场日度价格06:00 - 08:00每天 09:00 跑一次产地收购价15:00 - 18:00每天 19:00 跑一次气象墒情数据每小时每小时第 20 分钟跑一次实时行情接口连续更新每 10 分钟拉取一次增量这个表的意义在于把调度周期与数据发布日期对齐而不是让爬虫全程待命。数据发布时间是分析出来的不是拍脑袋定的。4.2 Scrapyd 调度的两种提交方式以及 cron 兜底Scrapyd 是部署 Scrapy 爬虫的经典方案它负责把爬虫项目打包成 egg 上传到服务器并提供 HTTP API 进行任务调度。部署后调度和监控都不再依赖本机终端。添加一个定时任务直接通过 API 提交curl http://your-server:6800/schedule.json \ -d projectagri_prices \ -d spidermulti_site_prices \ -d settingDOWNLOAD_DELAY0.8提交后 Scrapyd 返回任务 ID日志可以通过curl http://your-server:6800/logs/agri_prices/multi_site_prices/1.log查看。但 Scrapyd 本身不带定时能力它只负责“接收任务并执行”调度还需要靠 cron 或第三方调度器。生产环境我一般用两层结构cron 触发schedule.jsonScrapyd 执行任务Redis 负责指纹去重。# 每天 09:05 提交批发价格采集任务 5 9 * * * curl -s http://localhost:6800/schedule.json -d projectagri_prices -d spiderwholesale_prices # 每小时的第 20 分钟触发气象数据采集 20 * * * * curl -s http://localhost:6800/schedule.json -d projectagri_prices -d spiderweather_obsschedule.json接口接收到请求后会立即执行脚本本身没有任何返回值处理所以 cron 只是用来“提交”不需要等待爬虫跑完。如果同一个爬虫因为上一个任务还没结束就被再次调度Scrapyd 默认会排队执行对于时效性弱的日度数据这没问题但实时行情这类任务就要自己维护一个锁避免任务积压。4.3 新数据推送飞书群机器人足够替代邮件告警爬虫跑完谁来判断“这次的采集是否有新数据”比看日志更直观的做法是把结果推到群机器人飞书或钉钉都可以支持 webhook 的普通群就能满足。采集结束的close_spider事件里做推送这是最省事的接入位置import requests class NotifyPipeline: def __init__(self, webhook_url): self.webhook_url webhook_url self.item_count 0 self.spider_name None def open_spider(self, spider): self.spider_name spider.name self.item_count 0 def process_item(self, item, spider): self.item_count 1 return item def close_spider(self, spider): text f爬虫 {self.spider_name} 本次采集新增 {self.item_count} 条数据 requests.post(self.webhook_url, json{msg_type: text, content: {text: text}}) classmethod def from_crawler(cls, crawler): return cls(crawler.settings.get(NOTIFY_WEBHOOK_URL))这里用的是 item 通过去重 Pipeline 后数量也就是指纹判定为“新增”的数据量不是原始请求数。如果item_count一直挂零优先检查指纹键是否过期堆积——农业数据按日期生成指纹一旦 Redis 键里的日期永远不更新新增数据就会被误判为重复。5. Docker Scrapyd 部署把开发环境复刻到服务器5.1 Dockerfile 里最容易忽略的两个依赖时区与中文字体农业数据的发布和调度都依赖本地时间Docker 容器默认的 UTC 时区会导致“每天 09:00 跑批”变成“北京时间 17:00 跑批”。这是部署后最先踩的坑。此外部分农业数据网站依赖中文字体渲染验证码或动态图表如果后续接入 OCR 识别验证码容器里没有中文字体识别率会直线下降。一个可用的 Dockerfile 如下FROM python:3.11-slim WORKDIR /app # 先装系统依赖layers 优化改动频率低于 requirements 的层放前面 RUN apt-get update apt-get install -y --no-install-recommends \ tzdata \ fonts-noto-cjk \ curl \ ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \ echo Asia/Shanghai /etc/timezone \ rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 6800 CMD [scrapyd]时区的处理是ln -sf创建软链到本地时间文件再写入/etc/timezone。fonts-noto-cjk是文泉驿之外最常见的开源中文字体包体积约几十 MB但能避免后续 OCR 识别时中文乱码的问题。requirements.txt里至少要包含scrapy、scrapyd、scrapy-playwright、redis、requestsscrapyd-client只在发布机上需要不用打进运行镜像。5.2 scrapyd-deploy 上传的 target 配置与版本升级发布流程通常是在开发机执行scrapyd-deploy把项目打包上传到服务器的 Scrapyd。配置放在项目根目录的scrapy.cfg[deploy:production] url http://your-server:6800/ project agri_prices然后执行scrapyd-deploy production -p agri_prices --version 20250603--version参数如果不指定默认会取当前时间戳但显式指定版本号更利于回滚。上传成功后服务器端会保留历史版本通过 API 查询curl http://your-server:6800/listversions.json?projectagri_prices有个高频报错需要记住scrapyd-deploy提示Unhandled Error往往是因为服务器 Scrapyd 版本与开发机scrapyd-client版本不一致通用做法是两边都升级到当前稳定的 2.x 系列避免老版本对 egg 格式的兼容性问题。另外注意scrapy.cfg里的 url 不要带http://localhost:6800/否则发布时会打到开发机的 6800 端口而不是服务器的。5.3 部署后的访问控制只允许本机查看 job 日志Scrapyd 默认监听0.0.0.0:6800这意味着公网任何一台机器都可以往你的 Scrapyd 提交任务、删除版本风险极大。安全配置第一件事是把绑定地址改掉只监听内网或本机scrapyd-deploy production -p agri_prices --version 20250603这里配合的修改是在scrapyd.conf[scrapyd] bind_address 127.0.0.1 http_port 6800改完后调用方只能从本机访问。如果确实需要远程查看任务状态用 SSH 隧道转发 6800 端口而不是直接暴露ssh -N -L 6800:localhost:6800 useryour-server访问控制之外max_proc参数也值得关注它决定了 Scrapyd 同时能跑多少个进程。如果机器有 4 核 CPU可以设置max_proc 4 max_proc_per_cpu 1这个配置避免同一个爬虫被调度时疯狂开进程内存被多个 Scrapy 实例吃满。日志目录默认在/var/log/scrapyd用 Docker 部署时务必挂载宿主机目录否则容器重建后历史日志全部丢失排查数据质量问题时无据可查。6. 用自检抓取脚本验证数据新鲜度不依赖监控面板与其搭建一整套监控系统不如写一个轻量自检脚本每天检查“今天的目标站点是否产生了新数据”。脚本独立于 Scrapy 项目只做一件事查询 Redis 里的指纹集合和数据库表里的最新日期然后对比当前日期。import redis import pymysql from datetime import datetime r redis.Redis(hostlocalhost, port6379, db1) # 检查 pub_date 维度是否达到当日采集预期 db pymysql.connect(hostlocalhost, useragri, password****, databaseagri_mart) try: with db.cursor() as cursor: # 查出采集表中最新的发布日期 cursor.execute(SELECT MAX(publish_date) FROM price_daily WHERE sourcewholesale) latest_date cursor.fetchone()[0] today datetime.now().date() # latest_date 是 date 类型如果是昨天说明当日批任务漏跑 if latest_date today: print(f[警告] 数据停留在 {latest_date}今日尚未采集) else: print(f[正常] 最新数据日期 {latest_date}) finally: db.close()脚本配合 cron 每天执行一次输出追加到日志文件如果发现latest_date比当前日期晚一天以上就说明调度或解析出了问题。这个自检方法虽然粗糙但比看 Scrapyd 的任务状态要可靠得多——任务状态显示“已完成”不代表页面里的数据已经被正确解析入库只有数据库里的最大发布日期才能代表真实的数据新鲜度。线上爬虫出问题的环节一半以上不在请求失败而在解析规则因为页面改版失效后任务照常跑、数据照常抓但字段全部为空或错位最终入库数据量骤减问题被宏大的监控面板淹没在 200 状态码里。本文还有配套的精品资源点击获取 
长河君 · 2026/9/12 9:15:12 · 阅读 1.2k 农业数据爬虫实战：Scrapy字段建模、增量更新与Scrapyd部署
简介面向计算机专业学生和爬虫开发者的基于Python和Scrapy框架的农业数据爬虫项目针对农业科研、政策制定与市场分析等场景的数据采集需求完整覆盖爬虫结构设计、Spider编写、Item定义、Pipeline数据清洗与存储、中间件异常处理及系统部署等环节适用于课程设计、毕业设计或企业数据采集方案演示。整个压缩包共34个文件大小仅35KB包含10个Python源文件、9个编译后的pyc文件、6个HTML页面模板、cfg配置及Markdown说明文档等目录层次清楚便于按模块对照阅读。目前已有67人学习下载该项目源自个人高分答辩作品经导师认可代码经过完整测试运行稳定。资源内附部署文档与设计思路详细说明了不同环境下的服务器搭建、所需配置与安全注意事项即使基础薄弱也能按说明完成部署有经验的用户还可针对现有Spider、Pipeline等组件进行扩展融入更多数据源与定时调度策略快速构建属于自己的农业数据采集系统兼备学术参考与工程实用价值。1. 农业数据爬虫的难点不在“爬”而在数据的时效与清洗把“PythonScrapy实现的农业数据爬虫设计与部署”拆开看前半段是技术选型后半段是数据属性。多数农业数据页面并不复杂——行情价格、气象墒情、品种产地大部分是列表页加详情页的经典结构Scrapy 完全可以覆盖。真正的难点在于农业数据的时间属性一条玉米价格上午 10 点发布和下午 3 点发布的记录含义完全不同晚抓一小时拿到的可能是隔天的行情快照去重逻辑、存储粒度、调度周期全都会跟着变。这个标题要解决的不只是“怎么把网页抓下来”而是“怎么在三个月内持续稳定地抓到可入库、可分析、可追溯的新鲜数据”。适合的人群分两类一类是农业信息化项目里负责数据采集的工程师另一类是把公开农业数据当作竞品分析素材的数据从业者。这套设计更偏向于把采集做成一个低运维成本的生产任务而不是一次性的脚本。2. Scrapy Item 设计把农业数据的时间属性写进字段模型2.1 字段建模发布日期、行情类型、地区编码一个都不能少农业数据源的字段差异比想象中大。同一张批发价格表有的站点叫“品名”有的叫“品种”有的把产地写在品名里比如“山东大葱”有的单独有“产地”列价格单位更是混乱元/斤、元/公斤、元/100g 都见过。Item 设计的目标不是对齐字段名而是对齐字段语义否则进了数据库再进行清洗成本会翻倍。一个通用的农业行情 Item 模板可以做成下面这样import scrapy class AgriPriceItem(scrapy.Item): # 数据源标识方便追溯某条数据来自哪个站点 source scrapy.Field() # 详情页 URL作为溯源入口 url scrapy.Field() # 页面上的“信息发布/更新”时间不是爬虫抓取时间 publish_date scrapy.Field() publish_time scrapy.Field() # 行情类型批发价 / 零售价 / 产地收购价 price_type scrapy.Field() # 品种名称保持站点原始写法清洗动作放到 pipeline commodity scrapy.Field() # 地区字段只存站点给的最细粒度 region scrapy.Field() # 统一转成 float单位统一转成 元/公斤 price scrapy.Field() unit scrapy.Field() # 规格描述如“中等、混等”不参与计算但值得保留 spec scrapy.Field() # 抓取时间戳用于判断数据新鲜度 crawl_ts scrapy.Field()这里最容易犯的错误是拿抓取时间当发布时间用。爬虫跑在服务器上请求到页面的那一刻是crawl_ts页面正文里写的“统计日期 2025-06-03”才是publish_date。排错的时候如果只看抓取时间去判断数据是否过期会被页面本身的更新节奏误导。另外建议把unit单独留出来因为后续清洗时不同单位要换算单位一旦在源头上合并掉换算关系就查不回来了。2.2 去重指纹用“页面内容摘要 发布日期”替代简单 URL 哈希Scrapy 默认的request_fingerprint机制针对的是请求去重同一个 URL 不会请求两遍。但农业数据网页有个特点同样一个 URL页面内容每天在变。比如一个批发市场每日报价页URL 恒定今天请求和明天请求会拿到不同日期的价格数据。所以请求去了重不等于数据去了重数据层面的去重必须自己动手。我一般会在 Item Pipeline 里做指纹计算并把指纹直接写进 Redis 集合用SISMEMBER查重import hashlib import json import redis class DuplicatePipeline: def __init__(self, redis_host, redis_port, redis_db): self.r redis.Redis(hostredis_host, portredis_port, dbredis_db) # 指纹集合的 key按站点区分避免不同站点数据互相碰撞 self.fp_key agri:price:fingerprint classmethod def from_crawler(cls, crawler): return cls( redis_hostcrawler.settings.get(REDIS_HOST, localhost), redis_portcrawler.settings.get(REDIS_PORT, 6379), redis_dbcrawler.settings.get(REDIS_PIPE_DB, 1), ) def process_item(self, item, spider): # 指纹组合站点 品种 地区 发布日期天然适配每日更新的场景 raw json.dumps({ source: item.get(source), commodity: item.get(commodity), region: item.get(region), publish_date: item.get(publish_date), }, ensure_asciiFalse, sort_keysTrue) fp hashlib.sha1(raw.encode(utf-8)).hexdigest() # set 添加成功说明是首次出现添加失败说明已存在直接丢弃 if self.r.sadd(self.fp_key, fp): item[_fingerprint] fp return item raise DropItem(fduplicate item: {item.get(commodity)})这段逻辑里有几个细节值得说明。指纹用sha1而不是md5主要考虑是在分布式部署场景下避免极端碰撞性能差距可以忽略。sadd的返回值是关键Redis 的集合操作天然支持原子判重新增返回 1重复返回 0省掉了“先查再写”的竞态窗口。publish_date放进指纹意味着同一天重复抓到的同一条数据会被丢弃而第二天再次爬取时因为日期变化指纹变成新的数据能正常更新入库。这个方案核心解决了“URL 不变但内容变了”的页面更新跟踪问题。2.3 清洗 Pipeline单位、价格、规格乱成这样也能归一化清洗动作放在 Pipeline 的好处是后置且统一爬虫里不用改动解析代码。农业数据的清洗主要集中在三件事单位换算、价格类型归一化、产地解析。UNIT_MAP { 元/斤: 2.0, # 1 斤 0.5 公斤乘以 2 得到 元/公斤 元/公斤: 1.0, 元/kg: 1.0, 元/500g: 2.0, # 500g 即 1 市斤 元/千克: 1.0, } def normalize_price(item: dict): unit item.get(unit) price float(item.get(price)) if unit in UNIT_MAP: item[price] round(price * UNIT_MAP[unit], 2) item[unit] 元/公斤 return item TYPE_ALIAS { 批发: wholesale, 批发价: wholesale, 市场价: market, 零售: retail, 收购价: purchase, } def normalize_price_type(item: dict): raw_type item.get(price_type) if raw_type in TYPE_ALIAS: item[price_type] TYPE_ALIAS[raw_type] return itemUNIT_MAP里的换算系数需要人工确认特别是“元/500g”这种写法很多南方信息站习惯用这个单位报价不换算直接入库后续做同比分析时价格会整体差一倍。normalize_price_type把中文描述映射为英文枚举统一price_type之后才能用 SQL 做GROUP BY汇总。清洗规则不要写死在爬虫里Pipeline 阶段统一处理爬虫只保留原始值这样同样一套 Item 可以复用到不同站点只需替换 Pipeline 配置不用动解析代码。3. 用 scrapy-playwright 和中间件处理动态页面与访问节奏3.1 静态页面交给 Scrapyiframe 里的报价交给 Playwright农业站点里有一批老旧的政府信息平台页面结构还停留在 iframe 嵌套加表格。Scrapy 原生解析 iframe 比较麻烦需要先把这个页面发请求拿到 iframe 子页面的 URL再对子页面发起二次请求。如果 iframe 里再用 JavaScript 动态拼接表格纯 Scrapy 处理起来成本很高。这时用scrapy-playwright把动态页面交给浏览器引擎渲染。配置上做一次中间件启用即可DOWNLOAD_HANDLERS { http: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, https: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, } TWISTED_REACTOR twisted.internet.asyncioreactor.AsyncioSelectorReactor PLAYWRIGHT_LAUNCH_OPTIONS { headless: True, }请求时通过meta参数控制哪些页面走渲染yield scrapy.Request( urliframe_page_url, meta{ playwright: True, # 等待市场报价表格渲染完成 playwright_page_methods: [ {method: wait_for_selector, selector: table#priceTable}, ], }, callbackself.parse_price_table, )需要特别说的是路由策略要分流而不是全站 Playwright。价格列表页如果本身就是静态 HTML直接走默认下载器速度快且不占内存。只有遇到 iframe、动态填充表格、点击“更多”加载这类交互时才把请求交给 Playwright。全站无脑开 Playwright 的后果是爬虫并发数降到原来的五分之一而且服务器内存经常报警。3.2 Retry 中间件与 UA 轮换从日志判断该降速还是该换伪装反爬处理的原则是先看日志再动手不要一遇 403 就上重武器。农业信息站的反爬强度普遍不高很多 403 只是因为没有携带常规请求头或者请求频率超过了站点容忍度。在settings.py里做基础配置DOWNLOADER_MIDDLEWARES { scrapy.downloadermiddlewares.useragent.UserAgentMiddleware: None, scrapy.downloadermiddlewares.retry.RetryMiddleware: 550, my_project.middlewares.RotateUserAgentMiddleware: 500, } RETRY_ENABLED True RETRY_TIMES 2 # 最多重试 2 次重试太多次会放大请求压力 RETRY_HTTP_CODES [403, 500, 502, 503, 504] CONCURRENT_REQUESTS 8 DOWNLOAD_DELAY 1.0UA 轮换中间件本身不复杂从配置一个 UA 列表每次请求分配一个class RotateUserAgentMiddleware: def __init__(self, ua_list): self.ua_list ua_list classmethod def from_crawler(cls, crawler): return cls(crawler.settings.getlist(USER_AGENT_LIST)) def process_request(self, request, spider): request.headers[User-Agent] random.choice(self.ua_list) return None USER_AGENT_LIST [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15, ]排错的关键看日志。如果 403 出现在爬虫刚启动的前几个请求里大概率是 UA 被拦如果跑了十几分钟后才密集出现说明是频率问题这时候调低并发和增加DOWNLOAD_DELAY比换 UA 更有效。我处理的农业站点里一半以上的 403 调整完频率后就不再出现说明根本没有高级风控在拦纯粹是请求太密集。3.3 并发与限速参数按域名区分配置而不是全局一刀切一个爬虫往往同时采集多个农业站点各站点对访问频率的容忍度完全不同。政府公开数据接口一般能接受每秒几次请求而小型的民营行情站可能一两秒一次就触发限制。Scrapy 提供了一个容易被忽略的参数——每个域名的独立配置per-domain配置。在 Spider 里覆写custom_settingsclass MultiSitePriceSpider(scrapy.Spider): name agri_prices custom_settings { CONCURRENT_REQUESTS_PER_DOMAIN: { data.gov-example.cn: 4, vegmarket.example.com: 1, }, DOWNLOAD_DELAY: { data.gov-example.cn: 0.5, vegmarket.example.com: 3.0, }, AUTOTHROTTLE_ENABLED: True, AUTOTHROTTLE_TARGET_CONCURRENCY: 4.0, }AUTOTHROTTLE_ENABLED开启后Scrapy 会依据响应耗时动态调整延迟目标并发适中站点响应慢就自动降速。但要注意AUTOTHROTTLE和手动设置的DOWNLOAD_DELAY是叠加关系如果两个都设了实际请求间隔会比预期长。简单配置方式是用DOWNLOAD_DELAY做基础限速AUTOTHROTTLE只在目标站响应波动大时开启。4. 增量更新与调度按发布窗口而不是固定频率去跑4.1 先做站点时间窗分析再做调度计划农业数据的发布节奏高度依赖自然日和交易时段。批发市场价格一般在清晨 5 点到 8 点更新产地收购价集中在下午 3 点到 6 点而生鲜电商的实时价格则可能每隔 10 分钟刷新一次。固定频率每小时扫一遍最短的发布窗口往往扫到的是没有变化的数据还容易因为请求频繁触发限制。因此在写调度之前先做一轮时间窗分析方法很简单用带时间戳的下载器抓取目标站两到三天统计页面里publish_date分布。得到一个典型的更新规律后按规律决定跑批时间。数据类型典型更新时间建议调度策略批发市场日度价格06:00 - 08:00每天 09:00 跑一次产地收购价15:00 - 18:00每天 19:00 跑一次气象墒情数据每小时每小时第 20 分钟跑一次实时行情接口连续更新每 10 分钟拉取一次增量这个表的意义在于把调度周期与数据发布日期对齐而不是让爬虫全程待命。数据发布时间是分析出来的不是拍脑袋定的。4.2 Scrapyd 调度的两种提交方式以及 cron 兜底Scrapyd 是部署 Scrapy 爬虫的经典方案它负责把爬虫项目打包成 egg 上传到服务器并提供 HTTP API 进行任务调度。部署后调度和监控都不再依赖本机终端。添加一个定时任务直接通过 API 提交curl http://your-server:6800/schedule.json \ -d projectagri_prices \ -d spidermulti_site_prices \ -d settingDOWNLOAD_DELAY0.8提交后 Scrapyd 返回任务 ID日志可以通过curl http://your-server:6800/logs/agri_prices/multi_site_prices/1.log查看。但 Scrapyd 本身不带定时能力它只负责“接收任务并执行”调度还需要靠 cron 或第三方调度器。生产环境我一般用两层结构cron 触发schedule.jsonScrapyd 执行任务Redis 负责指纹去重。# 每天 09:05 提交批发价格采集任务 5 9 * * * curl -s http://localhost:6800/schedule.json -d projectagri_prices -d spiderwholesale_prices # 每小时的第 20 分钟触发气象数据采集 20 * * * * curl -s http://localhost:6800/schedule.json -d projectagri_prices -d spiderweather_obsschedule.json接口接收到请求后会立即执行脚本本身没有任何返回值处理所以 cron 只是用来“提交”不需要等待爬虫跑完。如果同一个爬虫因为上一个任务还没结束就被再次调度Scrapyd 默认会排队执行对于时效性弱的日度数据这没问题但实时行情这类任务就要自己维护一个锁避免任务积压。4.3 新数据推送飞书群机器人足够替代邮件告警爬虫跑完谁来判断“这次的采集是否有新数据”比看日志更直观的做法是把结果推到群机器人飞书或钉钉都可以支持 webhook 的普通群就能满足。采集结束的close_spider事件里做推送这是最省事的接入位置import requests class NotifyPipeline: def __init__(self, webhook_url): self.webhook_url webhook_url self.item_count 0 self.spider_name None def open_spider(self, spider): self.spider_name spider.name self.item_count 0 def process_item(self, item, spider): self.item_count 1 return item def close_spider(self, spider): text f爬虫 {self.spider_name} 本次采集新增 {self.item_count} 条数据 requests.post(self.webhook_url, json{msg_type: text, content: {text: text}}) classmethod def from_crawler(cls, crawler): return cls(crawler.settings.get(NOTIFY_WEBHOOK_URL))这里用的是 item 通过去重 Pipeline 后数量也就是指纹判定为“新增”的数据量不是原始请求数。如果item_count一直挂零优先检查指纹键是否过期堆积——农业数据按日期生成指纹一旦 Redis 键里的日期永远不更新新增数据就会被误判为重复。5. Docker Scrapyd 部署把开发环境复刻到服务器5.1 Dockerfile 里最容易忽略的两个依赖时区与中文字体农业数据的发布和调度都依赖本地时间Docker 容器默认的 UTC 时区会导致“每天 09:00 跑批”变成“北京时间 17:00 跑批”。这是部署后最先踩的坑。此外部分农业数据网站依赖中文字体渲染验证码或动态图表如果后续接入 OCR 识别验证码容器里没有中文字体识别率会直线下降。一个可用的 Dockerfile 如下FROM python:3.11-slim WORKDIR /app # 先装系统依赖layers 优化改动频率低于 requirements 的层放前面 RUN apt-get update apt-get install -y --no-install-recommends \ tzdata \ fonts-noto-cjk \ curl \ ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \ echo Asia/Shanghai /etc/timezone \ rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 6800 CMD [scrapyd]时区的处理是ln -sf创建软链到本地时间文件再写入/etc/timezone。fonts-noto-cjk是文泉驿之外最常见的开源中文字体包体积约几十 MB但能避免后续 OCR 识别时中文乱码的问题。requirements.txt里至少要包含scrapy、scrapyd、scrapy-playwright、redis、requestsscrapyd-client只在发布机上需要不用打进运行镜像。5.2 scrapyd-deploy 上传的 target 配置与版本升级发布流程通常是在开发机执行scrapyd-deploy把项目打包上传到服务器的 Scrapyd。配置放在项目根目录的scrapy.cfg[deploy:production] url http://your-server:6800/ project agri_prices然后执行scrapyd-deploy production -p agri_prices --version 20250603--version参数如果不指定默认会取当前时间戳但显式指定版本号更利于回滚。上传成功后服务器端会保留历史版本通过 API 查询curl http://your-server:6800/listversions.json?projectagri_prices有个高频报错需要记住scrapyd-deploy提示Unhandled Error往往是因为服务器 Scrapyd 版本与开发机scrapyd-client版本不一致通用做法是两边都升级到当前稳定的 2.x 系列避免老版本对 egg 格式的兼容性问题。另外注意scrapy.cfg里的 url 不要带http://localhost:6800/否则发布时会打到开发机的 6800 端口而不是服务器的。5.3 部署后的访问控制只允许本机查看 job 日志Scrapyd 默认监听0.0.0.0:6800这意味着公网任何一台机器都可以往你的 Scrapyd 提交任务、删除版本风险极大。安全配置第一件事是把绑定地址改掉只监听内网或本机scrapyd-deploy production -p agri_prices --version 20250603这里配合的修改是在scrapyd.conf[scrapyd] bind_address 127.0.0.1 http_port 6800改完后调用方只能从本机访问。如果确实需要远程查看任务状态用 SSH 隧道转发 6800 端口而不是直接暴露ssh -N -L 6800:localhost:6800 useryour-server访问控制之外max_proc参数也值得关注它决定了 Scrapyd 同时能跑多少个进程。如果机器有 4 核 CPU可以设置max_proc 4 max_proc_per_cpu 1这个配置避免同一个爬虫被调度时疯狂开进程内存被多个 Scrapy 实例吃满。日志目录默认在/var/log/scrapyd用 Docker 部署时务必挂载宿主机目录否则容器重建后历史日志全部丢失排查数据质量问题时无据可查。6. 用自检抓取脚本验证数据新鲜度不依赖监控面板与其搭建一整套监控系统不如写一个轻量自检脚本每天检查“今天的目标站点是否产生了新数据”。脚本独立于 Scrapy 项目只做一件事查询 Redis 里的指纹集合和数据库表里的最新日期然后对比当前日期。import redis import pymysql from datetime import datetime r redis.Redis(hostlocalhost, port6379, db1) # 检查 pub_date 维度是否达到当日采集预期 db pymysql.connect(hostlocalhost, useragri, password****, databaseagri_mart) try: with db.cursor() as cursor: # 查出采集表中最新的发布日期 cursor.execute(SELECT MAX(publish_date) FROM price_daily WHERE sourcewholesale) latest_date cursor.fetchone()[0] today datetime.now().date() # latest_date 是 date 类型如果是昨天说明当日批任务漏跑 if latest_date today: print(f[警告] 数据停留在 {latest_date}今日尚未采集) else: print(f[正常] 最新数据日期 {latest_date}) finally: db.close()脚本配合 cron 每天执行一次输出追加到日志文件如果发现latest_date比当前日期晚一天以上就说明调度或解析出了问题。这个自检方法虽然粗糙但比看 Scrapyd 的任务状态要可靠得多——任务状态显示“已完成”不代表页面里的数据已经被正确解析入库只有数据库里的最大发布日期才能代表真实的数据新鲜度。线上爬虫出问题的环节一半以上不在请求失败而在解析规则因为页面改版失效后任务照常跑、数据照常抓但字段全部为空或错位最终入库数据量骤减问题被宏大的监控面板淹没在 200 状态码里。本文还有配套的精品资源点击获取