Algolia API逆向实战：四大难点搞定展会数据爬虫

1. 项目背景与整体架构设计

上个月接了个活儿：抓取泰国InterPlas橡塑展的展商名录。刚接到手时我以为就是个普通爬虫项目，无非是列表页翻页、详情页解析、存库里。但打开展会官网的展商目录看了一眼，我立刻意识到这事没这么简单——整个展商搜索功能跑在Algolia这个第三方托管搜索服务上，页面里的展商列表全是异步加载的，而且站点同时维护着英文、泰文、中文三套语言版本，国家字段在语言切换后完全不是一回事。项目做到一半又接连撞上重复展商和批量入库失败的问题，最后整理下来，核心技术难点正好就是标题里写的四个：Algolia API逆向、多语言国家过滤、重复数据合并、批量插入回滚。

相关服务：泰国服务器

这篇内容适合三类人看：一是准备爬各类展会官网、行业协会会员名录的朋友，这类站点用Algolia做搜索的比例比我预想的高得多；二是做数据采集中间件、需要考虑多语言数据清洗和去重的开发者；三是在写批量写入数据库但总担心写坏数据的后端同学。我会把每一关的思考过程、关键代码、踩过的坑都写出来，尽量做到拿过去就能改改直接用。

先交代一下项目背景。InterPlas是泰国规模比较大的橡塑行业展会，每年在曼谷BITEC举办，参展商以机械设备、模具、原材料、化工助剂为主，展商数量大概在七八百家。我需要抓的核心字段是：展商英文名、泰文名、所属国家/地区、展位号、产品分类、一句话简介、官网地址。这些数据拿下来之后要进MySQL，后续做市场分析和客户画像用。

整体架构我设计得比较朴素，没有引入Scrapy这种重型框架，直接用的requests + pymysql，中间加了一层本地JSON落地缓存。原因也很简单：这个项目的数据量不大，七八百条记录，没必要上分布式；但网络请求和解析逻辑比较绕，需要反复调试，所以原始响应一定要先存本地，避免每次调试都重新打一遍Algolia接口，给自己留出容错空间。

项目根目录/
├── crawler/
│   ├── algolia_client.py   # Algolia请求封装
│   ├── country_map.py      # 多语言国家映射表
│   ├── dedup.py            # 重复数据识别与合并
│   ├── db_writer.py        # 批量入库与事务回滚
│   └── main.py             # 主流程编排
├── raw_data/               # 原始JSON响应缓存
├── logs/                   # 运行日志
└── requirements.txt

这个结构回头看不算最优，但胜在直观。四个核心模块刚好对应标题里的四个难点，出问题的时候能直接定位到文件，不用在整个工程里翻来翻去。

1.1 为什么展会官网会用Algolia

先说一个判断：你打开一个展会官网，看到展商目录页有搜索框、有按国家筛选、有按产品分类筛选，而且操作起来响应特别快、几乎无刷新，那这个站点的搜索功能大概率不是自己写的，而是接的Algolia这类托管搜索服务。原因也很现实，展会官网通常是非技术团队维护的，自己搞一套搜索引擎的成本太高，托管服务只需要前端接一个JavaScript库，把数据同步过去就能提供带错字容忍、多维度筛选的搜索体验。

Algolia在会展行业的使用率相当高，除了InterPlas，我还见过不少海外展会、行业门户都在用。所以这次逆向的经验是可以复用的——理解了一个，基本就理解了同类站点的数据获取方式。

1.2 四条技术主线的拆解逻辑

开工之前，我把整个项目的风险点列了个清单：第一，前端是异步加载的，数据接口藏在XHR请求里，需要先找到它；第二，数据包含大量非英文文本，国家字段在语言版本之间对不上；第三，展商信息在数据源里就存在重复，不做清洗会导致分析结果失真；第四，数据入库时要保证不会出现“写了一半、断在中间”的情况。

