复杂网页表格采集：合并单元格、多级表头与分页导出

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install beautifulsoup4 httpx

import csv
import re
import unicodedata
from decimal import Decimal
from urllib.parse import urljoin
from bs4 import BeautifulSoup
import httpx
 
HEAD = """<thead><tr><th rowspan="2">地区</th><th rowspan="2">渠道</th>
<th colspan="2">2026年</th></tr><tr><th>销售额（万元）</th><th>订单数</th></tr></thead>"""
PAGES = {
    "https://example.16yun.cn/stats?page=1":
        '<table id="stats">' + HEAD + """<tbody>
        <tr><td rowspan="2">华东</td><td>在线</td><td>1,234.50</td><td>10</td></tr>
        <tr><td>线下</td><td>—</td><td>3</td></tr></tbody></table>
        <a class="next" href="?page=2">Next</a>""",
    "https://example.16yun.cn/stats?page=2":
        '<table id="stats">' + HEAD + """<tbody>
        <tr><td>华南</td><td>在线</td><td>100.00</td><td>2</td></tr>
        </tbody></table>""",
}
 
def expand(section):
    rows = section.find_all("tr", recursive=False)
    grid = {}
    for r, row in enumerate(rows):
        c = 0
        for cell in row.find_all(["th", "td"], recursive=False):
            while (r, c) in grid:
                c += 1
            height, width = int(cell.get("rowspan", 1)), int(cell.get("colspan", 1))
            if height < 1 or width < 1 or r + height > len(rows):
                raise ValueError("合并范围超出当前行组")
            value = cell.get_text(" ", strip=True)
            for dr in range(height):
                for dc in range(width):
                    position = (r + dr, c + dc)
                    if position in grid:
                        raise ValueError("合并单元格重叠")
                    grid[position] = value
            c += width
    columns = max((c for _, c in grid), default=-1) + 1
    if any((r, c) not in grid for r in range(len(rows)) for c in range(columns)):
        raise ValueError("表格存在无法解释的缺格")
    return [[grid[r, c] for c in range(columns)] for r in range(len(rows))]
 
def flattened_headers(section):
    matrix = expand(section)
    return [
        "_".join(dict.fromkeys(row[c] for row in matrix if row[c]))
        for c in range(len(matrix[0]))
    ]
 
def amount_yuan(value):
    value = unicodedata.normalize("NFKC", value).strip()
    if value in {"—", "-", ""}:
        return None
    if not re.fullmatch(r"(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d{1,2})?", value):
        raise ValueError("金额格式不符合本例约定")
    return str(Decimal(value.replace(",", "")) * Decimal(10000))

url = "https://example.16yun.cn/stats?page=1"
visited, records, seen_keys = set(), [], set()
expected = None
while url:
    if url in visited:
        raise ValueError("分页出现循环")
    visited.add(url)
    if len(visited) > 10:
        raise ValueError("超出本次页数上限")
    if url in PAGES:
        html = PAGES[url]
    else:
        with httpx.Client(timeout=20, follow_redirects=False) as client:
            response = client.get(url)
            response.raise_for_status()
            html = response.text
    soup = BeautifulSoup(html, "html.parser")
    table = soup.select_one("table#stats")
    if table is None or table.thead is None or table.tbody is None:
        raise ValueError("没有识别到完整表格行组")
    headers = flattened_headers(table.thead)
    if expected is None:
        expected = headers
    if headers != expected:
        raise ValueError("分页表头变化，不直接拼接")
    for cells in expand(table.tbody):
        if len(cells) != len(headers):
            raise ValueError("数据列数与表头不一致")
        item = dict(zip(headers, cells))
        region, channel = item["地区"], item["渠道"]
        order_text = item["2026年_订单数"]
        if not order_text.isdigit():
            raise ValueError("订单数不是非负整数")
        key = (2026, region, channel)
        if key in seen_keys:
            raise ValueError("分页中出现重复业务行")
        seen_keys.add(key)
        records.append({
            "year": 2026, "region": region, "channel": channel,
            "sales_cny": amount_yuan(item["2026年_销售额（万元）"]),
            "orders": int(order_text), "source_url": url,
        })
    next_link = soup.select_one("a.next")
    url = urljoin(url, next_link["href"]) if next_link else None
 
with open("statistics.csv", "w", encoding="utf-8-sig", newline="") as stream:
    writer = csv.DictWriter(stream, fieldnames=[
        "year", "region", "channel", "sales_cny", "orders", "source_url"
    ])
    writer.writeheader()
    writer.writerows(records)
print({"pages": len(visited), "rows": len(records)})

python tables.py