突破网站查询限制:基于前缀模糊匹配的批量数据采集实战方案(Python+OCR)

13

一、问题背景

某大型网站服务平台的信息查询系统存在严格的前端展示限制:无论使用何种查询条件,系统每次仅返回固定条数(如10条),且未提供翻页或加载更多的交互入口。然而,后台公开提示该类目下实际存有上万条有效记录。对于需要批量获取企业资质、证照信息的用户而言,这种"看得见总数、拿不全明细"的设计构成了显著障碍。

二、技术难点分析

在尝试自动化采集过程中,主要遇到以下三个技术壁垒:

  • 分页参数失效:虽然接口形式上支持pagerows参数,但rows被硬限制为固定值,且翻页参数大于初始页时返回空数据,说明服务端并未真正开放连续分页能力。

  • 验证码机制:每次查询均需输入图形验证码,且验证码与当前会话(Session)强绑定,传统固定验证码或跳过的策略会很快失效。

  • 反爬拦截:请求频率过高会触发"请勿重复提交"或类似拦截提示,需要合理的请求间隔与会话保持机制。

三、核心解决思路

3.1 前缀模糊查询策略

企业类证照编号通常具有层级编码规则(如:类别码+地区码+顺序码+校验位)。通过观察发现,平台支持前缀模糊匹配。因此,可提取所有完整编号的前若干位作为基础前缀,每次查询一个前缀即可一次性召回该前缀下的全部可见记录(通常恰好为系统允许返回的上限条数),从而用数百次查询覆盖原本需要数千次精确查询才能触及的数据范围。

3.2 验证码本地识别

采用开源OCR库对平台返回的图片验证码进行实时本地识别。该方案无需调用第三方付费接口,在常规开发环境中即可完成纯数字验证码的识别,准确率较高。识别流程如下:

1. 请求平台验证码接口获取图片流
2. 使用本地OCR引擎识别为数字/字符
3. 立即携带验证码提交业务查询请求
4. 若返回"验证码错误"则自动重试

3.3 数据去重与断点续传

由于查询结果可能包含历史变更记录(同一证照的续期、变更或不同状态版本),系统可能返回多条。我们采用字典去重策略,以业务唯一主键(如证照编号)为键,确保同一条记录仅保留最新版本。同时引入JSON进度文件,每处理若干批次自动保存一次,支持随时中断与恢复,避免长时间任务因意外而前功尽弃。

四、通用Python代码框架

以下为经脱敏处理后的核心爬虫框架,实际使用时请将占位符替换为目标平台的真实接口地址与业务参数。依赖requestspandasddddocr库运行。

import requests
import pandas as pd
import time
import json
import os
import random
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

try:
    import ddddocr
    ocr = ddddocr.DdddOcr(show_ad=False)
except ImportError:
    print("请安装依赖: pip install ddddocr")
    exit()

