Python Scrapy校园集市数据爬取实战：从源码包到稳定采集链路

简介：这是一份基于Python与Scrapy框架构建的校园集市数据爬虫设计源码，面向具备一定Python基础、希望学习爬虫工程化实践的开发者与在校学生，可用于校园信息搜集、数据分析等场景的入门与二次开发。资源包共48个文件，约1.33MB，以24个py源文件为核心，涵盖爬虫逻辑、数据管道、中间件与配置模块；另有11个txt说明文档、2个sql建表脚本、1个cfg配置文件及许可证等，目录结构清晰，便于按模块阅读。项目围绕校园集市数据抓取展开，包含热度计算、情感评分、OCR识别、词向量相关性分析等扩展脚本，并配有数据库连接池与初始化脚本，能帮助读者理解从数据采集到持久化存储的完整链路。目前已有77人学习下载，适合作为爬虫课程设计或毕业设计的参考起点。

1. 校园集市数据爬取：从 Scrapy 源码包到能跑通的采集链路

校园集市这类站点，本质是一个信息高度结构化、但反爬强度中等的垂直社区：帖子列表分页、详情页字段固定、图片走 CDN，真正麻烦的是登录态、翻页边界和字段缺失。这份基于 Python Scrapy 框架的校园集市数据爬取设计源码，解决的就是「把散落在列表页和详情页里的商品/帖子信息，稳定落成结构化数据」这件事。它适合两类人：一类是刚学完 Scrapy 基础、想找一个完整项目练手的同学；另一类是需要快速搭一套校园二手/集市采集链路、又不想从零写中间件的从业者。源码包给的是骨架和关键实现，真正决定成败的是你对请求调度、字段解析和去重策略的理解，下面按「能跑起来 → 跑得稳 → 跑得久」的顺序拆。

2. 拆开源码包：Scrapy 工程结构与校园集市场景的对应关系

拿到一个 Scrapy 源码包，第一件事不是急着scrapy crawl，而是先看清目录结构和它对应的采集逻辑。校园集市的数据形态决定了这个工程必须处理列表翻页、详情补全、图片下载三件事，源码包的目录基本就是围绕这三件事展开的。

2.1 标准 Scrapy 目录与各文件职责

一个能跑的 Scrapy 工程，核心目录长这样：

campus_market_spider/ ├── scrapy.cfg # 部署配置，指定 settings 模块位置 ├── campus_market_spider/ │ ├── __init__.py │ ├── items.py # 定义要抓的字段结构 │ ├── middlewares.py # 下载中间件，处理请求头、重试、代理逻辑 │ ├── pipelines.py # 数据清洗、去重、入库 │ ├── settings.py # 全局配置：并发、延迟、管道开关 │ └── spiders/ │ └── market.py # 核心爬虫，定义 start_urls 和 parse 逻辑 └── requirements.txt # 依赖清单

items.py决定你最终能拿到哪些字段，校园集市常见的字段包括标题、价格、发布时间、卖家昵称、联系方式、图片链接、帖子详情。pipelines.py负责把这些字段做二次处理，比如价格去掉「¥」符号转成数字、发布时间统一成标准格式。settings.py是最容易被忽视但最影响稳定性的文件，DOWNLOAD_DELAY、CONCURRENT_REQUESTS、RETRY_TIMES这几个参数直接决定你会不会被限流。

2.2 items.py：字段设计决定后续清洗成本

字段设计是采集的地基，设计得糙，后面 pipeline 里全是补丁。校园集市的 item 我一般这样定义：

import scrapy class CampusMarketItem(scrapy.Item): # 帖子唯一标识，优先用站点自带的 id，没有就用 url 的 hash post_id = scrapy.Field() title = scrapy.Field() price = scrapy.Field() # 原始文本，如 "¥1200 可小刀" price_num = scrapy.Field() # 清洗后的数字，如 1200.0 publish_time = scrapy.Field() # 原始时间文本 publish_ts = scrapy.Field() # 转成时间戳，便于排序 seller = scrapy.Field() contact = scrapy.Field() detail_url = scrapy.Field() images = scrapy.Field() # 图片链接列表 description = scrapy.Field() crawl_time = scrapy.Field() # 采集时间，用于增量判断

