简介这是一份Python爬虫实战项目“spider-master”的压缩包面向有基础爬虫知识、想拓展多站点采集能力的开发者。资源围绕亚马逊、Confluence等网站的数据抓取展开同时涵盖贴吧、糗事百科等常见目标既能了解简单静态页面抓取也能接触动态加载站点和需登录系统的处理方式。压缩包共41个文件以31个py脚本为核心辅以3份md说明文档、3份cfg配置文件和2份json数据文件整体仅47KB结构清晰适合快速上手。目前已有466人学习。脚本涉及requests发送请求、BeautifulSoup解析、Scrapy中间件及代理池应用并配有amazonsims、confluence等专项抓取示例涉及商品信息采集、知识库页面遍历、登录状态维持与反爬应对等实战场景通过阅读代码还能了解Scrapy的调度流程、代理轮换和异常处理思路是一份不可多得的爬虫学习参考。1. 拿到 spider.zip 先别急着解压这个包到底能替你做什么做 Python 爬虫的人手里多少会攒几个“万能包”而这个 amazon、confluence 场景下的 spider.zip本质是一份已经跑通过两条采集线的工程模板一条面向 Amazon 这种强反爬电商站一条面向 Confluence 这种带权限体系的内部知识库。它不是给你一个现成的数据结果而是把“请求怎么发、字段怎么抽、数据怎么落”的完整链路打包好了。你拿到手能复用的是骨架和思路不是直接改个 URL 就能躺着抓。这个 zip 解决的问题很具体电商商品信息采集、内部文档空间备份与检索。适合三类人刚接触分布式采集、想找个规范工程起步的开发者需要把 Amazon 商品数据抓到本地做价格分析的选品人员以及要给 Confluence 空间做定期快照的运维或知识管理工程师。往下拆你会发现它真正值钱的部分不是那两个目标站而是中间那层可替换的请求调度和数据管道。2. 拆包与通用骨架这个 zip 里最该先看懂的是哪几个文件2.1 目录结构先分清“业务代码”和“框架代码”解压后第一件事不是运行而是理清目录。一个规范的爬虫工程包通常会把“跟具体网站相关的代码”和“跟采集流程相关的代码”分开。我从这个 zip 包里常见的组织方式看核心结构一般长这样spider_project/ ├── spiders/ │ ├── amazon_spider.py │ └── confluence_spider.py ├── core/ │ ├── base_spider.py │ ├── http_client.py │ └── pipeline.py ├── utils/ │ ├── user_agent.py │ └── text_cleaner.py ├── config/ │ └── settings.py └── requirements.txtspiders/目录是每个网站独有的采集逻辑core/目录是跟目标站无关的通用能力。你先打开core/base_spider.py和config/settings.py这两个文件决定了整个工程的上限。如果发现目录里只把两个网站的代码揉在一起、没有分层那这个包的价值就要打折你要花时间自己拆。从顺手程度讲我一般拿到包后第一件事是看requirements.txt用了哪些依赖。常见组合是requests加parsel或BeautifulSoup重型场景才会引入scrapy。如果这个 zip 里的依赖列表特别长反而要小心——很可能把不必要的东西都塞进来了。2.2 请求基类与重试机制把“稳定拿数据”这件事抽象出来两个目标站风格完全不同但它们的请求流程都能抽象成同一个模式构造请求 → 发送 → 检查响应 → 解析。这个 zip 里最有复用价值的往往是那个http_client.py它把重试、超时、请求头管理都封装进去了。我拆包时最看重这个文件因为它决定你换新目标站时要改多少代码。import random import time import requests class HttpClient: 带重试与降级策略的请求客户端适合 Amazon / Confluence 两类场景 def __init__(self, timeout10, retry_times3, retry_interval(1, 3)): self.timeout timeout self.retry_times retry_times self.retry_interval retry_interval # 重试间隔范围单位秒 def get(self, url, headersNone, paramsNone): for attempt in range(1, self.retry_times 1): try: resp requests.get( url, headersheaders, paramsparams, timeoutself.timeout, ) if resp.status_code 200: return resp # 429/503 是限流信号直接触发重试不要等业务层处理 if resp.status_code in (429, 503): raise requests.RequestException(f限流状态码: {resp.status_code}) except requests.RequestException as exc: if attempt self.retry_times: raise exc wait_time random.uniform(*self.retry_interval) * attempt time.sleep(wait_time) return None这段代码把重试策略做成了“重试次数递增等待时间”的模式第一次失败等 1 到 3 秒第二次等 2 到 6 秒最多重试 3 次。这么做比固定间隔更接近人的操作习惯被目标站限流识别的概率也低一些。429和503直接映射成“限流”处理是因为 Amazon 在触发风控时常返回这类状态码而 Confluence 的负载均衡器在服务压力大时也会吐 503。参数层面你要改的核心是retry_times和retry_interval。对 Confluence 这种内网服务重试次数可以给到 5 次以上因为内网抖动大多是瞬时的对 Amazon 这种强风控站点重试 2 次就够了再多只会加重账号或 IP 的封禁风险。random.uniform让每次等待时间落在区间内避免机器那种精准的规律性。2.3 数据管道的默认出口写文件与写库的取舍爬虫的价值最终要落在数据上。这个 zip 里pipeline.py通常提供两种出口JSON 文件或数据库。Amazon 场景数据量中等且字段嵌套深适合 JSON 文件Confluence 空间导出适合按页面存成独立文件方便后续检索。import json import sqlite3 from pathlib import Path class Pipeline: 统一数据出口支持 JSON 文件与 SQLite 两种方式 def __init__(self, output_diroutput, use_sqliteFalse): self.output_dir Path(output_dir) self.output_dir.mkdir(parentsTrue, exist_okTrue) self.use_sqlite use_sqlite if use_sqlite: self.conn sqlite3.connect(self.output_dir / spider_data.db) self.cursor self.conn.cursor() def save_json(self, data, name): file_path self.output_dir / f{name}.json with open(file_path, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) def save_sqlite(self, table_name, data): if not self.use_sqlite: return keys list(data.keys()) values [data[k] for k in keys] placeholders ,.join([?] * len(keys)) self.cursor.execute( fINSERT INTO {table_name} ({,.join(keys)}) VALUES ({placeholders}), values, ) self.conn.commit()save_json适合保存列表页的原始数据方便人工核对字段save_sqlite适合增量采集场景。注意save_sqlite这里默认是INSERT而非INSERT OR REPLACE这是个很容易踩的坑——断点续跑时同一条记录会重复入库。你在拆包后建议改成主键冲突时更新或者加上去重判断。3. 针对 Amazon 的采集反爬对抗与商品字段抽取3.1 Amazon 页面特征为什么通用爬虫到这里集体失效Amazon 是爬虫界的硬骨头主要难点有三层第一层是请求校验它会在 Cookie 里嵌入合法会话令牌直接裸请求拿不到任何商品数据第二层是页面动态化列表页和详情页的大量字段靠 JavaScript 渲染或延迟加载第三层是风控体系触发频率异常时直接返回验证码页或“机器人检测”页且这个检测阈值会跟随账号信用动态变化。所以从 spider.zip 里跑 Amazon 爬虫时你要处理的不是“怎么解析页面”而是“怎么让自己的请求看起来像真人浏览”。我拆这类包的经验是先看它有没有带 Cookie 池或账号池的设计再看它有没有处理验证码页面的分支。两个都没有的话这个爬虫基本只能在低频率下运行不适合批量采集。3.2 商品列表与详情采集的最小框架Amazon 的页面结构虽然有变化但核心字段分布相对固定。列表页的每个商品项在div容器里包含标题、价格、评分、评价数详情页则集中在productTitle、priceblock_ourprice等 id 节点里。这个 zip 里对应的amazon_spider.py一般会写成两个方法一个抓列表一个抓详情。import re import requests from parsel import Selector class AmazonSpider: Amazon 商品采集骨架列表页解析 详情页解析 BASE_URL https://www.amazon.com/dp/{asin} def __init__(self, http_client, headers): self.http_client http_client self.headers headers def fetch_search_page(self, keyword, page1): url https://www.amazon.com/s params {k: keyword, page: page} resp self.http_client.get(url, headersself.headers, paramsparams) if resp is None: return [] selector Selector(textresp.text) items [] for product in selector.css(div[data-component-types-search-result]): asin product.attrib.get(data-asin) title product.css(h2 span::text).get() price product.css(span.a-price span.a-offscreen::text).get() if asin and title: items.append({asin: asin, title: title, price: price}) return items def fetch_detail(self, asin): url self.BASE_URL.format(asinasin) resp self.http_client.get(url, headersself.headers) if resp is None: return None selector Selector(textresp.text) title selector.css(#productTitle::text).get() price selector.css(span.priceToPay span.a-offscreen::text).get() if title is None and price is None: return {asin: asin, blocked: True} return { asin: asin, title: title.strip() if title else None, price: self._clean_price(price), } def _clean_price(self, price_text): if not price_text: return None match re.search(r\d\.\d{2}, price_text) return float(match.group()) if match else None这段代码里fetch_search_page用 CSS 选择器按 Amazon 当前搜索结果页的 DOM 结构提取>def check_blocked(html_text): 判断 Amazon 响应页是否被风控拦截返回拦截原因或 None if captcha in html_text.lower() or api-services-supportamazon.com in html_text: return captcha if automated access in html_text.lower(): return bot_detected if Robot Check in html_text or Enter the characters in html_text: return captcha_alt return None这段拦截检测的价值在于给上游调度提供决策依据检测到captcha就降速、换代理或换账号检测到bot_detected就停掉整个采集任务等风控冷却。如果只是简单地在异常时重试你会陷入“被封 → 重试 → 被封更死”的恶性循环。我见过一个翻车的例子某开发者把 Amazon 搜索词列表页的请求频率设为每 2 秒一次跑了不到 20 分钟就被限制访问因为没有处理验证码页后续请求全部拿到的是同一个“机器人检测”页面数据直接从源头污染了。正确做法是给每次请求引入 3 到 6 秒的随机延迟并把check_blocked做成每个响应必经的过滤器。4. 针对 Confluence 的空间导出认证、分页与富文本清洗4.1 Confluence REST API 与认证方式的选择Confluence 和 Amazon 最大的不同在于它有正规的 REST API不需要你去啃 HTML。spider.zip 里的confluence_spider.py如果走了 API 路线说明作者对官方接口比较熟如果它还在解析 HTML 页面那大概率只适用于旧版本或权限受限的场景。用 API 的好处是返回 JSON字段稳定且分页逻辑由服务端保证。import base64 import requests class ConfluenceClient: Confluence REST API 客户端支持账号密码与 Personal Access Token def __init__(self, base_url, usernameNone, passwordNone, tokenNone): self.base_url base_url.rstrip(/) self.session requests.Session() if token: self.session.headers[Authorization] fBearer {token} elif username and password: raw f{username}:{password}.encode(utf-8) self.session.headers[Authorization] fBasic {base64.b64encode(raw).decode()} self.session.headers[Content-Type] application/json def get(self, path, paramsNone): url f{self.base_url}{path} resp self.session.get(url, paramsparams, timeout15) resp.raise_for_status() return resp.json()认证方式里token优先级大于账号密码因为 Personal Access Token 更安全且不受密码过期影响。你要根据自己部署的 Confluence 版本确认 Token 的生成入口新版通常在头像菜单里的“Personal Access Tokens”选项中创建。如果目标 Confluence 启用了 SSO 单点登录这两种方式都可能失效你需要改用 Session 登录后的 Cookie。4.2 按空间拉取页面树与正文分页与递归Confluence 的 API 设计里最常用的两个接口是获取空间列表、获取空间下的页面列表。但页面之间存在父子层级有些页面挂了子页面但子页面不在同一个扁平接口的返回里。常见的做法是先拉全部页面分页数据再按parentId手工组装成树。class ConfluenceSpider(ConfluenceClient): 按空间导出所有页面内容组织成层级结构 def fetch_all_pages(self, space_key, limit50): pages [] start 0 while True: data self.get( /rest/api/content, params{ spaceKey: space_key, limit: limit, start: start, expand: body.storage,version, }, ) results data.get(results, []) pages.extend(results) if len(results) limit: break start limit return pages def build_tree(self, pages): tree {} for page in pages: parent_id page.get(ancestors, [{}])[-1].get(id) if page.get(ancestors) else None tree.setdefault(page[id], page) if parent_id: tree.setdefault(children, {}).setdefault(parent_id, []).append(page[id]) return tree这个fetch_all_pages用了标准的 offset 分页模式start每次增加limit直到返回结果数小于limit就认为拉完了。注意expandbody.storage参数很关键没有它你拿到的只是页面元数据拿不到正文内容。build_tree从ancestors的最后一个元素取直接父页面 id避免把祖辈层级当成父级。我在实际导出中遇到过一个问题页面数量上千时一次带body.storage的请求可能非常慢甚至超时。这种情况下不要试图在一个请求里膨胀全部字段可以先拉不带正文的元数据再按页面 id 分批拉正文。时间换空间的取舍在这种场景下是值得的。4.3 富文本转纯文本把存储格式洗成可检索的内容Confluence 的body.storage返回的是富文本 HTML里面充满ac:开头的宏标签、ri:引用的附件标记、structured-macro这类渲染专用节点。直接存 HTML 会浪费存储空间而且后续全文检索时会被宏标签干扰。这个 zip 里的utils/text_cleaner.py如果实现得好应该是一个基于 HTML 解析器的清洗器而不是粗暴地replace(, )。from html.parser import HTMLParser class ConfluenceTextCleaner(HTMLParser): 将 Confluence 存储格式 HTML 转为纯文本丢弃宏标签与样式 def __init__(self): super().__init__() self.parts [] self.skip_depth 0 self.skip_macro False def handle_starttag(self, tag, attrs): if tag.startswith(ac:) or tag in (structured-macro, table): self.skip_depth 1 self.skip_macro True if tag br: self.parts.append(\n) if tag p: self.parts.append(\n) def handle_endtag(self, tag): if self.skip_macro and (tag.startswith(ac:) or tag in (structured-macro, table)): self.skip_depth - 1 if self.skip_depth 0: self.skip_macro False def handle_data(self, data): if not self.skip_macro: self.parts.append(data) def get_text(self): return .join(self.parts).strip()skip_macro标记解决了核心问题宏内部的文本是渲染参数或元数据不该进入正文。如果不去掉它们产出的“纯文本”里会混入body、atlassian-macro这些噪音词直接拖累后续的搜索和摘要提取。handle_starttag里把p和br转成换行保证标题、段落之间有基本的阅读结构。这个清洗器在真正进入生产前需要结合你的页面模板调优。有些页面大量使用代码宏来展示示例代码清洗后代码块的缩进会丢失如果这种页面占比高你要单独在handle_starttag里捕获pre标签保留代码块的原始文本。5. 爬虫避坑指南Amazon 与 Confluence 场景下最常见的 5 个问题5.1 现象Amazon 抓到的数据一会儿正常一会儿全是验证码页原因通常是请求频率没有随机化或者 Cookie 池太薄。Amazon 的风控系统会记录单位时间内的请求密度和来源 IP 的变化规律一旦你的请求呈现稳定节奏就直接触发验证码。解决方法是把固定 sleep 改成随机区间例如 3 到 7 秒之间浮动并给每个会话绑定独立 Cookie。5.2 现象Confluence 分页越拉越慢最后直接被限流原因大概率是每次请求都带上了expandbody.storage服务端要渲染大量富文本响应体膨胀导致资源占用飙高。解决方法是先拉元数据再用并发度较低的多线程按子任务拉正文。至少把并发限制在 3 到 5 的区间给内网服务留足缓冲。5.3 现象下载的 zip 解压后一运行就报缺少模块原因不是代码有问题而是依赖没装全。这个 zip 里如果requirements.txt是手写的很可能没把所有间接依赖列进去。解决方法是不要在项目根目录直接 run而是先建虚拟环境然后逐个 import 检查缺失项。遇到parsel报错时还要确认w3lib是否同步安装。5.4 现象采集到的 Confluence 正文里混着 CSS 类名和宏标记原因是用正则或朴素字符串替换清洗 HTML而不是用解析器。HTML 标签可以嵌套任意层ac:structured-macro ac:nameinfo ac:schema-version1这种标签里的属性不是正文但朴素正则拦截不干净。解决方法是换用HTMLParser或lxml按节点类型做白名单过滤。5.5 现象断点续跑后同一批数据重复入库原因是pipeline.py用的是纯INSERT没有幂等设计。Amazon 的 ASIN 和 Confluence 的页面 ID 都是天然主键入库前先查重或者把建表语句改成UNIQUE约束加INSERT OR IGNORE。这个改动要提前做否则数据量大了以后去重代价极高。6. 让爬虫能在生产环境长期跑并发控制与断点续爬6.1 用队列解耦采集与入库告别“边抓边写”的噩梦爬虫工程里最容易翻车的写法是“抓到一条就写一条”一旦目标站返回异常数据脏数据已经落库。更稳的做法是双队列任务队列放待抓 URL结果队列放解析完的记录入库单独消费。spider.zip 如果没做这层解耦你可以自己补一个简单的queue方案。import queue import threading import time class CrawlerScheduler: 任务队列 结果队列采与存分离天然支持断点 def __init__(self, worker_count3): self.task_queue queue.Queue() self.result_queue queue.Queue() self.worker_count worker_count def start(self): workers [ threading.Thread(targetself._worker, daemonTrue) for _ in range(self.worker_count) ] for w in workers: w.start() self._consume_results() def _worker(self): while True: task self.task_queue.get() if task is None: # 空任务哨兵 break result self._fetch_and_parse(task) self.result_queue.put(result) self.task_queue.task_done() time.sleep(1) def _consume_results(self): while True: result self.result_queue.get() self._save_result(result) self.result_queue.task_done()worker_count决定并发请求数Amazon 场景设 3 到 5Confluence 内网可以放到 8 到 10。None作为哨兵值让线程优雅退出避免while True结束后线程还悬挂。这里的_fetch_and_parse和_save_result需要你按实际业务补全这个调度器解决的是流程控制不是具体的抓取逻辑。6.2 断点续爬的最小落地把进度写进本地文件而不是内存爬虫跑挂了最难受的莫过于从头再来。解决思路是每隔一段时间把“已完成的任务 ID”持久化到文件里启动时加载这个集合已经在里面的任务直接跳过。对 Amazon 用 ASIN 做去重键对 Confluence 用页面 ID。import json from pathlib import Path class ProgressTracker: 断点续爬的进度追踪基于本地文件重启后自动加载 def __init__(self, progress_fileprogress.json): self.progress_file Path(progress_file) self.done set() if self.progress_file.exists(): data json.loads(self.progress_file.read_text(encodingutf-8)) self.done set(data.get(done, [])) def mark_done(self, task_id): self.done.add(task_id) def is_done(self, task_id): return task_id in self.done def persist(self): self.progress_file.write_text( json.dumps({done: list(self.done)}), encodingutf-8, )我一般会在每完成 50 个任务时调用一次persist()避免频繁写盘拖慢主流程。启动时先is_done过滤任务队列这样重启后能直接接续进度。这个文件在任务结束后要保留因为后续再跑同类型采集时可能还要作为白名单去重。6.3 验证采集质量的一个效率技巧抽 20 条人工核对而不是全量检查爬虫跑完别急着入库完事。我的习惯是两个维度抽查维度一是页面拦截率随机抽 20 条数据看是不是验证码页混进去了维度二是字段完整性重点看价格字段有没有None、标题有没有乱码。人工核对 20 条的时间成本大约 10 分钟但能避免几万条数据全是废品的尴尬。技巧是写一个抽样脚本从结果里随机挑 20 条把标题、URL、关键字段打印出来然后逐条打开原始页面比对。如果超过 2 条数据字段对不上就该回头检查解析逻辑而不是继续加任务。这个步骤虽然老派却是长期采集经验里最可靠的验证方式比任何自动化校验都能提前暴露问题。我在维护这类采集工程一年多后最大的习惯变化是不再追求跑得多快而是每次跑完后先看拦截率和字段完整率这两个数字。Amazon 场景拦截率超过 15% 就该停掉排查Confluence 场景正文清洗后的字数明显偏少时要回看富文本结构是不是改版了。稳定的采集系统都是靠这些细节堆出来的希望帮到你。本文还有配套的精品资源点击获取