class DataCrawler:
    def __init__(self):
        # 请替换为目标平台的实际域名与接口前缀
        self.base_url = "https://example.com/api"
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
            "X-Requested-With": "XMLHttpRequest",
            "Referer": "https://example.com/queryPage.html"
        })
        self.all_data = {}
        self.processed = set()
        self.progress_file = "progress.json"

    def load_prefix_list(self):
        """读取待查询的编号前缀列表(每行一个,存于 prefix_list.txt)"""
        with open("prefix_list.txt", "r", encoding="utf-8") as f:
            return [line.strip() for line in f if line.strip()]

    def get_captcha(self):
        """获取并识别验证码,请根据实际接口调整 URL 与参数"""
        ts = int(time.time() * 1000)
        captcha_url = f"{self.base_url}/captcha/get?ts={ts}"
        try:
            resp = self.session.get(captcha_url, timeout=10)
            code = "".join(filter(str.isdigit, ocr.classification(resp.content)))[:4]
            return code if len(code) == 4 else None
        except Exception:
            return None

    def query_by_prefix(self, prefix, max_retry=3):
        """使用前缀查询,返回该前缀下的全部可见记录"""
        for attempt in range(max_retry):
            captcha = self.get_captcha()
            if not captcha:
                continue
            
            payload = {
                "keyword": "",
                "licenseNo": prefix,      # 业务编号前缀查询字段
                "creditCode": "",
                "status": "",             # 如需查全部状态,可留空
                "captchaCode": captcha,
                "page": 0,
                "rows": 10
            }
            
            try:
                resp = self.session.post(
                    f"{self.base_url}/certificate/list",
                    data=payload,
                    timeout=30
                )
                result = resp.json()
                
                # 处理反爬提示
                if "重复提交" in str(result.get("msg", "")):
                    time.sleep(90)
                    return self.query_by_prefix(prefix, max_retry)
                
                if result.get("code") != 200:
                    if "验证码" in str(result.get("msg", "")):
                        continue
                    return []
                
                return result.get("data", [])
                
            except Exception:
                time.sleep(1)
        return []

    def crawl(self):
        prefixes = self.load_prefix_list()
        todo = [p for p in prefixes if p not in self.processed]
        print(f"待处理前缀: {len(todo)} 个")
        
        for idx, prefix in enumerate(todo, 1):
            print(f"[{idx}/{len(todo)}] 查询前缀 {prefix} ...", end=" ", flush=True)
            rows = self.query_by_prefix(prefix)
            
            if rows:
                names = []
                for row in rows:
                    biz_id = row.get("licenseNo")  # 业务主键字段
                    if biz_id and biz_id not in self.all_data:
                        self.all_data[biz_id] = row
                        names.append(row.get("orgName", "未知"))
                
                if len(names) == 1:
                    display = names[0][:20]
                else:
                    display = f"{names[0][:15]}等{len(names)}家"
                
                print(f"✓ 存入{len(rows)}条 [{display}] 累计{len(self.all_data)}条")
            else:
                print("✗ 无数据")
            
            self.processed.add(prefix)
            
            # 每10个保存进度
            if idx % 10 == 0:
                self.save_progress()
                pd.DataFrame(list(self.all_data.values())).to_excel(
                    f"进度_{len(self.all_data)}条.xlsx", index=False
                )
                print(f"\n>>> 已处理{idx}个前缀,累计{len(self.all_data)}条数据 <<<\n")
            
            # 控制请求间隔,避免触发反爬
            if idx < len(todo):
                time.sleep(random.uniform(60, 65))
        
        self.save_progress()
        if self.all_data:
            df = pd.DataFrame(list(self.all_data.values()))
            df.to_excel(f"最终_{len(self.all_data)}条.xlsx", index=False)
            print(f"\n✅ 采集完成!共获取 {len(self.all_data)} 条不重复记录")

    def save_progress(self):
        with open(self.progress_file, "w", encoding="utf-8") as f:
            json.dump({"processed": list(self.processed)}, f)
        with open("data_backup.json", "w", encoding="utf-8") as f:
            json.dump(self.all_data, f, ensure_ascii=False)

if __name__ == "__main__":
    crawler = DataCrawler()
    crawler.crawl()

五、运行效果与数据规模

按照每个前缀间隔60秒左右的策略运行,系统可稳定工作而不触发拦截。实测效果如下:

  • 读取数千个业务编号前缀,去重后约数百个唯一前缀;

  • 每个有效前缀平均召回系统允许返回的上限条数;

  • 最终成功采集近万条不重复的企业信息记录;

  • 数据自动导出为Excel,包含机构名称、证照编号、发证日期、有效期、当前状态等完整字段。

六、总结

针对网站服务平台常见的"前端截断、无真分页"限制,直接构造分页参数往往无效。通过深入分析业务主键的编码规律,采用前缀模糊查询配合验证码本地OCR识别低频稳态请求,是实现万级数据全量获取的可靠路径。该方法同样适用于其他具有类似编号体系的企业信息查询平台。

关键词:网站数据采集、验证码识别、Python爬虫、前缀模糊查询、断点续传、数据去重、反爬策略、企业信息查询

网友评论

访客信息

你已经3分钟没有访问该网站

这些可能是你需要的内容: