简介本资源是一款面向Python爬虫学习者与数据采集工程师的58同城全站信息抓取框架源码聚焦房产、招聘、二手车、二手交易等多类垂直信息的自动化采集解决人工收集效率低、覆盖面窄的问题。压缩包共26个文件含13个.py源码、10个.pyc字节码、1个.cfg配置文件、1个.txt说明文档及1个.gitignore总大小仅29KB轻量紧凑其中spiders、items、pipelines、settings等模块构成标准Scrapy风格架构agent_list.py与proxy_test.py体现反反爬策略设计test58.py提供快速验证入口。目前已有291人学习下载适合具备Python基础并了解HTTP协议、HTML解析的中级开发者用于实战复现、框架二次开发或反爬机制研究。 做一个能真正支撑“全站级”信息采集的Python爬虫框架远比写一个能跑通单页的脚本复杂得多。58同城这类分类信息网站页面结构不像新闻站那么单一城市分支、行业分类、列表页与详情页的层级关系叠在一起数据字段又散落在不同板块的不同HTML结构里如果一上来就写面条式脚本爬到一半基本就废了。这篇就把我近期实践的一套“58同城全站爬取信息框架”的核心设计思路和源码拆解开来讲。它不是一个只能抓单个页面的玩具而是把调度、下载、解析、存储、去重、断点续爬拆成分层模块的可扩展框架。适合正在做分类信息站采集、或者想把爬虫代码从“一次性脚本”升级成“可维护系统”的朋友参考。1. 项目整体设计与思路拆解1.1 为什么要做成框架而不是一次性脚本先说一个很现实的问题58同城这个站点的数据组织形式天然就是一棵深度树。根节点是主站首页往下分城市北京、上海、广州……每个城市下面又分一级分类房产、招聘、二手车、二手物品、生活服务……每个一级分类下面还有二级分类房产-二手房、房产-租房、房产-写字楼……再到具体列表页最后才是详情页。全站爬取意味着每一层都要遍历而这个“遍历”动作必须是有序、可控、可中断后恢复的。如果只是临时抓某一个城市、某一个分类下的几百条数据写个requests循环完全够用。但“全站”两个字就把复杂度提升了一个量级几百个城市、几十个分类交叉组合涉及成千上万个URL入口。一次性脚本在这个规模下会出现几个致命问题第一无法控制抓取节奏。全站遍历时如果所有请求挤在一起极容易触发目标站点的风控机制导致IP被限制访问。第二错误无法隔离。某一个分类页面结构变了脚本可能直接抛异常退出前面抓的数据全白干。第三无法增量更新。全站数据抓完之后下一次只需要抓新增和变更的部分一次性脚本做不到这种状态记录。所以框架的价值不在于“抓取”本身而在于把抓取过程拆成可管理、可监控、可恢复的组件。每一层只干一件事层与层之间用数据结构解耦。这套思路迁移到其他分类信息网站比如各类本地生活平台、招聘平台的时候只需要替换解析层调度和存储层可以原样复用。1.2 技术选型与理由框架底层的技术栈我最终确定的是requests parsel pymongo Redis并发用ThreadPoolExecutor。很多人会问为什么不用Scrapy。Scrapy确实是业界标准但如果目标不是分布式、不是千万级URL调度那么在Scrapy和一套轻量自研框架之间我倾向于后者。原因很简单Scrapy的组件模型虽然强大但学习曲线和定制成本都在“中间件”和“管道”的机制里对于58同城这种页面结构复杂但URL规模可控的场景自研框架反而更容易控制每一个环节的行为。当然如果你是团队协作、需要爬虫管理平台、需要对接分布式采集集群那直接上Scrapy或者基于Scrapy二次开发更合适。具体到每个组件的选择我是这么考虑的网络请求用requests而不是httpx。虽然httpx支持HTTP/2和异步但58同城PC端的页面响应并不依赖HTTP/2特性requests的会话管理和重试机制都非常成熟生态里各种插件也最全。异步方案aiohttp/httpx在这个场景下收益不明显因为瓶颈往往在目标站点的响应速度和服务端限制而不是本地IO。页面解析用parsel它是Scrapy底层用的解析库基于lxml同时支持XPath和CSS选择器。相比直接用lxmlparsel的extract_first()和getall()接口更符合写爬虫的习惯相比BeautifulSoupparsel的性能要好不少在大量列表页解析场景下差距很明显。数据存储正文数据结构复杂字段数量和层级在不同分类下都不一样房源的字段和二手物品的字段完全不同所以用MongoDB这种文档型数据库不需要提前定义强Schema每条数据按自己的结构存即可。如果必须用MySQL也可以但要做字段的JSON序列化存储或者建宽表灵活性差很多。URL去重与请求指纹用的是Redis的SET结构把所有请求过的URL指纹存进去利用SADD的原子性判断是否已经请求过。全站URL规模估计在百万级别Redis内存完全扛得住。如果要更大规模可以换成布隆过滤器但当前场景没必要增加这个复杂度。1.3 整体架构分层这套框架最终落地的架构是五层调度层负责任务队列管理维护“城市-分类-列表页-详情页”的遍历状态产出所有待抓取URL。调度层本身不发起请求。下载层负责发起HTTP请求、处理重试、随机延时、请求头管理。职责就是把URL变成HTML文本不关心内容是什么。解析层负责从HTML中提取结构化数据以及提取下一页链接和详情页链接。这一层是分站点、分板块定制的也是整个框架中改动最频繁的地方。存储层负责数据入库和去重。对外只暴露save(item)和is_duplicate(key)两个接口。监控层负责日志记录、统计抓取成功数和失败数并在异常时输出告警信息。数据流是单向的调度层产出URL - 下载层获取HTML - 解析层提取数据和新URL - 存储层持久化 - 调度层继续遍历新URL。这种分层有一个很直接的好处每一层都可以单独测试。我可以不启动抓取直接喂一个URL给解析层看解析结果是否正确也可以不加载解析规则先测试下载层对目标网站各种状态码的响应处理。这对排错来说效率提升非常明显。2. 核心细节解析与实操要点2.1 URL规划与遍历路径设计“全站爬取”的第一步是梳理URL规则这步做不好后面全是空中楼阁。58同城的URL结构大致可以归纳为三种模式城市入口北京列表页https://bj.58.com/上海列表页https://sh.58.com/城市前缀是固定的拼音缩写理论上只需要维护一份城市代码表就可以生成所有城市入口URL。但实际有例外比如一线城市的二手房是https://bj.58.com/ershoufang/而某些城市可能在主站域名下没有独立子域。所以城市列表不建议用规则硬拼而是从主站导航页解析出来。分类入口北京房产https://bj.58.com/ershoufang/北京招聘https://bj.58.com/zhaopin/北京二手车https://bj.58.com/ershouche/分类代码相对固定可以从每个城市首页的导航链接里提取。列表页与分页列表页第1页https://bj.58.com/ershoufang/列表页第2页https://bj.58.com/ershoufang/pn2/列表页第3页https://bj.58.com/ershoufang/pn3/分页规则非常规整pn{n}就是第n页。但需要注意部分分类没有分页或者只展示前50页后50页需要通过搜索接口才能触达。这些都需要在解析层做兜底判断。详情页典型详情页URLhttps://bj.58.com/ershoufang/1234567890x.shtml详情页URL从列表页解析出来数据中携带房源唯一编号。到这一步调度层的遍历路径就清晰了城市列表 - 城市首页 - 分类链接 - 分类列表页 - 分页处理 - 详情页URL - 详情页数据遍历路径必须在调度层维护成状态而不是临时现算。我设计了一个CrawlTask数据结构包含URL、任务类型CITY/CATEGORY/LIST/DETAIL、所属城市、所属分类、当前重试次数、时间戳。调度层根据任务类型决定下一步动作解析完列表页后会把详情页任务和下一页任务同时写回任务队列。2.2 请求头与访问节奏控制下载层是整个框架中和目标站点博弈最直接的部分。第一道门槛就是请求头。我在这套框架里采用了一份“会话级”的请求头池。每次新建会话时从池子里随机抽取一组User-Agent并把常用的Accept、Accept-Language、Accept-Encoding设置成与浏览器一致。USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36, Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0, ]请求头只是基础真正决定成败的是访问节奏。这里的原则是单位时间内的请求数必须低于目标站点正常人类用户的上限。我用了一个简单的限速器class RateLimiter: def __init__(self, min_interval1.5, max_interval3.5): self.min_interval min_interval self.max_interval max_interval def wait(self): interval random.uniform(self.min_interval, self.max_interval) time.sleep(interval)每个线程在发起请求前调用limiter.wait()把两次请求之间的间隔控制在1.5到3.5秒之间随机波动。这个波动很关键固定间隔比高频请求更容易被识别因为人类不可能每2.5秒精确地发一次请求。除了请求间隔还需要控制单城市的并发数量。我设置了“每城市单线程”策略即同一时间同一个城市只允许一个线程在跑不同城市可以并行。这样即使分类很多实际对目标站点的请求压力也被天然分散了。2.3 页面解析策略与常见坑位58同城的列表页HTML结构有一个特点列表项的数据绝大多数放在li标签里但不同分类的li内部结构完全不同。房产列表项包含小区名、户型、面积、总价、单价二手车列表项包含车龄、里程、变速箱类型招聘列表项包含公司名、薪资、经验要求。所以解析层不能写一个万能解析器而是要做“分类注册制”每个分类对应一个解析器实例。class ListParser: def parse(self, html): raise NotImplementedError class ErshoufangListParser(ListParser): def parse(self, html): selector parsel.Selector(texthtml) items [] for li in selector.css(ul.house-list-wrap li): items.append({ title: li.css(.title a::text).get(), url: li.css(.title a::attr(href)).get(), total_price: li.css(.total-price span::text).get(), unit_price: li.css(.unit-price span::text).get(), }) return items解析时最容易踩的坑是“取了空值但没报错”。CSS选择器匹配不到元素时parsel返回的是空列表get()返回None。如果直接忽略None会造成字段缺失但如果一遇到None就报错重试又会有大量URL白白浪费。我的做法是详情页关键字段标题、价格、链接为None时标记为解析失败非关键字段为None时记录为缺失字段。还有个很隐蔽的问题列表页的详情链接有两种一种是相对路径/ershoufang/123.shtml一种是绝对路径https://bj.58.com/ershoufang/123.shtml。入库前必须统一转成绝对路径否则后续调度层没法用。from urllib.parse import urljoin def normalize_url(base_url, url): if url.startswith(http): return url return urljoin(base_url, url)2.4 数据存储设计与去重键选择存储层我选了MongoDB但设计时还是有讲究。集合按分类维度拆分house_ershou存二手房、house_zufang存租房、job_zhaopin存招聘、car_ershou存二手车。这样做的原因是不同分类的查询模式完全不同按分类拆集合后后续做数据分析和导出都更干净。去重键的选择是另一个关键点。58同城详情页URL里的那串数字是信息唯一ID天然适合当去重键。我设置了两个去重键url_md5URL的MD5哈希和info_id提取出的数字ID。前者用于请求层去重后者用于数据层去重。为什么数据层去重不用URL因为同一个信息ID可能在不同时间点生成不同的URL比如加了跟踪参数但ID不变。如果只对URL去重可能出现同一条信息重复入库。def make_data_key(item): if item.get(info_id): return fdata:{item[city]}:{item[category]}:{item[info_id]} return fdata:{item[city]}:{item[category]}:{item[url]}Redis里维护两套SETreq:done存已经请求过的URL指纹data:done存已经入库的数据指纹。每次调度层从队列取URL时先查req:done每次存储层入库前先查data:done。3. 实操过程与核心环节实现3.1 框架目录结构最终落地的目录结构是这样的spider_framework/ ├── config.py # 全局配置数据库连接、Redis连接、限速参数 ├── models.py # 数据模型CrawlTask、Item ├── scheduler/ │ ├── __init__.py │ └── scheduler.py # 任务调度器任务队列、入队、出队 ├── downloader/ │ ├── __init__.py │ └── downloader.py # HTTP下载器请求头、重试、限速 ├── parser/ │ ├── __init__.py │ ├── base.py # 解析器基类 │ ├── list_parsers.py # 各分类列表页解析器 │ └── detail_parsers.py # 各分类详情页解析器 ├── storage/ │ ├── __init__.py │ └── mongo_storage.py # MongoDB存储与去重 ├── monitor/ │ ├── __init__.py │ └── logger.py # 日志与统计 └── main.py # 入口这个结构是按“层”划分目录的每个目录对应一个独立模块。后续如果要增加新的目标站点只需要新增一个parser包复用download和storage。3.2 调度器与任务模型实现models.py定义了核心的任务数据结构from dataclasses import dataclass, field import time dataclass class CrawlTask: url: str task_type: str # city / category / list / detail city: str category: str retry_count: int 0 created_at: float field(default_factorytime.time) extra: dict field(default_factorydict)task_type是调度器的判断依据。不同类型的任务产出不同的后续任务city任务解析城市首页产出category任务category任务解析分类列表页第一页产出list任务和detail任务list任务解析列表页产出下一页list任务和当前页所有detail任务detail任务解析详情页产出最终的Item数据调度器维护一个线程安全的任务队列和一个进度计数器import threading import queue class Scheduler: def __init__(self): self.task_queue queue.Queue() self.lock threading.Lock() self.total_count 0 self.done_count 0 def add_task(self, task): self.task_queue.put(task) with self.lock: self.total_count 1 def get_task(self): try: return self.task_queue.get(timeout1) except queue.Empty: return None def task_done(self): with self.lock: self.done_count 1 self.task_queue.task_done() def progress(self): with self.lock: return self.done_count, self.total_countmain.py里通过一个线程池来消费任务from concurrent.futures import ThreadPoolExecutor def worker(): while True: task scheduler.get_task() if not task: continue try: html downloader.fetch(task) if not html: scheduler.task_done() continue new_tasks dispatcher.dispatch(task, html) for t in new_tasks: scheduler.add_task(t) except Exception as e: logger.error(ftask failed: {task.url}, error: {e}) finally: scheduler.task_done() pool ThreadPoolExecutor(max_workers4) for _ in range(4): pool.submit(worker)这里有一个很关键的工程细节new_tasks里必须包含“下一页”任务。也就是说只要列表页有下一页调度器就会一直往下派任务直到没有下一页为止。这样全站遍历才不会漏。3.3 下载器的请求与重试实现下载器直接决定了你能不能在目标站点稳定抓数据。核心逻辑是import requests import random import time class Downloader: def __init__(self): self.session requests.Session() self.retry_count 3 self.timeout 10 def fetch(self, task): headers self._random_headers() for attempt in range(self.retry_count): try: resp self.session.get( task.url, headersheaders, timeoutself.timeout, allow_redirectsTrue ) if resp.status_code 200: resp.encoding resp.apparent_encoding return resp.text elif resp.status_code in (403, 429): # 风控限制等待较长时间再重试 time.sleep(30 random.randint(10, 30)) else: time.sleep(random.uniform(1, 3)) except requests.RequestException: time.sleep(random.uniform(2, 5)) return None有几个细节值得讲1. 编码处理。58同城的页面编码是UTF-8但某些分类或某些老页面可能声明GBK。直接用resp.text偶尔会出现乱码。保险做法是先用apparent_encoding探测再手动指定。2. 重定向跟随。列表页和详情页的URL偶尔会跳转到城市首页或登录页。如果发现跳转后的URL和原始URL域名不一致、或者URL变成了登录页面直接视为抓取失败不要继续传下去。3. 会话保持。用同一个requests.Session()实例可以让Cookie在请求间传递。有些列表页第一次访问会种下Cookie后续请求带上这个Cookie后详情页的访问成功率会高很多。3.4 列表页与详情页解析实现列表页解析器的用法前面已经展示过这里重点说详情页。class ErshoufangDetailParser: def parse(self, html, task): selector parsel.Selector(texthtml) info_id task.url.split(/)[-1].split(.)[0] title selector.css(h1::text).get().strip() total_price selector.xpath(//span[classtotal]/text()).get() unit_price selector.xpath(//span[classunit]/text()).get() info_panel selector.css(div.house-info-panel ul li) info_dict {} for li in info_panel: key li.css(span.label::text).get() value li.css(span.value::text).get() if key and value: info_dict[key.strip()] value.strip() item { info_id: info_id, url: task.url, title: title, total_price: total_price, unit_price: unit_price, info: info_dict, city: task.city, category: task.category, crawled_at: time.time(), } return item二手房详情页的房屋信息是一个键值对列表小区名称、建造年份、楼层、朝向等但不同房源的字段数量不一致有的房源没有配套信息。这里直接统一放进info字段将来需要做数据分析时再从中提取具体指标这是文档型数据库比较舒服的用法。另一种需要处理的页面是“信息已过期”或“信息已删除”的详情页。58同城的房源下架后会返回一个提示“房源信息已删除”的页面。这种页面解析出来标题为空、价格为空需要在解析阶段就识别并丢弃避免把垃圾数据存进库里。if not title or not total_price: return None3.5 存储与去重实现MongoDB存储层实现如下import hashlib import pymongo import redis class MongoStorage: def __init__(self, mongo_uri, db_name, redis_uri): self.client pymongo.MongoClient(mongo_uri) self.db self.client[db_name] self.redis redis.Redis.from_url(redis_uri) def is_url_done(self, url): url_key req: hashlib.md5(url.encode()).hexdigest() return self.redis.sadd(req:done, url_key) 0 def save(self, item): if not item: return False data_key fdata:{item[city]}:{item[category]}:{item[info_id]} if self.redis.sadd(data:done, data_key) 0: return False collection_name self._collection_name(item[category]) self.db[collection_name].update_one( {_id: item[info_id]}, {$set: item}, upsertTrue ) return True def _collection_name(self, category): mapping { ershoufang: house_ershou, zufang: house_zufang, zhaopin: job_zhaopin, ershouche: car_ershou, } return mapping.get(category, other)update_one配合upsertTrue可以实现“有则更新、无则插入”。这在增量抓取场景下非常有用同一套房源下次再被抓到不会重复插入而是更新价格、状态等变化字段。去重逻辑放在入库前避免重复的解析和存储开销。但这里有一个权衡点如果去重太严格会导致某些URL因为网络波动第一次没抓到详情第二次再请求时发现URL已在req:done里就直接跳过造成数据永久缺失。所以我将“请求去重”和“入库去重”分开请求去重在下载前检查但只对“已经成功返回200且已经完成解析”的URL标记完成请求失败或者解析失败的URL不会被标记保证了重试机会。4. 常见问题与排查技巧实录4.1 高频访问触发风控怎么办实际抓取过程中最容易遇到的现象是刚开始跑的时候一切正常跑了半小时后突然大量请求返回403或者跳转验证码页面。排查时要先区分是单IP触发风控还是被识别为异常流量。判断方法是看报错比例。如果某个时刻开始所有请求的返回状态全部变成403基本可以确定是该出口IP被限制了。此时继续跑没有任何意义只会加重限制。我的应对策略是在下载器里对403/429状态码做特殊计数连续触发3次就自动暂停该城市的所有任务暂停时间从30分钟起步之后逐步递增。同时在合规前提下降低抓取并发数、拉大请求间隔让流量特征更接近人工浏览。注意任何形式的采集行为都要遵守目标网站的服务条款和相关法律法规。不要尝试绕过网站的风控机制。合理限速、控制频率、在允许范围内抓取公开信息才是可持续的做法。另外本地调试时不要直接从列表页第1页扫到第100页。先抓一页确认解析结果正确后再放量这样能减少很多无效请求。4.2 列表页结构与分类不一致这是做全站爬取时最头疼的问题。58同城有数十个一级分类每个分类的列表页HTML结构都不同甚至同一个分类在不同城市也可能出现结构微调。比如二手房列表页的数据在ul.house-list-wrap li里但租房列表页的数据在div.house-cell里二手车列表页的数据则在ul.listUl li里。如果一个万能解析器在所有分类上跑结果就是大部分分类抓不到数据。解决思路是“解析器注册表”模式PARSER_REGISTRY { ershoufang: ErshoufangListParser, zufang: ZufangListParser, ershouche: ErshoucheListParser, zhaopin: ZhaopinListParser, } def get_parser(category): parser_cls PARSER_REGISTRY.get(category) if not parser_cls: raise ValueError(fno parser for category: {category}) return parser_cls()新增一个分类时只需要实现对应的解析器并在注册表里登记调度层和存储层不用改一行代码。这就是分层架构的好处。结构不一致的问题还会出现在详情页。有些房源标题在h1里有些在div.house-title里。我的做法是在解析器里编写多组候选选择器按优先级依次尝试直到命中非空值。def extract_first(selector, candidates): for css in candidates: value selector.css(css).get() if value: return value.strip() return None4.3 抓取中断如何断点续爬全站爬取通常会跑很长时间中途断电、程序崩溃、网络中断都可能导致进程退出。如果没有状态保存重启后就要从头开始代价很大。我在这套框架里把任务状态保存在Redis里所以天然支持断点续爬。具体实现是任务在入队前先写入task:queue队列的源头——一个备份的Redis列表任务被成功处理后才从列表里移除。重启时扫描这个列表把剩余任务重新放回内存队列。还有一种更简单的方案每处理完100个任务就把done_count写入一个本地文件或Redis键。重启后读取这个计数虽然不能精确恢复到每个任务的粒度但至少能知道大概进度。如果再配合前面说的“请求去重”即使任务重复执行也不会重复入库所以断点续爬其实是一个软性需求主要价值在于减少重启后的无谓请求。4.4 日志与监控怎么做爬虫跑起来之后最怕的不是出错而是出错了你不知道。我在这套框架里加了一个轻量级的统计模块每抓完100个详情页就打印一次进度内容包括已抓取详情页数、列表页数、当前队列积压数、失败数、平均响应时间。class ProgressMonitor: def __init__(self, interval100): self.count 0 self.fail_count 0 self.interval interval self.start_time time.time() def success(self): self.count 1 if self.count % self.interval 0: elapsed time.time() - self.start_time rate self.count / elapsed if elapsed 0 else 0 logger.info( fprogress: success{self.count}, fail{self.fail_count}, frate{rate:.2f}/s, elapsed{elapsed:.0f}s ) def fail(self): self.fail_count 1日志千万不要什么都打。请求成功不用打日志因为数量太大会刷屏但请求失败一定要打并且要带上URL、状态码、异常类型。我见过不少爬虫工程的问题就是因为日志只打了“request failed”却没打URL最后只能靠猜去定位问题。还有一个很实用的监控技巧实时监控MongoDB里的数据总量。如果跑了一个小时数据条数纹丝不动基本可以断定解析链路出了问题这个时候直接去看最近有没有解析成功的日志记录即可。5. 后续可以怎么扩展这套框架目前跑的是58同城但架构本身是通用的。我实际测试过换一个分类信息站只需要做三件事写一份新的城市入口URL规则、实现对应的列表页解析器和详情页解析器、修改存储层的分类映射。调度和下载完全不用动。如果后续数据量上来了建议做两件事第一引入消息队列。当任务量超过单机内存队列的承受能力时把调度器的任务队列替换成RabbitMQ或Kafka这样天然支持多机分布式抓取。第二解析规则配置化。把解析器的选择器抽成JSON配置存到数据库里这样改解析规则不需要改代码、不需要重启进程。对于维护成本来说这个改进是立竿见影的。最后再说一个体会爬虫项目最需要投入精力的地方往往不是写代码本身而是“发现规则变化”和“处理边界情况”。页面改版、字段调整、反爬策略升级这些都会让你的代码突然失效。所以不要把一切都写死在代码里尽量让规则可配置、可监控、可恢复。这也是为什么我坚持要做一个“框架”而不是写一个“脚本”的原因——脚本跑通了只是开始框架能撑住后续所有的变化才是真正能用的东西。本文还有配套的精品资源点击获取