企业官网建设流程全解析

简介：这套基于Python的分布式网络爬虫系统，专为中国裁判文书网全量案件数据采集而设计，面向法律研究者、数据分析师及有一定基础的爬虫开发者，解决司法文书批量获取、解析与本地化存储难题。系统内置反反爬机制与数据清洗模块，可自动化完成案件信息采集、文书详情页解析与下载，并具备多节点分布式扩展能力，适用于法律实证研究、类案检索或司法数据积累等场景。压缩包共17个文件，以13个Python脚本为核心，涵盖生产者、配置检测、日志记录与CLI入口等模块，另附带README说明、一份附赠资源Word文档及运行日志，整体仅42KB，轻量易部署。目前已有109人学习使用。通过源码可掌握请求策略、解析流程、清洗逻辑与异常处理思路，快速搭建个人司法数据采集工具，节省大量人工检索时间。

1. 中国裁判文书网全量数据爬取:这套 Python 分布式网络爬虫到底值不值得下

做法律研究的人大概都经历过这种崩溃:在某裁判文书网按关键词检索,一页一页翻,手动复制判决书,再把案号、案由、裁判日期逐个粘进 Excel,折腾几天数据还是不完整。真正要跑实证分析,你需要的是几万、几十万份文书的结构化字段,而不是网页上的连续文本。这套基于 Python 的分布式网络爬虫系统,就是把裁判文书网的案件信息采集、详情页解析、数据清洗与存储串成一条自动化流水线,内置反反爬机制与频率控制,下载解压后可以直接按模块使用。适合法学研究生做批量样本研究、律所做类案检索、司法数据分析从业者建语料库,也适合 Python 爬虫工程师参考工程结构。

2. 分布式爬虫架构与任务拆分:为什么单机跑不动,调度队列怎么设计

2.1 单机爬虫的瓶颈:速度、封禁和断点续爬

最早的版本我是拿 requests 加 BeautifulSoup 写单线程脚本,跑一个下午抓了两万条检索结果,看起来速度能接受,但问题紧接着就来了:请求频率一上来,触发访问限制后整个任务从头再来,抓到的数据全在内存里,进程一断什么都没留下。更麻烦的是,当你想按案由、地域、年份多个维度组合采集时,单机串行根本排不过来。

分布式要解决的不是"多几台机器",而是把任务拆分、队列缓冲、节点伸缩、断点续爬这几件事做对。我一般这样分层:调度节点负责把待爬的检索条件或详情页 ID 推入队列;抓取节点只从队列取任务,跑完再把结果交给清洗模块;存储节点负责写库。每个环节互相解耦,某一台机器挂了,其它节点继续消费队列,重启后补齐任务即可,不需要从头开始。

这套架构里最关键的假设是:裁判文书网的检索结果可以按年份、案由、地域等维度切成互不重叠的批次,每个批次独立入队、独立去重。只要任务切得足够细,节点数量从 2 加到 20 都只是改配置的事,代码不用动。

2.2 任务队列与消息中间件:用 Redis 当调度核心

任务队列的选型,我对比过 Redis、RabbitMQ 和 Kafka,最后在大多数项目里选了 Redis。原因很实际:爬虫任务的量级是几十万到几百万,单条任务就是一个 JSON 字符串,Redis 的 List 结构配合阻塞弹出就够用,不需要引入重型的消息系统。美中不足是 Redis 没有消费者确认机制,任务被取出后如果节点宕机会丢任务,所以要配合失败重试和落盘日志兜底。

三种方案的对比如下:

用 Redis 的LPUSH把任务推进去,消费者BRPOP阻塞弹出,谁空闲谁取,天然负载均衡。示例代码如下:

import redis import json r = redis.Redis(host="192.168.1.10", port=6379, db=0, password="yourpass") def push_task(channel, task_dict): # 把待爬的检索条件或详情页ID序列化成JSON,推入指定队列 task = json.dumps(task_dict, ensure_ascii=False) r.lpush(f"crawl:queue:{channel}", task)

