简介基于PythonScrapy的农业数据爬虫设计与部署方案面向计算机相关专业学生、教师及企业IT人员适用于课程设计、毕业设计或项目初期演示。资料包共34个文件包含10个Python源码、9个pyc编译文件、6个HTML页面以及cfg配置、Markdown文档、zbak备份和DS_Store等辅助文件整体仅35KB按模块组织便于查阅。目前已有67人学习下载。方案从农业数据需求出发系统讲解了数据源确定、爬虫架构设计、Item/Pipeline/Spider/Scheduler等核心组件的实现与协作并通过中间件处理异常、利用Scrapy工具优化抓取流程同时覆盖服务器搭建、环境配置、安全策略及系统监控等部署维护环节。代码经测试运行正常设计已获导师认可、答辩评分95分兼具完整性与实用性可直接作为学习范本也适合在此基础上进行功能扩展和二次开发。1. 农业数据爬虫为什么值得用 Scrapy 重写一遍农业数据采集的难点从来不是“没有数据”而是数据源太散农产品批发市场、气象站点、地方农业信息网各自用不同的表格结构和单位且不少站点把价格、成交量藏在分页和动态 iframe 里。如果每次都用 requests 手写一个脚本维护成本会很快超过数据本身的价值。用 Python Scrapy 统一做一版农业数据爬虫可以把“解析列表页、清洗字段、控制并发、定期部署”这些重复动作收敛到一套架构里。Scrapy 的异步并发模型配合下载中间件抓取效率并不输给手动调 ThreadPool 的方案更重要的是它自带重试、日志、统计和数据管道适合从单机脚本演进成持续运行的数据管道。这篇文章会从项目建模开始一直到服务器上的部署监控讲清每一步的关键参数和常见坑位。2. 从农业数据源到 Scrapy 项目结构设计与建模2.1 理清农业数据站点的三种页面特征农业数据爬虫和普通垂直爬虫最大的差异在于数据源本身的异构程度。常见的农产品价格站点大致分三类一类是政府农业信息网的“今日价格”列表页URL 规律但分类参数经常是中文编码另一类是批发市场的详情页价格表静态渲染在 HTML 中但同一行里既有元/斤也有元/公斤还有一类是集成了图表库的平台展示层使用 iframe 嵌套真实数据在一个独立的 JSON 接口里。这三类页面决定了爬虫不能只用一套 parse 函数走遍所有站点。更合理的做法是建一个基类 spider把公共的列表页翻页、详情页入口解析放进去再把不同站点的字段转换成统一的 Item 结构。这样后面加新数据源时只需要增加一个 spider 文件不需要改动 pipeline。反爬强度也完全不同政府类站点通常只要控制频率即可市场类站点会有简单的 User-Agent 校验图表类站点则往往需要先拿到 iframe 的 src 地址再发起二次请求。基础设施方面建议先用 conda 创建独立的 Python 环境避免和系统 Python 打架。Python 3.11 在 Scrapy 2.11 以上版本运行很稳定同时也能兼容 scrapy-playwright、pandas 等依赖。下面的命令创建环境并安装核心库conda create -n agri-scrapy python3.11 -y conda activate agri-scrapy pip install scrapy scrapyd scrapyd-client pandas pymongo scrapy startproject agridata这里的scrapyd是后续部署要用的守护进程scrapyd-client负责把项目打包上传。pandas和pymongo不一定所有项目需要但农业数据最终通常要落到数据库或报表里提前装上能省掉不少临时处理。执行scrapy startproject agridata后会生成标准的 Scrapy 目录结构其中items.py、middlewares.py、pipelines.py、settings.py是这一版爬虫的核心文件。2.2 定义一个能兼容多来源的农业数据 Item农业数据字段看起来简单实际对齐起来很麻烦。以农产品价格为例有的站点给的是“批发价”有的给的是“最高价/最低价/均价”有的甚至把产地信息混在备注里。如果每个站点写成不同的字典结构后面做分析时还要写一堆清洗脚本。所以在items.py中用一个统一结构是必须的字段可以多不能少。常见做法是先在代码中的items.py里定义清楚字段和清洗管线import scrapy from scrapy.loader import ItemLoader from scrapy.loader.processors import MapCompose, TakeFirst def parse_price(value): if isinstance(value, str): value value.replace(元/公斤, ).replace(元/斤, ).strip() try: return float(value) except (TypeError, ValueError): return None def parse_date(value): value value.replace(年, -).replace(月, -).replace(日, ) return value class AgriculturalDataItem(scrapy.Item): region scrapy.Field(output_processTakeFirst()) crop_name scrapy.Field(output_processTakeFirst()) date scrapy.Field( input_processMapCompose(parse_date), output_processTakeFirst() ) avg_price scrapy.Field( input_processMapCompose(parse_price), output_processTakeFirst() ) min_price scrapy.Field( input_processMapCompose(parse_price), output_processTakeFirst() ) max_price scrapy.Field( input_processMapCompose(parse_price), output_processTakeFirst() ) unit scrapy.Field(output_processTakeFirst()) source_url scrapy.Field(output_processTakeFirst()) crawled_at scrapy.Field(output_processTakeFirst())这里的MapCompose是 Scrapy 自带的输入处理函数它会把列表页传入的原始字符串依次经过parse_price、parse_date这些函数清洗再把结果交给TakeFirst取出第一个非空值。parse_price里做的单位替换很关键因为农业站点同类数据的单位经常混用统一成以公斤为单位可以省掉后续分析时的换算步骤。如果遇到无法解析的值函数返回None后续管道可以判断该字段是否为空再决定是否丢弃。2.3 用列表页 详情页两层解析实现数据抓取在spiders目录下创建具体的站点爬虫。下面这个例子模拟了一个常见的农产品价格站列表页有分页参数详情页是单独的 HTML 页面。爬虫的start_requests方法生成多个列表页请求parse_list提取详情页链接后交给parse_detail填充 Item。import scrapy from agridata.items import AgriculturalDataItem from scrapy.loader import ItemLoader class MarketPriceSpider(scrapy.Spider): name market_price allowed_domains [example.com] base_url https://example.com/market/list?page{} def start_requests(self): for page in range(1, 11): url self.base_url.format(page) yield scrapy.Request(url, callbackself.parse_list, meta{page: page}) def parse_list(self, response): rows response.css(table.price-table tr) for row in rows: detail_link row.css(a::attr(href)).get() if detail_link: yield response.follow(detail_link, callbackself.parse_detail) def parse_detail(self, response): loader ItemLoader(itemAgriculturalDataItem(), responseresponse) loader.add_css(region, span.region::text) loader.add_css(crop_name, h1.crop-name::text) loader.add_css(date, span.date::text) loader.add_css(avg_price, td.avg::text) loader.add_css(min_price, td.min::text) loader.add_css(max_price, td.max::text) loader.add_css(unit, td.unit::text) loader.add_value(source_url, response.url) loader.add_value(crawled_at, response.headers.get(date, b).decode(utf-8)) yield loader.load_item()response.follow会自动拼接详情页的绝对 URL同时继承当前请求的 meta 数据。ItemLoader配合add_css会将 CSS 选择器提取到的所有值放入该字段的输入管线最终输出的就是清洗好的字典。crawled_at在这里取了响应头里的日期实际上用datetime.now()更常见因为不同站点服务器时间不一定准确。字段在进入 Item 时没有经过parse_detail之外的硬编码好处是后续遇到一个新站点时只需要把 CSS 选择器替换掉即可。下面这张表总结了不同农业数据源的建模要点设计 Item 时可以对照检查数据源类型典型字段单位风险建议校验方式农产品价格地区、品种、价格、成交量元/斤与元/公斤混用统一转公斤后保留原始单位气象数据日最高/最低温、降雨量无单位问题检查日期是否连续农业政策标题、正文、发布日期无按标题 MD5 去重种植面积省份、作物、年份、面积亩与公顷混用统一为公顷3. Scrapy 中间件与并发参数农业数据抓取不被封的底牌3.1 并发与限速参数怎么调农业数据站的服务器一般不会上特别复杂的反爬系统但并发过大依然会触发 403 或连接超时。很多初学者把CONCURRENT_REQUESTS调到 32 以上结果反而是大量重试降低了整体效率。我一般在settings.py里首先确认下面这些参数CONCURRENT_REQUESTS 16 CONCURRENT_REQUESTS_PER_DOMAIN 4 CONCURRENT_REQUESTS_PER_IP 4 DOWNLOAD_DELAY 0.8 AUTOTHROTTLE_ENABLED True AUTOTHROTTLE_START_DELAY 1.0 AUTOTHROTTLE_MAX_DELAY 15.0 AUTOTHROTTLE_TARGET_CONCURRENCY 4.0 RANDOMIZE_DOWNLOAD_DELAY True RETRY_TIMES 3 RETRY_HTTP_CODES [500, 502, 503, 504, 408, 429] DOWNLOAD_TIMEOUT 20CONCURRENT_REQUESTS_PER_DOMAIN比全局并发数更重要它限制了同一域名下的最大并发请求这个值在 4 到 8 之间比较稳妥。DOWNLOAD_DELAY是相邻请求的固定延迟设成 0.8 秒左右即可。开启AUTOTHROTTLE_ENABLED后Scrapy 会根据响应时间动态调整延迟如果响应变慢会自动拉长间隔。AUTOTHROTTLE_TARGET_CONCURRENCY建议设为目标并发数的一半这样既不会太激进又能保证吞吐量。RANDOMIZE_DOWNLOAD_DELAY会在基础延迟上加上一个随机值避免请求节奏太规律被识别。下面这张参数表可以直接复制到笔记里调优时逐项检查参数名作用推荐初始值出现 403 时怎么改CONCURRENT_REQUESTS_PER_DOMAIN单域名并发4降到 2DOWNLOAD_DELAY固定延迟0.8提高到 2AUTOTHROTTLE_MAX_DELAY最大动态延迟15保持RETRY_TIMES失败重试次数3提高但别超过 5RETRY_HTTP_CODES需要重试的状态码500,502,503,504,408,429加上 403 但配合限速3.2 使用 Downloader Middleware 轮换 User-Agent 与代理池在middlewares.py中定义一个 User-Agent 轮换中间件是最常见的防治初级反爬的手段。Scrapy 默认使用scrapy.downloadermiddlewares.useragent.UserAgentMiddleware我们可以继承它并替换默认 UAimport random from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware class RotateUserAgentMiddleware(UserAgentMiddleware): user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15, Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X), ] def process_request(self, request, spider): request.headers.setdefault(User-Agent, random.choice(self.user_agents))process_request会在请求发出前被调用setdefault表示只有当请求头没有 UA 时才设置避免覆盖已在Request里手工指定的值。对于农业数据站点UA 轮换通常已经够用但如果遇到 IP 维度限流就需要接入代理中间件。常见做法是先把代理列表写在配置里然后在process_request中为每个请求轮换一个代理 IPrequest.meta[proxy] random.choice(proxy_list)。确保代理服务来自合规渠道不要使用任何未授权的代理工具。如果在抓取过程中频繁收到 429 状态码说明触发频控。此时应该在中间件里加一个retry_http_codes或直接在handle_httpcode_list中处理 429并配合延迟重试。最简单的实现是通过RETRY_HTTP_CODES配合RETRY_TIMES完成但要注意不要把重试间隔设成 0否则会雪崩。3.3 动态 iframe 与 scrapy-playwright 的桥接方案农业数据平台的图表模块很流行用 iframe 嵌套比如某市场信息网的价格走势图嵌入一个/charts/price?cropIdxxx的页面。这时候直接抓详情页是拿不到图表数据的必须先解析出 iframe 的src属性再请求该地址。结构简单时可以直接在 spider 中用response.css(iframe::attr(src)).get()获取真实地址然后yield scrapy.Request(url)继续解析。这是最稳定的方式不需要任何额外依赖。如果 iframe 内部还依赖 JavaScript 渲染用requests就会失效。这时候推荐接入scrapy-playwright让 Scrapy 驱动一个无头浏览器。安装步骤为pip install scrapy-playwright playwright然后执行playwright install chromium下载浏览器内核。在settings.py中加入以下配置DOWNLOAD_HANDLERS { http: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, https: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, }在 spider 中对需要动态渲染的请求添加meta{playwright: True}yield scrapy.Request( urlchart_url, callbackself.parse_chart, meta{playwright: True} )此时parse_chart拿到的response已经是浏览器执行完 JS 之后的 DOM。scrapy-playwright 也支持截图、等待选择器等操作通过meta{playwright_include_page: True}可以拿到页面对象。不过每开启一个浏览器实例都会占用 500MB 级内存生产环境要控制并发数最好只在真正需要渲染的页面使用不要全局启用。4. 用 Scrapy 部署方案落地Scrapyd Docker 与定时调度4.1 用 Scrapyd 发布项目并调用 API本地爬虫能跑通只是第一步农业数据要求每天定时抓取所以必须把项目部署到独立服务器上。Scrapyd 是 Scrapy 官方维护的进程守护工具它会把爬虫项目打包成 egg 文件然后通过 HTTP API 启动任务。先在服务器上安装依赖pip install scrapyd scrapyd-client scrapyd默认监听0.0.0.0:6800。然后在项目的scrapy.cfg中添加部署配置[deploy:agri] url http://your-server:6800/ project agridataagri是部署目标的别名可以随环境切换。url指向服务器的 Scrapyd 地址注意如果开启了防火墙要放行 6800 端口。发布命令scrapyd-deploy agri -p agridata-p参数指定项目名必须和settings.py中的BOT_NAME或实际项目名一致。部署完成后通过 API 触发爬虫curl http://your-server:6800/schedule.json -d projectagridata -d spidermarket_priceschedule.json接口会立即返回一个jobid这是该次任务运行实例的唯一标识。Scrapyd 还提供了listjobs.json用于查询任务状态cancel.json用于取消运行中的任务。下面是一个完整的查询命令示例curl http://your-server:6800/listjobs.json?projectagridata结果中的running、pending、finished三个数组分别表示当前运行、排队和已完成的任务列表start_time和end_time字段可以用来判断任务是否超时。4.2 用 Docker 部署 Scrapyd 与数据库联动Scrapyd 直接跑在宿主机上虽然简单但环境容易漂移。农业数据爬虫涉及的依赖包括 lxml、twisted、pymongo甚至可能包含 scrapy-playwright一旦换机器就要重新手动装一遍。用 Docker 把整个运行环境固化是最常见的部署方案。项目根目录下创建DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple COPY . /app EXPOSE 6800 CMD [scrapyd]python:3.11-slim镜像体积适中requirements.txt里应包含scrapy、scrapyd、pymongo、scrapyd-client等依赖。如果使用 scrapy-playwright还需要在镜像中安装 Chromium所以更稳妥的写法是FROM python:3.11并使用playwright install --with-deps chromium但基础镜像会大很多。用docker-compose.yml把 Scrapyd 和 MongoDB、Redis 一起管理起来version: 3.8 services: scrapyd: build: . container_name: agri-scrapyd ports: - 6800:6800 volumes: - ./eggs:/app/eggs depends_on: - mongo - redis mongo: image: mongo:6.0 container_name: agri-mongo volumes: - mongo_data:/data/db redis: image: redis:7-alpine container_name: agri-redis volumes: mongo_data:depends_on保证了 Scrapyd 启动时数据库已经就绪。把eggs目录挂在宿主机上可以避免容器重建后丢失已部署的代码包。启动命令是docker-compose up -d查看日志用docker-compose logs -f scrapyd。此时scrapyd-deploy agri -p agridata的目标地址仍然是http://your-server:6800/因为端口已经映射出来了。4.3 增量抓取与定时调度避免每天重复全量数据农产品价格数据每天都在更新如果每次部署都重新抓全站不仅浪费资源还容易被限制访问。常见的做法是在 spider 中维护一个增量字段比如页面上有当天的日期或者数据更新时间只有满足条件才进入详情页。更通用的是用 Redis 做一个去重集合把每次请求的 URL 或某个唯一字段加入集合如果已经存在则跳过。在middlewares.py或pipelines.py中用redis-py实现import redis r redis.Redis(hostredis, port6379, db0) def is_duplicate(url): return r.sadd(agri_urls, url) 0sadd返回 0 表示该 URL 已经存在返回 1 表示新增成功。这样重复任务只会抓取新的数据历史数据留在数据库中。定时调度不需要额外安装复杂调度平台官方推荐的搭配是用系统的 cron 配合 curl 调 Scrapyd API。也可以写一个 Python 调度脚本使用 APScheduler 在容器中运行from apscheduler.schedulers.blocking import BlockingScheduler import requests def run_market_price(): resp requests.post( http://scrapyd:6800/schedule.json, data{project: agridata, spider: market_price} ) print(schedule result:, resp.json()) scheduler BlockingScheduler() scheduler.add_job(run_market_price, cron, hour8, minute30, timezoneAsia/Shanghai) scheduler.start()这里容器内的scrapyd由于在同一 compose 网络中可以直接用服务名访问。timezone要显式指定否则默认使用容器 UTC 时间导致任务在下午才执行。APScheduler 的BlockingScheduler适合单任务场景如果调度任务很多可以改用BackgroundScheduler并集成到已有应用中。5. 验证农业数据爬虫的抓取质量与监控状态5.1 用 Scrapy 日志统计确认任务没有空跑爬虫部署后最怕的是每天都在跑但数据量突然跌到 0。Scrapy 每次任务结束都会输出 stats 信息其中最关键的是item_scraped_count、item_dropped_count和downloader/response_status_count/403。运行任务时加上--logfile参数可以持久化日志便于事后排查scrapy crawl market_price --logfile crawl.log任务结束后用grep提取关键统计grep item_scraped_count crawl.log如果发现item_scraped_count连续几天为 0优先检查站点是否改版其次检查DOWNLOAD_DELAY是否被反爬击中。log.log中的downloader/exception_count能帮助确认网络层问题。5.2 在 Pipeline 中做幂等和字段质量检查在pipelines.py中加入一个质量检查管道。每一条 item 先检查必填字段是否存在再计算空值比例。下面是一个精简的校验 pipelineclass QualityCheckPipeline: required_fields [region, crop_name, date, avg_price] def process_item(self, item, spider): for field in self.required_fields: if not item.get(field): raise DropItem(fMissing {field} in {item.get(crop_name, )}) return item结合第 4 章的 Redis 去重可以在is_duplicate之前先检查必填字段把脏数据挡在增量去重之前。对于农业价格数据建议把avg_price是否为空和date是否合法纳入必填检查否则数据分析时会出现单价 0 或未来日期。5.3 把任务心跳接入内部监控系统定时任务最怕不是没有数据而是任务根本没有启动。Scrapyd 的listjobs.json接口可以拿到最近任务列表写一个监控脚本每隔 10 分钟检查一次import requests from datetime import datetime, timedelta jobs requests.get(http://scrapyd:6800/listjobs.json?projectagridata).json() running jobs.get(running, []) finished jobs.get(finished, []) if len(running) 0 and len(finished) 0: last_time datetime.fromisoformat(finished[-1][start_time].replace(Z, 00:00)) if datetime.now(timezone.utc) - last_time timedelta(days1): # 触发 webhook 报警 pass判断逻辑是检查是否存在运行中的任务以及最近一次完成的任务是否距今超过一天。如果超过阈值就通过 Webhook 发送报警到企业微信或钉钉群。这个脚本可以放在独立的容器中也可以用宿主机的 cron 执行关键是报警不是靠人肉去看 Scrapyd 页面。最后再分享一个实用技巧在 Item 中始终保留date原始字符串和解析后的date_parsed字段不要直接覆盖原始值。这样当站点修改了日期格式导致清洗异常时可以通过原始字符串反推出正确的解析规则而不需要重新抓取全量历史数据。本文还有配套的精品资源点击获取 Scrapy农业数据爬虫实战：从项目建模到部署监控
简介基于PythonScrapy的农业数据爬虫设计与部署方案面向计算机相关专业学生、教师及企业IT人员适用于课程设计、毕业设计或项目初期演示。资料包共34个文件包含10个Python源码、9个pyc编译文件、6个HTML页面以及cfg配置、Markdown文档、zbak备份和DS_Store等辅助文件整体仅35KB按模块组织便于查阅。目前已有67人学习下载。方案从农业数据需求出发系统讲解了数据源确定、爬虫架构设计、Item/Pipeline/Spider/Scheduler等核心组件的实现与协作并通过中间件处理异常、利用Scrapy工具优化抓取流程同时覆盖服务器搭建、环境配置、安全策略及系统监控等部署维护环节。代码经测试运行正常设计已获导师认可、答辩评分95分兼具完整性与实用性可直接作为学习范本也适合在此基础上进行功能扩展和二次开发。1. 农业数据爬虫为什么值得用 Scrapy 重写一遍农业数据采集的难点从来不是“没有数据”而是数据源太散农产品批发市场、气象站点、地方农业信息网各自用不同的表格结构和单位且不少站点把价格、成交量藏在分页和动态 iframe 里。如果每次都用 requests 手写一个脚本维护成本会很快超过数据本身的价值。用 Python Scrapy 统一做一版农业数据爬虫可以把“解析列表页、清洗字段、控制并发、定期部署”这些重复动作收敛到一套架构里。Scrapy 的异步并发模型配合下载中间件抓取效率并不输给手动调 ThreadPool 的方案更重要的是它自带重试、日志、统计和数据管道适合从单机脚本演进成持续运行的数据管道。这篇文章会从项目建模开始一直到服务器上的部署监控讲清每一步的关键参数和常见坑位。2. 从农业数据源到 Scrapy 项目结构设计与建模2.1 理清农业数据站点的三种页面特征农业数据爬虫和普通垂直爬虫最大的差异在于数据源本身的异构程度。常见的农产品价格站点大致分三类一类是政府农业信息网的“今日价格”列表页URL 规律但分类参数经常是中文编码另一类是批发市场的详情页价格表静态渲染在 HTML 中但同一行里既有元/斤也有元/公斤还有一类是集成了图表库的平台展示层使用 iframe 嵌套真实数据在一个独立的 JSON 接口里。这三类页面决定了爬虫不能只用一套 parse 函数走遍所有站点。更合理的做法是建一个基类 spider把公共的列表页翻页、详情页入口解析放进去再把不同站点的字段转换成统一的 Item 结构。这样后面加新数据源时只需要增加一个 spider 文件不需要改动 pipeline。反爬强度也完全不同政府类站点通常只要控制频率即可市场类站点会有简单的 User-Agent 校验图表类站点则往往需要先拿到 iframe 的 src 地址再发起二次请求。基础设施方面建议先用 conda 创建独立的 Python 环境避免和系统 Python 打架。Python 3.11 在 Scrapy 2.11 以上版本运行很稳定同时也能兼容 scrapy-playwright、pandas 等依赖。下面的命令创建环境并安装核心库conda create -n agri-scrapy python3.11 -y conda activate agri-scrapy pip install scrapy scrapyd scrapyd-client pandas pymongo scrapy startproject agridata这里的scrapyd是后续部署要用的守护进程scrapyd-client负责把项目打包上传。pandas和pymongo不一定所有项目需要但农业数据最终通常要落到数据库或报表里提前装上能省掉不少临时处理。执行scrapy startproject agridata后会生成标准的 Scrapy 目录结构其中items.py、middlewares.py、pipelines.py、settings.py是这一版爬虫的核心文件。2.2 定义一个能兼容多来源的农业数据 Item农业数据字段看起来简单实际对齐起来很麻烦。以农产品价格为例有的站点给的是“批发价”有的给的是“最高价/最低价/均价”有的甚至把产地信息混在备注里。如果每个站点写成不同的字典结构后面做分析时还要写一堆清洗脚本。所以在items.py中用一个统一结构是必须的字段可以多不能少。常见做法是先在代码中的items.py里定义清楚字段和清洗管线import scrapy from scrapy.loader import ItemLoader from scrapy.loader.processors import MapCompose, TakeFirst def parse_price(value): if isinstance(value, str): value value.replace(元/公斤, ).replace(元/斤, ).strip() try: return float(value) except (TypeError, ValueError): return None def parse_date(value): value value.replace(年, -).replace(月, -).replace(日, ) return value class AgriculturalDataItem(scrapy.Item): region scrapy.Field(output_processTakeFirst()) crop_name scrapy.Field(output_processTakeFirst()) date scrapy.Field( input_processMapCompose(parse_date), output_processTakeFirst() ) avg_price scrapy.Field( input_processMapCompose(parse_price), output_processTakeFirst() ) min_price scrapy.Field( input_processMapCompose(parse_price), output_processTakeFirst() ) max_price scrapy.Field( input_processMapCompose(parse_price), output_processTakeFirst() ) unit scrapy.Field(output_processTakeFirst()) source_url scrapy.Field(output_processTakeFirst()) crawled_at scrapy.Field(output_processTakeFirst())这里的MapCompose是 Scrapy 自带的输入处理函数它会把列表页传入的原始字符串依次经过parse_price、parse_date这些函数清洗再把结果交给TakeFirst取出第一个非空值。parse_price里做的单位替换很关键因为农业站点同类数据的单位经常混用统一成以公斤为单位可以省掉后续分析时的换算步骤。如果遇到无法解析的值函数返回None后续管道可以判断该字段是否为空再决定是否丢弃。2.3 用列表页 详情页两层解析实现数据抓取在spiders目录下创建具体的站点爬虫。下面这个例子模拟了一个常见的农产品价格站列表页有分页参数详情页是单独的 HTML 页面。爬虫的start_requests方法生成多个列表页请求parse_list提取详情页链接后交给parse_detail填充 Item。import scrapy from agridata.items import AgriculturalDataItem from scrapy.loader import ItemLoader class MarketPriceSpider(scrapy.Spider): name market_price allowed_domains [example.com] base_url https://example.com/market/list?page{} def start_requests(self): for page in range(1, 11): url self.base_url.format(page) yield scrapy.Request(url, callbackself.parse_list, meta{page: page}) def parse_list(self, response): rows response.css(table.price-table tr) for row in rows: detail_link row.css(a::attr(href)).get() if detail_link: yield response.follow(detail_link, callbackself.parse_detail) def parse_detail(self, response): loader ItemLoader(itemAgriculturalDataItem(), responseresponse) loader.add_css(region, span.region::text) loader.add_css(crop_name, h1.crop-name::text) loader.add_css(date, span.date::text) loader.add_css(avg_price, td.avg::text) loader.add_css(min_price, td.min::text) loader.add_css(max_price, td.max::text) loader.add_css(unit, td.unit::text) loader.add_value(source_url, response.url) loader.add_value(crawled_at, response.headers.get(date, b).decode(utf-8)) yield loader.load_item()response.follow会自动拼接详情页的绝对 URL同时继承当前请求的 meta 数据。ItemLoader配合add_css会将 CSS 选择器提取到的所有值放入该字段的输入管线最终输出的就是清洗好的字典。crawled_at在这里取了响应头里的日期实际上用datetime.now()更常见因为不同站点服务器时间不一定准确。字段在进入 Item 时没有经过parse_detail之外的硬编码好处是后续遇到一个新站点时只需要把 CSS 选择器替换掉即可。下面这张表总结了不同农业数据源的建模要点设计 Item 时可以对照检查数据源类型典型字段单位风险建议校验方式农产品价格地区、品种、价格、成交量元/斤与元/公斤混用统一转公斤后保留原始单位气象数据日最高/最低温、降雨量无单位问题检查日期是否连续农业政策标题、正文、发布日期无按标题 MD5 去重种植面积省份、作物、年份、面积亩与公顷混用统一为公顷3. Scrapy 中间件与并发参数农业数据抓取不被封的底牌3.1 并发与限速参数怎么调农业数据站的服务器一般不会上特别复杂的反爬系统但并发过大依然会触发 403 或连接超时。很多初学者把CONCURRENT_REQUESTS调到 32 以上结果反而是大量重试降低了整体效率。我一般在settings.py里首先确认下面这些参数CONCURRENT_REQUESTS 16 CONCURRENT_REQUESTS_PER_DOMAIN 4 CONCURRENT_REQUESTS_PER_IP 4 DOWNLOAD_DELAY 0.8 AUTOTHROTTLE_ENABLED True AUTOTHROTTLE_START_DELAY 1.0 AUTOTHROTTLE_MAX_DELAY 15.0 AUTOTHROTTLE_TARGET_CONCURRENCY 4.0 RANDOMIZE_DOWNLOAD_DELAY True RETRY_TIMES 3 RETRY_HTTP_CODES [500, 502, 503, 504, 408, 429] DOWNLOAD_TIMEOUT 20CONCURRENT_REQUESTS_PER_DOMAIN比全局并发数更重要它限制了同一域名下的最大并发请求这个值在 4 到 8 之间比较稳妥。DOWNLOAD_DELAY是相邻请求的固定延迟设成 0.8 秒左右即可。开启AUTOTHROTTLE_ENABLED后Scrapy 会根据响应时间动态调整延迟如果响应变慢会自动拉长间隔。AUTOTHROTTLE_TARGET_CONCURRENCY建议设为目标并发数的一半这样既不会太激进又能保证吞吐量。RANDOMIZE_DOWNLOAD_DELAY会在基础延迟上加上一个随机值避免请求节奏太规律被识别。下面这张参数表可以直接复制到笔记里调优时逐项检查参数名作用推荐初始值出现 403 时怎么改CONCURRENT_REQUESTS_PER_DOMAIN单域名并发4降到 2DOWNLOAD_DELAY固定延迟0.8提高到 2AUTOTHROTTLE_MAX_DELAY最大动态延迟15保持RETRY_TIMES失败重试次数3提高但别超过 5RETRY_HTTP_CODES需要重试的状态码500,502,503,504,408,429加上 403 但配合限速3.2 使用 Downloader Middleware 轮换 User-Agent 与代理池在middlewares.py中定义一个 User-Agent 轮换中间件是最常见的防治初级反爬的手段。Scrapy 默认使用scrapy.downloadermiddlewares.useragent.UserAgentMiddleware我们可以继承它并替换默认 UAimport random from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware class RotateUserAgentMiddleware(UserAgentMiddleware): user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15, Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X), ] def process_request(self, request, spider): request.headers.setdefault(User-Agent, random.choice(self.user_agents))process_request会在请求发出前被调用setdefault表示只有当请求头没有 UA 时才设置避免覆盖已在Request里手工指定的值。对于农业数据站点UA 轮换通常已经够用但如果遇到 IP 维度限流就需要接入代理中间件。常见做法是先把代理列表写在配置里然后在process_request中为每个请求轮换一个代理 IPrequest.meta[proxy] random.choice(proxy_list)。确保代理服务来自合规渠道不要使用任何未授权的代理工具。如果在抓取过程中频繁收到 429 状态码说明触发频控。此时应该在中间件里加一个retry_http_codes或直接在handle_httpcode_list中处理 429并配合延迟重试。最简单的实现是通过RETRY_HTTP_CODES配合RETRY_TIMES完成但要注意不要把重试间隔设成 0否则会雪崩。3.3 动态 iframe 与 scrapy-playwright 的桥接方案农业数据平台的图表模块很流行用 iframe 嵌套比如某市场信息网的价格走势图嵌入一个/charts/price?cropIdxxx的页面。这时候直接抓详情页是拿不到图表数据的必须先解析出 iframe 的src属性再请求该地址。结构简单时可以直接在 spider 中用response.css(iframe::attr(src)).get()获取真实地址然后yield scrapy.Request(url)继续解析。这是最稳定的方式不需要任何额外依赖。如果 iframe 内部还依赖 JavaScript 渲染用requests就会失效。这时候推荐接入scrapy-playwright让 Scrapy 驱动一个无头浏览器。安装步骤为pip install scrapy-playwright playwright然后执行playwright install chromium下载浏览器内核。在settings.py中加入以下配置DOWNLOAD_HANDLERS { http: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, https: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, }在 spider 中对需要动态渲染的请求添加meta{playwright: True}yield scrapy.Request( urlchart_url, callbackself.parse_chart, meta{playwright: True} )此时parse_chart拿到的response已经是浏览器执行完 JS 之后的 DOM。scrapy-playwright 也支持截图、等待选择器等操作通过meta{playwright_include_page: True}可以拿到页面对象。不过每开启一个浏览器实例都会占用 500MB 级内存生产环境要控制并发数最好只在真正需要渲染的页面使用不要全局启用。4. 用 Scrapy 部署方案落地Scrapyd Docker 与定时调度4.1 用 Scrapyd 发布项目并调用 API本地爬虫能跑通只是第一步农业数据要求每天定时抓取所以必须把项目部署到独立服务器上。Scrapyd 是 Scrapy 官方维护的进程守护工具它会把爬虫项目打包成 egg 文件然后通过 HTTP API 启动任务。先在服务器上安装依赖pip install scrapyd scrapyd-client scrapyd默认监听0.0.0.0:6800。然后在项目的scrapy.cfg中添加部署配置[deploy:agri] url http://your-server:6800/ project agridataagri是部署目标的别名可以随环境切换。url指向服务器的 Scrapyd 地址注意如果开启了防火墙要放行 6800 端口。发布命令scrapyd-deploy agri -p agridata-p参数指定项目名必须和settings.py中的BOT_NAME或实际项目名一致。部署完成后通过 API 触发爬虫curl http://your-server:6800/schedule.json -d projectagridata -d spidermarket_priceschedule.json接口会立即返回一个jobid这是该次任务运行实例的唯一标识。Scrapyd 还提供了listjobs.json用于查询任务状态cancel.json用于取消运行中的任务。下面是一个完整的查询命令示例curl http://your-server:6800/listjobs.json?projectagridata结果中的running、pending、finished三个数组分别表示当前运行、排队和已完成的任务列表start_time和end_time字段可以用来判断任务是否超时。4.2 用 Docker 部署 Scrapyd 与数据库联动Scrapyd 直接跑在宿主机上虽然简单但环境容易漂移。农业数据爬虫涉及的依赖包括 lxml、twisted、pymongo甚至可能包含 scrapy-playwright一旦换机器就要重新手动装一遍。用 Docker 把整个运行环境固化是最常见的部署方案。项目根目录下创建DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple COPY . /app EXPOSE 6800 CMD [scrapyd]python:3.11-slim镜像体积适中requirements.txt里应包含scrapy、scrapyd、pymongo、scrapyd-client等依赖。如果使用 scrapy-playwright还需要在镜像中安装 Chromium所以更稳妥的写法是FROM python:3.11并使用playwright install --with-deps chromium但基础镜像会大很多。用docker-compose.yml把 Scrapyd 和 MongoDB、Redis 一起管理起来version: 3.8 services: scrapyd: build: . container_name: agri-scrapyd ports: - 6800:6800 volumes: - ./eggs:/app/eggs depends_on: - mongo - redis mongo: image: mongo:6.0 container_name: agri-mongo volumes: - mongo_data:/data/db redis: image: redis:7-alpine container_name: agri-redis volumes: mongo_data:depends_on保证了 Scrapyd 启动时数据库已经就绪。把eggs目录挂在宿主机上可以避免容器重建后丢失已部署的代码包。启动命令是docker-compose up -d查看日志用docker-compose logs -f scrapyd。此时scrapyd-deploy agri -p agridata的目标地址仍然是http://your-server:6800/因为端口已经映射出来了。4.3 增量抓取与定时调度避免每天重复全量数据农产品价格数据每天都在更新如果每次部署都重新抓全站不仅浪费资源还容易被限制访问。常见的做法是在 spider 中维护一个增量字段比如页面上有当天的日期或者数据更新时间只有满足条件才进入详情页。更通用的是用 Redis 做一个去重集合把每次请求的 URL 或某个唯一字段加入集合如果已经存在则跳过。在middlewares.py或pipelines.py中用redis-py实现import redis r redis.Redis(hostredis, port6379, db0) def is_duplicate(url): return r.sadd(agri_urls, url) 0sadd返回 0 表示该 URL 已经存在返回 1 表示新增成功。这样重复任务只会抓取新的数据历史数据留在数据库中。定时调度不需要额外安装复杂调度平台官方推荐的搭配是用系统的 cron 配合 curl 调 Scrapyd API。也可以写一个 Python 调度脚本使用 APScheduler 在容器中运行from apscheduler.schedulers.blocking import BlockingScheduler import requests def run_market_price(): resp requests.post( http://scrapyd:6800/schedule.json, data{project: agridata, spider: market_price} ) print(schedule result:, resp.json()) scheduler BlockingScheduler() scheduler.add_job(run_market_price, cron, hour8, minute30, timezoneAsia/Shanghai) scheduler.start()这里容器内的scrapyd由于在同一 compose 网络中可以直接用服务名访问。timezone要显式指定否则默认使用容器 UTC 时间导致任务在下午才执行。APScheduler 的BlockingScheduler适合单任务场景如果调度任务很多可以改用BackgroundScheduler并集成到已有应用中。5. 验证农业数据爬虫的抓取质量与监控状态5.1 用 Scrapy 日志统计确认任务没有空跑爬虫部署后最怕的是每天都在跑但数据量突然跌到 0。Scrapy 每次任务结束都会输出 stats 信息其中最关键的是item_scraped_count、item_dropped_count和downloader/response_status_count/403。运行任务时加上--logfile参数可以持久化日志便于事后排查scrapy crawl market_price --logfile crawl.log任务结束后用grep提取关键统计grep item_scraped_count crawl.log如果发现item_scraped_count连续几天为 0优先检查站点是否改版其次检查DOWNLOAD_DELAY是否被反爬击中。log.log中的downloader/exception_count能帮助确认网络层问题。5.2 在 Pipeline 中做幂等和字段质量检查在pipelines.py中加入一个质量检查管道。每一条 item 先检查必填字段是否存在再计算空值比例。下面是一个精简的校验 pipelineclass QualityCheckPipeline: required_fields [region, crop_name, date, avg_price] def process_item(self, item, spider): for field in self.required_fields: if not item.get(field): raise DropItem(fMissing {field} in {item.get(crop_name, )}) return item结合第 4 章的 Redis 去重可以在is_duplicate之前先检查必填字段把脏数据挡在增量去重之前。对于农业价格数据建议把avg_price是否为空和date是否合法纳入必填检查否则数据分析时会出现单价 0 或未来日期。5.3 把任务心跳接入内部监控系统定时任务最怕不是没有数据而是任务根本没有启动。Scrapyd 的listjobs.json接口可以拿到最近任务列表写一个监控脚本每隔 10 分钟检查一次import requests from datetime import datetime, timedelta jobs requests.get(http://scrapyd:6800/listjobs.json?projectagridata).json() running jobs.get(running, []) finished jobs.get(finished, []) if len(running) 0 and len(finished) 0: last_time datetime.fromisoformat(finished[-1][start_time].replace(Z, 00:00)) if datetime.now(timezone.utc) - last_time timedelta(days1): # 触发 webhook 报警 pass判断逻辑是检查是否存在运行中的任务以及最近一次完成的任务是否距今超过一天。如果超过阈值就通过 Webhook 发送报警到企业微信或钉钉群。这个脚本可以放在独立的容器中也可以用宿主机的 cron 执行关键是报警不是靠人肉去看 Scrapyd 页面。最后再分享一个实用技巧在 Item 中始终保留date原始字符串和解析后的date_parsed字段不要直接覆盖原始值。这样当站点修改了日期格式导致清洗异常时可以通过原始字符串反推出正确的解析规则而不需要重新抓取全量历史数据。本文还有配套的精品资源点击获取