这四个问题并不是独立的，它们之间有依赖关系：先要把Algolia接口调通，拿到原始数据；然后做多语言归一化，把国家字段洗干净；再去做重复识别和合并；最后才是设计可靠的数据入库流程。下面的章节就按这个顺序展开。

2. Algolia API逆向：把网站自己的接口变成数据源

2.1 从浏览器Network里揪出完整请求

逆向的第一步永远不是看代码，而是打开开发者工具。我用Chrome的F12进入Network面板，勾选XHR过滤，然后在展会官网的展商目录页里随便搜了一个词，比如"injection"，紧接着就看到一个发往 xxx-dsn.algolia.net 的POST请求，路径是 /1/indexes/exhibitor_prod/query 。

点开这个请求，Headers里有几个关键信息：

X-Algolia-Application-Id: XXXX
X-Algolia-API-Key: xxxxxxxxxxxxxxxx
Content-Type: application/json

这两个 X-Algolia- 开头的头就是访问Algolia接口的凭证。每个Algolia应用有一个Application ID（应用标识），API Key则分很多种权限，前端页面里嵌入的一般是Search-Only Key（仅搜索权限），它只能查数据，不能增删改，所以从安全角度讲，用它拉取公开数据属于正常使用，和用浏览器打开页面拿数据没有本质区别。

RequestBody长这样，Algolia的老版接口把参数都塞进一个 params 字符串里：

{
  "params": "query=injection&hitsPerPage=50&page=0&facets=%5B%22country%22%5D&facetFilters="
}

返回的JSON结构也很规整，核心字段是 hits 数组和 nbHits 总数：

{
  "hits": [
    {
      "name": "Yong Ming Machinery",
      "name_th": "หย่งหมิงแมชชีนเนอรี่",
      "country": "China",
      "booth": "B101",
      "category": ["Injection Molding Machine"],
      "website": "http://example.com",
      "description": "..."
    }
  ],
  "nbHits": 812,
  "page": 0,
  "hitsPerPage": 50
}

到这里，数据源已经找到了，剩下的就是把这个"网页内搜索"逻辑搬出来，自己用requests循环分页把全部数据拉下来。

2.2 翻页参数与1000条上限的坑

拿到接口之后，我第一个想法很简单：把 hitsPerPage 调大，然后翻页。Algolia单次查询最多返回1000条记录， hitsPerPage 最大也是1000。InterPlas展商总数是812家，小于1000，所以我理论上只需要一次查询就能拿全。但这里有个隐患：如果你遇到一个大型展会、展商数超过1000，直接设 hitsPerPage=1000 翻页到第二页，Algolia会返回空结果或者在某个阈值处截断，因为免费额度下搜索API默认只返回前1000个匹配。

解决这个问题的常规做法是分段查询：用 facetFilters 按国家、或按展商名称首字母A-Z分段，把总数据集拆成若干个子集，每个子集都小于1000条，再分别翻页合并。我这次的量级用不上，但这属于通用经验，提前写进代码里是对的。

我封装的那个 algolia_client.py 核心逻辑长这样：

import requests
import json
import time

class AlgoliaClient:
    def __init__(self, app_id, search_key, index_name):
        self.url = f"https://{app_id}-dsn.algolia.net/1/indexes/{index_name}/query"
        self.headers = {
            "X-Algolia-Application-Id": app_id,
            "X-Algolia-API-Key": search_key,
            "Content-Type": "application/json",
        }

    def fetch_all(self, query="", hits_per_page=100, attributes=None):
        all_hits = []
        page = 0
        while True:
            params = {
                "query": query,
                "hitsPerPage": hits_per_page,
                "page": page,
            }
            if attributes:
                params["attributesToRetrieve"] = attributes

            payload = {"params": json.dumps(params, ensure_ascii=False)}
            resp = requests.post(self.url, headers=self.headers, json=payload)
            resp.raise_for_status()
            data = resp.json()

            hits = data.get("hits", [])
            all_hits.extend(hits)
            print(f"page={page}, got={len(hits)}, total={data.get('nbHits')}")

            # 终止条件：当前页没拿满，或者page已经到末尾
            if len(hits) < hits_per_page or page >= data.get("nbPages", 0) - 1:
                break
            page += 1
            # 非激进爬取，每次请求间隔半秒
            time.sleep(0.5)

        return all_hits

