Python 实战：构建生态环境损害公开赔偿磋商与修复公告库——从公告采集到企业统一社会信用代码消歧

㊗️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～ ㊙️本期爬虫难度指数：⭐⭐⭐⭐⭐（专家级） 🉐福利： 一次订阅后，专栏内的所有文章可永久免费看，持续更新中，保底1000+(篇)硬核实战内容。

全文目录：

• 🌟 开篇语
• 0️⃣ 前言（Preface）
• 1️⃣ 摘要（Abstract）
• 2️⃣ 背景与需求（Why）
• • 2.1 为什么值得做成数据库
  • 2.2 本文目标字段
  • 2.3 为什么额外保留金额字段
• 3️⃣ 合规与注意事项
• • 3.1 robots.txt
  • 3.2 请求频率
  • 3.3 不采集非业务所需敏感信息
  • 3.4 不绕过登录、付费或访问控制
• 4️⃣ 技术选型与整体流程（What / How）
• • 4.1 静态、动态还是 API？
  • 4.2 总流程
  • 4.3 为什么不用 BeautifulSoup 作为主解析器
  • 4.4 为什么暂时不用 Scrapy
• 5️⃣ 环境准备与依赖安装
• • 5.1 Python
  • 5.2 创建虚拟环境
  • 5.3 requirements.txt
  • 5.4 项目结构
• 6️⃣ 核心实现：请求层（Fetcher）
• • 6.1 config.py
  • 6.2 headers 为什么要有 UA
  • 6.3 Referer
  • 6.4 timeout 必须设置
  • 6.5 Session
  • 6.6 为什么同时使用 Retry 和随机 sleep
  • 6.7 关于 Cookie
• 7️⃣ 核心实现：解析层（Parser）
• • 7.1 models.py
  • 7.2 列表页怎么拿详情链接
  • 7.3 保存列表快照
• 7.4 详情页解析
• • 7.5 文本标准化函数
  • 7.6 Table → Dict
  • 7.7 字段名称不是永远一致
• 7.8 发布时间
• 7.9 完整详情解析器
• 7.10 缺失字段怎么办
• 7.11 损害类型提取
• 7.12 企业名称提取
• 7.13 匿名主体识别
• 7.14 修复费用提取
• • 7.15 根据语义抽 restoration_cost
• 7.16 状态归一化
• 7.17 case_id 怎么生成
• 7.18 content_hash
• 8️⃣ 数据存储与导出（Storage）
• 8.1 schema.sql
• 8.2 storage.py
• 8.3 为什么 CSV 用 utf-8-sig
• 8.4 字段映射表
• 8.5 去重策略
• 8.6 企业主数据
• 8.7 企业名称标准化
• 8.8 统一社会信用代码合法性校验
• 8.9 RapidFuzz 企业消歧
• • 8.10 为什么先精确匹配再模糊匹配
• 8.11 entity_matcher.py
• 8.12 为什么阈值推荐 96 / 85
• 8.13 更稳妥的二级消歧
• 8.14 cleaner.py 完整版
• 9️⃣ 运行方式与结果展示
• • 9.1 main.py
• 9.2 启动方式
• 9.3 示例输出
• 🔟 常见问题与排错
• • 10.1 403
• 10.2 429 Too Many Requests
• 10.3 HTML 抓到空壳
• 10.4 XPath 突然失效
• 10.5 网站字段名称改变
• 10.6 中文乱码
• 10.7 金额解析不准
• 10.8 为什么抓到了企业，却匹配错了
• 10.9 “某公司”怎么办
• 1️⃣1️⃣ 进阶优化
• 11.1 断点续跑
• 11.2 增量窗口
• 11.3 并发
• 11.4 asyncio
• 11.5 Scrapy
• 11.6 Playwright
• 11.7 日志
• 11.8 数据质量报告
• 11.9 review 企业人工审核队列
• 11.10 Alias 优先级
• 11.11 SQLite 升级 MySQL
• 11.12 定时任务
• 11.13 Airflow
• 11.14 原始层与清洗层分离
• • • ODS
    • DWD
    • ADS
• 11.15 Raw HTML 存档
• 11.16 增加 source 字段
• 11.17 企业主体多值问题
• • cases
  • companies
  • case_companies
• 11.18 状态历史
• 11.19 单元测试
• 11.20 可观察性
• 1️⃣2️⃣ 总结与延伸阅读
• • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

哈喽，各位小伙伴们你们好呀～我是【喵手】。 运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO 欢迎大家常来逛逛，一起学习，一起进步～🌟

  我长期专注 Python 爬虫工程化实战，主理专栏👉 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。

  📌 专栏食用指南（建议收藏）

• ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
• ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
• ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
• ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅专栏👉《Python爬虫实战》👈，一次订阅后，专栏内的所有文章可永久免费阅读，持续更新中。    💕订阅后更新会优先推送，按目录学习更高效💯～

0️⃣ 前言（Preface）

这篇文章做一件比较具体的事情：使用 Python + requests + lxml + SQLite + RapidFuzz，采集公开发布的生态环境损害赔偿、磋商和修复类公告，将网页里的非结构化信息整理成统一的数据集，并进一步解决企业主体名称不统一、简称混用、括号差异等问题，最终形成一套可以持续增量更新的生态环境损害赔偿案例库。

最终重点保留五个业务字段：

case_id
damage_type
company_involved
restoration_cost
status

同时，为了保证数据真正可追溯，我还会增加：

company_normalized
unified_social_credit_code
match_score
publish_date
source_url
content_hash
created_at

这些字段不会改变原始需求，反而能解决真实项目里最容易遇到的问题：一条记录到底从哪里来的、同一家企业为什么出现三个名字、信用代码是不是可靠匹配出来的、公告修改以后怎么知道数据发生了变化。

本文选择昌吉州生态环境局公开“通知公告”栏目作为可运行示例来源。该栏目目前确实存在生态环境损害赔偿案件基本信息公示；例如 2026 年 5 月发布的案件详情页中，可以直接读取案件名称、损害调查结果、鉴定评估结论、修复方案以及最终处理结果等信息。

读完本文，你应该能得到三样东西：

我尤其想强调第三点。

很多爬虫项目真正困难的地方，并不是“网页能不能下载下来”，而是：

网页里的名称
↓
能不能变成数据库中唯一、稳定、可关联的实体

比如：

新疆天康原种猪育种有限公司
新疆天康原种猪育种有限公司（昌吉）
天康原种猪公司
新疆天康原种猪育种有限责任公司

如果不做主体消歧，数据库看起来有四家公司。

实际上，它们可能是在描述同一个主体，也可能并不是。

所以这一篇不会把“企业名称清洗”理解成简单的 .replace("有限公司", "")，而会采用：

原始名称
↓
标准化
↓
完全匹配
↓
别名匹配
↓
模糊候选匹配
↓
信用代码合法性校验
↓
置信度判断
↓
自动确认 / 人工复核 / unresolved

这才是我认为更接近生产环境的数据工程思路。

1️⃣ 摘要（Abstract）

本文以公开生态环境损害赔偿公告为数据源，通过 requests 负责 HTTP 请求、lxml 负责 HTML 解析、正则表达式承担中文公告中的半结构化字段提取、SQLite 负责持久化，再使用 RapidFuzz 对企业主体名称进行候选匹配，最终建立可追溯、可增量更新的生态环境损害赔偿公告数据库。

文章中的示范站点采用：

https://www.cj.gov.cn/p122/tzgg.html

即昌吉州生态环境局公开通知公告栏目。

公开页面中确实存在较为规整的生态环境损害赔偿案件信息。例如某 2026 年案例明确提供：

案件名称
损害调查结果
损害鉴定评估结论
损害赔偿协议主要条款
赔偿资金使用情况
损害修复方案简要
生态环境修复效果
诉讼裁判文书
案件最终处理结果

这样的页面结构，非常适合拿来说明“政府公告 → 结构化数据库”的完整过程。

与此同时，公开案例并不保证每一次都完整披露企业实名。

生态环境部公开的典型案例中就经常出现“某公司”等匿名表达，所以企业统一社会信用代码对齐必须接受一个现实：

没有实名企业，就不应该为了提高匹配率而猜测企业。

生态环境部目前仍设有“生态环境损害赔偿制度”公开栏目，并持续发布相关制度文件和典型案例。

因此本文在实体匹配环节设置三类结果：

matched
review
unresolved

而不是简单输出一个看起来很完整、实际上可能错误的信用代码。

完成本文后，可以掌握：

• 政府公告列表页与详情页采集；
• 中文半结构化文本字段抽取；
• 金额标准化；
• 污染类型分类；
• 案件状态归一化；
• 企业名称清洗；
• 统一社会信用代码校验；
• RapidFuzz 主体消歧；
• SQLite 去重和增量写入；
• 失败重试、限速、断点续跑和日志设计。

