企业搜索页解析与代账线索筛选：从 HTML 结构到批处理表格

这篇文章记录我编写“天眼查企业搜索页解析与代账客户筛选”脚本的过程。脚本的工作方式是：在授权的浏览器会话和允许的访问范围内获取页面，保存 HTML，解析搜索卡片与详情页中的公开字段，最后输出到线索筛选表。当前版本没有实现接口加密或解密，主要处理的是服务端渲染 HTML、__NEXT_DATA__ 和页面中的结构化字段。

一、项目文件

我把当前目录里的文件分成三部分：

a.py                    当前使用的搜索页和详情页解析脚本
a-备份.py               早期备份版本
a备份0429.py            更早的备份版本

代账客户筛选表模板.csv   最终导入的字段模板
代账客户筛选表使用说明.md 线索字段和评分规则

先查看文件：

ls -lh
file *
python3 -m py_compile a.py

a.py 不是浏览器自动化脚本，主要依赖 requests 获取 HTML，使用 lxml.html 做解析，再用 Python 字典组织结果。

二、先确定页面数据从哪里来

我在调试时先把搜索页保存成 HTML，不直接在解析函数里发请求：

from pathlib import Path


HTML_FILE = Path(__file__).with_name("a.html")


def save_html(text: str, path: Path = HTML_FILE) -> None:
    path.write_text(text, encoding="utf-8")


def load_html(path: Path = HTML_FILE) -> str:
    return path.read_text(encoding="utf-8")

这样做以后，网络请求和 HTML 解析可以分别测试。页面访问失败时先处理请求问题，解析结果不对时直接使用已经保存的 HTML 重放，不需要反复访问页面。

解析之前先判断页面类型：

from lxml import html


def page_kind(text: str) -> str:
    document = html.fromstring(text)
    title = " ".join(document.xpath("//title//text()")).strip()
    if document.xpath("//script[@id='__NEXT_DATA__']"):
        return f"next-page:{title}"
    if "请进行身份验证以继续使用" in text:
        return "verification-page"
    return f"html-page:{title}"

实际调试时我先看标题、状态提示和关键节点，再决定进入搜索卡片解析还是详情页解析。

三、当前项目没有接口加密解密逻辑

我检查了当前目录的所有 Python 文件，发现它们没有 AES、RSA、签名算法、WASM 解密或接口密文还原流程。当前处理链路是：

请求 HTML
  -> lxml 解析 DOM
  -> 读取搜索卡片
  -> 读取 __NEXT_DATA__
  -> 统一字段
  -> 输出 JSON

详情页里的 __NEXT_DATA__ 是 Next.js 页面预置的数据。项目只从其中查找已经出现在页面响应里的结构化内容，没有再执行 JavaScript，也没有调用隐藏接口。

import json
from lxml import html


def extract_next_data(html_text: str) -> dict:
    document = html.fromstring(html_text)
    raw = document.xpath("string(//script[@id='__NEXT_DATA__'])")
    if not raw:
        raise ValueError("页面中未找到 __NEXT_DATA__")
    return json.loads(raw)

如果以后拿到的响应是密文，解密过程应该以数据所有者提供的接口文档为准，先确认算法、编码、IV、签名和错误码，再在本地测试夹具里验证。不能通过猜测密钥或复制第三方登录态来处理。

四、请求层的实现

当前脚本的请求层包含四个动作：

1. 复制基础请求头；
2. 控制相邻请求的时间间隔；
3. 判断是否返回验证码页；
4. 对临时网络错误做有限次数重试。

基础常量如下：

import time


REQUEST_INTERVAL_SECONDS = 3.0
REQUEST_RETRY_COUNT = 3
REQUEST_RETRY_BACKOFF_SECONDS = 5.0
RETRY_STATUS_CODES = {403, 429, 433, 500, 502, 503, 504}
CAPTCHA_URL_MARKERS = ("/sorry/captcha",)
CAPTCHA_TEXT_MARKERS = (
    "请进行身份验证以继续使用",
    "点击按钮开始验证",
    "识别码",
)
LAST_REQUEST_AT = 0.0
CAPTCHA_REQUIRED = False