两个细节值得说明。第一， query 传空字符串时，Algolia会返回索引里的全部记录，这正好是我们想要的"全量导出"。第二，请求间隔我固定了0.5秒，数据量小无所谓，但养成习惯是好的，避免给别人的服务造成压力。

2.3 分析返回字段与索引结构

拿到第一批数据后，我先做了个字段盘点，而不是急着写入库逻辑。这一步很重要，因为Algolia索引里的字段名、字段类型，直接决定了后面解析和清洗怎么写。

我的做法是把JSON原样保存到本地，然后用一段小脚本统计每个字段的覆盖率：

import json
from collections import Counter

with open("raw_data/exhibitors.json") as f:
    hits = json.load(f)

field_count = Counter()
for hit in hits:
    for key in hit.keys():
        field_count[key] += 1

for field, count in field_count.most_common():
    print(f"{field}: {count}/{len(hits)} ({count/len(hits)*100:.1f}%)")

跑完之后，我注意到几个关键数据质量信号： name 字段覆盖率100%，但 name_th 只有60%左右，说明不是每家展商都提供了泰文名； country 字段覆盖率高但有脏值；最重要的是，同样的 name 在索引里出现了多次，而且 objectID 还不一样，这就是后面重复数据合并要解决的问题。

2.4 关于Key权限边界的安全提醒

这里必须多说一句。前端暴露的Search-Only Key是网站主动公开的，它只能做搜索和读取，所以拿它抓公开的展商信息没有越权问题。但如果你发现 X-Algolia-API-Key 是一个高权限的Admin Key，千万不要顺手拿去调删除或写索引的接口——那属于越权访问，性质就变了。正确做法是继续使用站点的公开接口，或者联系主办方获取数据授权。

另外，Algolia官方对请求频率有约束，短时间大量请求会触发429限流。我的经验是每次请求之间至少间隔300到500毫秒，加上失败重试和指数退避，基本不会触发限制。爬虫这个行当，心态摆正比技术重要：拿自己该拿的公开数据，控制频率，不添乱，才能长久。

3. 多语言国家过滤：别让"中国"和"China"分成两拨

3.1 国家字段的真实脏数据形态

Algolia数据调通之后，我原本以为国家过滤就是个简单的 if country == "China" 判断。但实际跑了一遍统计，发现country字段的取值多到让人头疼：

原始值	实际指向	所在语言
China	中国	英文
CHINA	中国	英文大写
P.R. China	中国	英文全称
中国	中国	中文
PRC	中国	英文缩写
THAILAND	泰国	英文大写
泰国	泰国	泰文
Deutschland	德国	德文
Germany	德国	英文
日本	日本	中文
Taiwan	中国台湾	英文

问题比想象中复杂：同一国家，英文大小写、缩写、中文名、本地语言名、历史习惯名会在不同展商记录里交替出现。如果直接拿原始字符串做分组统计，德国会被拆成"Germany"和"Deutschland"两组，中国会被拆成五六组，后面做市场分析就完全没法看。

3.2 归一化映射表设计

我的解法是建一张国家别名映射表，把所有可能串门的写法统一到ISO 3166-1的两位国家代码上，这个代码是国际标准，后续做图表、对比都方便。