2️⃣ 背景与需求（Why）

2.1 为什么值得做成数据库

生态环境损害赔偿相关公开信息有一个非常典型的特点：

信息是公开的，但数据并不天然结构化。

人打开网页，可以很容易看懂：

某企业排放了什么
损害了什么环境要素
损害金额是多少
是否已经修复
案件最后怎么处理

但计算机并不知道。

网页上很可能只有：

<p>损害鉴定评估结论：通过鉴定评估……</p>

或者：

<tr>
<td>案件最终处理结果</td>
<td>某企业已履行赔偿义务</td>
</tr>

如果要进一步做：

年度趋势分析
污染类型统计
修复成本分布
企业案例关联
地区比较
案件履约状态统计

就必须先把它转换成稳定的行列结构。

从工程上看：

公开网页 ≠ 数据库

真正需要构建的是：

公开网页
↓
原始HTML
↓
公告正文
↓
字段抽取
↓
字段标准化
↓
实体消歧
↓
结构化数据

2.2 本文目标字段

用户要求的核心字段为：

不过，在真实项目中只保存这五列并不够。

所以数据库内部推荐使用：

case_id
case_name
damage_type
company_involved
company_normalized
unified_social_credit_code
match_score
match_status
restoration_cost
assessment_cost
service_loss
total_compensation
status
publish_date
source_url
content_hash
raw_text
created_at
updated_at

最终如果业务系统只要五列，再进行投影即可：

SELECT
case_id,
damage_type,
company_involved,
restoration_cost,
status
FROM eco_damage_cases;

2.3 为什么额外保留金额字段

这是一个非常容易踩坑的地方。

例如昌吉州公开的一起案例里，鉴定评估信息同时包含：

生态环境服务功能损失
调查鉴定评估费用
清除污染、修复生态环境费用

而且三者并不是同一个概念。

某公开案例中，大气生态环境服务功能损失为 8294300 元，鉴定评估费用为 300000 元，但“清除污染、修复生态环境费用”为 0 元。

如果我们见到：

8294300元

就直接写：

restoration_cost = 8294300

数据语义实际上已经错了。

所以代码必须优先寻找：

清除污染、修复生态环境费用
修复生态环境费用
生态修复费用
修复费用

只有业务定义明确允许时，才把总赔偿金额作为 restoration_cost 的替代值。

本文默认：

restoration_cost = 真正用于清除污染或生态修复的金额

不拿服务功能损失冒充修复费用。

3️⃣ 合规与注意事项

写爬虫之前，我更愿意先把边界说清楚。

因为“代码能请求到”与“应该怎么采集”从来不是一回事。

3.1 robots.txt

启动采集前，应该检查目标网站的：

/robots.txt

了解网站对于不同路径和 User-Agent 的公开抓取约束。

示例：

import requests

url = "https://example.gov.cn/robots.txt"

resp = requests.get(url, timeout=10)

if resp.ok:
print(resp.text)
else:
print("robots.txt unavailable:", resp.status_code)

robots.txt 不是网站全部规则的替代品，但它应该成为采集程序启动前的基本检查项之一。

如果网站无法正常返回 robots.txt，也不代表可以无限制采集。

更合理的处理是：

降低频率
限定公开页面
不访问后台路径
不尝试绕过技术限制

3.2 请求频率

本文示例默认：

REQUEST_INTERVAL = (1.2, 2.5)

即两个详情请求之间随机等待约 1.2～2.5 秒。

政府公开网站的数据量通常没有大到需要：

100线程
500协程
1000并发

那样做收益很小，风险却明显增加。

正确的思路应该是：

增量抓取 + 本地缓存 + 去重

而不是：

每次运行都重新把全站轰一遍

3.3 不采集非业务所需敏感信息

本文只关注公开公告里的：

案件
企业主体
损害类型
修复费用
案件状态

正文里如果出现：

个人电话
详细个人住址
身份证信息
私人联系方式

均不进入业务数据表。

即使页面公开展示，也不意味着我们的数据库就必须复制保存。

原则很简单：

数据最小化。

3.4 不绕过登录、付费或访问控制

如果目标信息必须：

登录
验证码
专用账号
付费权限
授权接口

才能获得，那么本文程序不尝试绕过这些限制。

对于企业统一社会信用代码，也是一样。

本文不编写针对商业企业信息平台的验证码破解、登录绕过或访问控制规避逻辑。

企业主体对齐推荐使用：

依法取得的企业主数据
公开授权数据集
企业内部已有主体库
人工整理的企业基表

然后在本地完成实体匹配。

4️⃣ 技术选型与整体流程（What / How）

4.1 静态、动态还是 API？

本文示范案例属于：

服务端输出 HTML 的公告型网页。

因此首选：

requests + lxml

而不是一上来使用 Playwright。

昌吉州生态环境局的公开页面经过普通 HTTP 请求即可获得公告正文结构，现有公开搜索结果同样能够解析出列表及详情文本。

4.2 总流程

完整流程可以概括成：

┌─────────────────┐
│ 列表页 URL │
└────────┬────────┘
│
▼
┌─────────────────┐
│ Fetcher │
│ requests/session│
└────────┬────────┘
│
▼
┌─────────────────┐
│ 列表页 Parser │
│ 提取详情页链接 │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 详情页请求 │
└────────┬────────┘
│
▼
┌─────────────────┐
│ 详情页 Parser │
│ 表格/正文 │
└────────┬────────┘
│
▼
┌─────────────────┐
│ Cleaner │
│ 金额/状态/类型 │
└────────┬────────┘
│
▼
┌─────────────────┐
│ Entity Matcher │
│ 企业主体消歧 │
└────────┬────────┘
│
▼
┌─────────────────┐
│ SQLite │
│ 去重/增量/导出 │
└─────────────────┘

对应四个核心词就是：

采集 → 解析 → 清洗 → 存储

但工程上实际上多了一层：

采集 → 解析 → 清洗 → 实体对齐 → 存储

4.3 为什么不用 BeautifulSoup 作为主解析器

BeautifulSoup 很好用。

不过这个项目大量涉及：

父节点
兄弟节点
表格
文本归一化
XPath contains

所以我更倾向：

lxml.html

比如：

tree.xpath("//table//tr")

以及：

tree.xpath("//a[contains(normalize-space(.), '生态环境损害赔偿')]")

会很自然。

4.4 为什么暂时不用 Scrapy

如果只是：

一个栏目
数百条公告
每天增量几条

requests + SQLite 足够。

当规模扩大到：

几十个生态环境部门
几万个页面
跨省采集
每日自动更新

再迁移到 Scrapy 会更合理。

工程不是工具越重越专业。

合适才专业。

5️⃣ 环境准备与依赖安装

5.1 Python

推荐：

Python 3.11+

本文示例以：

Python 3.12

为参考。

检查：

python –version

5.2 创建虚拟环境

Windows：

python -m venv .venv
.venv\\Scripts\\activate

macOS / Linux：

python3 -m venv .venv
source .venv/bin/activate

5.3 requirements.txt

requests>=2.32
lxml>=5.2
rapidfuzz>=3.9
tenacity>=8.3
loguru>=0.7
tqdm>=4.66

安装：

pip install -r requirements.txt

5.4 项目结构

推荐：

eco_damage_spider/
│
├── main.py
├── config.py
├── fetcher.py
├── parser.py
├── cleaner.py
├── entity_matcher.py
├── storage.py
├── models.py
├── utils.py
│
├── data/
│ ├── raw/
│ ├── master/
│ │ └── companies.csv
│ └── output/
│
├── logs/
│
├── sql/
│ └── schema.sql
│
├── requirements.txt
└── README.md

创建目录：

mkdir eco_damage_spider
cd eco_damage_spider

mkdir data
mkdir logs
mkdir sql

6️⃣ 核心实现：请求层（Fetcher）

一个长期运行的采集项目，我不建议在所有文件里到处写：

requests.get(...)

这样以后：

改 UA
改 timeout
加重试
加代理
统计请求次数

都会变得很难。

因此统一封装 Fetcher。

6.1 config.py

from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent

DATA_DIR = BASE_DIR / "data"
RAW_DIR = DATA_DIR / "raw"
OUTPUT_DIR = DATA_DIR / "output"
MASTER_DIR = DATA_DIR / "master"
LOG_DIR = BASE_DIR / "logs"

for directory in [
DATA_DIR,
RAW_DIR,
OUTPUT_DIR,
MASTER_DIR,
LOG_DIR,
]:
directory.mkdir(parents=True, exist_ok=True)

BASE_URL = "https://www.cj.gov.cn"

LIST_URL = "https://www.cj.gov.cn/p122/tzgg.html"