LAST_REQUEST_AT 使用单调时钟计算间隔，避免系统时间被调整后出现负数：

def wait_for_request_slot() -> None:
    global LAST_REQUEST_AT

    elapsed = time.monotonic() - LAST_REQUEST_AT
    remaining = REQUEST_INTERVAL_SECONDS - elapsed
    if remaining > 0:
        time.sleep(remaining)


def mark_request_time() -> None:
    global LAST_REQUEST_AT
    LAST_REQUEST_AT = time.monotonic()

这个间隔控制的作用是让任务按照允许的访问频率执行，不是用来绕过限制。遇到服务端明确返回验证码或拒绝页面时，脚本应该停止自动请求。

五、验证码页面识别

我把验证码判断单独写成函数，分别检查最终 URL 和响应正文：

import requests


def is_captcha_response(response: requests.Response) -> bool:
    response_url = response.url.lower()
    if any(marker in response_url for marker in CAPTCHA_URL_MARKERS):
        return True

    return any(
        marker in response.text
        for marker in CAPTCHA_TEXT_MARKERS
    )

在请求函数里，一旦识别到验证码页，设置全局状态并抛出错误：

def ensure_not_blocked(response: requests.Response) -> None:
    global CAPTCHA_REQUIRED

    if is_captcha_response(response):
        CAPTCHA_REQUIRED = True
        raise RuntimeError(
            "已触发验证码页，请先在浏览器完成人工验证后再重试"
        )

后续请求开始前再次检查：

if CAPTCHA_REQUIRED:
    raise RuntimeError(
        "已触发验证码页，请先在浏览器完成人工验证后再重试"
    )

这里的处理顺序是停止任务、保存当前状态、由人工完成验证或确认访问权限，再决定是否继续。脚本不自动点击验证码、不修改验证参数，也不通过更换设备指纹来规避页面保护。

六、请求重试

请求函数只对临时错误进行有限次重试：

from requests import RequestException


def fetch_html(
    session,
    url: str,
    *,
    params: dict | None = None,
    headers: dict | None = None,
) -> str:
    global CAPTCHA_REQUIRED

    if CAPTCHA_REQUIRED:
        raise RuntimeError("任务已暂停，请先处理验证状态")

    last_error = None
    for attempt in range(1, REQUEST_RETRY_COUNT + 1):
        wait_for_request_slot()
        try:
            response = session.get(
                url,
                params=params,
                headers=headers,
                timeout=30,
            )
            mark_request_time()
            ensure_not_blocked(response)

            if (
                response.status_code in RETRY_STATUS_CODES
                and attempt < REQUEST_RETRY_COUNT
            ):
                time.sleep(
                    REQUEST_RETRY_BACKOFF_SECONDS * attempt
                )
                continue

            response.raise_for_status()
            return response.text
        except RuntimeError:
            raise
        except RequestException as exc:
            mark_request_time()
            last_error = exc
            if attempt < REQUEST_RETRY_COUNT:
                time.sleep(
                    REQUEST_RETRY_BACKOFF_SECONDS * attempt
                )
                continue
            break

    raise RuntimeError(f"请求失败: {last_error}") from last_error

对于 401、403、429 或验证码页，必须先检查授权和访问规则。不能无限重试，也不能把重试写成并发请求。

七、搜索卡片解析

搜索结果页面里的每条企业信息位于带有 index_search-box__ 类名的卡片中。我的解析函数先找到卡片，再逐项取字段：

from lxml import html


def clean_text(value: str) -> str:
    return " ".join(value.replace("\xa0", " ").split())


def normalize_value(value) -> str:
    if hasattr(value, "text_content"):
        return clean_text(value.text_content())
    return clean_text(str(value))


def first_text(node, xpath: str) -> str:
    values = [
        normalize_value(value)
        for value in node.xpath(xpath)
    ]
    values = [value for value in values if value]
    return values[0] if values else ""