# country_map.py
ALIAS_TO_ISO = {
    # 中国
    "china": "CN", "p.r. china": "CN", "prc": "CN",
    "chinese mainland": "CN", "mainland china": "CN",
    "中国": "CN", "中国大陆": "CN", "内地": "CN",
    # 泰国
    "thailand": "TH", "thai": "TH", "thailandia": "TH",
    "泰国": "TH", "ไทย": "TH",
    # 日本
    "japan": "JP", "nippon": "JP", "nihon": "JP",
    "日本": "JP", "ญี่ปุ่น": "JP",
    # 德国
    "germany": "DE", "deutschland": "DE", "german": "DE",
    "德国": "DE", "เยอรมนี": "DE",
    # 意大利
    "italy": "IT", "italia": "IT", "italian": "IT",
    "意大利": "IT", "อิตาลี": "IT",
}

def normalize_country(raw: str) -> str:
    """把各种写法的国家名映射成ISO两位代码，映射不到返回原值"""
    if not raw:
        return ""
    key = raw.strip().lower()
    if key in ALIAS_TO_ISO:
        return ALIAS_TO_ISO[key]
    # 兜底：本身就是ISO代码的情况
    if key.upper() in {"CN", "TH", "JP", "DE", "IT", "SG", "MY", "VN", "IN", "TW", "KR", "US", "UK"}:
        return key.upper()
    return raw.strip()

这里有个处理顺序的细节：先 lower() 再做映射，可以同时处理大写和小写；把泰文、中文别名也放进来，是因为这个站点有泰文数据源， ไทย 就是泰语里的"泰国"。

映射不到的保留原值，不要强行置空——保留现场数据永远比丢数据安全，等统计时再看哪些值没覆盖到，再补充映射表。

3.3 双层过滤策略：查询期粗筛加落地后精筛

国家过滤我做了两层，这两层承担的任务完全不同。

第一层是查询期粗筛。Algolia索引里本身有 country 这个可过滤字段，前端之所以能做"按国家筛选"，就是因为它在请求里带了 facetFilters 。我在拉数据时可以按国家分段抓取，比如只抓 facetFilters=[["country:China"]] ，减少单次返回量。但这一层的价值在于分段，不在于清洗——因为索引里的country值本身就是脏的，"China"、"P.R.China"、"中国"在Algolia看来是不同的facet值，分段会漏数据。

第二层才是落地后精筛。把全量原始数据拉到本地之后，对所有记录统一过一遍 normalize_country() ，生成一个新的 country_iso 字段。后续所有分组、统计、筛选全部基于这个新字段，不再碰原始值。

原始 country → normalize_country() → country_iso
"China"  → "CN"
"PRC"    → "CN"
"中国"   → "CN"
"Deutschland" → "DE"

这样处理完，国家分布统计就是干净的了。

3.4 编码与乱码处理

多语言数据绕不开编码问题。这个项目里的泰文文本是正常的UTF-8编码，Python的json模块处理没有问题。但有一个容易翻车的点：泰文里没有空格分词，字符串处理时按空格切分会导致一个完整词被拆开，好在本次我不需要对泰文做分词，只保留原值，所以影响不大。

真正需要盯紧的是数据库写入环节。MySQL连接时如果不显式指定 charset="utf8mb4" ，泰文和中文在入库时会变成乱码。还有，如果表结构是 utf8 而不是 utf8mb4 ，遇到某些特殊emoji或四字节字符会直接报错。我在建表时统一用了 utf8mb4 ，这是处理多语言数据的基本配置，后面在入库章节还会再提。

4. 重复数据合并：同一家展商别出现三次

4.1 重复数据的三个来源

清洗完国家字段后，我开始统计展商的唯一性，结果发现一个大问题：去重前的记录数是812，但按 name 去重后只剩711，有上百条重复记录。这些重复是怎么来的？我在数据里找到了三处来源。

第一，语言版本造成的"一人两名"。同一家泰国本地展商，英文名是"S.B. Plastics"，泰文名是"เอส.บี. พลาสติก"，Algolia索引里给这两条记录分别建了 objectID ，看起来就是两条完全不同的数据。

