Watch-IP

使用 Python 批量获取 IP 地理位置

以每批 100 个为单位,针对 POST /v1/lookup 批量补充一份 IP 列表的信息,包含行对应关系、检查点式输出以及 429 退避处理——由于官方 Python SDK 尚未覆盖该端点,因此直接通过 HTTP 调用。

开始之前

  • 一个 Watch-IP 账户,以及一个私密查询密钥——POST /v1/lookup 仅限服务器到服务器调用;切勿从浏览器调用它,也不要将该密钥打包进客户端。
  • Python 3.9 或更高版本。
  • 一份需要补充信息的本地 IP 列表——本指南从每行一个 IP 的 CSV 文件中读取这些地址。

1. 为什么本指南直接使用 HTTP 而非 SDK

watch-ip-sdk 这一官方 Python 包,只实现了 GET /v1/geo(面向浏览器的访问者端点)——它没有针对 POST /v1/lookup 的方法。本指南使用标准库中的 urllib 直接调用该 API,这与 SDK 自身零依赖的思路保持一致,而不是为此专门引入一个新的依赖。

2. 读取并校验输入列表

保留每个 IP 的原始行位置,以便将结果与其对应的原始记录进行匹配——API 会按接收 IP 时的相同顺序返回结果,但这只有在您自己的列表也保持该顺序时才有意义。

Python

import csv

with open("ips.csv", newline="") as f:
    ips = [row[0].strip() for row in csv.reader(f) if row and row[0].strip()]

print(f"Loaded {len(ips)} IPs")

3. 拆分为每批 100 个的批次

POST /v1/lookup 每次调用接受 1 至 100 个 IP,并会在查询任何内容之前校验整个批次——一个格式错误的地址会导致整个批次失败,因此如果您的输入列表可能存在脏数据,请先在本地校验 IP 格式。

Python

def chunk(items, size=100):
    for i in range(0, len(items), size):
        yield items[i : i + size]

4. 提交每个批次并检查点式保存结果

在每个批次完成时就写入结果,而不是把所有内容都保留在内存中直到最后——这样即使长列表在处理到一半时失败,也不会丢失已经成功的批次。found: false 表示该 IP 语法有效但在数据库中没有对应记录——这是正常结果,不是错误。

Python

import json
import time
import urllib.error
import urllib.request

API_URL = "https://api.watch-ip.com/v1/lookup"
SECRET_KEY = "YOUR_SECRET_KEY"

def lookup_batch(ips):
    request = urllib.request.Request(
        API_URL,
        data=json.dumps({"ips": ips}).encode(),
        headers={
            "Authorization": f"Bearer {SECRET_KEY}",
            "Content-Type": "application/json",
        },
    )
    with urllib.request.urlopen(request) as response:
        return json.load(response)["results"]

with open("results.jsonl", "a") as out:
    for batch in chunk(ips):
        try:
            results = lookup_batch(batch)
        except urllib.error.HTTPError as err:
            if err.code == 429:
                time.sleep(60)  # 在当前这一分钟窗口的剩余时间内退避等待
                results = lookup_batch(batch)
            else:
                raise
        for result in results:
            out.write(json.dumps(result) + "\n")

5. 恢复一个未完成的任务

由于结果是逐批次追加写入的,崩溃后重新启动只需跳过已经写入 results.jsonl 的 IP,而不必重新提交整个列表——重复提交同一个 IP 会被再次计费,因为批次中的每个输入都是单独计量的,而不会去重。

Python

done = set()
with open("results.jsonl") as f:
    for line in f:
        done.add(json.loads(line)["ip"])

remaining = [ip for ip in ips if ip not in done]

错误与边界情况

400 表示该批次包含超过 100 个 IP、一个空列表,或其中某处存在格式错误的地址——整个批次都会失败,而不仅仅是错误的那一条。403 表示使用了公开可用密钥而不是私密密钥。同一批次中混合使用 IPv4 与 IPv6 地址是允许的;每个地址都会被独立查询,可能为空的字段在两种情况下是一致的。

下一步

具体的响应结构请参阅批量 IP 查询参考,该端点可能返回的完整状态码列表请参阅错误与速率限制。

相关页面