简介这是一款面向法律从业者、研究者和数据分析人员的裁判文书网批量爬取工具用于自动获取案件文书docid并下载完整文书同时提取正文概要、解析法律依据解决人工逐条检索效率低下的问题。工具采用多线程并发抓取并集成代理IP轮换策略可有效规避访问限制提升采集稳定性配套的源码、配置参数与登录脚本可支持二次开发和功能定制。资源包共20个文件以Python脚本、txt说明性文档为主包含9个py核心脚本、7个txt参数及状态清单以及docx使用指南、js登录辅助模块、Markdown说明等整包266KB轻量易部署。目前已有247人学习下载。通过附赠的docx使用说明和txt操作指南用户可快速完成代理IP配置、爬取参数设置等步骤并借助源码理解docid获取、文书下载及法律依据解析的完整流程适合需要批量采集裁判文书开展实证研究或数据分析的用户。1. 裁判文书网爬虫先搞清楚 docid 才是整个抓取的关键做法律数据分析的人多半有过这种经历想批量拿裁判文书结果在网页上翻页翻到手酸复制粘贴到 Excel 里再清洗一晚上只弄了几百份。市面上能搜到的爬虫脚本不少但能稳定跑到几千份、几万份的很少绝大多数死在一个环节——没把 docid 这个“文书身份证号”单独拆出来管理。docid 是文书在裁判文书网内部的唯一标识批量搜索拿到 docid 列表再按 docid 逐个下载正文整个过程才算真正解耦。这篇笔记要拆的资源就是一套围绕 docid 调度的 Python 爬虫搜索接口拿 id、多线程下载正文、解析正文概要与法律依据配合代理 IP 降低被限流的概率。适合正在做法律文本挖掘、案件数据分析或者单纯需要批量语料的人目标不是“能跑”而是“能持续跑到上万条不翻车”。2. 先从接口层拆解搜索响应里的 data 结构决定整条流水线怎么设计2.1 为什么不能直接爬 HTML 页面裁判文书网的搜索结果是异步渲染的直接请求列表页拿不到完整数据拿到手的只是一个空壳 DOM真正的文书列表藏在 XHR 请求的 JSON 响应里。很多新手踩的第一个坑就是对着页面源码找标题、找案号找了半天一无所获最后误以为是被反爬挡住了其实是抓错了对象。正确做法是打开浏览器开发者工具切到 Network 面板在搜索框输入关键词后观察发起的那条 POST 请求。响应体是 JSON里面有一个列表字段每项包含 docid、案件名称、案号、法院、裁判日期等核心元数据。docid 就是后面所有下载动作的输入参数这一步相当于整个爬虫的地基。拿到的 JSON 里还有分页信息比如 total、pageCount 这类字段。我的习惯是先把响应完整落盘成 JSON 文件存一份再写解析逻辑。因为接口字段可能变留存原始数据能让你在字段调整时不用重新抓一遍。2.2 用 requests 维护一个常驻会话常见做法是建一个 requests.Session把必要的请求头固定挂在会话上避免每次请求都重复构造 headers。Session 会自动管理 Cookie这在搜索接口场景下很重要因为一次搜索会话的 Cookie 里通常会携带搜索条件的状态。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Referer: https://wenshu.court.gov.cn/, Accept: application/json, text/plain, */*, Content-Type: application/json }) search_url https://wenshu.court.gov.cn/website/wenshu/181107ANFZ0BX5SK4/index.html # 注意列表数据的真实地址以抓包为准这里用变量占位 list_api https://wenshu.court.gov.cn/website/parse/rest.q4w def search_docid(keyword: str, page: int, page_size: int 10) - dict: payload { keyword: keyword, pageNum: page, pageSize: page_size, sortFields: s50:desc, queryCondition: {} } resp session.post(list_api, jsonpayload, timeout15) resp.raise_for_status() return resp.json()逻辑说明Session 对象复用 TCP 连接比每次 requests.get 新建连接快不少headers 里的 Referer 和 UA 是防反爬的第一道门槛没有 Referer 的请求很容易被识别为异常流量。payload 里的 pageNum 和 pageSize 控制翻页节奏sortFields 控制排序方式实际字段名以抓包为准。timeout 必须显式设置否则网络抖动时线程会一直挂在那里拖垮整个任务队列。2.3 把分页总数算出来再分配任务拿到响应后先看 data 里的 total 字段除以 pageSize 向上取整就是总页数。这里有个小坑裁判文书网的搜索接口在某些关键词下会返回 fixed 的“最多前 500 页”之类的限制具体以实际响应为准。我一般会把总页数打日志同时跟数据库里已有数据量做对比判断是增量抓取还是全量重建。import math def calc_total_pages(resp_json: dict, page_size: int) - int: total resp_json.get(data, {}).get(total, 0) return math.ceil(total / page_size)参数说明page_size 不建议设太大常见经验值是 10 到 20。设太大时单次响应体膨胀解析变慢而且更容易触发接口的限流规则设太小又会导致请求次数剧增代理 IP 消耗变快。3. 把 docid 调度拆成生产者-消费者多线程下载不再互相踩脚3.1 为什么不用简单的 for 循环加 ThreadPool很多人写多线程爬虫是这么干的for 循环里把每个任务丢进 ThreadPoolExecutor然后 map 一下等结果。这种方式在小规模任务上没问题但场景换成“搜索 2000 页拿 docid再下载 2 万篇文书”时就会出现问题搜索和下载是两类完全不同的任务前者快、后者慢放在同一个线程池里会让搜索线程被下载任务阻塞整体吞吐上不去。更合理的模型是生产者-消费者生产者线程只负责翻页搜索、把拿到的 docid 放进队列消费者线程从队列里取 docid去下载正文、解析字段。两队之间通过队列解耦队列长度可以控制内存占用也天然实现了任务缓冲。3.2 用 queue.Queue 做 docid 中转站import queue import threading import time from concurrent.futures import ThreadPoolExecutor docid_queue queue.Queue(maxsize2000) seen_docid set() lock threading.Lock() def produce_docid(keyword: str, total_pages: int): for page in range(1, total_pages 1): resp_json search_docid(keyword, page, page_size20) records resp_json.get(data, {}).get(content, []) for item in records: docid item.get(docid) if not docid: continue with lock: if docid in seen_docid: continue seen_docid.add(docid) docid_queue.put(docid) time.sleep(1) # 控制搜索接口的请求频率逻辑说明seen_docid 集合负责去重。翻页时不同页之间偶尔会有重复数据如果不去重消费者就会重复下载浪费带宽也浪费代理 IP。lock 保护集合的并发读写避免多线程下 set 出现脏读。docid_queue 的 maxsize 设为 2000是因为消费者下载速度远慢于搜索速度如果队列无限增长内存会被 docid 元数据撑爆。3.3 消费者线程里每个线程独立 Session消费者这边的关键点Session 不是线程安全的绝不能让多个线程共享同一个 Session 实例。我见过有人把全局 Session 丢给线程池用结果 cookie 互相覆盖请求一会儿 200 一会儿 401查了半天发现是 Session 打架。正确做法是每个消费者线程内部自己 new 一个 Session。def consume_docid(): local_session requests.Session() local_session.headers.update({ User-Agent: Mozilla/5.0, Referer: https://wenshu.court.gov.cn/ }) while True: try: docid docid_queue.get(timeout3) except queue.Empty: break try: content fetch_doc_content(local_session, docid) if content: save_doc(docid, content) except Exception as e: log_failure(docid, str(e)) finally: docid_queue.task_done() def run_consumer(worker_count: int 8): with ThreadPoolExecutor(max_workersworker_count) as pool: for _ in range(worker_count): pool.submit(consume_docid)参数说明worker_count 我常用的区间是 4 到 12。太小下载速度跟不上搜索速度队列越积越长太大对目标站点压力过高触发限流后反而全被 403。timeout3 是队列读取超时当生产者结束、队列被消费空后消费者能自动退出不会永久阻塞。save_doc 和 log_failure 建议做成独立函数方便后续把失败任务单独重跑。4. 正文解析与法律依据提取xpath 文本函数、字段清洗和入库4.1 正文页面的 DOM 结构没那么规整拿到 docid 后通常需要请求一个详情接口来拿正文。正文页可能是直接返回 HTML 片段也可能是被转义过的 JSON 字符串。常见做法是请求详情接口拿 JSON再取里面的 htmlContent 字段。正文结构在不同法院、不同文书类型之间差异很大有的文书带“本院认为”段落有的只是简单的事实陈述。解析时要抓住几个稳定存在的锚点标题、案号、法院名称、正文主体、裁判日期。正文主体通常是某个特定 id 或 class 的 div 内部文本但你不能保证它永远不变所以解析层要做容错。4.2 用 lxml 的 xpath 和 text() 函数提取关键字段from lxml import html def parse_doc_content(raw_html: str) - dict: tree html.fromstring(raw_html) title tree.xpath(//div[idtitle]//text()) title .join(title).strip() if title else content_parts tree.xpath(//div[idcontent]//text()) content .join(part.strip() for part in content_parts if part.strip()) court tree.xpath(//span[classcourt-name]//text()) court court[0].strip() if court else date_text tree.xpath(//span[classcase-date]//text()) date_text date_text[0].strip() if date_text else return { title: title, content: content, court: court, case_date: date_text }逻辑说明xpath 的 //text() 会取节点下所有文本节点包括子节点里的文字这对提取大段正文很关键。如果只用 /text() 只能取直接子节点正文内容会被截断。所有字段提取后都做 strip 清洗正文里残留的空行、空格会影响后续分词和向量化效果。4.3 法律依据解析不靠模型靠正则就行法律依据的提取其实不需要上 NLP 模型裁判文书的引用格式相对固定无非是“依照《中华人民共和国刑法》第二百六十四条”或者“根据《中华人民共和国民法典》第一千一百六十五条”这类句式。用正则把《》里的法律名称和具体条数抓出来就够用。import re pattern re.compile( r[依根结适]?据?\s*《([^》])》\s*(?:第[一二三四五六七八九十百千万零〇\d]条)? r(?:之)?(?:第[一二三四五六七八九十百千万零〇\d]款)? ) def extract_law_basis(content: str) - list: matches pattern.findall(content) # 去掉重复但保留出现顺序 seen set() result [] for law in matches: if law not in seen: seen.add(law) result.append(law) return result参数说明这个正则不是万能的文书里偶尔会出现“依照《中华人民共和国刑法》第X条、第Y条”的合并写法此时正则只能抓到第一个。我的处理方式是先抓所有含《》的片段再对每个片段单独抽条数宁可多抓也不要漏抓后续可以人工抽样审核。4.4 字段落库与增量更新解析结果建议直接进 SQLite 或 MySQL。SQLite 适合个人研究单文件迁移方便MySQL 适合多人协作或数据量超过几十万条。表结构至少要包含 docid 主键、标题、案号、法院、裁判日期、正文、法律依据、抓取时间、来源关键词。import sqlite3 def init_db(db_path: str): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS docs ( docid TEXT PRIMARY KEY, title TEXT, court TEXT, case_date TEXT, content TEXT, law_basis TEXT, fetch_time TEXT, source_keyword TEXT ) ) conn.commit() return conn参数说明docid 设为主键意义重大因为同一篇文书可能被不同关键词搜出来重复抓取主键冲突时可以直接跳过或更新天然实现幂等。fetch_time 存的是 ISO 格式时间字符串方便按天统计入库量判断爬虫是否还在正常工作。5. 代理IP与踩坑排查高频翻车点、反爬识别和失败重试策略5.1 代理IP怎么接进 requests代理 IP 是这个项目里争议最大也最容易写错的模块。常见做法是维护一个代理池每次请求前从池子里随机挑一个代理挂到 requests 的 proxies 参数上。这里要明确一点代理 IP 是降低单 IP 请求频率的手段让请求分散到多个出口 IP 上而不是绕过正常访问限制。import random PROXY_POOL [ {http: http://user:passip1:port, https: http://user:passip1:port}, {http: http://user:passip2:port, https: http://user:passip2:port}, ] def get_random_proxy() - dict: return random.choice(PROXY_POOL) def fetch_with_proxy(session: requests.Session, url: str, **kwargs): proxy get_random_proxy() kwargs[proxies] proxy try: resp session.get(url, timeout15, **kwargs) return resp except requests.exceptions.ProxyError as e: log_failure(proxy_error, str(e)) return None参数说明proxies 字典的 key 必须区分 http 和 https否则 requests 不知道该用哪个代理。带认证的代理需要在地址里写 user:pass。这里我把代理逻辑单独包了一层而不是塞在业务代码里方便后续做质量检测——比如某个代理连续失败三次就自动拉黑。5.2 高频踩坑403 和空白响应现象脚本跑前几百条数据一切正常跑到 1000 条左右开始大量 403或者返回的正文字段为空。原因目标接口对单 IP 的请求频次做了限制短时间集中请求触发风控另外请求头过于单一也会被识别比如所有请求的 User-Agent 完全一样。解决第一给每个请求之间的间隔引入随机性不要固定 sleep(1)改成 sleep(0.5) 到 sleep(2) 之间随机第二代理池至少准备 10 个以上的 IP每次请求随机换第三UA 池也要做多套轮换让 User-Agent 和 IP 不要形成固定映射。import random import time def random_sleep(): time.sleep(random.uniform(0.5, 2.0))5.3 高频踩坑队列消费空但任务没跑完现象消费者线程全部退出了但数据库里的数据量远小于预期日志里也没有报错。原因docid_queue.get(timeout3) 在队列为空时会抛 queue.Empty消费者就 break 退出了。问题是生产者的搜索请求可能还在路上队列暂时为空不代表任务结束。解决引入一个全局的 producer_done 标志位。消费者在队列超时后先检查 producer_done 是否为 False如果生产者还没结束就继续循环等待而不是立刻退出。producer_done False def consume_docid_v2(): local_session requests.Session() # 初始化 headers 省略 while True: try: docid docid_queue.get(timeout3) except queue.Empty: if producer_done and docid_queue.empty(): break continue try: content fetch_doc_content(local_session, docid) if content: save_doc(docid, content) except Exception as e: log_failure(docid, str(e)) finally: docid_queue.task_done()5.4 高频踩坑编码错乱正文变成乱码现象下载下来的文书正文出现大量“锟斤拷”或者繁体字乱码。原因响应内容被 requests 默认按 ISO-8859-1 解码而页面实际是 UTF-8 或 GBK。requests 只有在响应头里明确声明 charset 时才会用对编码很多接口不声明。解决优先用 resp.apparent_encoding 自动检测如果检测结果不对再按已知站点编码硬解码。更稳妥的做法是对照文档开头的那几个固定字符判断编码。def decode_response(resp: requests.Response) - str: if resp.apparent_encoding: resp.encoding resp.apparent_encoding return resp.text5.5 高频踩坑失败任务没有“后悔药”现象全量跑完后发现某些 docid 下载失败想单独重跑结果找不到是哪些 docid 失败了只能全量重来。解决从第一天起就把失败任务单独记录。log_failure 里除了写错误原因还要把 docid、失败时间、当前代理 IP 都写进去。重跑时直接读失败日志里的 docid 列表构造一个小队列单独处理不要重新搜索。def log_failure(docid: str, reason: str): with open(failed.log, a, encodingutf-8) as f: f.write(f{docid}\t{reason}\t{time.time()}\n)6. 从“能跑”到“敢上量”用样本校验和断点续抓验证数据完整性到了这一步代码已经能跑通全流程但离“敢上量”还差一道工序验证。我自己的习惯是任何一个爬虫项目都要先跑一个 dry-run 模式只抓 20 条样本人工核对字段是否完整再决定是否放开全量。这不是浪费时间是给后面的上量省时间。dry-run 要检查的字段清单docid 是否唯一、标题是否为空、正文长度是否正常太短可能是被反爬挡了比如只剩一个空壳、法律依据提取结果是否合理。我见过一次翻车案例某次改版后正文接口返回了一段 JSON 字符串而非 HTML解析器没有报错但所有正文都变成了一大段带引号和花括号的文本入库后 NLP 模型直接被打爆。从那以后我每次上量前都会强制跑一遍 20 条样本 人工抽查正文前 200 字。另一个验证点是断点续抓。全量抓取过程中难免中断比如代理 IP 全部失效、网络抖动、服务器重启。续抓策略就是数据库里已有的 docid 集合作为过滤条件搜索拿到的新 docid 先跟已入库 docid 求差集剩下的才进队列。这样断点后重新跑脚本不会重复下载已经入库的文书。def filter_existing_docid(docid_list: list, conn: sqlite3.Connection) - list: existing set() for row in conn.execute(SELECT docid FROM docs): existing.add(row[0]) return [d for d in docid_list if d not in existing]逻辑说明这个函数把全库 docid 一次性加载到内存集合里几十万条规模下内存占用完全可接受。每次搜索请求返回后立即做一次差集过滤把重复 docid 挡在队列外消费者就不会做无用功。如果 docid 已经超过百万级别建议把过滤逻辑改成分批 IN 查询避免一次加载过多。最后补一个上量前的“体检清单”检查项合格标准20 条样本中字段完整率不低于 95%正文长度分布无空正文中位数在 1000 字以上法律依据提取结果人工抽查 10 条准确率不低于 80%连续请求 1000 次的 403 率低于 2%失败任务可重跑failed.log 里 docid 能被重新消费这条流水线跑熟之后你会发现 docid 调度才是整个爬虫的地基代理 IP 和编码处理只是外围的“玄学”加成。从那以后我写任何爬虫都先做小样本验证再谈上量这套习惯帮我避开了好几次大规模返工的坑。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/26 16:23:01 · 阅读 1.2k 裁判文书网爬虫实战：docid调度与多线程下载全解析
简介这是一款面向法律从业者、研究者和数据分析人员的裁判文书网批量爬取工具用于自动获取案件文书docid并下载完整文书同时提取正文概要、解析法律依据解决人工逐条检索效率低下的问题。工具采用多线程并发抓取并集成代理IP轮换策略可有效规避访问限制提升采集稳定性配套的源码、配置参数与登录脚本可支持二次开发和功能定制。资源包共20个文件以Python脚本、txt说明性文档为主包含9个py核心脚本、7个txt参数及状态清单以及docx使用指南、js登录辅助模块、Markdown说明等整包266KB轻量易部署。目前已有247人学习下载。通过附赠的docx使用说明和txt操作指南用户可快速完成代理IP配置、爬取参数设置等步骤并借助源码理解docid获取、文书下载及法律依据解析的完整流程适合需要批量采集裁判文书开展实证研究或数据分析的用户。1. 裁判文书网爬虫先搞清楚 docid 才是整个抓取的关键做法律数据分析的人多半有过这种经历想批量拿裁判文书结果在网页上翻页翻到手酸复制粘贴到 Excel 里再清洗一晚上只弄了几百份。市面上能搜到的爬虫脚本不少但能稳定跑到几千份、几万份的很少绝大多数死在一个环节——没把 docid 这个“文书身份证号”单独拆出来管理。docid 是文书在裁判文书网内部的唯一标识批量搜索拿到 docid 列表再按 docid 逐个下载正文整个过程才算真正解耦。这篇笔记要拆的资源就是一套围绕 docid 调度的 Python 爬虫搜索接口拿 id、多线程下载正文、解析正文概要与法律依据配合代理 IP 降低被限流的概率。适合正在做法律文本挖掘、案件数据分析或者单纯需要批量语料的人目标不是“能跑”而是“能持续跑到上万条不翻车”。2. 先从接口层拆解搜索响应里的 data 结构决定整条流水线怎么设计2.1 为什么不能直接爬 HTML 页面裁判文书网的搜索结果是异步渲染的直接请求列表页拿不到完整数据拿到手的只是一个空壳 DOM真正的文书列表藏在 XHR 请求的 JSON 响应里。很多新手踩的第一个坑就是对着页面源码找标题、找案号找了半天一无所获最后误以为是被反爬挡住了其实是抓错了对象。正确做法是打开浏览器开发者工具切到 Network 面板在搜索框输入关键词后观察发起的那条 POST 请求。响应体是 JSON里面有一个列表字段每项包含 docid、案件名称、案号、法院、裁判日期等核心元数据。docid 就是后面所有下载动作的输入参数这一步相当于整个爬虫的地基。拿到的 JSON 里还有分页信息比如 total、pageCount 这类字段。我的习惯是先把响应完整落盘成 JSON 文件存一份再写解析逻辑。因为接口字段可能变留存原始数据能让你在字段调整时不用重新抓一遍。2.2 用 requests 维护一个常驻会话常见做法是建一个 requests.Session把必要的请求头固定挂在会话上避免每次请求都重复构造 headers。Session 会自动管理 Cookie这在搜索接口场景下很重要因为一次搜索会话的 Cookie 里通常会携带搜索条件的状态。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Referer: https://wenshu.court.gov.cn/, Accept: application/json, text/plain, */*, Content-Type: application/json }) search_url https://wenshu.court.gov.cn/website/wenshu/181107ANFZ0BX5SK4/index.html # 注意列表数据的真实地址以抓包为准这里用变量占位 list_api https://wenshu.court.gov.cn/website/parse/rest.q4w def search_docid(keyword: str, page: int, page_size: int 10) - dict: payload { keyword: keyword, pageNum: page, pageSize: page_size, sortFields: s50:desc, queryCondition: {} } resp session.post(list_api, jsonpayload, timeout15) resp.raise_for_status() return resp.json()逻辑说明Session 对象复用 TCP 连接比每次 requests.get 新建连接快不少headers 里的 Referer 和 UA 是防反爬的第一道门槛没有 Referer 的请求很容易被识别为异常流量。payload 里的 pageNum 和 pageSize 控制翻页节奏sortFields 控制排序方式实际字段名以抓包为准。timeout 必须显式设置否则网络抖动时线程会一直挂在那里拖垮整个任务队列。2.3 把分页总数算出来再分配任务拿到响应后先看 data 里的 total 字段除以 pageSize 向上取整就是总页数。这里有个小坑裁判文书网的搜索接口在某些关键词下会返回 fixed 的“最多前 500 页”之类的限制具体以实际响应为准。我一般会把总页数打日志同时跟数据库里已有数据量做对比判断是增量抓取还是全量重建。import math def calc_total_pages(resp_json: dict, page_size: int) - int: total resp_json.get(data, {}).get(total, 0) return math.ceil(total / page_size)参数说明page_size 不建议设太大常见经验值是 10 到 20。设太大时单次响应体膨胀解析变慢而且更容易触发接口的限流规则设太小又会导致请求次数剧增代理 IP 消耗变快。3. 把 docid 调度拆成生产者-消费者多线程下载不再互相踩脚3.1 为什么不用简单的 for 循环加 ThreadPool很多人写多线程爬虫是这么干的for 循环里把每个任务丢进 ThreadPoolExecutor然后 map 一下等结果。这种方式在小规模任务上没问题但场景换成“搜索 2000 页拿 docid再下载 2 万篇文书”时就会出现问题搜索和下载是两类完全不同的任务前者快、后者慢放在同一个线程池里会让搜索线程被下载任务阻塞整体吞吐上不去。更合理的模型是生产者-消费者生产者线程只负责翻页搜索、把拿到的 docid 放进队列消费者线程从队列里取 docid去下载正文、解析字段。两队之间通过队列解耦队列长度可以控制内存占用也天然实现了任务缓冲。3.2 用 queue.Queue 做 docid 中转站import queue import threading import time from concurrent.futures import ThreadPoolExecutor docid_queue queue.Queue(maxsize2000) seen_docid set() lock threading.Lock() def produce_docid(keyword: str, total_pages: int): for page in range(1, total_pages 1): resp_json search_docid(keyword, page, page_size20) records resp_json.get(data, {}).get(content, []) for item in records: docid item.get(docid) if not docid: continue with lock: if docid in seen_docid: continue seen_docid.add(docid) docid_queue.put(docid) time.sleep(1) # 控制搜索接口的请求频率逻辑说明seen_docid 集合负责去重。翻页时不同页之间偶尔会有重复数据如果不去重消费者就会重复下载浪费带宽也浪费代理 IP。lock 保护集合的并发读写避免多线程下 set 出现脏读。docid_queue 的 maxsize 设为 2000是因为消费者下载速度远慢于搜索速度如果队列无限增长内存会被 docid 元数据撑爆。3.3 消费者线程里每个线程独立 Session消费者这边的关键点Session 不是线程安全的绝不能让多个线程共享同一个 Session 实例。我见过有人把全局 Session 丢给线程池用结果 cookie 互相覆盖请求一会儿 200 一会儿 401查了半天发现是 Session 打架。正确做法是每个消费者线程内部自己 new 一个 Session。def consume_docid(): local_session requests.Session() local_session.headers.update({ User-Agent: Mozilla/5.0, Referer: https://wenshu.court.gov.cn/ }) while True: try: docid docid_queue.get(timeout3) except queue.Empty: break try: content fetch_doc_content(local_session, docid) if content: save_doc(docid, content) except Exception as e: log_failure(docid, str(e)) finally: docid_queue.task_done() def run_consumer(worker_count: int 8): with ThreadPoolExecutor(max_workersworker_count) as pool: for _ in range(worker_count): pool.submit(consume_docid)参数说明worker_count 我常用的区间是 4 到 12。太小下载速度跟不上搜索速度队列越积越长太大对目标站点压力过高触发限流后反而全被 403。timeout3 是队列读取超时当生产者结束、队列被消费空后消费者能自动退出不会永久阻塞。save_doc 和 log_failure 建议做成独立函数方便后续把失败任务单独重跑。4. 正文解析与法律依据提取xpath 文本函数、字段清洗和入库4.1 正文页面的 DOM 结构没那么规整拿到 docid 后通常需要请求一个详情接口来拿正文。正文页可能是直接返回 HTML 片段也可能是被转义过的 JSON 字符串。常见做法是请求详情接口拿 JSON再取里面的 htmlContent 字段。正文结构在不同法院、不同文书类型之间差异很大有的文书带“本院认为”段落有的只是简单的事实陈述。解析时要抓住几个稳定存在的锚点标题、案号、法院名称、正文主体、裁判日期。正文主体通常是某个特定 id 或 class 的 div 内部文本但你不能保证它永远不变所以解析层要做容错。4.2 用 lxml 的 xpath 和 text() 函数提取关键字段from lxml import html def parse_doc_content(raw_html: str) - dict: tree html.fromstring(raw_html) title tree.xpath(//div[idtitle]//text()) title .join(title).strip() if title else content_parts tree.xpath(//div[idcontent]//text()) content .join(part.strip() for part in content_parts if part.strip()) court tree.xpath(//span[classcourt-name]//text()) court court[0].strip() if court else date_text tree.xpath(//span[classcase-date]//text()) date_text date_text[0].strip() if date_text else return { title: title, content: content, court: court, case_date: date_text }逻辑说明xpath 的 //text() 会取节点下所有文本节点包括子节点里的文字这对提取大段正文很关键。如果只用 /text() 只能取直接子节点正文内容会被截断。所有字段提取后都做 strip 清洗正文里残留的空行、空格会影响后续分词和向量化效果。4.3 法律依据解析不靠模型靠正则就行法律依据的提取其实不需要上 NLP 模型裁判文书的引用格式相对固定无非是“依照《中华人民共和国刑法》第二百六十四条”或者“根据《中华人民共和国民法典》第一千一百六十五条”这类句式。用正则把《》里的法律名称和具体条数抓出来就够用。import re pattern re.compile( r[依根结适]?据?\s*《([^》])》\s*(?:第[一二三四五六七八九十百千万零〇\d]条)? r(?:之)?(?:第[一二三四五六七八九十百千万零〇\d]款)? ) def extract_law_basis(content: str) - list: matches pattern.findall(content) # 去掉重复但保留出现顺序 seen set() result [] for law in matches: if law not in seen: seen.add(law) result.append(law) return result参数说明这个正则不是万能的文书里偶尔会出现“依照《中华人民共和国刑法》第X条、第Y条”的合并写法此时正则只能抓到第一个。我的处理方式是先抓所有含《》的片段再对每个片段单独抽条数宁可多抓也不要漏抓后续可以人工抽样审核。4.4 字段落库与增量更新解析结果建议直接进 SQLite 或 MySQL。SQLite 适合个人研究单文件迁移方便MySQL 适合多人协作或数据量超过几十万条。表结构至少要包含 docid 主键、标题、案号、法院、裁判日期、正文、法律依据、抓取时间、来源关键词。import sqlite3 def init_db(db_path: str): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS docs ( docid TEXT PRIMARY KEY, title TEXT, court TEXT, case_date TEXT, content TEXT, law_basis TEXT, fetch_time TEXT, source_keyword TEXT ) ) conn.commit() return conn参数说明docid 设为主键意义重大因为同一篇文书可能被不同关键词搜出来重复抓取主键冲突时可以直接跳过或更新天然实现幂等。fetch_time 存的是 ISO 格式时间字符串方便按天统计入库量判断爬虫是否还在正常工作。5. 代理IP与踩坑排查高频翻车点、反爬识别和失败重试策略5.1 代理IP怎么接进 requests代理 IP 是这个项目里争议最大也最容易写错的模块。常见做法是维护一个代理池每次请求前从池子里随机挑一个代理挂到 requests 的 proxies 参数上。这里要明确一点代理 IP 是降低单 IP 请求频率的手段让请求分散到多个出口 IP 上而不是绕过正常访问限制。import random PROXY_POOL [ {http: http://user:passip1:port, https: http://user:passip1:port}, {http: http://user:passip2:port, https: http://user:passip2:port}, ] def get_random_proxy() - dict: return random.choice(PROXY_POOL) def fetch_with_proxy(session: requests.Session, url: str, **kwargs): proxy get_random_proxy() kwargs[proxies] proxy try: resp session.get(url, timeout15, **kwargs) return resp except requests.exceptions.ProxyError as e: log_failure(proxy_error, str(e)) return None参数说明proxies 字典的 key 必须区分 http 和 https否则 requests 不知道该用哪个代理。带认证的代理需要在地址里写 user:pass。这里我把代理逻辑单独包了一层而不是塞在业务代码里方便后续做质量检测——比如某个代理连续失败三次就自动拉黑。5.2 高频踩坑403 和空白响应现象脚本跑前几百条数据一切正常跑到 1000 条左右开始大量 403或者返回的正文字段为空。原因目标接口对单 IP 的请求频次做了限制短时间集中请求触发风控另外请求头过于单一也会被识别比如所有请求的 User-Agent 完全一样。解决第一给每个请求之间的间隔引入随机性不要固定 sleep(1)改成 sleep(0.5) 到 sleep(2) 之间随机第二代理池至少准备 10 个以上的 IP每次请求随机换第三UA 池也要做多套轮换让 User-Agent 和 IP 不要形成固定映射。import random import time def random_sleep(): time.sleep(random.uniform(0.5, 2.0))5.3 高频踩坑队列消费空但任务没跑完现象消费者线程全部退出了但数据库里的数据量远小于预期日志里也没有报错。原因docid_queue.get(timeout3) 在队列为空时会抛 queue.Empty消费者就 break 退出了。问题是生产者的搜索请求可能还在路上队列暂时为空不代表任务结束。解决引入一个全局的 producer_done 标志位。消费者在队列超时后先检查 producer_done 是否为 False如果生产者还没结束就继续循环等待而不是立刻退出。producer_done False def consume_docid_v2(): local_session requests.Session() # 初始化 headers 省略 while True: try: docid docid_queue.get(timeout3) except queue.Empty: if producer_done and docid_queue.empty(): break continue try: content fetch_doc_content(local_session, docid) if content: save_doc(docid, content) except Exception as e: log_failure(docid, str(e)) finally: docid_queue.task_done()5.4 高频踩坑编码错乱正文变成乱码现象下载下来的文书正文出现大量“锟斤拷”或者繁体字乱码。原因响应内容被 requests 默认按 ISO-8859-1 解码而页面实际是 UTF-8 或 GBK。requests 只有在响应头里明确声明 charset 时才会用对编码很多接口不声明。解决优先用 resp.apparent_encoding 自动检测如果检测结果不对再按已知站点编码硬解码。更稳妥的做法是对照文档开头的那几个固定字符判断编码。def decode_response(resp: requests.Response) - str: if resp.apparent_encoding: resp.encoding resp.apparent_encoding return resp.text5.5 高频踩坑失败任务没有“后悔药”现象全量跑完后发现某些 docid 下载失败想单独重跑结果找不到是哪些 docid 失败了只能全量重来。解决从第一天起就把失败任务单独记录。log_failure 里除了写错误原因还要把 docid、失败时间、当前代理 IP 都写进去。重跑时直接读失败日志里的 docid 列表构造一个小队列单独处理不要重新搜索。def log_failure(docid: str, reason: str): with open(failed.log, a, encodingutf-8) as f: f.write(f{docid}\t{reason}\t{time.time()}\n)6. 从“能跑”到“敢上量”用样本校验和断点续抓验证数据完整性到了这一步代码已经能跑通全流程但离“敢上量”还差一道工序验证。我自己的习惯是任何一个爬虫项目都要先跑一个 dry-run 模式只抓 20 条样本人工核对字段是否完整再决定是否放开全量。这不是浪费时间是给后面的上量省时间。dry-run 要检查的字段清单docid 是否唯一、标题是否为空、正文长度是否正常太短可能是被反爬挡了比如只剩一个空壳、法律依据提取结果是否合理。我见过一次翻车案例某次改版后正文接口返回了一段 JSON 字符串而非 HTML解析器没有报错但所有正文都变成了一大段带引号和花括号的文本入库后 NLP 模型直接被打爆。从那以后我每次上量前都会强制跑一遍 20 条样本 人工抽查正文前 200 字。另一个验证点是断点续抓。全量抓取过程中难免中断比如代理 IP 全部失效、网络抖动、服务器重启。续抓策略就是数据库里已有的 docid 集合作为过滤条件搜索拿到的新 docid 先跟已入库 docid 求差集剩下的才进队列。这样断点后重新跑脚本不会重复下载已经入库的文书。def filter_existing_docid(docid_list: list, conn: sqlite3.Connection) - list: existing set() for row in conn.execute(SELECT docid FROM docs): existing.add(row[0]) return [d for d in docid_list if d not in existing]逻辑说明这个函数把全库 docid 一次性加载到内存集合里几十万条规模下内存占用完全可接受。每次搜索请求返回后立即做一次差集过滤把重复 docid 挡在队列外消费者就不会做无用功。如果 docid 已经超过百万级别建议把过滤逻辑改成分批 IN 查询避免一次加载过多。最后补一个上量前的“体检清单”检查项合格标准20 条样本中字段完整率不低于 95%正文长度分布无空正文中位数在 1000 字以上法律依据提取结果人工抽查 10 条准确率不低于 80%连续请求 1000 次的 403 率低于 2%失败任务可重跑failed.log 里 docid 能被重新消费这条流水线跑熟之后你会发现 docid 调度才是整个爬虫的地基代理 IP 和编码处理只是外围的“玄学”加成。从那以后我写任何爬虫都先做小样本验证再谈上量这套习惯帮我避开了好几次大规模返工的坑。希望帮到你。本文还有配套的精品资源点击获取