Python正则表达式实战：从招生网页到结构化数据的精准提取

最近在帮一个朋友处理招生数据时，遇到了一个典型问题——他需要从某高校招生网站上提取近几年的录取分数线信息，但网站没有提供直接的API接口，数据都嵌在HTML表格里。手动复制粘贴几十个页面显然不现实，而用现成的爬虫框架又显得有点“杀鸡用牛刀”。这时候，Python的正则表达式就成了最趁手的工具。

正则表达式常被初学者视为“天书”，但实际上，一旦掌握了它的核心逻辑，你会发现它就像一把精准的手术刀，能从杂乱的文本中快速切出你需要的数据。特别是处理那些结构相对规整但又不完全标准的网页内容时，正则表达式的灵活性和直接性优势明显。今天，我们就以从招生网页提取结构化数据这个实际场景为例，一步步拆解正则表达式的应用技巧，让你不仅能完成任务，还能真正理解背后的原理。

1. 理解网页数据的本质：从HTML到可处理文本

在开始写任何代码之前，我们需要先搞清楚网页数据到底长什么样。很多初学者直接跳进正则表达式的编写，却忽略了数据源的特性，结果写出的规则脆弱不堪，页面稍有变动就失效。

当你用浏览器打开一个招生信息页面，看到的是整洁的表格和清晰的数字。但如果你右键点击“查看页面源代码”，看到的却是完全不同的景象——大量的HTML标签、属性、样式定义，以及可能夹杂着的JavaScript代码。我们需要的录取分数数据，就藏在这些标签构成的“森林”里。

提示：现代网页很多内容是通过JavaScript动态加载的，这种情况下直接获取HTML源码可能看不到数据。但招生信息这类相对静态的内容，通常直接嵌入在初始HTML中，适合用正则表达式提取。

让我们先看看一个典型的招生数据表格在HTML中的结构：

<table class="score-table" border="1">
  <tr>
    <th>年份</th>
    <th>省份</th>
    <th>文科最低分</th>
    <th>理科最低分</th>
  </tr>
  <tr>
    <td>2023</td>
    <td>北京</td>
    <td>625</td>
    <td>638</td>
  </tr>
  <tr>
    <td>2023</td>
    <td>上海</td>
    <td>618</td>
    <td>632</td>
  </tr>
</table>

这个结构看起来清晰，但实际网页中往往复杂得多。可能会有colspan、rowspan属性合并单元格，可能有<span>标签包裹具体数值，还可能有无处不在的 （HTML空格实体）。更麻烦的是，不同年份的页面模板可能还不一致。

获取网页内容的基础操作，通常使用Python的urllib或更友好的requests库：

import urllib.request
import re

def fetch_webpage(url):
    """获取网页内容并解码为字符串"""
    try:
        # 创建请求对象，可添加headers模拟浏览器
        headers = {'User-Agent': 'Mozilla/5.0'}
        request = urllib.request.Request(url, headers=headers)
        
        # 打开URL并读取内容
        response = urllib.request.urlopen(request)
        html_content = response.read()
        
        # 解码为UTF-8字符串（注意编码可能不同）
        return html_content.decode('utf-8')
    except Exception as e:
        print(f"获取网页失败: {e}")
        return None

# 示例：获取招生页面
url = "http://example.com/admission-scores"
html_text = fetch_webpage(url)
if html_text:
    print(f"成功获取页面，长度: {len(html_text)} 字符")

这里有几个关键点需要注意：

• 有些网站会检查User-Agent，直接使用默认的Python-urllib可能被拒绝
• 编码不一定是UTF-8，可能需要根据网页的<meta charset>标签判断
• 网络请求应该有异常处理，避免程序因网络问题崩溃

2. 正则表达式核心语法：从零到精准匹配

正则表达式之所以让人望而生畏，很大程度上是因为它那套特殊的语法符号。但如果我们按功能分类理解，其实并不复杂。对于网页数据提取，真正常用的模式就那么几种。

2.1 基础匹配模式

先看一个最简单的需求：从文本中找出所有的年份数字，比如“2021”、“2022”、“2023”。对应的正则表达式是\d{4}，这里的\d表示数字，{4}表示精确匹配4次。

但实际网页中，年份可能出现在各种上下文中：

• <td>2023</td>（被标签包裹）
• 2023年录取分数（与中文混合）
• 年份：2023（前面有中文冒号）

这时候就需要更灵活的模式。下面这个表格总结了网页数据提取中最常用的正则表达式元字符：