键值块的格式是“字段名：字段值”，统一解析：

def parse_key_value_blocks(nodes) -> dict[str, str]:
    result = {}
    for node in nodes:
        text = clean_text(
            "".join(node.xpath(".//text()"))
        )
        if "：" not in text:
            continue
        key, value = text.split("：", 1)
        result[clean_text(key)] = clean_text(value)
    return result

联系方式里可能有“更多”提示，当前代码只保留提示前的可见值：

def clean_contact_value(value: str) -> str:
    return clean_text(
        value.split("更多", 1)[0]
    ).lstrip("")

完整的单卡片解析：

def parse_company_cards_from_document(document):
    cards = document.xpath(
        "//div[contains(@class, 'index_search-box__')]"
    )
    results = []

    for card in cards:
        name = first_text(
            card,
            ".//div[contains(@class, 'index_name__')]//a",
        )
        if not name:
            continue

        info_blocks = parse_key_value_blocks(
            card.xpath(
                ".//div[contains(@class, 'index_info-col__')]"
            )
        )
        contact_blocks = parse_key_value_blocks(
            card.xpath(
                ".//div[contains(@class, 'index_contact-col__')]"
            )
        )
        tags = [
            clean_text(text)
            for text in card.xpath(
                ".//div[contains("
                "@class, 'index_tag-list-container__'"
                ")]//span[contains("
                "@class, 'index_tag-item__'"
                ")]//text()"
            )
            if clean_text(text)
        ]

        results.append({
            "company_name": name,
            "company_url": first_text(
                card,
                ".//div[contains(@class, 'index_name__')]"
                "//a/@href",
            ),
            "status": first_text(
                card,
                ".//span[contains(@class, "
                "'index_tag-status__')]//span[contains("
                "@class, 'index_tag-btn__')][1]//text()",
            ),
            "tags": list(dict.fromkeys(tags)),
            "legal_person": info_blocks.get("法定代表人", ""),
            "registered_capital": info_blocks.get("注册资本", ""),
            "establish_date": info_blocks.get("成立日期", ""),
            "credit_code": info_blocks.get(
                "统一社会信用代码", ""
            ),
            "phone": clean_contact_value(
                contact_blocks.get("电话", "")
            ),
            "email": clean_contact_value(
                contact_blocks.get("邮箱", "")
            ),
            "website": clean_contact_value(
                contact_blocks.get("网址", "")
            ),
            "address": contact_blocks.get("地址", ""),
            "match_fields": parse_match_fields(card),
        })

    return results

页面类名可能随前端版本变化，所以我把字段解析都集中在一个函数中。页面升级后只需要更新 XPath 和本地 HTML fixture，不需要修改请求、输出和评分逻辑。

八、匹配字段解析

搜索结果中可能有匹配原因或匹配字段，我把它们保存成列表，而不是拼成一段字符串：

def parse_match_fields(card):
    matches = []
    nodes = card.xpath(
        ".//div[contains("
        "@class, 'index_mult-match-fileid-wrap__'"
        ")]"
    )
    for node in nodes:
        field_name = first_text(
            node,
            ".//span[contains("
            "@class, 'index_match-field-name__'"
            ")]/text()",
        )
        field_value = first_text(
            node,
            ".//span[contains("
            "@class, 'index_jump-other__'"
            ")]//text()",
        )
        if field_name or field_value:
            matches.append({
                "field": field_name.rstrip("："),
                "value": field_value,
            })
    return matches

这样输出后可以继续做筛选，也可以在人工复核时知道企业为什么出现在结果里。

九、详情页里的 Next.js 数据

搜索卡片只包含基础信息，详情页需要再解析 __NEXT_DATA__。当前代码先提取整个 JSON，再找到 dehydratedState.queries：

def extract_dehydrated_queries(next_data):
    queries = (
        next_data.get("props", {})
        .get("pageProps", {})
        .get("dehydratedState", {})
        .get("queries", [])
    )
    result = []
    for query in queries:
        state_data = (
            query.get("state", {})
            .get("data")
        )
        if state_data is None:
            continue
        result.append({
            "query_key": query.get("queryKey"),
            "data": state_data,
        })
    return result

