Geolocalización de IP en lote con Python

Antes de empezar

• Una cuenta de Watch-IP y una clave secreta de lookup: POST /v1/lookup es exclusivamente de servidor a servidor; nunca la llames desde un navegador ni la incluyas en un cliente.
• Python 3.9 o superior.
• Una lista local de IP para enriquecer: esta guía las lee desde un CSV con una IP por línea.

1. Por qué esto usa HTTP directo en lugar del SDK

watch-ip-sdk, el paquete oficial de Python, solo implementa GET /v1/geo, el endpoint de visitante orientado al navegador; no tiene ningún método para POST /v1/lookup. Esta guía llama a la API directamente con urllib de la biblioteca estándar, siguiendo el mismo enfoque sin dependencias del propio SDK en lugar de añadir una nueva solo para esto.

2. Lee y valida la lista de entrada

Conserva la posición de fila original de cada IP para que un resultado pueda asociarse de vuelta con el registro del que provino; la API devuelve los resultados en el mismo orden en que recibió las IP, lo cual solo ayuda si tu propia lista mantiene ese mismo orden.

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"Se cargaron {len(ips)} IP")

3. Divide en lotes de 100

POST /v1/lookup acepta de 1 a 100 IP por llamada y valida todo el lote antes de consultar nada: una dirección malformada hace fallar todo el fragmento, así que valida la sintaxis de las IP localmente primero si tu lista de entrada podría estar sucia.

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

4. Envía cada lote y registra los resultados con puntos de control

Escribe los resultados a medida que cada lote se completa en lugar de mantener todo en memoria hasta el final, para que una falla a mitad de una lista larga no pierda los lotes que ya tuvieron éxito. found: false significa que la IP es sintácticamente válida pero no tiene ninguna entrada en la base de datos: un resultado normal, no un error.

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)  # esperar a que termine la ventana de un minuto actual
                results = lookup_batch(batch)
            else:
                raise
        for result in results:
            out.write(json.dumps(result) + "\n")

5. Reanudar un trabajo parcial

Como los resultados se añaden un lote a la vez, reiniciar después de una falla solo necesita omitir las IP ya escritas en results.jsonl en lugar de volver a enviar toda la lista; volver a enviar una IP duplicada se factura de nuevo, ya que cada entrada de un lote se mide individualmente en lugar de deduplicarse.

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]

Errores y casos límite

Un 400 significa que el lote tenía más de 100 IP, una lista vacía, o una dirección malformada en algún punto; todo el fragmento falla, no solo la entrada incorrecta. Un 403 significa que se usó una clave publicable en lugar de una clave secreta. Mezclar direcciones IPv4 e IPv6 en el mismo lote no supone ningún problema; cada una se consulta de forma independiente, con los mismos campos anulables en ambos casos.

Siguiente paso

Consulta la referencia de bulk IP lookup para conocer el esquema exacto de la respuesta, y errores y límites de tasa para la lista completa de códigos de estado que puede devolver este endpoint.