元字符	含义	示例	匹配结果
.	匹配任意单个字符（除换行符）	a.c	"abc"、"a-c"、"a c"
\d	匹配数字	\d+	"123"、"45"、"6"
\w	匹配字母、数字、下划线	\w+	"hello"、"user123"
\s	匹配空白字符（空格、制表符等）	\s+	" "、"\t"
*	匹配前一个字符0次或多次	a*b	"b"、"ab"、"aab"
+	匹配前一个字符1次或多次	a+b	"ab"、"aab"
?	匹配前一个字符0次或1次	colou?r	"color"、"colour"
{n}	匹配前一个字符恰好n次	\d{4}	"2023"
{n,}	匹配前一个字符至少n次	\d{3,}	"123"、"1234"
{n,m}	匹配前一个字符n到m次	\d{2,4}	"12"、"123"、"1234"
[]	字符集合，匹配其中任意一个	[aeiou]	匹配任何元音字母
()	分组，提取匹配内容	(\d{4})年	从"2023年"中提取"2023"
`	`	或操作	`北京

2.2 贪婪与非贪婪匹配

这是正则表达式中最容易出错的概念之一。看这个例子：要从<td>分数：625</td><td>排名：120</td>中提取第一个<td>标签的内容。

如果写<td>.*</td>，会匹配到分数：625</td><td>排名：120——因为.*是贪婪的，它会尽可能匹配更多的字符，直到最后一个</td>。

正确的写法是<td>.*?</td>，这里的?使.*变为非贪婪模式，匹配到第一个</td>就停止，结果就是分数：625。

在实际的HTML提取中，几乎总是使用非贪婪模式，除非你确定要匹配大段的跨标签内容。

import re

# 贪婪匹配 vs 非贪婪匹配示例
html_example = '<td>理科：638</td><td>文科：625</td>'

# 贪婪匹配 - 错误
greedy_pattern = r'<td>.*</td>'
greedy_match = re.search(greedy_pattern, html_example)
print(f"贪婪匹配结果: {greedy_match.group() if greedy_match else '无匹配'}")

# 非贪婪匹配 - 正确
non_greedy_pattern = r'<td>.*?</td>'
non_greedy_match = re.search(non_greedy_pattern, html_example)
print(f"非贪婪匹配结果: {non_greedy_match.group() if non_greedy_match else '无匹配'}")

# 输出：
# 贪婪匹配结果: <td>理科：638</td><td>文科：625</td>
# 非贪婪匹配结果: <td>理科：638</td>

2.3 实战：提取表格行数据

现在我们来解决一个实际问题：从HTML表格中提取所有行数据。假设我们有这样的HTML片段：

<table>
  <tr class="header">
    <th>省份</th><th>最低分</th><th>平均分</th>
  </tr>
  <tr>
    <td>北京</td><td>638</td><td>645</td>
  </tr>
  <tr>
    <td>上海</td><td>632</td><td>640</td>
  </tr>
</table>

提取思路应该是：

1. 先匹配整个表格：<table.*?>(.*?)</table>
2. 再匹配每一行：<tr.*?>(.*?)</tr>
3. 最后匹配每个单元格：<td.*?>(.*?)</td>或<th.*?>(.*?)</th>

但这里有个细节：re.findall()默认不会匹配换行符，而HTML中的标签经常跨越多行。这时候需要re.S标志（使.匹配包括换行符在内的所有字符）：

def extract_table_data(html_text):
    """从HTML文本中提取表格数据"""
    
    # 匹配整个表格（非贪婪，支持跨行）
    table_pattern = r'<table.*?>(.*?)</table>'
    table_match = re.search(table_pattern, html_text, re.S | re.I)
    
    if not table_match:
        return []
    
    table_content = table_match.group(1)
    
    # 匹配所有行
    rows_pattern = r'<tr.*?>(.*?)</tr>'
    rows = re.findall(rows_pattern, table_content, re.S | re.I)
    
    all_data = []
    for row in rows:
        # 匹配行内所有单元格（包括th和td）
        cells_pattern = r'<(?:td|th).*?>(.*?)</(?:td|th)>'
        cells = re.findall(cells_pattern, row, re.S | re.I)
        
        if cells:  # 跳过空行
            # 清理单元格内容：去除HTML实体、多余空白
            cleaned_cells = []
            for cell in cells:
                # 移除HTML标签（如果有嵌套）
                cell_text = re.sub(r'<.*?>', '', cell)
                # 替换HTML实体
                cell_text = cell_text.replace('&nbsp;', ' ')
                cell_text = cell_text.replace('&amp;', '&')
                cell_text = cell_text.replace('&lt;', '<')
                cell_text = cell_text.replace('&gt;', '>')
                # 去除首尾空白
                cell_text = cell_text.strip()
                cleaned_cells.append(cell_text)
            
            all_data.append(cleaned_cells)
    
    return all_data

# 测试提取
sample_html = """
<html>
<body>
<table border="1">
  <tr><th>省份</th><th>分数</th></tr>
  <tr><td>北京</td><td>638</td></tr>
  <tr><td>上海</td><td>632</td></tr>
</table>
</body>
</html>
"""

table_data = extract_table_data(sample_html)
for i, row in enumerate(table_data):
    print(f"第{i}行: {row}")

3. 数据清洗与规范化：从原始文本到可用数据

提取出来的数据往往不能直接使用，需要经过清洗和规范化处理。这是数据提取过程中最繁琐但也最关键的一步。

3.1 常见的数据清洗问题

招生数据提取中，我遇到过各种奇怪的数据格式问题：

1. 空白字符混乱：除了普通的空格，还有 、全角空格、制表符、多个连续空格等
2. 数字格式不一致：有的分数带小数点，有的是整数；有的千位有逗号分隔，有的没有
3. 特殊字符干扰：<br>标签、&实体、不可见字符等
4. 数据缺失或异常：某些单元格可能是"暂无"、"—"、"-"或直接为空
5. 编码问题：中文乱码、特殊符号显示异常

3.2 系统化的清洗流程

我通常按照以下流程清洗提取的数据：

def clean_extracted_data(raw_data):
    """清洗提取的原始数据"""
    
    cleaned_results = []
    
    for row in raw_data:
        cleaned_row = []
        for cell in row:
            # 1. 去除所有HTML标签（处理嵌套情况）
            cell = re.sub(r'<[^>]+>', '', cell)
            
            # 2. 处理HTML实体
            html_entities = {
                '&nbsp;': ' ', '&amp;': '&', '&lt;': '<', 
                '&gt;': '>', '&quot;': '"', '&#39;': "'",
                '&ensp;': ' ', '&emsp;': '  '
            }
            for entity, replacement in html_entities.items():
                cell = cell.replace(entity, replacement)
            
            # 3. 统一空白字符
            # 替换全角空格为半角空格
            cell = cell.replace('　', ' ')
            # 替换多个连续空白为单个空格
            cell = re.sub(r'\s+', ' ', cell)
            # 去除首尾空白
            cell = cell.strip()
            
            # 4. 处理特殊占位符
            if cell in ['暂无', '—', '-', 'NULL', 'null', 'NaN']:
                cell = ''  # 或根据需求设置为None或0
            
            # 5. 规范化数字格式
            # 移除数字中的逗号（如"1,235" -> "1235"）
            if re.match(r'^[\d,]+(\.\d+)?$', cell.replace(',', '')):
                cell = cell.replace(',', '')
            
            cleaned_row.append(cell)
        
        # 跳过完全空的行
        if any(cell for cell in cleaned_row):
            cleaned_results.append(cleaned_row)
    
    return cleaned_results

# 测试清洗函数
dirty_data = [
    ['<span>北京</span>', '638  ', '&nbsp;645'],
    ['上海', '632', '640.5'],
    ['广东', '暂无', '625'],
    ['浙江', '1,235', '1,240.5']
]

cleaned = clean_extracted_data(dirty_data)
print("清洗前:")
for row in dirty_data:
    print(row)

print("\n清洗后:")
for row in cleaned:
    print(row)

3.3 数据类型转换与验证

清洗后的文本数据还需要转换为适当的数据类型：

def convert_data_types(cleaned_data):
    """将清洗后的文本数据转换为适当类型"""
    
    converted_data = []
    
    for row in cleaned_data:
        converted_row = []
        for cell in cleaned_row:
            # 尝试转换为整数
            if re.match(r'^\d+$', cell):
                try:
                    converted_row.append(int(cell))
                    continue
                except ValueError:
                    pass
            
            # 尝试转换为浮点数
            if re.match(r'^\d+\.\d+$', cell) or re.match(r'^\d+$', cell):
                try:
                    converted_row.append(float(cell))
                    continue
                except ValueError:
                    pass
            
            # 保留字符串
            converted_row.append(cell)
        
        converted_data.append(converted_row)
    
    return converted_data

# 添加数据验证
def validate_admission_data(data):
    """验证招生数据的合理性"""
    
    validation_errors = []
    
    for i, row in enumerate(data):
        # 检查行数据完整性
        if len(row) < 3:  # 假设至少需要省份、最低分、平均分三列
            validation_errors.append(f"第{i}行数据不完整: {row}")
            continue
        
        # 检查分数范围合理性（假设满分750）
        if isinstance(row[1], (int, float)) and (row[1] < 0 or row[1] > 750):
            validation_errors.append(f"第{i}行最低分异常: {row[1]}")
        
        if len(row) > 2 and isinstance(row[2], (int, float)):
            if row[2] < 0 or row[2] > 750:
                validation_errors.append(f"第{i}行平均分异常: {row[2]}")
            
            # 平均分不应低于最低分
            if isinstance(row[1], (int, float)) and row[2] < row[1]:
                validation_errors.append(f"第{i}行平均分低于最低分: {row}")
    
    return validation_errors

4. 完整实战：构建健壮的招生数据提取系统

现在我们把所有部分组合起来，构建一个完整的招生数据提取系统。这个系统需要处理多个页面、多种表格结构，并生成结构化的输出（如CSV文件）。

4.1 系统架构设计

我设计的提取系统包含以下模块：

招生数据提取系统
├── 网页获取模块 (Fetcher)
│   ├── URL管理
│   ├── 请求发送（带重试机制）
│   └── 编码检测与解码
├── 内容解析模块 (Parser)
│   ├── 表格定位器
│   ├── 正则表达式提取器
│   └── 备用解析策略（如BeautifulSoup）
├── 数据清洗模块 (Cleaner)
│   ├── 文本规范化
│   ├── 类型转换
│   └── 异常值处理
├── 数据验证模块 (Validator)
│   ├── 完整性检查
│   ├── 逻辑验证
│   └── 一致性检查
└── 输出模块 (Exporter)
    ├── CSV导出
    ├── JSON导出
    └── 数据库存储

4.2 实现核心提取器

import re
import csv
import json
import time
from typing import List, Dict, Any, Optional
import urllib.request
from urllib.error import URLError, HTTPError

class AdmissionDataExtractor:
    """招生数据提取器"""
    
    def __init__(self, base_url: str, start_year: int, end_year: int):
        self.base_url = base_url
        self.years = range(start_year, end_year + 1)
        self.all_data = {}  # 按年份存储数据
        
    def fetch_year_page(self, year: int) -> Optional[str]:
        """获取指定年份的招生页面"""
        
        # 构建年份页面URL（实际应用中需要根据网站结构调整）
        year_url = f"{self.base_url}/admission/{year}.html"
        
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        }
        
        max_retries = 3
        for attempt in range(max_retries):
            try:
                req = urllib.request.Request(year_url, headers=headers)
                with urllib.request.urlopen(req, timeout=10) as response:
                    content = response.read()
                    
                    # 检测编码
                    charset = 'utf-8'  # 默认
                    # 可以从Content-Type头或HTML meta标签检测实际编码
                    
                    return content.decode(charset)
                    
            except (URLError, HTTPError) as e:
                print(f"第{attempt + 1}次尝试获取{year}年数据失败: {e}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数退避
                else:
                    return None
            except Exception as e:
                print(f"获取{year}年数据时发生未知错误: {e}")
                return None
    
    def locate_score_table(self, html: str) -> Optional[str]:
        """定位包含分数数据的表格"""
        
        # 策略1：通过表格的class或id属性定位
        table_patterns = [
            r'<table[^>]*class=".*?score.*?"[^>]*>(.*?)</table>',
            r'<table[^>]*id=".*?score.*?"[^>]*>(.*?)</table>',
            r'<table[^>]*class=".*?admission.*?"[^>]*>(.*?)</table>',
            r'<table[^>]*class=".*?data.*?"[^>]*>(.*?)</table>',
        ]
        
        for pattern in table_patterns:
            match = re.search(pattern, html, re.S | re.I)
            if match:
                return match.group(1)
        
        # 策略2：通过表格内容特征定位（包含"省份"、"分数"等关键词）
        # 查找所有表格，然后筛选包含关键字的
        all_tables = re.findall(r'<table[^>]*>(.*?)</table>', html, re.S | re.I)
        
        for table in all_tables:
            # 检查表格是否包含招生相关关键词
            keywords = ['省份', '分数', '录取', '最低分', '平均分', '文科', '理科']
            table_text = re.sub(r'<[^>]+>', '', table)
            if any(keyword in table_text for keyword in keywords):
                return table
        
        return None
    
    def extract_table_data(self, table_html: str) -> List[List[str]]:
        """从表格HTML中提取结构化数据"""
        
        # 提取所有行
        rows = re.findall(r'<tr[^>]*>(.*?)</tr>', table_html, re.S | re.I)
        
        extracted_data = []
        for row in rows:
            # 提取所有单元格（包括td和th）
            cells = re.findall(r'<(?:td|th)[^>]*>(.*?)</(?:td|th)>', row, re.S | re.I)
            
            if not cells:
                continue
            
            # 清理每个单元格
            cleaned_cells = []
            for cell in cells:
                # 移除内部HTML标签
                cell = re.sub(r'<[^>]+>', '', cell)
                # 处理HTML实体
                cell = self._clean_html_entities(cell)
                # 规范化空白
                cell = re.sub(r'\s+', ' ', cell).strip()
                
                cleaned_cells.append(cell)
            
            # 跳过完全空的行或表头行（如果不需要表头）
            if cleaned_cells and any(cell for cell in cleaned_cells):
                extracted_data.append(cleaned_cells)
        
        return extracted_data
    
    def _clean_html_entities(self, text: str) -> str:
        """清理HTML实体"""
        
        entities = {
            '&nbsp;': ' ', '&amp;': '&', '&lt;': '<', '&gt;': '>',
            '&quot;': '"', '&#39;': "'", '&ensp;': ' ', '&emsp;': '  ',
            '&#160;': ' ', '&#xa0;': ' ', '&ndash;': '-', '&mdash;': '—',
        }
        
        for entity, replacement in entities.items():
            text = text.replace(entity, replacement)
        
        # 处理数字实体（如&#1234;）
        def replace_decimal_entity(match):
            code = int(match.group(1))
            return chr(code) if 32 <= code <= 126 else ''
        
        text = re.sub(r'&#(\d+);', replace_decimal_entity, text)
        
        # 处理十六进制实体（如&#x4e2d;）
        def replace_hex_entity(match):
            code = int(match.group(1), 16)
            return chr(code) if 32 <= code <= 126 else ''
        
        text = re.sub(r'&#x([0-9a-fA-F]+);', replace_hex_entity, text)
        
        return text
    
    def process_year(self, year: int) -> bool:
        """处理单个年份的数据"""
        
        print(f"正在处理{year}年数据...")
        
        # 1. 获取页面
        html = self.fetch_year_page(year)
        if not html:
            print(f"  获取{year}年页面失败")
            return False
        
        # 2. 定位表格
        table_html = self.locate_score_table(html)
        if not table_html:
            print(f"  未找到{year}年的分数表格")
            # 可以尝试备用解析策略
            return False
        
        # 3. 提取数据
        raw_data = self.extract_table_data(table_html)
        if not raw_data:
            print(f"  {year}年表格数据提取失败")
            return False
        
        # 4. 数据清洗和转换
        cleaned_data = []
        for row in raw_data:
            cleaned_row = []
            for cell in row:
                # 进一步清洗
                cell = cell.replace(' ', '')  # 移除所有空格（分数通常不需要空格）
                cell = cell.replace('　', '')  # 移除全角空格
                
                # 处理特殊占位符
                if cell in ['—', '-', '暂无', 'NULL', 'null', '']:
                    cell = None
                elif cell.isdigit():
                    cell = int(cell)
                elif re.match(r'^\d+\.\d+$', cell):
                    cell = float(cell)
                
                cleaned_row.append(cell)
            
            # 只保留有效行（至少有一个非None值）
            if any(cell is not None for cell in cleaned_row):
                cleaned_data.append(cleaned_row)
        
        # 5. 存储数据
        self.all_data[year] = cleaned_data
        print(f"  {year}年数据提取完成，共{len(cleaned_data)}行")
        
        return True
    
    def export_to_csv(self, filename: str):
        """导出所有数据到CSV文件"""
        
        with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
            writer = csv.writer(f)
            
            # 写入标题行
            writer.writerow(['年份', '省份', '最低分', '平均分', '备注'])
            
            # 写入数据行
            for year, data in self.all_data.items():
                for row in data:
                    # 假设每行数据格式为[省份, 最低分, 平均分, ...]
                    if len(row) >= 3:
                        csv_row = [year, row[0], row[1], row[2] if len(row) > 2 else '']
                        # 添加其他列作为备注
                        if len(row) > 3:
                            csv_row.append(' '.join(str(x) for x in row[3:] if x is not None))
                        else:
                            csv_row.append('')
                        
                        writer.writerow(csv_row)
        
        print(f"数据已导出到 {filename}")
    
    def export_to_json(self, filename: str):
        """导出所有数据到JSON文件"""
        
        # 转换为字典格式
        json_data = {}
        for year, data in self.all_data.items():
            year_data = []
            for row in data:
                if len(row) >= 2:
                    item = {
                        'province': row[0],
                        'min_score': row[1] if len(row) > 1 else None,
                        'avg_score': row[2] if len(row) > 2 else None,
                    }
                    year_data.append(item)
            
            json_data[str(year)] = year_data
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(json_data, f, ensure_ascii=False, indent=2)
        
        print(f"数据已导出到 {filename}")
    
    def run(self):
        """运行完整的提取流程"""
        
        success_count = 0
        for year in self.years:
            if self.process_year(year):
                success_count += 1
            time.sleep(1)  # 礼貌性延迟，避免对服务器造成压力
        
        print(f"\n处理完成: {success_count}/{len(self.years)} 个年份数据提取成功")
        
        if success_count > 0:
            # 导出数据
            timestamp = time.strftime("%Y%m%d_%H%M%S")
            self.export_to_csv(f"admission_scores_{timestamp}.csv")
            self.export_to_json(f"admission_scores_{timestamp}.json")
            
            # 打印统计信息
            self.print_statistics()
    
    def print_statistics(self):
        """打印数据统计信息"""
        
        print("\n=== 数据统计 ===")
        for year, data in self.all_data.items():
            if data:
                provinces = len(data)
                scores = [row[1] for row in data if len(row) > 1 and isinstance(row[1], (int, float))]
                if scores:
                    avg_score = sum(scores) / len(scores)
                    max_score = max(scores)
                    min_score = min(scores)
                    print(f"{year}年: {provinces}个省份，平均分{avg_score:.1f}，最高{max_score}，最低{min_score}")

# 使用示例
if __name__ == "__main__":
    # 注意：这里使用示例URL，实际使用时需要替换为真实的招生网站URL
    extractor = AdmissionDataExtractor(
        base_url="http://example-university.edu.cn",
        start_year=2020,
        end_year=2023
    )
    
    # 运行提取
    extractor.run()

4.3 处理复杂情况和边缘案例

在实际项目中，你可能会遇到各种复杂情况：

情况1：表格有合并单元格

def handle_merged_cells(table_html):
    """处理包含rowspan和colspan的表格"""
    
    # 使用更复杂的解析逻辑
    # 这里可以使用BeautifulSoup等库辅助，或者编写更复杂的正则表达式
    # 基本思路：跟踪行和列的偏移量
    
    pass

情况2：数据分布在多个表格中

def extract_from_multiple_tables(html):
    """从多个相关表格中提取数据"""
    
    # 查找所有可能的分数表格
    score_tables = []
    
    # 通过不同特征定位表格
    table_patterns = [
        (r'文科.*?录取.*?分数', '文科'),
        (r'理科.*?录取.*?分数', '理科'),
        (r'综合.*?录取.*?分数', '综合'),
    ]
    
    all_tables = re.findall(r'<table[^>]*>(.*?)</table>', html, re.S | re.I)
    
    for table_html in all_tables:
        table_text = re.sub(r'<[^>]+>', '', table_html)
        
        for pattern, table_type in table_patterns:
            if re.search(pattern, table_text, re.I):
                score_tables.append({
                    'type': table_type,
                    'html': table_html,
                    'text': table_text
                })
                break
    
    return score_tables

情况3：数据通过JavaScript动态加载

# 对于动态加载的数据，正则表达式可能不够用
# 可以考虑使用Selenium或Playwright模拟浏览器

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

def extract_dynamic_content(url):
    """提取JavaScript动态加载的内容"""
    
    options = webdriver.ChromeOptions()
    options.add_argument('--headless')  # 无头模式
    driver = webdriver.Chrome(options=options)
    
    try:
        driver.get(url)
        # 等待表格加载
        wait = WebDriverWait(driver, 10)
        table = wait.until(
            EC.presence_of_element_located((By.CSS_SELECTOR, "table.score-table"))
        )
        
        # 获取渲染后的HTML
        html = driver.page_source
        return html
        
    finally:
        driver.quit()

4.4 性能优化与错误处理

当处理大量页面时，性能变得重要：

import concurrent.futures
import threading

class ConcurrentExtractor:
    """并发数据提取器"""
    
    def __init__(self, max_workers=5):
        self.max_workers = max_workers
        self.lock = threading.Lock()
        self.results = {}
    
    def process_single_year(self, year, base_url):
        """处理单个年份（可在不同线程中运行）"""
        
        extractor = AdmissionDataExtractor(base_url, year, year)
        html = extractor.fetch_year_page(year)
        
        if html:
            table_html = extractor.locate_score_table(html)
            if table_html:
                data = extractor.extract_table_data(table_html)
                with self.lock:
                    self.results[year] = data
                return True
        
        return False
    
    def process_years_concurrently(self, years, base_url):
        """并发处理多个年份"""
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有任务
            future_to_year = {
                executor.submit(self.process_single_year, year, base_url): year 
                for year in years
            }
            
            # 收集结果
            for future in concurrent.futures.as_completed(future_to_year):
                year = future_to_year[future]
                try:
                    success = future.result()
                    print(f"{year}年处理{'成功' if success else '失败'}")
                except Exception as e:
                    print(f"{year}年处理出错: {e}")

5. 正则表达式的替代方案与选择策略

虽然正则表达式在提取结构化数据时非常强大，但它并不是万能的。在某些情况下，其他工具可能更合适。

5.1 何时使用正则表达式

正则表达式最适合以下场景：

• 数据模式规整：要提取的数据有清晰、一致的模式
• 性能要求高：需要快速处理大量文本
• 依赖最小化：不希望引入额外的库依赖
• 简单提取任务：只需要提取少量字段，不需要完整的DOM解析

5.2 何时考虑其他工具

BeautifulSoup - 当HTML结构复杂、不规则时：

from bs4 import BeautifulSoup

def extract_with_bs4(html):
    """使用BeautifulSoup提取表格数据"""
    
    soup = BeautifulSoup(html, 'html.parser')
    
    # 查找所有表格
    tables = soup.find_all('table')
    
    for table in tables:
        # 检查是否是分数表格
        if '分数' in table.text or '录取' in table.text:
            data = []
            rows = table.find_all('tr')
            
            for row in rows:
                cells = row.find_all(['td', 'th'])
                row_data = [cell.get_text(strip=True) for cell in cells]
                if row_data:
                    data.append(row_data)
            
            return data
    
    return []

lxml - 当需要最高性能时：

from lxml import html

def extract_with_lxml(html_content):
    """使用lxml提取数据（性能最好）"""
    
    tree = html.fromstring(html_content)
    
    # 使用XPath定位表格
    tables = tree.xpath('//table[contains(@class, "score") or contains(@id, "score")]')
    
    for table in tables:
        data = []
        rows = table.xpath('.//tr')
        
        for row in rows:
            cells = row.xpath('.//td|.//th')
            row_data = [cell.text_content().strip() for cell in cells]
            if row_data:
                data.append(row_data)
        
        if data:
            return data
    
    return []

5.3 混合策略：正则表达式与其他工具结合

在实际项目中，我经常混合使用多种技术：

def hybrid_extraction(html_text):
    """混合使用多种技术提取数据"""
    
    # 先用正则表达式快速定位可能包含数据的区域
    # 这样可以减少需要解析的HTML量
    
    # 查找包含"录取分数"等关键词的区域
    score_sections = re.findall(
        r'<div[^>]*>.*?(?:录取|分数|分数线).*?</div>',
        html_text, 
        re.S | re.I
    )
    
    all_data = []
    
    for section in score_sections:
        # 在区域内查找表格
        tables = re.findall(r'<table[^>]*>(.*?)</table>', section, re.S | re.I)
        
        for table_html in tables:
            # 简单表格用正则表达式快速提取
            if table_html.count('<tr>') < 20:  # 行数少，结构简单
                data = extract_with_regex(table_html)
            else:  # 复杂表格用BeautifulSoup
                data = extract_with_bs4(f"<table>{table_html}</table>")
            
            if data:
                all_data.extend(data)
    
    return all_data

5.4 选择策略总结

我通常根据以下因素选择提取方法：

场景	推荐工具	理由
简单、规整的表格	正则表达式	速度快，代码简洁
复杂HTML结构	BeautifulSoup	容错性好，API友好
超大规模处理	lxml	性能最优
动态加载内容	Selenium/Playwright	能执行JavaScript
未知结构探索	混合策略	灵活适应各种情况

6. 最佳实践与常见陷阱

经过多次实战，我总结了一些正则表达式提取网页数据的最佳实践：

6.1 编写健壮的正则表达式

不要写得太具体：

# 不好：过于依赖具体的class名
pattern = r'<table class="score-table border-1">(.*?)</table>'

# 好：更通用，适应变化
pattern = r'<table[^>]*class="[^"]*score[^"]*"[^>]*>(.*?)</table>'

使用非贪婪匹配：

# 几乎总是使用非贪婪匹配
pattern = r'<td.*?>(.*?)</td>'  # 正确
pattern = r'<td.*>(.*)</td>'    # 危险，可能匹配过多内容

考虑多种可能情况：

# 考虑标签可能有属性，也可能没有
pattern = r'<td\b[^>]*>(.*?)</td>'

# 考虑可能有换行
pattern = r'<td\b[^>]*>(.*?)</td>'
matches = re.findall(pattern, html, re.S)  # re.S让.匹配换行符

6.2 处理编码问题

网页编码不一致是常见问题：

def detect_encoding(html_bytes):
    """检测HTML编码"""
    
    # 方法1：从HTTP头获取
    # 方法2：从HTML meta标签获取
    encoding_patterns = [
        r'<meta[^>]*charset=["\']?([\w-]+)["\']?',
        r'<meta[^>]*content=["\'][^"\']*charset=([\w-]+)',
    ]
    
    html_text = html_bytes[:5000].decode('ascii', errors='ignore')
    
    for pattern in encoding_patterns:
        match = re.search(pattern, html_text, re.I)
        if match:
            return match.group(1).lower()
    
    # 默认使用UTF-8，但可以尝试其他常见编码
    common_encodings = ['utf-8', 'gbk', 'gb2312', 'big5', 'iso-8859-1']
    
    for encoding in common_encodings:
        try:
            html_bytes.decode(encoding)
            return encoding
        except UnicodeDecodeError:
            continue
    
    return 'utf-8'  # 默认

6.3 错误处理与日志记录

健壮的程序需要完善的错误处理：

import logging
from datetime import datetime

def setup_logging():
    """配置日志记录"""
    
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(f'extraction_{datetime.now():%Y%m%d}.log'),
            logging.StreamHandler()
        ]
    )
    
    return logging.getLogger(__name__)

class RobustExtractor:
    """带错误处理和日志的提取器"""
    
    def __init__(self):
        self.logger = setup_logging()
        self.stats = {
            'total_pages': 0,
            'successful': 0,
            'failed': 0,
            'errors': []
        }
    
    def extract_with_retry(self, url, max_retries=3):
        """带重试的提取"""
        
        for attempt in range(max_retries):
            try:
                self.logger.info(f"尝试提取 {url} (第{attempt + 1}次)")
                result = self._extract_single(url)
                self.stats['successful'] += 1
                return result
                
            except Exception as e:
                self.logger.error(f"第{attempt + 1}次尝试失败: {e}")
                self.stats['errors'].append({
                    'url': url,
                    'attempt': attempt + 1,
                    'error': str(e),
                    'time': datetime.now()
                })
                
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # 指数退避
                    self.logger.info(f"等待{wait_time}秒后重试")
                    time.sleep(wait_time)
                else:
                    self.logger.error(f"所有尝试均失败: {url}")
                    self.stats['failed'] += 1
                    return None
        
        return None

6.4 测试与验证

编写测试用例确保提取器正常工作：

import unittest

class TestAdmissionExtractor(unittest.TestCase):
    """测试招生数据提取器"""
    
    def test_table_extraction(self):
        """测试表格提取"""
        
        html = """
        <table>
            <tr><th>省份</th><th>分数</th></tr>
            <tr><td>北京</td><td>638</td></tr>
            <tr><td>上海</td><td>632</td></tr>
        </table>
        """
        
        extractor = AdmissionDataExtractor("http://test.com", 2023, 2023)
        data = extractor.extract_table_data(html)
        
        self.assertEqual(len(data), 3)  # 表头+2行数据
        self.assertEqual(data[1], ['北京', '638'])
    
    def test_clean_html_entities(self):
        """测试HTML实体清理"""
        
        extractor = AdmissionDataExtractor("http://test.com", 2023, 2023)
        
        test_cases = [
            ("北京&nbsp;上海", "北京 上海"),
            ("分数&gt;600", "分数>600"),
            ("a&amp;b", "a&b"),
        ]
        
        for input_text, expected in test_cases:
            result = extractor._clean_html_entities(input_text)
            self.assertEqual(result, expected)
    
    def test_empty_table(self):
        """测试空表格处理"""
        
        html = "<table></table>"
        extractor = AdmissionDataExtractor("http://test.com", 2023, 2023)
        data = extractor.extract_table_data(html)
        
        self.assertEqual(data, [])

if __name__ == '__main__':
    unittest.main()

7. 实际项目中的经验分享

在真实的数据提取项目中，我遇到过各种预料之外的情况。这里分享几个实际案例和解决方案：

案例1：网站改版导致提取失败

去年我维护的一个招生数据提取脚本突然失效了。调查后发现，学校网站进行了改版，表格的CSS类名从score-table改成了admission-score。解决方案是增加多个匹配模式，并添加版本检测：

def adaptive_table_locator(html):
    """自适应表格定位器"""
    
    # 尝试多种可能的表格标识
    patterns = [
        (r'录取.*?分数.*?表', '通过标题识别'),
        (r'<table[^>]*data-type="score"[^>]*>', '通过data属性识别'),
        (r'<table[^>]*id=".*?(score|admission).*?"[^>]*>', '通过ID识别'),
        (r'<table[^>]*class=".*?(score|admission|录取|分数).*?"[^>]*>', '通过class识别'),
    ]
    
    for pattern, method in patterns:
        match = re.search(pattern, html, re.S | re.I)
        if match:
            print(f"使用{method}定位表格")
            # 提取整个表格
            table_pattern = r'<table[^>]*>.*?</table>'
            table_match = re.search(table_pattern, html[match.start():], re.S)
            if table_match:
                return table_match.group()
    
    return None

案例2：数据分页加载

有些网站将数据分页显示，每页只显示部分省份。需要先提取分页链接，然后合并数据：

def extract_paginated_data(base_url):
    """提取分页数据"""
    
    all_data = []
    page = 1
    
    while True:
        # 构建分页URL
        if '?' in base_url:
            page_url = f"{base_url}&page={page}"
        else:
            page_url = f"{base_url}?page={page}"
        
        html = fetch_webpage(page_url)
        if not html:
            break
        
        # 提取当前页数据
        page_data = extract_table_data(html)
        if not page_data:
            break
        
        all_data.extend(page_data)
        
        # 检查是否有下一页
        if '下一页' not in html and 'next' not in html.lower():
            break
        
        page += 1
        time.sleep(0.5)  # 礼貌延迟
    
    return all_data

案例3：反爬虫机制

一些网站有简单的反爬虫措施。应对方法包括：

def create_stealth_headers():
    """创建模拟真实浏览器的请求头"""
    
    user_agents = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15',
        'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
    ]
    
    import random
    import time
    
    headers = {
        'User-Agent': random.choice(user_agents),
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        'Accept-Encoding': 'gzip, deflate',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1',
        'Cache-Control': 'max-age=0',
        'Referer': 'https://www.google.com/',  # 模拟从Google跳转
    }
    
    return headers

def fetch_with_stealth(url):
    """带伪装的数据获取"""
    
    headers = create_stealth_headers()
    
    # 随机延迟，模拟人类操作
    time.sleep(random.uniform(1, 3))
    
    # 使用会话保持cookies
    session = requests.Session()
    response = session.get(url, headers=headers, timeout=10)
    
    return response.text

正则表达式提取网页数据虽然需要面对各种挑战，但一旦掌握了正确的方法和技巧，它就成为了一种高效、灵活的数据获取手段。关键是要理解数据的结构特征，编写健壮的匹配模式，并准备好处理各种边缘情况。对于招生数据提取这类相对规整但又不完全标准化的任务，正则表达式往往是最直接有效的解决方案。