第二，企业法律后缀变体。同一家中国展商，有的记录叫"Guangdong Kaisen Plastics Co., Ltd."，有的叫"Guangdong Kaisen Plastics"，有的叫"Guangdong Kaisen Plastics Co.",公司名主体相同，后缀不同。

第三，官方数据源本身就脏。展会主办方在录入时可能同时从不同表单渠道导入了数据，导致同一展商被重复录入，甚至展位号都不同——这种情况最常见于大型展团。

4.2 归一化与相似度计算

既然重复来源不同，我的去重策略也分两步走。

第一步是精确归一化，用规则把"看起来一样"的名字统一成一个key：

import re

def normalize_company_name(name: str) -> str:
    """企业名归一化：去后缀、去符号、统一大小写"""
    if not name:
        return ""
    s = name.strip().lower()
    # 去掉常见法律后缀
    for suffix in [
        r"co\.?,?\s*ltd\.?$", r"ltd\.?$", r"limited$",
        r"inc\.?$", r"corporation$", r"corp\.?$",
        r"gmbh$", r"ag$", r"pte\.?\s*ltd\.?$",
        r"public company limited$", r"plc$",
        r"company limited$", r"company$",
    ]:
        s = re.sub(suffix, "", s, flags=re.IGNORECASE)
    # 去标点、去空格，中文也保留
    s = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", "", s)
    return s

归一化之后，同一公司所有后缀变体都会落到同一个key上，精确去重率非常高。

第二步是模糊匹配，针对"主体名有细微差异"的情况。比如"Yong Ming Machinery"和"Yongming Machinery"，归一化后仍然不同，但显然是同一家。我用的是Python标准库 difflib 的 SequenceMatcher ，算相似度，阈值定在0.92以上视为同一公司：

from difflib import SequenceMatcher

def is_similar(a: str, b: str, threshold: float = 0.92) -> bool:
    a2 = normalize_company_name(a)
    b2 = normalize_company_name(b)
    if not a2 or not b2:
        return False
    if a2 == b2:
        return True
    return SequenceMatcher(None, a2, b2).ratio() >= threshold

阈值0.92是我测试后的取值。调低了会把"Ming Thai Machinery"和"Ming Thai Trading"误判成一家，调高了又漏掉一些真实重复。0.92在当前数据集上精准度和召回率都还过得去，但这不是万能值，换数据集后要重新调。

4.3 合并策略：主记录加补全

识别出重复组之后，下一步是合并。我的原则很简单：每条记录保留身份字段最完整的作为主记录，其余记录的缺失字段用来补全，多出来的展位号、产品分类合并进列表。

def merge_group(group: list) -> dict:
    """把一组重复记录合并成一条完整记录"""
    # 按字段非空数量排序，最完整的作为主记录
    group.sort(key=lambda r: sum(1 for v in r.values() if v), reverse=True)
    merged = dict(group[0])

    # 补全缺失字段
    for field in ["website", "email", "description", "name_th"]:
        if not merged.get(field):
            for r in group[1:]:
                if r.get(field):
                    merged[field] = r[field]
                    break

    # 合并展位号和产品分类
    booths = set()
    categories = set()
    for r in group:
        if r.get("booth"):
            booths.add(r["booth"])
        for c in (r.get("category") or []):
            categories.add(c)
    merged["booth_list"] = sorted(booths)
    merged["category_list"] = sorted(categories)
    return merged

合并后的输出结构比原始记录多了一个 booth_list 和 category_list ，这样既保住了"这家公司占两个展位"的客观事实，又不会因为展位号不同而把它们拆成两条。

4.4 人工复核清单

自动去重不可能100%正确，所以在入库前，我加了一道人工复核环节，把不确定的记录导出成CSV检查。

我导出的是三类记录：一是模糊匹配相似度在0.85到0.92之间、被我判为"不重复"的边界情况；二是展位号相同但公司名不同的记录；三是公司名相同但国家字段不同的记录（这种要小心，可能是两家重名公司，也可能是一家公司在两地注册主体）。人工复核不用全查，抽这几类高风险的就够了。