REQUEST_TIMEOUT = 15

MIN_DELAY = 1.2
MAX_DELAY = 2.5

MAX_RETRIES = 3

USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36"
)

HEADERS = {
"User-Agent": USER_AGENT,
"Accept": (
"text/html,application/xhtml+xml,"
"application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8"
),
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
}

6.2 headers 为什么要有 UA

不少站点对于空 User-Agent 或明显异常客户端会做基本限制。

我们只是模拟正常浏览器的基础请求头：

HEADERS = {
"User-Agent": "…",
}

不要每次请求随机生成几十种奇怪 UA。

稳定、透明、低频采集更容易维护。

6.3 Referer

详情页请求可以带列表页作为 Referer：

headers = HEADERS.copy()

headers["Referer"] = LIST_URL

这更接近正常浏览行为。

6.4 timeout 必须设置

绝对不要：

requests.get(url)

然后无限等待。

推荐：

requests.get(
url,
timeout=15,
)

或者区分连接与读取：

timeout=(5, 20)

即：

连接超时 5 秒
读取超时 20 秒

6.5 Session

完整的 fetcher.py：

import random
import time
from pathlib import Path

import requests
from loguru import logger
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from config import (
HEADERS,
MIN_DELAY,
MAX_DELAY,
REQUEST_TIMEOUT,
RAW_DIR,
)

class Fetcher:

def __init__(self):
self.session = requests.Session()

self.session.headers.update(HEADERS)

retry = Retry(
total=3,
connect=3,
read=3,
status=3,
backoff_factor=1.0,
status_forcelist=[
429,
500,
502,
503,
504,
],
allowed_methods=["GET"],
respect_retry_after_header=True,
)

adapter = HTTPAdapter(
max_retries=retry,
pool_connections=5,
pool_maxsize=5,
)

self.session.mount("http://", adapter)
self.session.mount("https://", adapter)

def sleep(self):
delay = random.uniform(
MIN_DELAY,
MAX_DELAY,
)

time.sleep(delay)

def get(
self,
url: str,
referer: str | None = None,
) –> str:

headers = {}

if referer:
headers["Referer"] = referer

logger.info(
"GET {}",
url,
)

response = self.session.get(
url,
headers=headers,
timeout=REQUEST_TIMEOUT,
)

response.raise_for_status()

if not response.encoding:
response.encoding = response.apparent_encoding

html = response.text

self.sleep()

return html

def save_raw(
self,
html: str,
filename: str,
) –> Path:

path = RAW_DIR / filename

path.write_text(
html,
encoding="utf-8",
)

return path

def close(self):
self.session.close()

6.6 为什么同时使用 Retry 和随机 sleep

两者解决的不是同一个问题。

Retry 是：

请求失败以后怎么办

随机 sleep 是：

正常情况下请求之间隔多久

例如：

请求 A
↓
sleep 1.8 秒
↓
请求 B
↓
502
↓
backoff
↓
Retry

两套机制最好都保留。

6.7 关于 Cookie

当前示范的公开公告页面不需要登录 Cookie。

所以：

requests.Session()

主要用于连接复用。

如果公开站点后续返回必要的普通会话 Cookie，Session 也会自动保存：

session.cookies

但不要人为构造绕过访问限制的 Cookie。

7️⃣ 核心实现：解析层（Parser）

这是整个项目最值得花时间的地方。

因为公告网站经常会发生：

模板升级
编辑器变化
table → div
<p> → span
增加附件
标题略有变化

如果选择器写得太死：

//*[@id="content"]/div[2]/table/tr[3]/td[2]

页面稍微调整就挂。

更稳妥的方法是：

先把页面转换成“标签名 → 内容”的半结构化字典，再做业务字段解析。

7.1 models.py

from dataclasses import dataclass
from typing import Optional

@dataclass
class RawCase:

title: str

case_name: Optional[str] = None

damage_investigation: Optional[str] = None

assessment_conclusion: Optional[str] = None

agreement: Optional[str] = None

fund_usage: Optional[str] = None

restoration_plan: Optional[str] = None

restoration_effect: Optional[str] = None

final_result: Optional[str] = None

publish_date: Optional[str] = None

source_url: Optional[str] = None

raw_text: Optional[str] = None

清洗后的：

@dataclass
class CleanCase:

case_id: str

case_name: str

damage_type: str

company_involved: Optional[str]

restoration_cost: Optional[float]

status: str

publish_date: Optional[str]

source_url: str

content_hash: str

company_normalized: Optional[str] = None

unified_social_credit_code: Optional[str] = None

match_score: Optional[float] = None

match_status: str = "unresolved"

7.2 列表页怎么拿详情链接

不要假定一定是：

<ul class="news-list">

我们直接扫描 <a>：

from urllib.parse import urljoin

from lxml import html

KEYWORDS = (
"生态环境损害赔偿",
"损害赔偿案件",
"赔偿磋商",
)

def parse_list_page(
html_text: str,
base_url: str,
):

tree = html.fromstring(html_text)

result = []

for a in tree.xpath("//a[@href]"):

text = " ".join(
a.xpath(".//text()")
)

text = " ".join(
text.split()
)

if not text:
continue

if not any(
keyword in text
for keyword in KEYWORDS
):
continue

href = a.get("href")

if not href:
continue

url = urljoin(
base_url,
href,
)

result.append(
{
"title": text,
"url": url,
}
)

unique = {}

for item in result:
unique[item["url"]] = item

return list(unique.values())

这里最重要的一句反而不是 XPath：

if not any(keyword in text for keyword in KEYWORDS):
continue

因为通知公告里还有：

环评审批
企业名单
危险废物
通知
政策

我们不希望全部抓回来。

7.3 保存列表快照

实际项目最好：

list_html = fetcher.get(LIST_URL)

fetcher.save_raw(
list_html,
"list_latest.html",
)

为什么？

因为某天解析突然变成：

0 条数据

你首先要判断：

网页变了？
网络失败？
还是 Parser 写错了？

本地保存原始 HTML 后定位会快很多。

7.4 详情页解析

详情页的核心思想：

先从 table 中解析键值对
↓
失败后再从全文文本兜底

7.5 文本标准化函数

import re

def clean_text(text: str | None) –> str:

if not text:
return ""

text = text.replace("\\xa0", " ")

text = re.sub(
r"\\s+",
" ",
text,
)

return text.strip()

7.6 Table → Dict

from lxml import html

def extract_table_fields(
html_text: str,
) –> dict[str, str]:

tree = html.fromstring(html_text)

fields = {}

for tr in tree.xpath("//tr"):

cells = tr.xpath("./th|./td")

if len(cells) < 2:
continue

key = clean_text(
" ".join(
cells[0].xpath(".//text()")
)
)

value = clean_text(
" ".join(
cells[1].xpath(".//text()")
)
)

if key and value:
fields[key] = value

return fields

例如可以得到：

{
"年份": "2026年",
"案件名称": "…生态环境损害赔偿案",
"损害调查结果": "…",
"损害鉴定评估结论": "…",
"损害赔偿协议主要条款": "…",
"赔偿资金使用情况": "无",
"损害修复方案简要": "…",
"生态环境修复效果（修复评估报告结论）": "无",
"案件最终处理结果": "终止索赔程序"
}

这种形式与昌吉州目前公开案例展示的字段结构相吻合。

7.7 字段名称不是永远一致

可能出现：

年份
年 份

案件最终处理结果
最终处理结果

生态环境修复效果
生态环境修复效果（修复评估报告结论）

所以不要：

fields["案件最终处理结果"]

写死。

定义别名字典。

FIELD_ALIASES = {

"case_name": [
"案件名称",
"案例名称",
],

"damage_investigation": [
"损害调查结果",
"调查结果",
],

"assessment_conclusion": [
"损害鉴定评估结论",
"鉴定评估结论",
],

"agreement": [
"损害赔偿协议主要条款",
"赔偿协议主要条款",
],

"fund_usage": [
"赔偿资金使用情况",
],

"restoration_plan": [
"损害修复方案简要",
"生态修复方案",
"修复方案简要",
],

"restoration_effect": [
"生态环境修复效果（修复评估报告结论）",
"生态环境修复效果",
"修复效果",
],

"final_result": [
"案件最终处理结果",
"最终处理结果",
],
}

取字段：

def get_alias_value(
fields: dict,
aliases: list[str],
) –> str | None:

normalized = {
clean_text(k).replace(" ", ""): v
for k, v in fields.items()
}

for alias in aliases:

key = alias.replace(" ", "")

if key in normalized:
return normalized[key]

return None

7.8 发布时间

发布时间通常在正文顶部。

可以从全文中提取：

DATE_PATTERN = re.compile(
r"发布日期[：:\\s]*"
r"(\\d{4}-\\d{1,2}-\\d{1,2})"
)

def extract_publish_date(text: str):

match = DATE_PATTERN.search(text)

if match:
return match.group(1)

return None

兼容：

2026-05-26 18:25:29

只保存：

2026-05-26

7.9 完整详情解析器

parser.py：

import re

from lxml import html

from models import RawCase

FIELD_ALIASES = {

"case_name": [
"案件名称",
"案例名称",
],

"damage_investigation": [
"损害调查结果",
"调查结果",
],

"assessment_conclusion": [
"损害鉴定评估结论",
"鉴定评估结论",
],

"agreement": [
"损害赔偿协议主要条款",
"赔偿协议主要条款",
],

"fund_usage": [
"赔偿资金使用情况",
],

"restoration_plan": [
"损害修复方案简要",
"生态修复方案",
"修复方案简要",
],

"restoration_effect": [
"生态环境修复效果（修复评估报告结论）",
"生态环境修复效果",
"修复效果",
],

"final_result": [
"案件最终处理结果",
"最终处理结果",
],
}

def clean_text(text):

if not text:
return ""

text = text.replace(
"\\xa0",
" ",
)

text = re.sub(
r"\\s+",
" ",
text,
)

return text.strip()

def node_text(node):

return clean_text(
" ".join(
node.xpath(".//text()")
)
)

def extract_table_fields(
tree,
):

fields = {}

for tr in tree.xpath("//tr"):

cells = tr.xpath("./th|./td")

if len(cells) < 2:
continue

key = node_text(cells[0])

values = [
node_text(cell)
for cell in cells[1:]
]

value = " ".join(
value
for value in values
if value
)

if key and value:
fields[key] = value

return fields

def normalize_key(key):

return (
clean_text(key)
.replace(" ", "")
.replace("：", "")
.replace(":", "")
)

def get_alias_value(
fields,
aliases,
):

normalized_fields = {
normalize_key(k): v
for k, v in fields.items()
}

for alias in aliases:

key = normalize_key(alias)

if key in normalized_fields:
return normalized_fields[key]

return None

def parse_publish_date(
full_text,
):

patterns = [
r"发布日期[：:\\s]*(\\d{4}-\\d{2}-\\d{2})",
r"发布时间[：:\\s]*(\\d{4}-\\d{2}-\\d{2})",
r"(\\d{4}-\\d{2}-\\d{2})",
]

for pattern in patterns:

match = re.search(
pattern,
full_text,
)

if match:
return match.group(1)

return None

def parse_detail(
html_text,
source_url,
):

tree = html.fromstring(
html_text
)

title_candidates = tree.xpath(
"//h1//text() | "
"//h2//text() | "
"//title//text()"
)

title = ""

for candidate in title_candidates:

candidate = clean_text(candidate)

if (
candidate
and "生态环境损害赔偿"
in candidate
):
title = candidate
break

if not title:

title = clean_text(
" ".join(
tree.xpath(
"//title//text()"
)
)
)

fields = extract_table_fields(
tree
)

full_text = clean_text(
" ".join(
tree.xpath(
"//body//text()"
)
)
)

return RawCase(

title=title,

case_name=get_alias_value(
fields,
FIELD_ALIASES[
"case_name"
],
),

damage_investigation=get_alias_value(
fields,
FIELD_ALIASES[
"damage_investigation"
],
),

assessment_conclusion=get_alias_value(
fields,
FIELD_ALIASES[
"assessment_conclusion"
],
),

agreement=get_alias_value(
fields,
FIELD_ALIASES[
"agreement"
],
),

fund_usage=get_alias_value(
fields,
FIELD_ALIASES[
"fund_usage"
],
),

restoration_plan=get_alias_value(
fields,
FIELD_ALIASES[
"restoration_plan"
],
),

restoration_effect=get_alias_value(
fields,
FIELD_ALIASES[
"restoration_effect"
],
),

final_result=get_alias_value(
fields,
FIELD_ALIASES[
"final_result"
],
),

publish_date=parse_publish_date(
full_text
),

source_url=source_url,

raw_text=full_text,
)

7.10 缺失字段怎么办

爬虫解析最大的错误之一就是认为：

字段不存在 = 程序出错

其实并不是。

例如某些案件：

无需修复

或者：

不需签订协议

或者：

诉讼裁判文书：无

都属于业务事实。

昌吉州 2026 年公开的一起案件最终处理结果即为“终止索赔程序”，损害修复方案也明确无需进一步开展损害修复。

因此：

restoration_cost = None

和：

restoration_cost = 0

语义完全不同。

None 表示：

没找到。

0 表示：

找到了明确的零元。

一定不要混。

7.11 损害类型提取

我们定义：

water
air
soil
groundwater
solid_waste
ecological
mining
multiple
unknown

中文输出：

DAMAGE_LABELS = {
"water": "水环境",
"air": "大气环境",
"soil": "土壤环境",
"groundwater": "地下水",
"solid_waste": "固体废物",
"mining": "矿山生态",
"ecological": "生态破坏",
"multiple": "复合型",
"unknown": "未分类",
}

规则：

DAMAGE_RULES = {

"water": [
"废水",
"水污染",
"河流",
"水体",
"地表水",
],

"air": [
"废气",
"大气",
"粉尘",
"扬尘",
"烟气",
],

"soil": [
"土壤",
"污染土",
],

"groundwater": [
"地下水",
"渗坑",
],

"solid_waste": [
"固体废物",
"危险废物",
"危废",
"废渣",
"垃圾",
],

"mining": [
"采矿",
"矿山",
"矿坑",
],

"ecological": [
"植被",
"林地",
"湿地",
"草地",
"生态破坏",
],
}

实现：

def detect_damage_type(
text: str,
) –> str:

scores = {}

for damage_type, keywords \\
in DAMAGE_RULES.items():

score = sum(
text.count(keyword)
for keyword in keywords
)

if score:
scores[
damage_type
] = score

if not scores:
return "未分类"

ranked = sorted(
scores.items(),
key=lambda x: x[1],
reverse=True,
)

if len(ranked) >= 2:

first = ranked[0]
second = ranked[1]

if second[1] >= first[1] * 0.8:
return "复合型"

key = ranked[0][0]

return DAMAGE_LABELS[key]

这不是机器学习。

但对于公开公告库的第一版，规则分类往往比“为了高级而高级”的模型更容易解释。

7.12 企业名称提取

案件名称通常是非常有价值的信息源。

例如：

新疆天康原种猪育种有限公司排放水污染物生态环境损害赔偿案

直接提取有限公司部分：

COMPANY_PATTERN = re.compile(
r"([\\u4e00-\\u9fa5A-Za-z0-9（）()·\\-]+?"
r"(?:有限责任公司|股份有限公司|有限公司|"
r"集团有限公司|专业合作社))"
)

实现：

def extract_company(
text: str,
):

if not text:
return None

matches = COMPANY_PATTERN.findall(
text
)

if not matches:
return None

candidates = []

for company in matches:

company = clean_text(company)

if len(company) < 4:
continue

candidates.append(company)

if not candidates:
return None

return max(
candidates,
key=len,
)

7.13 匿名主体识别

这是一个非常重要的小函数。

ANON_PATTERNS = (
"某公司",
"某企业",
"某有限公司",
"某集团",
)

def is_anonymous_company(
company,
):

if not company:
return True

return any(
p in company
for p in ANON_PATTERNS
)

如果：

is_anonymous_company(company)

为 True：

match_status = unresolved

不要继续 RapidFuzz。

因为：

湖南某公司

与真实世界中几十万家公司之间进行模糊匹配没有任何统计意义。

生态环境部公开典型案例中确实存在大量“某公司”式匿名主体，因此将这类记录保留为 unresolved 是必要的数据质量控制，而不是程序能力不足。

7.14 修复费用提取

先写金额转换。

import re
from decimal import Decimal

def parse_money_number(
value: str,
):

if not value:
return None

value = value.replace(
",",
"",
)

match = re.search(
r"(\\d+(?:\\.\\d+)?)"
r"\\s*"
r"(亿元|万元|元)",
value,
)

if not match:
return None

number = Decimal(
match.group(1)
)

unit = match.group(2)

multiplier = {
"元": Decimal("1"),
"万元": Decimal("10000"),
"亿元": Decimal("100000000"),
}[unit]

return float(
number * multiplier
)

测试：

assert parse_money_number(
"167.12元"
) == 167.12

assert parse_money_number(
"30万元"
) == 300000.0

assert parse_money_number(
"1.2亿元"
) == 120000000.0

7.15 根据语义抽 restoration_cost

RESTORATION_PATTERNS = [

r"清除污染[、，,\\s]*"
r"修复生态环境费用"
r"(?:为|：|:|，|,|\\s)*"
r"(\\d+(?:\\.\\d+)?)"
r"\\s*(亿元|万元|元)",

r"生态修复费用"
r"(?:为|：|:|，|,|\\s)*"
r"(\\d+(?:\\.\\d+)?)"
r"\\s*(亿元|万元|元)",

r"修复费用"
r"(?:为|：|:|，|,|\\s)*"
r"(\\d+(?:\\.\\d+)?)"
r"\\s*(亿元|万元|元)",
]

def extract_restoration_cost(
text,
):

if not text:
return None

multiplier = {
"元": 1,
"万元": 10000,
"亿元": 100000000,
}

for pattern in RESTORATION_PATTERNS:

match = re.search(
pattern,
text,
)

if match:

number = float(
match.group(1)
)

unit = match.group(2)

return (
number
* multiplier[unit]
)

return None

这里故意没有写：

找到第一个 xxx 元就返回

因为公告里往往同时有很多金额。

7.16 状态归一化

公告里可能写：

终止索赔程序
已履行赔偿义务
已完成修复
达成赔偿协议
磋商中
已进入诉讼

数据库最好统一。

STATUS_RULES = [

(
"已履行",
[
"已履行赔偿义务",
"履行完毕",
],
),

(
"已修复",
[
"完成修复",
"已完成修复",
"修复目标",
],
),

(
"终止",
[
"终止索赔程序",
"终止索赔",
],
),

(
"已协议",
[
"签订赔偿协议",
"达成赔偿协议",
],
),

(
"诉讼中",
[
"提起诉讼",
"进入诉讼",
],
),

(
"磋商中",
[
"开展磋商",
"磋商会议",
],
),
]

def normalize_status(
text,
):

text = text or ""

for status, keywords \\
in STATUS_RULES:

if any(
word in text
for word in keywords
):
return status

return "未知"

例如昌吉州两个 2026 年公开案例的最终处理结果分别明确为“终止索赔程序”；另一个公开案例则显示相关企业已经履行赔偿义务。

7.17 case_id 怎么生成

网页通常不会提供我们想要的：

case_id

所以自行生成稳定 ID。

不要：

case_id = random.randint(...)

否则下一次抓同一个案件 ID 又变了。

推荐：

source_url + case_name

做 SHA256。

import hashlib

def generate_case_id(
source_url,
case_name,
):

raw = (
f"{source_url}|"
f"{case_name or ''}"
)

digest = hashlib.sha256(
raw.encode("utf-8")
).hexdigest()

return (
"ECO-"
+ digest[:16].upper()
)

例如：

ECO-22AB73603F7110D1

只要 URL 和案件名称不变：

case_id 永远不变

7.18 content_hash

case_id 判断：

是不是同一条案件

content_hash 判断：

公告内容有没有变化

实现：

def generate_content_hash(
text,
):

text = clean_text(
text or ""
)

return hashlib.sha256(
text.encode("utf-8")
).hexdigest()

有了它，以后可以做：

公告发生修改 → updated_at 更新

8️⃣ 数据存储与导出（Storage）

第一版我推荐 SQLite。

理由不是“它最强”，而是：

不用部署数据库
支持 SQL
支持唯一索引
支持事务
支持增量更新
可以直接复制
Python 内置支持

非常适合几百到几十万条公告。

8.1 schema.sql

CREATE TABLE IF NOT EXISTS eco_damage_cases (

id INTEGER PRIMARY KEY AUTOINCREMENT,

case_id TEXT NOT NULL UNIQUE,

case_name TEXT,

damage_type TEXT,

company_involved TEXT,

company_normalized TEXT,

unified_social_credit_code TEXT,

match_score REAL,

match_status TEXT,

restoration_cost REAL,

assessment_cost REAL,

service_loss REAL,

total_compensation REAL,

status TEXT,

publish_date TEXT,

source_url TEXT NOT NULL UNIQUE,

content_hash TEXT,

raw_text TEXT,

created_at TEXT NOT NULL,

updated_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS
idx_company_code
ON eco_damage_cases(
unified_social_credit_code
);

CREATE INDEX IF NOT EXISTS
idx_publish_date
ON eco_damage_cases(
publish_date
);

CREATE INDEX IF NOT EXISTS
idx_damage_type
ON eco_damage_cases(
damage_type
);

8.2 storage.py

import csv
import sqlite3
from datetime import datetime
from pathlib import Path

from config import OUTPUT_DIR

class Storage:

def __init__(
self,
db_path="data/eco_damage.db",
):

self.db_path = db_path

self.conn = sqlite3.connect(
db_path
)

self.conn.row_factory = \\
sqlite3.Row

self.init_db()

def init_db(self):

sql = """
CREATE TABLE IF NOT EXISTS
eco_damage_cases (

id INTEGER
PRIMARY KEY AUTOINCREMENT,

case_id TEXT
NOT NULL UNIQUE,

case_name TEXT,

damage_type TEXT,

company_involved TEXT,

company_normalized TEXT,

unified_social_credit_code TEXT,

match_score REAL,

match_status TEXT,

restoration_cost REAL,

status TEXT,

publish_date TEXT,

source_url TEXT
NOT NULL UNIQUE,

content_hash TEXT,

raw_text TEXT,

created_at TEXT NOT NULL,

updated_at TEXT NOT NULL
);
"""

self.conn.execute(sql)

self.conn.commit()

def exists_url(
self,
url,
):

cursor = self.conn.execute(
"""
SELECT 1
FROM eco_damage_cases
WHERE source_url = ?
LIMIT 1
""",
(url,),
)

return (
cursor.fetchone()
is not None
)

def get_hash_by_url(
self,
url,
):

cursor = self.conn.execute(
"""
SELECT content_hash
FROM eco_damage_cases
WHERE source_url = ?
""",
(url,),
)

row = cursor.fetchone()

if not row:
return None

return row[
"content_hash"
]

def upsert(
self,
item,
):

now = datetime.now().isoformat(
timespec="seconds"
)

sql = """
INSERT INTO eco_damage_cases (

case_id,
case_name,
damage_type,
company_involved,
company_normalized,
unified_social_credit_code,
match_score,
match_status,
restoration_cost,
status,
publish_date,
source_url,
content_hash,
raw_text,
created_at,
updated_at

)
VALUES (
?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?, ?,
?, ?, ?, ?
)

ON CONFLICT(source_url)
DO UPDATE SET

case_name = excluded.case_name,

damage_type =
excluded.damage_type,

company_involved =
excluded.company_involved,

company_normalized =
excluded.company_normalized,

unified_social_credit_code =
excluded.unified_social_credit_code,

match_score =
excluded.match_score,

match_status =
excluded.match_status,

restoration_cost =
excluded.restoration_cost,

status =
excluded.status,

publish_date =
excluded.publish_date,

content_hash =
excluded.content_hash,

raw_text =
excluded.raw_text,

updated_at =
excluded.updated_at
"""

values = (

item.case_id,

item.case_name,

item.damage_type,

item.company_involved,

item.company_normalized,

item.unified_social_credit_code,

item.match_score,

item.match_status,

item.restoration_cost,

item.status,

item.publish_date,

item.source_url,

item.content_hash,

item.raw_text,

now,

now,
)

self.conn.execute(
sql,
values,
)

self.conn.commit()

def export_csv(
self,
filename="eco_damage_cases.csv",
):

path = OUTPUT_DIR / filename

cursor = self.conn.execute(
"""
SELECT
case_id,
damage_type,
company_involved,
restoration_cost,
status,
unified_social_credit_code,
match_score,
publish_date,
source_url
FROM eco_damage_cases
ORDER BY publish_date DESC
"""
)

rows = cursor.fetchall()

if not rows:
return path

with path.open(
"w",
encoding="utf-8-sig",
newline="",
) as f:

writer = csv.writer(f)

writer.writerow(
rows[0].keys()
)

for row in rows:
writer.writerow(
list(row)
)

return path

def close(self):
self.conn.close()

8.3 为什么 CSV 用 utf-8-sig

因为：

encoding="utf-8-sig"

对于很多 Windows Excel 用户更友好。

普通：

UTF-8

文件本身没有问题，但部分 Excel 环境直接打开可能出现乱码。

8.4 字段映射表

8.5 去重策略

我推荐“双层去重”。

第一层：

source_url UNIQUE

判断：

同一网页。

第二层：

case_id

判断：

同一案件。

第三个辅助：

content_hash

判断：

页面内容有没有变化。

组合起来：

URL 一样 + hash 一样
→ 跳过

URL 一样 + hash 不一样
→ 更新

URL 不一样 + case_id 一样
→ 可能迁移页面，进入复核

URL 不一样 + case_id 不一样
→ 新案件

8.6 企业主数据

下面进入全文最重要的一块：

企业名称 → 统一社会信用代码。

首先准备：

data/master/companies.csv

格式：

company_name,unified_social_credit_code
新疆ABC环保科技有限公司,9165XXXXXXXXXXXXXX
新疆XYZ新型建材有限公司,9165YYYYYYYYYYYYYY

注意：

这里应该填你依法取得并且允许使用的真实企业主数据。

本文不会通过绕过企业查询网站访问限制去获取信用代码。

8.7 企业名称标准化

很多人会这样干：

name = name.replace(
"有限公司",
"",
)

我不建议。

因为：

A科技有限公司
A科技股份有限公司

不一定是同一主体。

所以标准化主要处理：

空格
全半角括号
异常标点
地区括号
常见格式差异

不随意删除企业组织形式。

import re
import unicodedata

def normalize_company_name(
name,
):

if not name:
return ""

name = unicodedata.normalize(
"NFKC",
name,
)

name = name.strip()

name = re.sub(
r"\\s+",
"",
name,
)

name = (
name
.replace("（", "(")
.replace("）", ")")
)

name = re.sub(
r"[·•]",
"",
name,
)

name = re.sub(
r"[，,。；;：:]",
"",
name,
)

return name

8.8 统一社会信用代码合法性校验

18 位统一社会信用代码不是普通随机字符串。

我们至少可以做格式与校验位检查。

字符集：

USCC_CHARSET = (
"0123456789"
"ABCDEFGHJKLMNPQRTUWXY"
)

注意里面没有：

I
O
S
V
Z

权重：

USCC_WEIGHTS = [
1,
3,
9,
27,
19,
26,
16,
17,
20,
29,
25,
13,
8,
24,
10,
30,
28,
]

完整实现：

USCC_CHARSET = (
"0123456789"
"ABCDEFGHJKLMNPQRTUWXY"
)

USCC_WEIGHTS = [
1,
3,
9,
27,
19,
26,
16,
17,
20,
29,
25,
13,
8,
24,
10,
30,
28,
]

def validate_uscc(
code,
):

if not code:
return False

code = (
code
.strip()
.upper()
)

if len(code) != 18:
return False

if any(
ch not in USCC_CHARSET
for ch in code
):
return False

total = 0

for i in range(17):

char_index = (
USCC_CHARSET.index(
code[i]
)
)

total += (
char_index
* USCC_WEIGHTS[i]
)

check_index = (
31
– total % 31
) % 31

expected = (
USCC_CHARSET[
check_index
]
)

return (
code[–1]
== expected
)

企业主数据载入时：

if not validate_uscc(code):
logger.warning(
"invalid USCC: {}",
code,
)

这样可以提前挡住很多脏数据。

8.9 RapidFuzz 企业消歧

安装：

pip install rapidfuzz

导入：

from rapidfuzz import (
fuzz,
process,
)

8.10 为什么先精确匹配再模糊匹配

假设有十万家企业。

如果名称完全一样：

新疆天康原种猪育种有限公司

就没有必要计算 fuzzy score。

流程应该是：

标准化名称
↓
hash map 完全匹配
↓
没找到
↓
RapidFuzz

而不是：

所有企业都 fuzzy

8.11 entity_matcher.py

import csv
from dataclasses import dataclass

from rapidfuzz import (
fuzz,
process,
)

from cleaner import (
normalize_company_name,
is_anonymous_company,
validate_uscc,
)

@dataclass
class CompanyRecord:

company_name: str

normalized_name: str

uscc: str

@dataclass
class MatchResult:

original_name: str | None

normalized_name: str | None

matched_company: str | None

uscc: str | None

score: float | None

status: str

class CompanyMatcher:

def __init__(
self,
csv_path,
auto_threshold=96,
review_threshold=85,
):

self.auto_threshold = \\
auto_threshold

self.review_threshold = \\
review_threshold

self.records = []

self.exact_map = {}

self.choice_map = {}

self._load(
csv_path
)

def _load(
self,
csv_path,
):

with open(
csv_path,
"r",
encoding="utf-8-sig",
) as f:

reader = csv.DictReader(
f
)

for row in reader:

name = (
row[
"company_name"
]
.strip()
)

code = (
row[
"unified_social_credit_code"
]
.strip()
.upper()
)

if not validate_uscc(
code
):
continue

normalized = (
normalize_company_name(
name
)
)

record = CompanyRecord(
company_name=name,
normalized_name=normalized,
uscc=code,
)

self.records.append(
record
)

self.exact_map[
normalized
] = record

self.choice_map[
normalized
] = record

def match(
self,
company_name,
):

if not company_name:

return MatchResult(
original_name=None,
normalized_name=None,
matched_company=None,
uscc=None,
score=None,
status="unresolved",
)

if is_anonymous_company(
company_name
):

return MatchResult(
original_name=company_name,
normalized_name=None,
matched_company=None,
uscc=None,
score=None,
status="unresolved",
)

normalized = (
normalize_company_name(
company_name
)
)

exact = self.exact_map.get(
normalized
)

if exact:

return MatchResult(
original_name=company_name,
normalized_name=normalized,
matched_company=exact.company_name,
uscc=exact.uscc,
score=100.0,
status="matched",
)

result = process.extractOne(

normalized,

self.choice_map.keys(),

scorer=fuzz.ratio,
)

if not result:

return MatchResult(
original_name=company_name,
normalized_name=normalized,
matched_company=None,
uscc=None,
score=None,
status="unresolved",
)

matched_name, score, _ = result

record = self.choice_map[
matched_name
]

if score >= self.auto_threshold:

status = "matched"

elif score >= self.review_threshold:

status = "review"

else:

return MatchResult(
original_name=company_name,
normalized_name=normalized,
matched_company=None,
uscc=None,
score=float(score),
status="unresolved",
)

return MatchResult(
original_name=company_name,
normalized_name=normalized,
matched_company=record.company_name,
uscc=record.uscc,
score=float(score),
status=status,
)

8.12 为什么阈值推荐 96 / 85

这是经验起点，不是普适真理。

我推荐：

100
→ 精确匹配

96～100
→ 自动匹配

85～96
→ 人工复核

<85
→ unresolved

原因在于企业名中的一个字符就可能产生完全不同的主体。

例如：

新疆宏达环保有限公司
新疆宏大环保有限公司

相似度非常高。

但可能是两家公司。

所以企业实体解析不是：

相似度越高越好。

而是：

高相似度只是候选证据。

8.13 更稳妥的二级消歧

如果主数据包含：

省
市
注册地址
法人
成立日期

可以进一步打分：

final_score = (
name_score * 0.75
+ region_score * 0.15
+ address_score * 0.10
)

即：

名称
+
地区
+
地址

联合判断。

这会比只用企业名稳定很多。

8.14 cleaner.py 完整版

import hashlib
import re
import unicodedata

DAMAGE_LABELS = {
"water": "水环境",
"air": "大气环境",
"soil": "土壤环境",
"groundwater": "地下水",
"solid_waste": "固体废物",
"mining": "矿山生态",
"ecological": "生态破坏",
}

DAMAGE_RULES = {

"water": [
"废水",
"水污染",
"地表水",
"河流",
"水体",
],

"air": [
"废气",
"大气",
"粉尘",
"扬尘",
"烟气",
],

"soil": [
"土壤",
"污染土",
],

"groundwater": [
"地下水",
"渗坑",
],

"solid_waste": [
"固体废物",
"危险废物",
"危废",
"废渣",
"垃圾",
],

"mining": [
"采矿",
"矿坑",
"矿山",
],

"ecological": [
"植被",
"林地",
"湿地",
"草地",
"生态破坏",
],
}

STATUS_RULES = [

(
"已履行",
[
"已履行赔偿义务",
"履行完毕",
],
),

(
"终止",
[
"终止索赔程序",
"终止索赔",
],
),

(
"已修复",
[
"已完成修复",
"完成生态修复",
],
),

(
"已协议",
[
"签订赔偿协议",
"达成赔偿协议",
],
),

(
"诉讼中",
[
"提起诉讼",
"进入诉讼",
],
),

(
"磋商中",
[
"开展磋商",
"磋商会议",
],
),
]

ANON_PATTERNS = (
"某公司",
"某企业",
"某有限公司",
"某集团",
)

COMPANY_PATTERN = re.compile(
r"([\\u4e00-\\u9fa5"
r"A-Za-z0-9（）()·\\-]+?"
r"(?:股份有限公司|"
r"有限责任公司|"
r"集团有限公司|"
r"有限公司|"
r"专业合作社))"
)

RESTORATION_PATTERNS = [

r"清除污染[、，,\\s]*"
r"修复生态环境费用"
r"(?:为|：|:|，|,|\\s)*"
r"(\\d+(?:\\.\\d+)?)"
r"\\s*(亿元|万元|元)",

r"生态修复费用"
r"(?:为|：|:|，|,|\\s)*"
r"(\\d+(?:\\.\\d+)?)"
r"\\s*(亿元|万元|元)",

r"修复费用"
r"(?:为|：|:|，|,|\\s)*"
r"(\\d+(?:\\.\\d+)?)"
r"\\s*(亿元|万元|元)",
]

USCC_CHARSET = (
"0123456789"
"ABCDEFGHJKLMNPQRTUWXY"
)

USCC_WEIGHTS = [
1,
3,
9,
27,
19,
26,
16,
17,
20,
29,
25,
13,
8,
24,
10,
30,
28,
]

def normalize_company_name(
name,
):

if not name:
return ""

name = unicodedata.normalize(
"NFKC",
name,
)

name = re.sub(
r"\\s+",
"",
name.strip(),
)

name = (
name
.replace("（", "(")
.replace("）", ")")
)

name = re.sub(
r"[·•]",
"",
name,
)

name = re.sub(
r"[，,。；;：:]",
"",
name,
)

return name

def is_anonymous_company(
company,
):

if not company:
return True

return any(
keyword in company
for keyword in ANON_PATTERNS
)

def extract_company(
text,
):

if not text:
return None

matches = (
COMPANY_PATTERN.findall(
text
)
)

candidates = []

for name in matches:

name = normalize_company_name(
name
)

if len(name) >= 4:
candidates.append(
name
)

if not candidates:
return None

return max(
candidates,
key=len,
)

def detect_damage_type(
text,
):

text = text or ""

scores = {}

for category, keywords \\
in DAMAGE_RULES.items():

score = sum(
text.count(keyword)
for keyword in keywords
)

if score:
scores[
category
] = score

if not scores:
return "未分类"

ranked = sorted(
scores.items(),
key=lambda x: x[1],
reverse=True,
)

if len(ranked) > 1:

if (
ranked[1][1]
>= ranked[0][1] * 0.8
):
return "复合型"

return DAMAGE_LABELS[
ranked[0][0]
]

def extract_restoration_cost(
text,
):

if not text:
return None

multiplier = {
"元": 1,
"万元": 10_000,
"亿元": 100_000_000,
}

for pattern in \\
RESTORATION_PATTERNS:

match = re.search(
pattern,
text,
)

if not match:
continue

number = float(
match.group(1)
)

unit = match.group(2)

return (
number
* multiplier[unit]
)

return None

def normalize_status(
text,
):

text = text or ""

for status, keywords \\
in STATUS_RULES:

for keyword in keywords:

if keyword in text:
return status

return "未知"

def generate_case_id(
source_url,
case_name,
):

value = (
f"{source_url}|"
f"{case_name or ''}"
)

digest = hashlib.sha256(
value.encode("utf-8")
).hexdigest()

return (
"ECO-"
+ digest[:16].upper()
)

def generate_content_hash(
text,
):

value = (
text
or ""
).strip()

return hashlib.sha256(
value.encode("utf-8")
).hexdigest()

def validate_uscc(
code,
):

if not code:
return False

code = (
code.strip().upper()
)

if len(code) != 18:
return False

if any(
ch not in USCC_CHARSET
for ch in code
):
return False

total = 0

for index in range(17):

char_value = (
USCC_CHARSET.index(
code[index]
)
)

total += (
char_value
* USCC_WEIGHTS[index]
)

check_index = (
31
– total % 31
) % 31

check_char = (
USCC_CHARSET[
check_index
]
)

return (
code[–1]
== check_char
)

9️⃣ 运行方式与结果展示

现在把所有模块串起来。

9.1 main.py

from loguru import logger
from tqdm import tqdm

from cleaner import (
detect_damage_type,
extract_company,
extract_restoration_cost,
generate_case_id,
generate_content_hash,
normalize_status,
)

from config import (
BASE_URL,
LIST_URL,
LOG_DIR,
)

from entity_matcher import (
CompanyMatcher,
)

from fetcher import Fetcher

from models import CleanCase

from parser import (
parse_detail,
parse_list_page,
)

from storage import Storage

def build_clean_case(
raw,
matcher,
):

combined_text = " ".join(
filter(
None,
[
raw.title,
raw.case_name,
raw.damage_investigation,
raw.assessment_conclusion,
raw.agreement,
raw.restoration_plan,
raw.restoration_effect,
raw.final_result,
],
)
)

case_name = (
raw.case_name
or raw.title
)

company = extract_company(
case_name
)

if not company:

company = extract_company(
combined_text
)

damage_type = (
detect_damage_type(
combined_text
)
)

restoration_cost = (
extract_restoration_cost(
raw.assessment_conclusion
or combined_text
)
)

status_text = " ".join(
filter(
None,
[
raw.final_result,
raw.restoration_effect,
raw.agreement,
],
)
)

status = normalize_status(
status_text
)

match = matcher.match(
company
)

case_id = generate_case_id(
raw.source_url,
case_name,
)

content_hash = (
generate_content_hash(
raw.raw_text
)
)

return CleanCase(

case_id=case_id,

case_name=case_name,

damage_type=damage_type,

company_involved=company,

restoration_cost=(
restoration_cost
),

status=status,

publish_date=(
raw.publish_date
),

source_url=(
raw.source_url
),

content_hash=(
content_hash
),

company_normalized=(
match.normalized_name
),

unified_social_credit_code=(
match.uscc
),

match_score=(
match.score
),

match_status=(
match.status
),
)

def main():

logger.add(
LOG_DIR
/ "spider_{time:YYYYMMDD}.log",

rotation="10 MB",

retention="30 days",

encoding="utf-8",
)

fetcher = Fetcher()

storage = Storage()

matcher = CompanyMatcher(
"data/master/companies.csv"
)

try:

logger.info(
"fetch list page"
)

list_html = fetcher.get(
LIST_URL
)

fetcher.save_raw(
list_html,
"list_latest.html",
)

links = parse_list_page(
list_html,
BASE_URL,
)

logger.info(
"found {} candidate urls",
len(links),
)

for item in tqdm(
links,
desc="Crawling",
):

url = item["url"]

try:

detail_html = (
fetcher.get(
url,
referer=LIST_URL,
)
)

raw = parse_detail(
detail_html,
url,
)

clean = (
build_clean_case(
raw,
matcher,
)
)

old_hash = (
storage
.get_hash_by_url(
url
)
)

if (
old_hash
== clean.content_hash
):

logger.info(
"unchanged: {}",
url,
)

continue

storage.upsert(
clean
)

logger.info(
"saved {} | {}",
clean.case_id,
clean.case_name,
)

except Exception:

logger.exception(
"detail failed: {}",
url,
)

csv_path = (
storage.export_csv()
)

logger.info(
"exported: {}",
csv_path,
)

finally:

fetcher.close()

storage.close()

if __name__ == "__main__":
main()

9.2 启动方式

项目根目录执行：

python main.py

或者：

python3 main.py

日志：

logs/spider_20260810.log

SQLite：

data/eco_damage.db

CSV：

data/output/eco_damage_cases.csv

9.3 示例输出

下面的示例用于展示数据格式，其中案件信息对应公开案例中可见的字段；信用代码列只有在你自己的企业主数据中成功匹配后才会出现，不在文章里虚构真实代码。昌吉州公开页面中可以确认新疆天康原种猪育种有限公司案件最终处理结果为终止索赔程序，其鉴定评估费用为 10000 元、修复费用为 0 元。

case_id,damage_type,company_involved,restoration_cost,status,unified_social_credit_code
ECO-A1…,复合型,奇台县顺新隆农牧业专业合作社,0,终止,
ECO-B2…,水环境,新疆天康原种猪育种有限公司,0,终止,
ECO-C3…,大气环境,新疆准东经济技术开发区昌吉吉盛新型建材有限公司,0,已履行,

另一个 2024 年公开案例中，新疆闽新钢铁（集团）闽航特钢有限责任公司案件涉及粉尘无组织排放，公开鉴定信息显示生态服务功能损失 167.12 元、鉴定评估费用 10000 元，而清除污染和修复生态环境费用为 0 元，最终结果为企业已履行赔偿义务。这恰好说明为什么 restoration_cost 不能简单取页面中最大的金额。

🔟 常见问题与排错

10.1 403

看到：

403 Forbidden

第一反应不应该是：

上代理池
改指纹
伪造浏览器

先检查：

请求是否太快
UA 是否为空
URL 是否正确
Referer 是否合理
目标是否允许普通公开访问

建议：

HEADERS = {
"User-Agent": "…",
"Referer": LIST_URL,
}

并降低频率。

如果正常浏览器无需登录可以打开，而程序偶尔出现 403，优先把请求频率继续往下降。

如果网站已经明确建立访问控制，不应通过技术手段规避。

10.2 429 Too Many Requests

429 的意思已经非常直接：

请求太频繁。

正确做法：

停止高频请求
增加 sleep
尊重 Retry-After
减少重复采集
做增量

本文 Retry 中：

respect_retry_after_header=True

就是为此准备的。

退避：

1 秒
2 秒
4 秒
8 秒

比：

失败后立刻再打 10 次

友好得多。

10.3 HTML 抓到空壳

典型现象：

浏览器：

看得到完整列表

requests：

<div id="app"></div>

这说明可能使用：

JavaScript 动态渲染

这时按顺序排查：

开发者工具
→ Network
→ Fetch/XHR
→ 刷新页面
→ 找 JSON 接口

如果存在公开接口：

API > Playwright

因为 API 通常更加：

稳定
快
结构化
易维护

只有无法通过公开接口获得必要内容时，再考虑 Playwright。

10.4 XPath 突然失效

比如以前：

tree.xpath(
'//div[@class="content"]'
)

突然返回：

[]

不要马上改成另一个绝对路径。

先把 HTML 保存：

Path(
"debug.html"
).write_text(
html,
encoding="utf-8",
)

然后：

浏览器打开
检查 DOM

我比较推荐这种选择器：

//table//tr

配合：

文本字段名

而不是：

/html/body/div[3]/div[2]/div[5]/table/tr[2]

后者极其脆弱。

10.5 网站字段名称改变

比如：

损害鉴定评估结论

以后改成：

鉴定评估结果

处理方法不是改所有代码。

只改：

FIELD_ALIASES

例如：

"assessment_conclusion": [
"损害鉴定评估结论",
"鉴定评估结论",
"鉴定评估结果",
]

这就是为什么“业务字段”和“网页字段”应该解耦。

10.6 中文乱码

requests：

response.encoding

不一定总正确。

可以：

if not response.encoding:
response.encoding = (
response.apparent_encoding
)

但如果站点声明错误，也可以根据实际情况：

response.encoding = "utf-8"

不要把乱码以后才修。

越早统一：

UTF-8

越省事。

10.7 金额解析不准

比如：

捌佰万元

正文没有阿拉伯数字。

第一版可以：

返回 None
人工复核

而不是急着手搓完整中文大写金额解析器。

真实项目更重要的是：

错误可见

而不是：

任何字符串都强行解析成一个数字

10.8 为什么抓到了企业，却匹配错了

例如：

A环保有限公司

匹配到：

A环保科技有限公司

score：

96.2

这并不能直接证明它们相同。

解决：

提高自动阈值
增加地区约束
增加企业状态
增加注册地址
人工确认 review 队列

我宁可数据库里有：

unresolved

也不愿意有一个假的统一社会信用代码。

10.9 “某公司”怎么办

直接：

unresolved

比如：

if is_anonymous_company(
company_name
):
return unresolved

不要用公告内容：

行业
地区
案件金额

反推到底是哪家公司。

一方面技术上不可靠，另一方面公开发布者已经选择了匿名表达，我们的数据工程没有必要反向猜测。

1️⃣1️⃣ 进阶优化

做到这里，其实已经是一套完整的小型采集系统了。

但如果准备长期运行，还可以继续升级。

11.1 断点续跑

目前数据库已经有：

source_url UNIQUE

所以可以在抓详情页前判断：

if storage.exists_url(url):
...

不过我更推荐不要简单：

抓过 = 永久跳过

因为公告可能更新。

更好的策略：

最近30天
→ 每次检查

30～180天
→ 每周检查

180天以前
→ 很少检查

11.2 增量窗口

伪代码：

if publish_date >= recent_date:
crawl()

例如：

每天只检查最新5页

而历史数据：

每月复查一次

比每天全量重抓更合理。

11.3 并发

如果未来扩展到很多站点，可以使用线程：

from concurrent.futures import (
ThreadPoolExecutor,
as_completed,
)

示意：

with ThreadPoolExecutor(
max_workers=3
) as executor:

futures = [
executor.submit(
fetcher.get,
url,
)
for url in urls
]

for future in as_completed(
futures
):
html = future.result()

注意：

max_workers = 3

已经足够。

政府公告采集不需要追求：

1000 QPS

11.4 asyncio

如果未来真的变成：

大量不同域名
几十万页面

可以考虑：

aiohttp
httpx

但还是那句话：

并发能力不等于应该使用的并发量。

11.5 Scrapy

扩展到全国后，Scrapy 的优势会明显起来：

Scheduler
Downloader Middleware
Retry
Pipeline
AutoThrottle
Request fingerprint
Job persistence

结构可以升级为：

scrapy project
│
├── spiders/
│ ├── changji.py
│ ├── province_a.py
│ └── province_b.py
│
├── pipelines.py
├── middlewares.py
└── items.py

所有站点最终统一输出：

EcoDamageItem

11.6 Playwright

只给真正动态的站点用。

例如：

from playwright.sync_api import (
sync_playwright
)

def fetch_dynamic(
url,
):

with sync_playwright() as p:

browser = (
p.chromium.launch(
headless=True
)
)

page = browser.new_page()

page.goto(
url,
wait_until=(
"networkidle"
),
)

html = page.content()

browser.close()

return html

不要出现：

requests 能跑
↓
因为 Playwright 看起来高级
↓
硬换 Playwright

毫无必要。

11.7 日志

建议至少记录：

请求成功数
请求失败数
解析成功数
解析失败数
新增案件数
更新案件数
企业匹配成功数
企业 review 数
unresolved 数

比如：

stats = {
"request_ok": 0,
"request_failed": 0,
"parse_ok": 0,
"parse_failed": 0,
"matched": 0,
"review": 0,
"unresolved": 0,
}

最后输出：

logger.info(
"stats={}",
stats,
)

11.8 数据质量报告

这一步特别值得做。

例如：

SELECT
match_status,
COUNT(*)
FROM eco_damage_cases
GROUP BY match_status;

得到：

matched 382
review 27
unresolved 51

还可以：

SELECT
damage_type,
COUNT(*)
FROM eco_damage_cases
GROUP BY damage_type;

如果：

未分类 = 70%

说明污染分类规则需要改。

如果：

restoration_cost IS NULL = 80%

说明金额提取器需要检查。

好的爬虫不是：

不报错

而是：

知道自己的数据质量怎么样

11.9 review 企业人工审核队列

把：

85 <= score < 96

单独导出。

def export_review(
conn,
):

sql = """
SELECT
case_id,
company_involved,
company_normalized,
unified_social_credit_code,
match_score,
source_url
FROM eco_damage_cases
WHERE match_status = 'review'
ORDER BY match_score DESC
"""

return conn.execute(
sql
).fetchall()

人工审核以后维护：

company_alias.csv

格式：

alias_name,standard_name,uscc
某某环保公司,某某环保有限公司,91…

下一次：

别名精确命中

就不需要继续 fuzzy。

这叫：

人工反馈闭环。

11.10 Alias 优先级

最终企业匹配流程推荐：

原名称
↓
匿名检测
↓
标准化
↓
正式名称 exact
↓
alias exact
↓
fuzzy
↓
score >= 96
├─ matched
↓
85 <= score < 96
├─ review
↓
score < 85
└─ unresolved

这个结构相当实用。

11.11 SQLite 升级 MySQL

当出现：

多机器写入
多人查询
后台接口
BI系统
Web应用

再切 MySQL/PostgreSQL。

表结构不用大改。

例如 MySQL：

CREATE TABLE eco_damage_cases (

id BIGINT
PRIMARY KEY AUTO_INCREMENT,

case_id VARCHAR(64)
NOT NULL UNIQUE,

case_name VARCHAR(1000),

damage_type VARCHAR(100),

company_involved VARCHAR(500),

company_normalized VARCHAR(500),

unified_social_credit_code
CHAR(18),

match_score DECIMAL(5,2),

match_status VARCHAR(32),

restoration_cost
DECIMAL(18,2),

status VARCHAR(100),

publish_date DATE,

source_url VARCHAR(2000)
NOT NULL,

content_hash CHAR(64),

raw_text LONGTEXT,

created_at DATETIME,

updated_at DATETIME,

UNIQUE KEY uk_source_url(
source_url(500)
)
);

11.12 定时任务

Linux cron：

crontab -e

例如每天凌晨 03:20：

20 3 * * * cd /opt/eco_damage_spider && /opt/eco_damage_spider/.venv/bin/python main.py >> logs/cron.log 2>&1

为什么选凌晨只是工程示例。

真正部署时应该根据目标网站访问情况合理安排，而且仍然保持低频请求。