تحديد الموقع الجغرافي لعناوين IP دفعةً واحدة باستخدام Python

قبل أن تبدأ

• حساب Watch-IP ومفتاح بحث سري — فـ POST /v1/lookup مخصص للاتصال من خادم إلى خادم فقط؛ لا تستدعِه أبدًا من متصفح ولا تُضمِّن المفتاح في أي عميل (client).
• Python 3.9 أو أحدث.
• قائمة محلية بعناوين IP لإثرائها — يقرأها هذا الدليل من ملف CSV يحتوي على عنوان IP واحد في كل سطر.

1. لماذا يستخدم هذا الدليل HTTP مباشرة بدلًا من الحزمة (SDK)

لا تُنفِّذ watch-ip-sdk، وهي الحزمة الرسمية للغة Python، سوى GET /v1/geo، نقطة نهاية الزوار الموجَّهة للمتصفح — وليس لديها أي دالة لـ POST /v1/lookup. يستدعي هذا الدليل الواجهة البرمجية مباشرة باستخدام urllib من المكتبة القياسية، بما يتوافق مع نهج الحزمة نفسها الخالي من التبعيات بدلًا من إضافة تبعية جديدة لهذا الغرض فقط.

2. قراءة قائمة المُدخلات والتحقق منها

احتفظ بموضع كل عنوان IP الأصلي في الصف حتى يمكن مطابقة النتيجة مرة أخرى بالسجل الذي جاءت منه — تُعيد الواجهة البرمجية النتائج بالترتيب نفسه الذي استلمت به عناوين IP، وهذا لا يُفيد إلا إذا بقيت قائمتك الخاصة بالترتيب ذاته.

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"تم تحميل {len(ips)} عنوان IP")

3. التقسيم إلى دفعات من 100

يقبل POST /v1/lookup من 1 إلى 100 عنوان IP في كل استدعاء، ويتحقق من الدفعة بأكملها قبل البحث عن أي شيء — عنوان واحد غير صحيح التنسيق يُفشل الدفعة بأكملها، لذا تحقق من صيغة عناوين IP محليًا أولًا إذا كانت قائمة المُدخلات لديك قد تحتوي على أخطاء.

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

4. إرسال كل دفعة وتسجيل نقطة تفتيش للنتائج

اكتب النتائج فور اكتمال كل دفعة بدلًا من الاحتفاظ بكل شيء في الذاكرة حتى النهاية، حتى لا يؤدي فشل في منتصف قائمة طويلة إلى فقدان الدفعات التي نجحت بالفعل. تعني found: false أن عنوان IP صحيح من حيث الصياغة لكن لا يوجد له سجل في قاعدة البيانات — وهي نتيجة طبيعية، لا خطأ.

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. استئناف مهمة جزئية

بما أن النتائج تُضاف دفعة تلو الأخرى، فإن إعادة التشغيل بعد تعطُّل لا تحتاج سوى إلى تخطي عناوين IP المكتوبة بالفعل في results.jsonl بدلًا من إعادة إرسال القائمة كاملة — إعادة إرسال عنوان IP مكرر تُحتسب فوترته مرة أخرى، لأن كل مُدخل في الدفعة يُقاس (metered) بشكل فردي بدلًا من إزالة التكرار.

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 في الدفعة نفسها؛ فكل عنوان يُبحث عنه بشكل مستقل، بنفس الحقول القابلة لأن تكون فارغة (null) في كلتا الحالتين.

الخطوة التالية

راجع مرجع البحث المجمّع عن عناوين IP لمعرفة مخطط الاستجابة الدقيق، وصفحة الأخطاء وحدود المعدل للاطلاع على القائمة الكاملة لرموز الحالة التي يمكن أن تُعيدها نقطة النهاية هذه.