基础信息从指定的 query key 中取：

def extract_base_info_from_queries(detail_queries):
    target = (
        "/cloud-other-information/"
        "companyinfo/baseinfo/web"
    )
    for item in detail_queries:
        query_key = item.get("query_key")
        if not isinstance(query_key, list):
            continue
        if not query_key or query_key[0] != target:
            continue

        data = item.get("data")
        if (
            isinstance(data, dict)
            and isinstance(data.get("data"), dict)
        ):
            return data["data"]
    return {}

详情页解析函数保留三类结果：

• base_info：解析到的基础信息；
• detail_queries：用于后续本地分析的结构化 query；
• error：当前页面失败原因。

def parse_company_detail(company_url: str) -> dict:
    try:
        text = fetch_company_detail_html(company_url)
        next_data = extract_next_data(text)
        queries = extract_dehydrated_queries(next_data)
        return {
            "detail_url": company_url,
            "query_count": len(queries),
            "base_info": extract_base_info_from_queries(
                queries
            ),
            "detail_queries": queries,
            "error": "",
        }
    except Exception as exc:
        return {
            "detail_url": company_url,
            "query_count": 0,
            "base_info": {},
            "detail_queries": [],
            "error": str(exc),
        }

十、搜索页和详情页的组合

搜索页解析出企业卡片后，我再按卡片里的详情链接补充详情页：

def enrich_company_list_with_details(company_list):
    for company in company_list:
        company_url = company.get("company_url", "")
        if not company_url:
            company["detail_page"] = {}
            continue

        if CAPTCHA_REQUIRED:
            company["detail_page"] = {
                "detail_url": str(company_url),
                "query_count": 0,
                "base_info": {},
                "detail_queries": [],
                "error": (
                    "已跳过：前序请求已触发验证码页"
                ),
            }
            continue

        company["detail_page"] = parse_company_detail(
            str(company_url)
        )
    return company_list

发生一次验证状态后，后面的详情页全部跳过，避免已经被拒绝后继续产生请求。

十一、输出 JSON

主函数保持简单：

import json
import sys


def main():
    try:
        html_text = fetch_search_html()
    except RuntimeError as exc:
        print(
            f"搜索页请求失败: {exc}",
            file=sys.stderr,
        )
        print("[]")
        return

    company_list = parse_company_cards_from_html(
        html_text
    )
    company_list = enrich_company_list_with_details(
        company_list
    )
    print(json.dumps(
        company_list,
        ensure_ascii=False,
        indent=2,
    ))

本地读取已经保存的 HTML 时：

def parse_company_cards(file_path: Path):
    document = html.fromstring(
        file_path.read_text(encoding="utf-8")
    )
    return parse_company_cards_from_document(document)

推荐先用本地 HTML 跑解析，再接入网络请求：

python3 -m py_compile a.py
python3 parse_fixture.py > work/result.json
python3 -m json.tool work/result.json > /dev/null

十二、线索筛选表

最终表格模板中的字段按三部分组织：

基础信息：
公司名称、统一社会信用代码、法人姓名、成立日期、
企业状态、企业类型、注册资本、所属行业

联系和区域：
联系电话、注册地址、所属区域、是否本地、
经营范围关键词、线索来源

跟进记录：
需求判断、当前痛点、首次沟通结果、跟进状态、
下次联系日期、线索评分、优先级、备注

CSV 表头保持稳定：

CSV_FIELDS = [
    "序号",
    "公司名称",
    "统一社会信用代码",
    "法人姓名",
    "联系电话",
    "成立日期",
    "企业状态",
    "企业类型",
    "注册资本",
    "所属行业",
    "经营范围关键词",
    "注册地址",
    "所属区域",
    "是否本地",
    "是否新设(90天内)",
    "是否已开票",
    "是否有专职会计",
    "是否有社保需求",
    "是否工商变更频繁",
    "是否异常经营",
    "线索来源",
    "需求判断",
    "当前痛点",
    "首次沟通结果",
    "跟进状态",
    "下次联系日期",
    "线索评分",
    "优先级",
    "备注",
]