这里的关键是「原始字段 + 清洗字段」并存。price保留原文，price_num存数字，这样一旦清洗规则出错，还能回溯原始数据重跑，不用重新爬。post_id一定要有，它是后面去重和增量采集的锚点，没有稳定 id 的站点就用detail_url做 md5 当 id。crawl_time是增量采集的后悔药，下次跑的时候可以只处理比上次新的帖子。

2.3 spiders/market.py：列表页与详情页的两级解析

校园集市的数据分散在两级页面：列表页给标题、价格、缩略图，详情页给描述、联系方式、大图。核心爬虫要处理这个跳转关系：

import scrapy from campus_market_spider.items import CampusMarketItem class MarketSpider(scrapy.Spider): name = "market" allowed_domains = ["example-campus.com"] start_urls = ["https://example-campus.com/market?page=1"] def parse(self, response): # 列表页：提取每个帖子的详情链接 for card in response.css("div.post-card"): detail_url = card.css("a.title::attr(href)").get() if detail_url: yield response.follow( detail_url, callback=self.parse_detail, # 把列表页能拿到的字段透传下去，减少详情页解析压力 cb_kwargs={"list_title": card.css("a.title::text").get(default="").strip()} ) # 翻页：找到下一页按钮，递归调度 next_page = response.css("a.next-page::attr(href)").get() if next_page: yield response.follow(next_page, callback=self.parse) def parse_detail(self, response, list_title): item = CampusMarketItem() item["title"] = list_title or response.css("h1.post-title::text").get(default="").strip() item["price"] = response.css("span.price::text").get(default="").strip() item["publish_time"] = response.css("span.time::text").get(default="").strip() item["seller"] = response.css("div.seller-name::text").get(default="").strip() item["contact"] = response.css("div.contact::text").get(default="").strip() item["description"] = "".join(response.css("div.desc ::text").getall()).strip() item["images"] = response.css("div.gallery img::attr(src)").getall() item["detail_url"] = response.url item["post_id"] = response.url.rstrip("/").split("/")[-1] yield item

这段代码有两个设计点值得说。第一，cb_kwargs把列表页的标题透传到详情页，因为有些站点详情页标题被 JS 改写或者带前缀，列表页的反而更干净，两边做兜底能提高字段完整率。第二，翻页用response.follow递归，而不是手动拼page=N，因为校园集市的翻页链接经常带签名参数，手动拼容易 403。parse_detail里所有字段都给了default=""，这是血泪经验——字段缺失时 Scrapy 默认返回 None，写进数据库会报错，给空字符串能让 pipeline 统一处理。

2.4 settings.py：并发与延迟的取舍

默认配置直接跑，大概率跑几十页就被限流。校园集市这类站点，我一般这样调：

# 并发请求数，单域名控制在 4~8，别贪 CONCURRENT_REQUESTS = 8 CONCURRENT_REQUESTS_PER_DOMAIN = 4 # 下载延迟，配合随机化，避免固定间隔被识别 DOWNLOAD_DELAY = 1.5 RANDOMIZE_DOWNLOAD_DELAY = True # 重试配置，遇到 429/503 自动退避 RETRY_ENABLED = True RETRY_TIMES = 3 RETRY_HTTP_CODES = [429, 500, 502, 503, 504] # 自动限流，让 Scrapy 根据响应延迟动态调整 AUTOTHROTTLE_ENABLED = True AUTOTHROTTLE_START_DELAY = 1.0 AUTOTHROTTLE_MAX_DELAY = 10.0 AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0 # 管道与中间件 ITEM_PIPELINES = { "campus_market_spider.pipelines.CleanPipeline": 300, "campus_market_spider.pipelines.DuplicatePipeline": 400, "campus_market_spider.pipelines.MysqlPipeline": 500, }

AUTOTHROTTLE是这套配置里最值钱的部分，它让 Scrapy 根据服务端响应速度自动放慢，比手动设死延迟聪明得多。CONCURRENT_REQUESTS_PER_DOMAIN设成 4 是保守值，实测校园集市这类站点超过 8 就容易触发验证码。管道优先级数字越小越先执行，清洗要在去重之前，去重要在入库之前，顺序错了会出现「脏数据进了去重集合」的问题。