这里host和port指向调度节点,db=0是 Redis 默认库,password按实际配置填,没有可以不填。ensure_ascii=False必须带上,否则中文案由会被转成\uXXXX形式,写入队列后肉眼排查困难。队列名带channel前缀是为了按年份、案由分开调度,不同优先级可以控制不同的消费速度。

消费者侧的核心逻辑如下:

def worker(queue_name): while True: task = r.brpop(f"crawl:queue:{queue_name}", timeout=5) if not task: continue data = json.loads(task[1]) # 这里调用抓取函数,失败时重新入队 try: crawl_one(data) except Exception: r.lpush(f"crawl:queue:{queue_name}", task[1])

brpop返回的是元组,第一个元素是队列名,第二个才是任务 JSON。timeout=5表示阻塞等待 5 秒,队列为空时返回None,避免死循环空转。失败重新入队有个隐患:如果目标地址永久失效,这条任务会在队列里反复横跳,所以重试前要判断失败次数,超过 3 次直接丢进crawl:dead:channel做人工复核。

部署上,调度节点和抓取节点分开跑,抓取节点可以开多台,每台启动一个 worker 进程:

# 在三台服务器上分别启动消费者 nohup python worker.py --queue detail --worker-id node01 > logs/node01.log 2>&1 & nohup python worker.py --queue detail --worker-id node02 > logs/node02.log 2>&1 &

nohup防止 SSH 断开后进程被杀,日志重定向到以节点命名的文件里方便排错。--worker-id写进日志和监控,出问题时能快速定位是哪台机器、哪个轮次的请求出了问题。

2.3 多节点协作与去重:基于 Redis Set 的指纹去重

分布式场景下最怕的是同一个文书被多个节点重复抓取,所以入队前必须做一次全局去重。我用的方案是 Redis Set:已经抓过的文书 ID 放进一个 Set,爬取前先查一下,存在就不再入队。

def is_dup(doc_id): # 用文书ID做指纹,已抓过的不再入队 return r.sismember("crawl:done:docid", doc_id) def mark_done(doc_id): r.sadd("crawl:done:docid", doc_id) def dispatch(urls): for doc_id, url in urls: if not is_dup(doc_id): push_task("detail", {"doc_id": doc_id, "url": url})

sismember是 O(1) 操作,百万级文书 ID 去重时 Redis 毫无压力。初版我看过一种误用:把整套 URL 字符串直接当 key 去重,一旦 URL 带着随机的_t时间戳参数,同一篇文书每次都生成新 URL,去重彻底失效。正确做法是用文书 ID 或案号做指纹,URL 只作为待抓地址。

内存占用也要心里有数:一百万条文书 ID,每个大约 20 到 30 字节,总共占用几十 MB,Redis 完全扛得住。如果未来做到千万级,可以换成 Bloom Filter,用少量误判换内存减半,但初期采集追求精确去重,Set 更合适。

还有一处细节:详情页抓取成功后我要先落原始 HTML 再解析,这样即使后面清洗逻辑写错了,也能重新解析而不必重新爬。原始文件按doc_id分目录存储,文件名就是文书 ID,好处是审计、重试、断点续爬都很直观。

3. 反反爬机制与请求调度:把访问频率控制得像真人

3.1 请求伪装的要害:UA 轮换、Referer 与 Session 保持

反爬判断的不只是 IP,还有请求头稳定性和行为规律。很多爬虫只改 User-Agent,其余请求头一尘不变,照样被识别。这套系统里的反反爬机制不是玄学,核心思路是降低请求特征的稳定性,让每次请求看起来来自不同的普通浏览器。

请求头构造我一般这样做:

import random from fake_useragent import UserAgent def build_headers(): ua = UserAgent() return { "User-Agent": ua.random, "Referer": "https://wenshu.court.gov.cn/", "Accept-Language": "zh-CN,zh;q=0.9", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Connection": "keep-alive", }

fake_useragent会从内置列表随机返回一个 UA,但注意这个库联网更新时可能变慢,我一般会先跑一次生成 100 条 UA 放到本地配置里,代码里随机取,不依赖外部接口。Referer统一指向站点首页,是为了避免被识别成跨站请求。Accept和Accept-Language要保持完整,只改 UA 不改其它头的请求,特征反而更明显。

登录态和搜索参数通常依赖 Cookie,这里用requests.Session保持连接池和 Cookie:

import requests session = requests.Session() session.headers.update(build_headers()) session.cookies.update({"cookie_key": "从浏览器复制的值"})

Session 会把响应里的 Cookie 自动更新到本地,后续请求不需要手动拼接。需要提醒的是,Cookie 有效期通常是几天到几周,过期后表现为搜索结果异常或返回登录页,排错时要先检查 Cookie 是否失效。

3.2 请求频率控制与重试:带抖动的限速器

固定间隔是反爬最容易抓的特征:每 2 秒一次,持续一整晚,规律明显得像计时器。正确的做法是给限速器加随机抖动,让间隔落在某个区间内。

import time import random class RateLimiter: def __init__(self, min_interval=1.0, max_interval=3.0): self.min_interval = min_interval self.max_interval = max_interval self.last_request = 0 def wait(self): # 随机抖动,避免出现机器式的固定间隔 delay = random.uniform(self.min_interval, self.max_interval) elapsed = time.time() - self.last_request if elapsed < delay: time.sleep(delay - elapsed) self.last_request = time.time() limiter = RateLimiter(min_interval=1.5, max_interval=4.0)

random.uniform的随机区间建议至少 2 到 3 秒,并且把实际间隔和期望间隔的偏差控制住:如果上次请求已经过了 6 秒(比如解析耗时较长),wait不会再额外等待,避免整体节奏越拉越慢。分布式节点越多,单节点间隔应该拉得越长,这是经验值,不是精确公式。

请求失败时的重试要采用指数退避,而不是固定间隔重试:

def fetch_with_retry(url, session, max_retries=3): for attempt in range(max_retries): try: resp = session.get(url, timeout=10) if resp.status_code == 200: return resp.text if resp.status_code in (403, 429): # 排查:触发访问限制,等待10秒 * 2^attempt 再重试 wait_time = 10 * (2 ** attempt) time.sleep(wait_time) except requests.RequestException as e: # 网络层异常,同样按指数退避 time.sleep(2 ** attempt) return None

403和429是典型的被限速信号,这时候加大并发只会让封禁更严重,正确反应是越退越慢。timeout=10指单次连接超过 10 秒就放弃,避免连接挂在半空占用线程。注意requests.RequestException只捕获网络层异常,解析层错误不应该在这里重试,否则可能把脏数据反复入库。

提示:反反爬机制设计的前提是合规采集。公开文书用于法律研究没有问题,但单个 IP 访问频率必须控制在合理范围内,不要对目标站点造成压力。

3.3 验证码与封禁处理:检测到异常先降速而不是硬刚

访问过于频繁时,站点会返回验证码页或安全验证页。这类页面通常有特征关键词,代码里做一次快速检测:

def detect_ban(text): # 常见关键词命中说明被引导到验证或拦截页 markers = ["验证码", "访问过于频繁", "安全验证", "verify"] return any(m in text for m in markers) def on_ban_detected(worker_id): # 通知调度端暂停该节点10分钟,而不是继续硬撞 r.set(f"crawl:pause:{worker_id}", "1", ex=600)

detect_ban的逻辑很简单但实用:页面正常时不会出现这些词,一旦出现就说明 IP 或账号被盯上了。on_ban_detected把暂停标志写进 Redis,消费者在取任务前先检查标志,命中则休息 10 分钟再继续。比起在本地time.sleep,把暂停状态写到 Redis 的好处是:即使这个节点重启,调度端也能读到状态,不会重启后立刻疯狂请求。

封禁处理的完整流程是:检测到拦截页 → 当前节点暂停 → 人工检查日志确认原因 → 降低全局限速参数 → 从断点继续。硬撞验证码没有任何意义,这套系统也不做任何破解动作,只做合规降速。

真实项目中还有一类隐蔽情况:不返回验证码,而是返回 200 但页面内容是被截断的空壳。我踩过一次,原因是请求头缺少某个字段导致服务端渲染不全,排查时先看页面长度,再看是否包含裁判文书正文特征词,比盯着状态码更可靠。

4. 数据清洗与文书详情页解析:从 HTML 到结构化字段

4.1 详情页字段抽取:lxml 配合正则的常规打法

详情页解析是整套系统的核心。页面结构在不同年份、不同法院间有差异,但字段基本固定:标题、案号、法院名称、裁判日期、案由、当事人、裁判结果、正文。我用的组合是lxml的 XPath 提取节点,再用正则做案号、日期的精修。

from lxml import etree import re def parse_detail(html): tree = etree.HTML(html) # 标题通常在版式调整后仍有稳定节点,取不到再走备用路径 title = tree.xpath("//h1//text()") or tree.xpath("//div[@class='title']//text()") full_text = "".join(tree.xpath("//div[@id='content']//text()")).strip() # 案号格式示例:（2023）京01民初123号 case_no = re.search(r"（\d{4}）[^0-9]{0,20}\d+号", full_text) # 法院名称几乎总在文书正文前200字内 court = re.search(r".+法院", full_text[:200]) return { "title": title[0].strip() if title else "", "case_no": case_no.group(0) if case_no else "", "court": court.group(0) if court else "", "full_text": full_text, }

XPath 取文本时最容易漏掉的是嵌套标签里的内容,所以每次取文本都要加//text()再用join拼回。正则匹配案号时,年份是四位数字,中间法院简称可能包含汉字和数字,最后以"号"结尾。这个正则只覆盖了最常见的格式,遇到"（2023）京01民初123号之一"这种带后缀的情况,需要在清洗阶段单独处理。

页面解析前,原始 HTML 会先落盘存储,我会在解析函数里直接传入落盘文件路径:

def save_html(doc_id, html): with open(f"data/raw/{doc_id}.html", "w", encoding="utf-8") as f: f.write(html) def parse_from_file(doc_id): with open(f"data/raw/{doc_id}.html", "r", encoding="utf-8") as f: return parse_detail(f.read())

这样做的原因是:解析逻辑写错时,不需要重新请求页面,直接改清洗脚本重新解析原始文件。这套"先落盘、后解析"的顺序,是我在实际项目里最后悔没有一开始就执行的方案。

4.2 pandas 数据清洗:去重、缺失值与字段标准化

原始解析结果通常是几千个字典,先汇总成 DataFrame,再用 pandas 做清洗。最容易出现的问题有三类:案号里的全半角括号不统一、日期格式五花八门、同一文书被多次抓取。

import pandas as pd df = pd.read_csv("raw_cases.csv") df["case_no"] = df["case_no"].str.replace(r"\s+", "", regex=True) df["case_no"] = df["case_no"].str.replace("（", "(", regex=False).str.replace("）", ")", regex=False) df = df.drop_duplicates(subset=["case_no"]) df["judge_date"] = pd.to_datetime(df["judge_date"], errors="coerce") df.dropna(subset=["case_no"], inplace=True)

第一步是去掉案号里所有空白字符,很多页面模板会在案号中间插入换行或空格。第二步把中文全角括号统一成半角,这一步直接影响去重效果:"（2023）京01民初123号"和"(2023)京01民初123号"如果不统一,会被 pandas 当成两条完全不同的数据。drop_duplicates只按case_no去重,保留第一条;如果你希望保留最新裁判日期那条,需要先排序再去重。

日期标准化用pd.to_datetime,errors="coerce"会把无法解析的值转成NaT,而不是抛异常中断整个任务。清洗后统计缺失值,案号为空的直接删除,裁判日期缺失的保留但标记,后续分析时按业务决定是否剔除。

常见脏数据和处理动作整理如下:

如果裁判日期缺失比例超过 20%,要回到解析脚本检查是不是某个页面模板的定位失效,而不是在清洗阶段强行填充。用均值或众数填充日期会直接污染后续的时间序列分析,这个坑我踩过。

4.3 存储方案:MySQL 还是 MongoDB

清洗后的结构化数据要落库,选型主要看下游怎么用。如果要做 SQL 查询、按案由聚合统计,MySQL 更顺手;如果字段经常变动、需要快速迭代,可以考虑 MongoDB。我用表格对比一下:

本系统默认走 MySQL,写入时用INSERT ... ON DUPLICATE KEY UPDATE实现幂等更新:

import pymysql conn = pymysql.connect( host="127.0.0.1", user="root", password="yourpass", database="legal", charset="utf8mb4" ) sql = """INSERT INTO cases (case_no, court, judge_date, full_text) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE full_text=VALUES(full_text)""" records = [(r["case_no"], r["court"], r["judge_date"], r["full_text"]) for r in cleaned] with conn.cursor() as cur: cur.executemany(sql, records) conn.commit()

ON DUPLICATE KEY UPDATE依赖case_no的唯一索引,第一次插入,后续重复时只更新正文,不产生重复记录。charset="utf8mb4"必须用,文书里有大量生僻字和特殊符号,utf8 会报错或存成乱码。executemany批量提交比逐条commit快一个数量级,我一般每攒 100 条就 commit 一次,防止意外断电丢太多数据。

存储层的坑主要集中在编码和字段长度:正文超过 6 万字时 TEXT 类型会截断,字段要提前建好为 MEDIUMTEXT;连接串、库表字符集、Python 脚本的编码三处必须保持一致,任何一处是 utf8 都可能出现乱码。

5. 避坑指南:裁判文书网爬取中常见的五个翻车现场

5.1 案例一:爬了一半全部超时

现象:运行两三个小时后,所有节点请求全部超时或直接返回 403,日志里全是连接错误,重启后依然如此。

原因:单节点请求频率虽然不高,但多个节点共用同一个出口 IP,聚合后的频率远超单机阈值,被识别为异常流量。

解决:把节点分散到多个出口 IP,或者在只有单 IP 的机器上只开两到三个 worker,同时把限速器的min_interval拉大到 3 秒以上。超时后走指数退避,不要立刻重试,冷却半小时再继续。

提示:分布式爬虫不等于多开几个进程。共用一个 IP 时,开 5 个节点等于把单机限速放大了 5 倍,这是最常见的误用。

5.2 案例二:详情页解析出来全是空字段

现象:抓回几千个 HTML 文件,但title、case_no解析出来全是空值,页面保存下来一看,正文长度正常。

原因:页面关键内容是 JS 异步加载的,requests直接请求拿到的是空壳 HTML;或者节点结构在不同案件类型下用了不同的 class 名,XPath 匹配失效。

解决:先检查原始 HTML 是否包含完整正文文本,如果只有几百字节,说明需要寻找真实的数据接口。常见做法是打开浏览器开发者工具,在 Network 面板里找返回 JSON 的链接,直接请求接口再解析。XPath 匹配不到时,优先使用文本特征定位节点,而不是依赖具体 class 名。

5.3 案例三:案号去重之后数据少了一半

现象:原始抓了 2 万条,drop_duplicates之后只剩 1 万条,怀疑是清洗规则误删。

原因:同一案件的一审、二审、再审文书可能共用一个案号前缀,或案号里带"之一""之二"这类后缀,去重时被当成重复数据;还有一种情况是全半角括号没统一,导致相同文书被拆成两条,数量虚高。

解决:清洗前先打印去重前后行数对比,抽查被删掉的样本。去重subset从单一case_no改成["case_no", "judge_date"]组合,同一案号不同裁判日期的文书保留。带后缀的案号在标准化阶段单独处理,不能直接删。

5.4 案例四:内存暴涨然后进程被系统杀掉

现象:跑了三天,内存占用从 800MB 涨到 6GB,然后 OOM 被杀,之前所有数据全丢。

原因:把详情页文本全部攒在内存列表里,打算最后统一入库,结果清洗和入库速度跟不上抓取速度,内存持续累积。

解决:改成边爬边写,每攒 100 条就 commit 一次;原始 HTML 先落盘,内存里只保留解析后的字典;如果入库节点性能差,中间用 SQLite 或 CSV 做落盘缓冲,而不是让抓取进程扛全部数据。进程崩溃不可怕,可怕的是崩溃前没落盘的东西全没了。

5.5 案例五:入库后全文乱码和截断

现象:MySQL 里查询文书,生僻字显示成??,长文书正文写到一半被截断。

原因:连接字符集不是utf8mb4,或者字段类型用了 TEXT,单个字段只能存 65535 字节,文书正文超过这个长度就被截断。

解决:统一三处字符集:Python 脚本顶部声明# -*- coding: utf-8 -*-,MySQL 连接参数写charset="utf8mb4",数据库建表语句里表字符集和字段字符集都指定utf8mb4。正文和裁判理由这类大字段用 MEDIUMTEXT,不要用 TEXT。

6. 数据质量验证与增量更新:采集前先过一遍这道校验关卡

6.1 四步校验脚本与增量爬取的思路

数据入库只是第一步,我更愿意说"跑完之后你信不信这批数据"才是关键。每次采集完成后,我会强制跑一遍校验脚本,四个数字过目确认:总数、缺案号数、重复数、长文书占比。脚本很短,但救过我很多次:

def validate(df): total = len(df) no_case = df["case_no"].isna().sum() dup = df["case_no"].duplicated().sum() complete = (df["full_text"].str.len() > 500).mean() print(f"总数: {total}, 缺案号: {no_case}, 重复: {dup}, 长文书占比: {complete:.2%}") return dup == 0 and no_case == 0

长文书占比是个很敏锐的信号:正常情况下绝大多数裁判文书正文超过 500 字,如果这个比例掉到 60% 以下,说明解析阶段大概率出了错,而不是真实数据如此。我经历过一次清洗规则写得太激进,把含空格和异体字的合法案号当成脏数据删了,等到下游分析做完才发现数据少了将近一半,只能从头补采,那个浪费的时间至今还记得。

增量更新的逻辑比全量简单很多:记录每次采集的最大裁判日期,下一次只抓取这个日期之后的数据,入队前再按案号去重一次。考虑到文书上网有滞后,增量窗口会往前多留一个月缓冲,避免漏掉延迟发布的文书。运行方式是把调度命令参数化:

python dispatcher.py --start 2024-01-01 --end 2024-03-31 --channel by_date

--start和--end决定批次的时间窗口,窗口越小任务越细,越容易分布到多个节点并行。增量模式下,旧文书不会再被抓一遍,Redis 里的去重 Set 会不断累积,偶尔要检查内存增长情况。

从那以后,我每次启动采集前都强制跑一遍校验四件套:总数对不对、缺案号多不多、去重后数量是否符合预期、抽样看几篇文书正文。验证通过才开始增量入库,入库后还要对日期边界做二次核对。这套系统最大的价值,不是帮你把爬虫代码跑起来,而是让你在跑起来之后还能知道数据到底哪里出了问题。希望帮到你。

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