输出 CSV 时使用 UTF-8 with BOM，方便在 Windows Excel 中直接打开：

import csv


def write_csv(rows, path):
    with open(
        path,
        "w",
        encoding="utf-8-sig",
        newline="",
    ) as fp:
        writer = csv.DictWriter(
            fp,
            fieldnames=CSV_FIELDS,
            extrasaction="ignore",
        )
        writer.writeheader()
        writer.writerows(rows)

十三、字段清洗和人工补充

搜索页能解析出基础字段，但“是否已开票”“是否有专职会计”“当前痛点”“首次沟通结果”等内容不能从企业公开页面直接推断，需要在沟通后人工填写。

我先把机器解析结果映射成表格：

def to_lead_row(company: dict, index: int) -> dict:
    base = company.get("detail_page", {}).get(
        "base_info", {}
    )
    return {
        "序号": index,
        "公司名称": company.get("company_name", ""),
        "统一社会信用代码": company.get(
            "credit_code", ""
        ),
        "法人姓名": company.get("legal_person", ""),
        "联系电话": company.get("phone", ""),
        "成立日期": company.get(
            "establish_date", ""
        ),
        "企业状态": company.get("status", ""),
        "注册资本": company.get(
            "registered_capital", ""
        ),
        "注册地址": company.get("address", ""),
        "线索来源": "企业搜索页",
        "需求判断": "",
        "首次沟通结果": "",
        "跟进状态": "未联系",
        "备注": "",
        **base,
    }

机器字段只作为线索整理，电话联系前再人工确认企业名称、联系人身份和公开联系方式。

十四、评分规则

项目说明文件里使用的初始规则如下：

def score_lead(row: dict) -> int:
    score = 0

    if row.get("是否新设(90天内)") == "是":
        score += 30
    if row.get("是否本地") == "是":
        score += 20
    if row.get("所属行业") in {
        "电商", "贸易", "咨询", "广告",
        "设计", "科技服务", "建筑劳务", "餐饮",
    }:
        score += 20
    if row.get("是否有专职会计") == "否":
        score += 15
    if row.get("是否已开票") == "是":
        score += 10
    if row.get("联系电话"):
        score += 10
    if row.get("是否异常经营") == "是":
        score -= 30

    return score

根据分数设置优先级：

def priority(score: int) -> str:
    if score >= 70:
        return "高"
    if score >= 50:
        return "中"
    return "低"

企业规模、注册资本和行业字段只能作为辅助排序条件，最终结果以人工核验和沟通结果为准。

十五、批量任务的实现

批处理时我先读取关键词，再去重：

def normalize_keywords(values):
    result = []
    seen = set()
    for value in values:
        keyword = " ".join(value.split())
        if not keyword or keyword in seen:
            continue
        seen.add(keyword)
        result.append(keyword)
    return result

每个关键词单独处理，失败项写入错误列表：

def run_keywords(keywords):
    results = []
    errors = []

    for keyword in keywords:
        try:
            page_html = fetch_search_html(
                params={"key": keyword}
            )
            companies = parse_company_cards_from_html(
                page_html
            )
            results.extend(companies)
        except RuntimeError as exc:
            errors.append({
                "keyword": keyword,
                "error": str(exc),
            })
            if CAPTCHA_REQUIRED:
                break

    return results, errors

任务中断后，不重复已经成功保存的条目：

from pathlib import Path
import json


def append_jsonl(path: Path, record: dict):
    with path.open("a", encoding="utf-8") as fp:
        fp.write(json.dumps(
            record,
            ensure_ascii=False,
        ) + "\n")


def completed_keywords(path: Path) -> set[str]:
    completed = set()
    if not path.exists():
        return completed
    for line in path.read_text(
        encoding="utf-8"
    ).splitlines():
        item = json.loads(line)
        if item.get("status") == "ok":
            completed.add(item["keyword"])
    return completed