# 生成复核清单
candidates = []
for group in groups:
    if len(group) < 2:
        continue
    countries = {normalize_country(r.get("country", "")) for r in group}
    if len(countries) > 1:
        candidates.append({"type": "country_mismatch", "records": group})
    # 其他类型同理...

这一步花不了多少时间，但能避免把两家名字相近的独立公司硬合成一家，导致后续分析数据失真。我个人的经验是：宁可漏掉一次合并，也不要错误合并——错误合并产生的数据污染，比几条重复记录难清理得多。

5. 批量插入回滚：让数据库永远不留半截数据

5.1 为什么必须用事务

数据清洗工作做完，拿到的是约700条干净记录。这时候就到了写入环节。一般的想法是循环 INSERT INTO 一条条插，或者拼一个大SQL批量插。但这里有个风险：如果第300条记录里有一个超长字段、或者违反唯一键约束，插入会中断，前面299条已经写进去了，后面400条还没写——数据库里留下一堆"半截数据"，而且你很难判断哪些入库了、哪些没有。

解决这类问题的标准手段是数据库事务。把一批插入操作包在一个事务里，要么全部成功提交，要么全部回滚，绝不留中间状态。MySQL的InnoDB引擎原生支持事务，Python的 pymysql 库也提供了 begin/commit/rollback 接口，实现成本很低，收益却非常大。

5.2 分批插入与事务边界设计

分批策略需要考虑一个平衡：批次太大，单次事务执行时间长，锁竞争加剧，出问题回滚的代价也大；批次太小，事务开销占比高，插入速度慢。我这边实测下来，200条一批是比较舒服的，700条数据分4批就能入库，每批事务执行时间都在秒级，完全可以接受。

import pymysql
import logging

logger = logging.getLogger("db")

def batch_insert(rows, batch_size=200):
    """分批插入，每批一个事务，失败自动回滚并记录"""
    conn = pymysql.connect(
        host=DB_HOST,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME,
        charset="utf8mb4",
    )
    inserted_total = 0
    for i in range(0, len(rows), batch_size):
        batch = rows[i:i + batch_size]
        try:
            conn.begin()
            with conn.cursor() as cursor:
                sql = """
                    INSERT INTO exhibitors
                    (name, name_th, country_iso, booth_list, category_list,
                     website, description, raw_data)
                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                """
                cursor.executemany(sql, [
                    (
                        r["name"],
                        r.get("name_th", ""),
                        r.get("country_iso", ""),
                        ",".join(r.get("booth_list", [])),
                        ",".join(r.get("category_list", [])),
                        r.get("website", ""),
                        r.get("description", ""),
                        json.dumps(r, ensure_ascii=False),
                    )
                    for r in batch
                ])
                # 校验实际影响行数，防止部分记录被静默跳过
                if cursor.rowcount != len(batch):
                    raise Exception(f"rowcount mismatch: {cursor.rowcount} != {len(batch)}")
            conn.commit()
            inserted_total += len(batch)
            logger.info(f"batch {i//batch_size}: insert {len(batch)} rows OK")
        except Exception as e:
            conn.rollback()
            logger.error(f"batch {i//batch_size} failed: {e}")
            # 逐条插入，定位是哪条记录引发的问题
            for r in batch:
                try:
                    conn.begin()
                    with conn.cursor() as cursor:
                        cursor.execute(sql, (...))
                    conn.commit()
                except Exception as e2:
                    conn.rollback()
                    logger.error(f"bad row: {r.get('name')}, error: {e2}")
                    # 这里把坏行单独落盘，方便后续修复
                    with open("logs/bad_rows.jsonl", "a", encoding="utf-8") as f:
                        f.write(json.dumps(r, ensure_ascii=False) + "\n")
            raise
        finally:
            # 事务状态复位
            pass
    conn.close()
    return inserted_total

