如何完整抓取 Wenomad 搜索接口的全部 162 条城市数据

雨芳吖_5939

发布时间：2026-10-22 13:39:02

776人浏览过

来源于php中文网

原创

如何完整抓取 Wenomad 搜索接口的全部 162 条城市数据

本文详解如何逆向分析 wenomad 网站的 aes 加密请求参数（x/y/z），动态生成合法 payload 并分页拉取全部 elasticsearch 搜索结果，解决仅能获取约 1/3 数据的问题。

本文详解如何逆向分析 wenomad 网站的 aes 加密请求参数（x/y/z），动态生成合法 payload 并分页拉取全部 elasticsearch 搜索结果，解决仅能获取约 1/3 数据的问题。

Wenomad（wenomad.so）的城市搜索接口 /elasticsearch/search 并非普通 REST API，而是采用前端 JavaScript 动态加密后发起的受控请求。原始脚本中硬编码的 x、y、z 参数实为 AES-CBC 加密结果，且包含时间戳、随机 IV 和应用标识等动态因子——这正是你只能获取前 69 条（而非全部 162 条）的根本原因：静态 payload 仅对应单次请求，且服务端通过 from 偏移量实现分页，而原始脚本未携带分页逻辑，也未刷新加密上下文。

要完整抓取全部结果，需两步协同：

复现前端加密逻辑：解析 run.js 中 encode3 函数，用 Python 实现兼容的 AES-PBKDF2-MD5 加密流程；
构建可迭代的分页请求循环：每次请求后更新 data['from']，并持续调用新生成的动态 payload，直至响应返回 "at_end": true。

以下是完整、可运行的解决方案：

from Crypto.Cipher import AES
from Crypto.Protocol.KDF import PBKDF2
from Crypto.Hash import MD5
from Crypto.Util.Padding import pad
import base64
import json
import random
import time
import requests


def encode(key: str, iv: str, text: str, appname: str) -> str:
    """复现前端 encode3 函数：AES-CBC + PBKDF2(MD5) 密钥派生"""
    derived_key = PBKDF2(key, appname.encode(), dkLen=32, count=7, hmac_hash_module=MD5)
    derived_iv = PBKDF2(iv, appname.encode(), dkLen=16, count=7, hmac_hash_module=MD5)

    cipher = AES.new(derived_key, AES.MODE_CBC, iv=derived_iv)
    text_bytes = pad(text.encode(), AES.block_size)
    encrypted_text = cipher.encrypt(text_bytes)
    return base64.b64encode(encrypted_text).decode()


def generate_payload(data: dict) -> dict:
    """生成含时间戳、随机 IV 和版本号的动态加密 payload"""
    v = "1"
    appname = 'fie'
    cur_timestamp = str(int(time.time() * 1000))
    timestamp_version = f'{cur_timestamp}_{v}'
    key = appname + cur_timestamp
    iv = str(random.random())

    text = json.dumps(data, separators=(',', ':'))
    return {
        'z': encode(key, iv, text, appname),
        'y': encode(appname, "po9", timestamp_version, appname),
        'x': encode(appname, "fl1", iv, appname)
    }


def fetch_all_search_results(data: dict) -> list:
    """自动分页拉取全部 hits，支持断点续传语义"""
    headers = {
        'x-requested-with': 'XMLHttpRequest',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36',
        'accept': 'application/json, text/javascript, */*; q=0.01',
        'origin': 'https://wenomad.so',
        'referer': 'https://wenomad.so/'
    }

    results = []
    while True:
        payload = generate_payload(data)
        try:
            response = requests.post(
                'https://wenomad.so/elasticsearch/search',
                headers=headers,
                json=payload,
                timeout=15
            )
            response.raise_for_status()
            res_json = response.json()

            hits = res_json.get('hits', {}).get('hits', [])
            results.extend(hits)
            data['from'] += len(hits)

            print(f"Fetched {len(hits)} items → total: {len(results)}")

            if res_json.get('at_end', False):
                break

            # 防爬友好：避免高频请求
            time.sleep(0.3)

        except (requests.RequestException, json.JSONDecodeError, KeyError) as e:
            print(f"Request failed: {e}")
            break

    return results


# 初始化搜索参数（与前端一致）
data_template = {
    "appname": "fie",
    "app_version": "live",
    "type": "custom.town",
    "constraints": [{
        "key": "active_boolean",
        "value": True,
        "constraint_type": "equals"
    }],
    "sorts_list": [
        {"sort_field": "overall_rating_number", "descending": True},
        {"sort_field": "overall_rating_number", "descending": True}
    ],
    "from": 0,
    "n": 9999,  # 单次最大条数（实际受服务端限制，但设高便于收敛）
    "search_path": "{\"constructor_name\":\"DataSource\",\"args\":[{\"type\":\"json\",\"value\":\"%p3.AAV.%el.cmQus.%el.cmSJO0.%p.%ds\"},{\"type\":\"node\",\"value\":{\"constructor_name\":\"Element\",\"args\":[{\"type\":\"json\",\"value\":\"%p3.AAV.%el.cmQus.%el.cmSJO0\"}]}},{\"type\":\"raw\",\"value\":\"Search\"}]}",
    "situation": "unknown"
}

# 执行全量抓取
all_hits = fetch_all_search_results(data_template.copy())
print(f"\n✅ Total fetched: {len(all_hits)} items")

# 解析并打印结构化结果（示例）
inner_link = 'https://wenomad.so/city/{}'
for hit in all_hits[:5]:  # 仅展示前5条
    src = hit['_source']
    print((
        src.get('town_text', 'N/A'),
        src.get('continent__text__text', 'N/A'),
        src.get('country__text__text', 'N/A'),
        inner_link.format(src.get('Slug', ''))
    ))

⚠️ 关键注意事项：