十六、为什么不使用并发请求

当前脚本选择串行执行。原因是搜索页和详情页属于连续访问，详情页还依赖搜索结果中的链接。串行处理可以：

每次只处理一个关键词
  -> 保存搜索结果
  -> 按需处理详情页
  -> 写入结果
  -> 记录最后请求时间

发生验证码页时，串行流程可以立即停止，保存已经处理的条目。如果使用并发，多个请求可能在同一时间继续发送，既不利于任务恢复，也容易超出服务方的访问限制。

十七、常见页面问题

1. 页面中没有搜索卡片

先检查是不是登录页、验证码页或错误页：

def diagnose(text: str) -> dict:
    document = html.fromstring(text)
    return {
        "title": " ".join(
            document.xpath("//title//text()")
        ).strip(),
        "has_cards": bool(document.xpath(
            "//div[contains(@class, "
            "'index_search-box__')]"
        )),
        "has_next_data": bool(document.xpath(
            "//script[@id='__NEXT_DATA__']"
        )),
        "looks_like_captcha": any(
            marker in text
            for marker in CAPTCHA_TEXT_MARKERS
        ),
    }

2. 页面字段为空

保存原始 HTML，查看字段所在节点是否改名。不要只根据一个固定 class 的完整字符串匹配，可以继续使用 contains(@class, '前缀')，并为每个字段准备本地 fixture。

3. 详情页没有 __NEXT_DATA__

说明当前详情页可能不是 Next.js 服务端渲染，或者返回的不是正常详情页。记录状态和 URL，停止后续自动请求，再根据页面所有者提供的公开接口文档处理。

4. 解析出重复企业

可以使用统一社会信用代码去重，没有信用代码时再使用“公司名称 + 注册地址”组合：

def deduplicate(rows):
    result = []
    seen = set()
    for row in rows:
        key = row.get("统一社会信用代码") or (
            row.get("公司名称", ""),
            row.get("注册地址", ""),
        )
        if key in seen:
            continue
        seen.add(key)
        result.append(row)
    return result

十八、运行顺序

我实际运行时按下面的顺序执行：

# 1. 检查语法
python3 -m py_compile a.py

# 2. 先用本地 HTML fixture 测试解析
python3 parse_fixture.py

# 3. 检查 JSON 结构
python3 -m json.tool work/result.json

# 4. 在获得授权的情况下运行单个关键词
python3 a.py > work/one-keyword.json

# 5. 通过人工复核后再写入筛选表
python3 export_leads.py work/one-keyword.json \
  --output work/代账客户筛选表.csv

任务结束后检查：

wc -l work/代账客户筛选表.csv
grep -n "异常" work/代账客户筛选表.csv

十九、配置和凭据整理

早期脚本中包含浏览器 Cookie、认证 Token、用户标识和设备标识。正式使用时不能把这些值继续放在源码里，应该改成环境变量或本地未提交配置：

import os


def required_env(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(
            f"缺少环境变量: {name}"
        )
    return value


session_cookie = required_env(
    "AUTHORIZED_SESSION_COOKIE"
)

检查 Git 历史和当前目录：

git grep -nE \
  "auth_token|csrfToken|Cookie|session|token" \
  -- '*.py' '*.js' '*.txt'

发现真实凭据后，应先在服务端撤销，再清理本地文件和提交历史。文章、截图、日志和 CSV 都不能带出完整凭据。

二十、总结

这个项目的完整链路是：

授权会话
  -> 串行请求搜索页
  -> 验证页面类型
  -> 解析企业卡片
  -> 按详情链接补充 Next 数据
  -> 字段清洗与去重
  -> 输出 JSON / CSV
  -> 人工筛选和跟进

当前版本的重点是 HTML 解析、Next.js 数据提取、验证码识别、限速重试和线索表格输出，并没有接口加密解密代码。反爬处理部分按照“识别限制后停止”的方式实现，任务在授权范围内运行。