这段代码里最关键的设计，是"失败后自动降级为逐条插入定位"：当整个批次回滚后，不直接放弃，而是把批次拆开逐条尝试，定位出真正有问题的行，把好行先入库，坏行单独写到 bad_rows.jsonl 文件里。这样既保住了大部分数据，又能精确锁定脏数据，而不是因为一条坏记录反复重跑全量。

5.3 幂等设计：唯一键与UPSERT

事务解决的是"半截数据"问题，但还有一个更隐蔽的问题：脚本如果中途崩溃，重启后重新跑，会不会插入大量重复数据？

解决办法是给表加唯一键，让插入操作变成幂等的。我用的是 (name_normalized) 作为唯一键，这是归一化后的公司名，相当于"一个公司只允许入库一次"：

CREATE TABLE IF NOT EXISTS exhibitors (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    name_th VARCHAR(255) DEFAULT '',
    country_iso CHAR(2) DEFAULT '',
    booth_list VARCHAR(500) DEFAULT '',
    category_list VARCHAR(500) DEFAULT '',
    website VARCHAR(500) DEFAULT '',
    description TEXT,
    raw_data JSON,
    name_normalized VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_name_normalized (name_normalized)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

写入SQL也改成 INSERT ... ON DUPLICATE KEY UPDATE ，这样重复执行同一批数据时，已存在的记录会被更新而不是报错：

INSERT INTO exhibitors
    (name, name_th, country_iso, booth_list, category_list, website, description, raw_data, name_normalized)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
    booth_list = VALUES(booth_list),
    category_list = VALUES(category_list),
    website = VALUES(website),
    updated_at = CURRENT_TIMESTAMP;

幂等设计是数据管道的生命线。你永远不知道脚本会在第几步被打断（断电、网络断开、服务器重启），只有保证"重跑无害"，才能安心地把定时任务挂起来。

5.4 回滚场景实战与断点续跑

项目实际运行中，我确实遇到过一次回滚。有一批数据里混进了一条description超长的记录——原始数据是泰文长文本，超过了 TEXT 类型64KB的限制（虽然通常不会，但那次确实有一家展商把整份公司手册的简介都塞进去了），入库时报错。

事务回滚后， batch_insert 的降级逻辑自动逐条定位，把这条坏记录单独隔离到 bad_rows.jsonl 里，其余199条正常入库。我拿到坏记录文件后，手动把description截断到合理长度，再补跑一条插入，整个过程不到5分钟，数据管道没有断掉。

还有一个容易被忽略的点：事务里如果查询时间过长，可能触发 innodb_lock_wait_timeout 。我这次是单进程写入，没有并发写冲突，但如果你的方案里有多进程并行写库，一定要对写入做串行化处理，或者用 SELECT ... FOR UPDATE 控制锁，否则会遇到死锁。

6. 踩坑实录与排查速查表

6.1 Algolia 401与Key相关的坑

逆向阶段最容易碰到的问题是401。我调试时有一次从浏览器里复制的 X-Algolia-API-Key 是对的，但因为 X-Algolia-Application-Id 写错了，直接报401。Algolia的鉴权是两个头配合使用的，只对一个是过不了的。另外有些站点会走代理转发，请求URL不是 algolia.net 而是站点自己的域名下的 /api/algolia ，这种情况需要先找清楚真正发出去的请求是哪种。

现象	可能原因	处理办法
401 Unauthorized	App ID或API Key错误	核对浏览器Network里的两个X-Algolia头是否完全一致
401 且提示incorrect api key	Key对但对应的应用不对	确认这个Key是从哪个域名下请求里复制的
403 Forbidden	当前IP被限制，或Key权限不足	检查是否用高权限Key调了越权接口；控制请求频率
429 Too Many Requests	触发限流	增加请求间隔，加指数退避重试
返回空hits但nbHits>0	query 参数被错误传值	用空字符串query全量拉取，或用facetFilters分段

6.2 请求频率与重试

Algolia限流是按应用维度的，如果这个App ID同时被官网自己的用户在用，你的高频请求可能会影响到线上搜索，反过来也会被限流处置。我这边把间隔设到0.5秒，全量800条数据几分钟就拉完了，完全够用。

重试逻辑我写成了带指数退避的版本：第一次失败等1秒，第二次2秒，第三次4秒，最多重试5次。注意重试只对网络错误和429有效，401这种鉴权错误重试多少次都一样，应该直接报错暴露出来，别让脚本静默重试掩盖真问题。

import time
import requests

def request_with_retry(url, headers, payload, max_retries=5):
    for attempt in range(max_retries):
        try:
            resp = requests.post(url, headers=headers, json=payload, timeout=30)
            if resp.status_code == 429:
                time.sleep(2 ** attempt)
                continue
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.Timeout:
            time.sleep(2 ** attempt)
        except requests.exceptions.ConnectionError:
            time.sleep(2 ** attempt)
    raise RuntimeError(f"failed after {max_retries} attempts, url={url}")

6.3 多语言入库编码问题

入库编码问题我前面提过，这里再补充一个细节： pymysql 连接时指定 charset="utf8mb4" ，但 pymysql.connect 有一个坑， read_timeout 和 write_timeout 在MySQL 8.0下默认是空的，个别环境会出现连接长时间挂起。我习惯显式设置 connect_timeout=10, read_timeout=30, write_timeout=30 ，避免程序在网络异常时卡死。

另外， raw_data 字段我用了 JSON 类型，MySQL 5.7以上才支持。如果你的环境是MySQL 5.6，要用 TEXT 类型存储JSON字符串，或者在应用层序列化后写入文本字段。

6.4 断点续跑的兜底方案

最后一个让我夜里睡安稳的设计，是"原始响应先落盘、解析入库后置"。整个流程分两步：第一步，跑通Algolia接口后，把全量原始JSON先存到 raw_data/ 目录；第二步，再从本地JSON文件读数据、清洗、入库。这样即使解析逻辑写错了、数据库崩了，只需要重新跑第二步，不用再去打Algolia接口。网络请求是不可控的，本地文件是可控的，把不可控的部分和可控的部分拆开，维护成本立刻降下来。

日志也要记全。我每个阶段都打了日志：拉取多少条、清洗后多少条、去重后多少条、入库多少条、坏行多少条。这个"数量漏斗"看起来简单，但在排查问题时能立刻定位数据是在哪个环节丢的。

7. 项目复盘与个人经验

整个项目跑下来，我最真实的体会是：爬虫项目的技术难点从来不是"写代码"，而是"理解数据"。Algolia逆向看起来是技术活，实际只是找到网站自己用的公开接口；多语言国家过滤看起来麻烦，动手建一张映射表就能解决；重复数据合并和事务回滚，也都不是什么高深算法，关键在于你愿不愿意在设计阶段多想一层。

有几个习惯是这次踩坑之后固定下来的。第一，任何外部接口的数据，先原样落盘再处理，绝不边拉边算。第二，入库逻辑必须幂等，重跑一万次和跑一次结果一样，数据库状态才可控。第三，给自己留"退出通道"——事务回滚、坏行隔离、断点续跑，这些都是在出问题时救命的机制，平时看着多余，真出事的时候就知道值了。

如果后续要把这个项目扩展成通用能力，我会把Algolia逆向部分抽成通用工具，因为海外展会、行业门户、电商站用Algolia的实在太多了，写一个支持任意App ID和Index Name的"Algolia全量数据导出器"，下次遇到类似站点就能十分钟接入。国内展会数据源复杂度更高一些，尤其是登录态、验证码、动态Token这些，那属于另一个量级的工程，等真遇到了再聊。