3. 让采集跑稳：中间件、管道与增量策略的落地实现

源码包能跑通只是起点，真正拉开差距的是稳定性。校园集市的反爬不算狠，但架不住你跑几千页，几个细节没处理好就会中途翻车。

3.1 下载中间件：请求头轮换与异常兜底

默认的 User-Agent 是Scrapy/x.x，等于告诉对方「我是爬虫」。中间件里做两件事：轮换 UA、处理异常响应。

import random from scrapy import signals class RotateUAMiddleware: def __init__(self): self.ua_list = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0 Safari/537.36", ] def process_request(self, request, spider): request.headers["User-Agent"] = random.choice(self.ua_list) # 补上 Referer，很多站点校验来源 request.headers.setdefault("Referer", "https://example-campus.com/market") return None def process_response(self, request, response, spider): # 被限流时，把请求重新塞回调度器，配合 RETRY 生效 if response.status in (429, 503): spider.logger.warning(f"被限流: {request.url} status={response.status}") return request return response

process_request里setdefault而不是直接赋值，是为了不覆盖爬虫里已经设好的 Referer。process_response里返回request对象会让 Scrapy 重新调度这个请求，配合RETRY_TIMES实现退避重试。注意别在这里无限重试，否则遇到永久 403 会死循环，加个计数器更稳妥。

3.2 管道：清洗、去重、入库三段式

管道是数据质量的最后一道关。我一般拆成三个，各管一段：

import re import hashlib import pymysql from itemadapter import ItemAdapter class CleanPipeline: """清洗价格、时间，补全派生字段""" def process_item(self, item, spider): adapter = ItemAdapter(item) # 价格：从 "¥1200 可小刀" 里抠出 1200 raw_price = adapter.get("price", "") match = re.search(r"(\d+(?:\.\d+)?)", raw_price.replace(",", "")) adapter["price_num"] = float(match.group(1)) if match else None # 时间：统一成时间戳，这里假设格式为 "2024-05-01 12:30" raw_time = adapter.get("publish_time", "") adapter["publish_ts"] = self._to_ts(raw_time) return item def _to_ts(self, text): from datetime import datetime for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d", "%m-%d %H:%M"): try: dt = datetime.strptime(text.strip(), fmt) if fmt == "%m-%d %H:%M": dt = dt.replace(year=datetime.now().year) return int(dt.timestamp()) except (ValueError, AttributeError): continue return None class DuplicatePipeline: """基于 post_id 去重，内存集合 + 可选 Redis""" def __init__(self): self.seen = set() def process_item(self, item, spider): adapter = ItemAdapter(item) pid = adapter.get("post_id") if not pid: # 没有 id 就用 url 的 hash 兜底 pid = hashlib.md5(adapter.get("detail_url", "").encode()).hexdigest() adapter["post_id"] = pid if pid in self.seen: from scrapy.exceptions import DropItem raise DropItem(f"重复帖子: {pid}") self.seen.add(pid) return item class MysqlPipeline: """入库，用 INSERT ... ON DUPLICATE KEY UPDATE 做幂等""" def open_spider(self, spider): self.conn = pymysql.connect( host="127.0.0.1", user="root", password="yourpass", database="campus_market", charset="utf8mb4" ) def process_item(self, item, spider): adapter = ItemAdapter(item) sql = """ INSERT INTO posts (post_id, title, price_num, publish_ts, seller, contact, detail_url, description, crawl_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,NOW()) ON DUPLICATE KEY UPDATE price_num=VALUES(price_num), crawl_time=NOW() """ with self.conn.cursor() as cur: cur.execute(sql, ( adapter.get("post_id"), adapter.get("title"), adapter.get("price_num"), adapter.get("publish_ts"), adapter.get("seller"), adapter.get("contact"), adapter.get("detail_url"), adapter.get("description"), )) self.conn.commit() return item def close_spider(self, spider): self.conn.close()

CleanPipeline里价格正则用replace(",", "")先去掉千分位，否则「1,200」会被抠成 1。时间解析给了三种格式兜底，校园集市的时间显示经常是「05-01 12:30」这种省略年份的，补当前年份是常见做法，但跨年采集时会有误差，严谨场景要结合帖子 id 递增规律判断。DuplicatePipeline用内存集合，跑几万条没问题，上百万条要换 Redis 的SADD，否则内存扛不住。MysqlPipeline用ON DUPLICATE KEY UPDATE保证重跑不产生重复行，前提是post_id建了唯一索引。

3.3 增量采集：用 crawl_time 和 publish_ts 做水位线

全量重爬既慢又容易被封，增量才是长期跑的正道。思路是记录上次采集的最大publish_ts，下次只处理比它新的帖子：

# 爬虫启动时从数据库读水位线 def start_requests(self): import pymysql conn = pymysql.connect(host="127.0.0.1", user="root", password="yourpass", database="campus_market") with conn.cursor() as cur: cur.execute("SELECT MAX(publish_ts) FROM posts") last_ts = cur.fetchone()[0] or 0 conn.close() self.last_ts = last_ts yield scrapy.Request(self.start_urls[0], callback=self.parse) # 在 parse_detail 里判断 def parse_detail(self, response, list_title): item = CampusMarketItem() # ... 字段解析 ... if item.get("publish_ts") and item["publish_ts"] <= self.last_ts: # 比水位线旧，跳过，但列表页翻页要继续 return yield item

这里有个坑：不能因为遇到旧帖子就停止翻页，因为列表页排序可能不是严格按时间倒序，置顶帖、推广帖会打乱顺序。正确做法是「跳过旧帖但继续翻页」，直到连续 N 页都是旧帖再停，N 一般取 3。水位线方案的前提是publish_ts解析准确，解析失败返回 None 的帖子会被当成新帖重复入库，所以清洗管道里时间解析的兜底逻辑必须扎实。

4. 避坑与排查：校园集市采集最常见的五类翻车

采集项目跑不起来，八成不是 Scrapy 本身的问题，而是下面这些细节。每条按「现象 → 原因 → 解决」写，都是实际踩过的。

4.1 现象：跑几十页后全部返回 403 或验证码页

原因：并发和延迟配置太激进，或者 UA 固定不变被识别。校园集市这类站点通常有基于 IP + UA 的频率统计，超过阈值就拦截。

解决：把CONCURRENT_REQUESTS_PER_DOMAIN降到 2~4，DOWNLOAD_DELAY提到 2 秒以上，开启AUTOTHROTTLE。UA 池至少准备 5 个以上真实浏览器 UA，别用网上抄来的假 UA。如果还是被封，检查是不是漏了Accept-Language、Accept-Encoding这些头，有些站点会校验头字段完整性。

4.2 现象：详情页字段大量为空，但浏览器里明明有数据

原因：数据是 JS 渲染的，Scrapy 拿到的响应里没有。校园集市的部分字段（尤其是联系方式和图片懒加载）经常走 AJAX。

解决：先确认是不是动态渲染。用scrapy shell打开详情页，view(response)看渲染后的 DOM，如果字段不在里面就是 JS 渲染。两条路：一是找 AJAX 接口直接请求 JSON，这是首选，效率高；二是上scrapy-playwright做浏览器渲染，但会拖慢速度，只对必要的详情页开。判断方法很简单，F12 看 Network 里 XHR 请求，能找到返回数据的接口就别用浏览器渲染。

4.3 现象：翻页到最后一页后无限循环，或者提前停止

原因：翻页判断逻辑太简单。a.next-page在最后一页可能仍然存在（只是 disabled），或者站点用 JS 控制翻页，静态 HTML 里没有下一页链接。

解决：翻页判断加多重条件——检查next_page是否存在、检查当前页码是否超过预设上限、检查本页是否解析到新帖子。如果本页parse没提取到任何详情链接，直接return停止。对于 JS 翻页，找翻页接口的 URL 规律，手动构造page=N请求，但要注意签名参数。

4.4 现象：数据库里出现重复帖子，post_id 相同但内容不同

原因：post_id提取逻辑不稳定。有些站点详情页 URL 带查询参数（如?from=list），直接取最后一段会拿到带参数的 id，同一帖子不同入口进来 id 不一样。

解决：post_id提取要规范化，去掉查询参数和尾部斜杠。用urlparse取 path 部分再切分：

from urllib.parse import urlparse path = urlparse(response.url).path.rstrip("/") post_id = path.split("/")[-1]

如果站点本身有稳定的帖子 id（在 HTML 的>import pymysql conn = pymysql.connect(host="127.0.0.1", user="root", password="yourpass", database="campus_market") with conn.cursor() as cur: # 指标一：字段完整率，title 和 price_num 为空的比例 cur.execute("SELECT COUNT(*) FROM posts") total = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM posts WHERE title IS NULL OR title=''") no_title = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM posts WHERE price_num IS NULL") no_price = cur.fetchone()[0] print(f"总数 {total}, 标题缺失率 {no_title/total:.2%}, 价格缺失率 {no_price/total:.2%}") # 指标二：时间分布，看是否集中在某几天（可能是解析错误） cur.execute("SELECT DATE(FROM_UNIXTIME(publish_ts)) d, COUNT(*) FROM posts GROUP BY d ORDER BY d DESC LIMIT 10") for row in cur.fetchall(): print(row) # 指标三：价格异常值，看有没有明显离谱的数字 cur.execute("SELECT MIN(price_num), MAX(price_num), AVG(price_num) FROM posts WHERE price_num IS NOT NULL") print("价格区间:", cur.fetchone()) conn.close()

标题缺失率超过 10% 说明列表页解析有问题，价格缺失率超过 20% 说明价格正则没覆盖全。时间分布如果全挤在采集当天，说明publish_ts解析失败被兜底成了当前时间，得回去查时间格式。价格最大值如果是几百万，多半是把「原价 9999 现价 100」里的原价抠出来了，正则要加约束。

5.2 导出与分析：从 MySQL 到 CSV 再到可视化

数据校验通过后，导出成 CSV 给非技术同学用，或者直接接 pandas 做分析：

import pandas as pd import pymysql conn = pymysql.connect(host="127.0.0.1", user="root", password="yourpass", database="campus_market") df = pd.read_sql("SELECT title, price_num, publish_ts, seller FROM posts WHERE price_num IS NOT NULL", conn) conn.close() # 时间戳转可读时间 df["publish_time"] = pd.to_datetime(df["publish_ts"], unit="s") # 按天统计发帖量 daily = df.groupby(df["publish_time"].dt.date).size() print(daily.tail(10)) # 导出 df.to_csv("campus_market_export.csv", index=False, encoding="utf-8-sig")

encoding="utf-8-sig"是为了 Excel 打开不乱码，这个细节坑过很多人。按天统计发帖量能看出校园集市的活跃规律，比如开学季、毕业季是高峰，这些规律反过来能指导你调整采集频率——高峰期降低并发避免被封，低谷期可以适当提速。

5.3 一个具体技巧：用 Scrapy 的close信号做采集报告

每次采集结束，自动生成一份报告，记录本次抓了多少、跳过多少、失败多少。用 Scrapy 的信号机制：

from scrapy import signals class StatsReportExtension: def __init__(self, stats): self.stats = stats @classmethod def from_crawler(cls, crawler): ext = cls(crawler.stats) crawler.signals.connect(ext.spider_closed, signal=signals.spider_closed) return ext def spider_closed(self, spider): stats = self.stats.get_stats() print("=== 采集报告 ===") print(f"请求总数: {stats.get('downloader/request_count', 0)}") print(f"响应总数: {stats.get('downloader/response_count', 0)}") print(f"抓取条目: {stats.get('item_scraped_count', 0)}") print(f"丢弃条目: {stats.get('item_dropped_count', 0)}") print(f"重试次数: {stats.get('retry/count', 0)}") print(f"耗时: {stats.get('elapsed_time_seconds', 0):.1f}s")

在settings.py里注册这个扩展，每次跑完自动打印。丢弃条目突然变多，说明去重逻辑可能误杀；重试次数飙升，说明被限流了。这个报告是我每次采集后必看的，比翻日志快得多。

从那以后我每次跑采集任务，都强制先跑一遍数据校验脚本再导出，宁可多花五分钟，也不愿把一份缺字段、时间错乱的数据交出去。希望这份拆解帮到你，源码包里的骨架加上这些细节，基本能撑起一个稳定的校园集市采集链路。

本文还有配套的精品资源，点击获取