التحقق المجمع من العناوين باستخدام Geocoding API: معالجة 10,000 عنوان دفعة واحدة

كل مؤسسة تجمع بيانات العناوين منذ أكثر من بضع سنوات تواجه المشكلة نفسها: جدول عناوين كبير وغير منظم، وجودته مجهولة. عناوين أدخلها مستخدمون عبر نماذج الويب دون انتباه. سجلات نُقلت من نظام CRM قديم لم يكن يتحقق من المدخلات. عمليات استيراد مجمعة من ملف Excel لدى شريك بتنسيق غير متوافق. ومواقع أنشطة تجارية انتقلت إلى أماكن أخرى منذ جمع البيانات.

تشغيل عمليات اللوجستيات على بيانات عناوين رديئة يؤدي إلى فشل التسليم. واستخدامها في التسويق يهدر ميزانية البريد المباشر. وبناء تقييمات العقارات عليها ينتج تحليلات خاطئة لمناطق الاستقطاب. قبل أي من هذه العمليات، يجب تنظيف العناوين والتحقق من صحتها.

الترميز الجغرافي (geocoding) هو الطريقة الأكثر فعالية لإنجاز ذلك على نطاق واسع. أرسل نص العنوان الخام إلى Geocoding API، وستحصل على مكونات منظمة وإحداثيات جغرافية ودرجة ثقة. العناوين التي تُرمَّز بدرجة ثقة عالية صحيحة في الغالب شبه المؤكد. أما العناوين التي تعود بدرجة ثقة منخفضة أو دون أي تطابق فتحتاج إلى مراجعة بشرية.

في هذا الدرس نبني سكربت Python كاملًا يقرأ ملف CSV يحتوي على عناوين، ويرمّز كل عنوان عبر MapAtlas Geocoding API، ويكتب النتائج مع درجات الثقة، ويضع علامة على السجلات منخفضة الثقة لمراجعتها يدويًا. ويتعامل السكربت مع تحديد معدل الطلبات، وأخطاء الشبكة المؤقتة، وخصوصيات تنسيق العناوين الأوروبية التي تتجاهلها الدروس المكتوبة للسوق الأمريكية.

لماذا تفسد بيانات العناوين

تتراجع جودة العناوين لأسباب يمكن توقعها:

أخطاء الإدخال اليدوي. يكتب مستخدمو نماذج الويب بسرعة، ويشوّه التصحيح التلقائي أسماء الشوارع، ويسمح التحقق الذي يقبل أي نص غير فارغ بمرور بيانات عديمة القيمة. وفي اختبارات الدفع واسعة النطاق التي أجراها Baymard Institute، احتوت 9% من عناوين الشحن المُدخلة يدويًا على أخطاء كتابية أو إملائية.

انتقال الأنشطة التجارية. تنتقل الشركات وتندمج وتُغلق باستمرار. ويقدّر Eurostat معدل زوال المؤسسات في الاتحاد الأوروبي بنسبة 8.5% في عام 2023، وذلك قبل احتساب عمليات الانتقال، لذا ستضم قاعدة بيانات B2B جُمعت قبل عامين نسبة ملموسة من العناوين التي لم تعد تطابق نشاطًا تجاريًا قائمًا.

عدم اتساق التنسيق. تستخدم البيانات المجمعة من مصادر متعددة أعرافًا مختلفة: أسماء الدول الكاملة مقابل رموز ISO، و"St." مقابل "Street"، ورقم المبنى قبل اسم الشارع أو بعده، و"flat" مقابل "apt" مقابل "wohnung". ويوحّد نظام الترميز الجغرافي كل ذلك في مخرجات منظمة.

الترحيل من الأنظمة القديمة. حقول العنوان التي كانت موزعة على عدة أعمدة في قاعدة البيانات تُدمج غالبًا أثناء الترحيل، فتفقد بنيتها. وقد تتضمن حقول العنوان النصية الحرة في الأنظمة القديمة ملاحظات أو مراجع أو تنسيقات ليست جزءًا من العنوان الفعلي.

يتعامل الترميز الجغرافي مع كل هذه الحالات لأنه يعتمد على المطابقة الجغرافية، لا على مطابقة النصوص. فالعنوان المنسق بشكل خاطئ يمكن ترميزه بشكل صحيح إذا كانت بيانات الموقع الأساسية مطابقة.

فهم درجات الثقة

يُرجع MapAtlas Geocoding API الخاصية confidence لكل نتيجة، بقيمة تتراوح بين 0.0 و1.0. وتمثل مدى تطابق النتيجة مع الاستعلام المدخل، مع مراعاة اختلافات التنسيق والاختصارات والغموض.

هذه العتبات نقطة انطلاق. في عمليات اللوجستيات التي يكلف فيها فشل التسليم كثيرًا، ارفع عتبة القبول التلقائي إلى 0.92 أو أكثر. وفي قائمة بريدية تسويقية تتجاوز فيها تكلفة المراجعة اليدوية تكلفة بضعة عناوين خاطئة، خفّضها إلى 0.80.

ويُرجع الـ API أيضًا الخاصية match_type: exact (تطابق العنوان كاملًا)، أو interpolated (موقع مقدَّر بين أرقام مبانٍ معروفة)، أو fallback (لم يُعثر على العنوان الكامل فتطابق مستوى أعم بدلًا منه). وتوضح الخاصية layer ما هو هذا المستوى، مثل address أو street أو locality، بينما تبين accuracy ما إذا كانت النقطة point دقيقة أم centroid.

سكربت التحقق بلغة Python

ثبّت الاعتماديات:

pip install requests pandas tqdm

يقرأ السكربت ملف CSV يحتوي على عمود address (أو أعمدة منفصلة street وcity وcountry)، ويرمّز كل صف، ثم يكتب ملف CSV جديدًا تُضاف إليه نتائج التحقق.

#!/usr/bin/env python3
"""
bulk_geocode.py, Validate a CSV of addresses using the MapAtlas Geocoding API.

Input CSV must have either:
  - An 'address' column (full address string), or
  - 'street', 'city', and 'country' columns (will be concatenated)

Outputs a new CSV with added columns:
  geocoded_label, geocoded_lat, geocoded_lng, confidence, match_type, status
"""

import csv
import time
import logging
import requests
import pandas as pd
from tqdm import tqdm
from pathlib import Path

# ── الإعدادات ──────────────────────────────────────────────────────────────
API_KEY         = 'YOUR_API_KEY'
API_BASE        = 'https://gateway.mapmetrics-atlas.net/forward-geocode/'
INPUT_CSV       = 'addresses.csv'
OUTPUT_CSV      = 'addresses_validated.csv'
RATE_LIMIT_RPS  = 5          # عدد الطلبات في الثانية (التزم بحدود خطتك)
RETRY_ATTEMPTS  = 3          # عدد المحاولات عند الأخطاء المؤقتة
RETRY_DELAY_S   = 2.0        # الثواني بين المحاولات

# عتبات الثقة
ACCEPT_THRESHOLD = 0.85
REVIEW_THRESHOLD = 0.60

logging.basicConfig(level=logging.INFO, format='%(levelname)s %(message)s')
log = logging.getLogger(__name__)

# ── دالة الترميز الجغرافي ─────────────────────────────────────────────────────────
def geocode_address(address_str: str) -> dict:
    """
    Geocode a single address string. Returns a dict with result fields.
    Retries on network errors and 429 rate-limit responses.
    """
    params = {
        'text': address_str,
        'token': API_KEY,
        'size': 1,
    }

    for attempt in range(RETRY_ATTEMPTS):
        try:
            resp = requests.get(API_BASE, params=params, timeout=10)

            if resp.status_code == 429:
                # تجاوز حد المعدل، انتظر ثم أعد المحاولة
                wait = float(resp.headers.get('Retry-After', RETRY_DELAY_S * (attempt + 1)))
                log.warning(f'Rate limited. Waiting {wait:.1f}s before retry {attempt + 1}.')
                time.sleep(wait)
                continue

            resp.raise_for_status()
            data = resp.json()

            if not data.get('features'):
                return {'geocoded_label': '', 'geocoded_lat': None, 'geocoded_lng': None,
                        'confidence': 0.0, 'match_type': 'no_match', 'status': 'reject'}

            feature    = data['features'][0]
            props      = feature['properties']
            coords     = feature['geometry']['coordinates']  # [lng, lat]
            confidence = round(float(props.get('confidence', 0.0)), 4)
            match_type = props.get('match_type', 'unknown')

            if confidence >= ACCEPT_THRESHOLD:
                status = 'accept'
            elif confidence >= REVIEW_THRESHOLD:
                status = 'review'
            else:
                status = 'reject'

            return {
                'geocoded_label': props.get('label', ''),
                'geocoded_lat':   round(coords[1], 6),
                'geocoded_lng':   round(coords[0], 6),
                'confidence':     confidence,
                'match_type':     match_type,
                'status':         status,
            }

        except requests.exceptions.RequestException as exc:
            log.warning(f'Network error on attempt {attempt + 1}: {exc}')
            if attempt < RETRY_ATTEMPTS - 1:
                time.sleep(RETRY_DELAY_S * (attempt + 1))

    # استُنفدت كل المحاولات
    return {'geocoded_label': '', 'geocoded_lat': None, 'geocoded_lng': None,
            'confidence': 0.0, 'match_type': 'error', 'status': 'error'}


# ── حلقة المعالجة الرئيسية ───────────────────────────────────────────────────────
def main():
    df = pd.read_csv(INPUT_CSV, dtype=str).fillna('')

    # بناء نص العنوان من الأعمدة المتاحة
    if 'address' in df.columns:
        df['_query'] = df['address']
    elif all(c in df.columns for c in ['street', 'city', 'country']):
        df['_query'] = df['street'] + ', ' + df['city'] + ', ' + df['country']
    else:
        raise ValueError("CSV must have 'address' or 'street'+'city'+'country' columns.")

    results = []
    sleep_interval = 1.0 / RATE_LIMIT_RPS

    for query in tqdm(df['_query'], desc='Geocoding', unit='addr'):
        result = geocode_address(query.strip())
        results.append(result)
        time.sleep(sleep_interval)

    results_df = pd.DataFrame(results)
    output_df  = pd.concat([df.drop(columns=['_query']), results_df], axis=1)
    output_df.to_csv(OUTPUT_CSV, index=False, quoting=csv.QUOTE_NONNUMERIC)

    # الملخص
    total   = len(output_df)
    accept  = (output_df['status'] == 'accept').sum()
    review  = (output_df['status'] == 'review').sum()
    reject  = (output_df['status'] == 'reject').sum()
    errors  = (output_df['status'] == 'error').sum()

    log.info(f'\n── Results ────────────────────────────────')
    log.info(f'Total processed : {total:,}')
    log.info(f'Accept (≥{ACCEPT_THRESHOLD}) : {accept:,}  ({accept/total:.1%})')
    log.info(f'Review           : {review:,}  ({review/total:.1%})')
    log.info(f'Reject           : {reject:,}  ({reject/total:.1%})')
    log.info(f'Errors           : {errors:,}  ({errors/total:.1%})')
    log.info(f'Output written to {OUTPUT_CSV}')


if __name__ == '__main__':
    main()

شغّله:

python bulk_geocode.py

مع ملف CSV يضم 10,000 عنوان وبمعدل 5 طلبات في الثانية، يحتاج السكربت إلى 33 دقيقة على الأقل من الانتظار الذي يفرضه حد المعدل، إضافة إلى زمن استجابة كل طلب، أي من 35 إلى 50 دقيقة في المجمل عادةً. ويعرض شريط التقدم (عبر tqdm) معدل المعالجة الفعلي والوقت المتوقع للانتهاء.

التعامل مع المخرجات

يضيف ملف CSV الناتج ستة أعمدة إلى بياناتك المدخلة:

address, ..., geocoded_label, geocoded_lat, geocoded_lng, confidence, match_type, status

صفِّ النتائج حسب الحالة لإنشاء ثلاثة ملفات مخرجات:

# بعد تشغيل main()، قسّم النتائج إلى مستويات القبول:
df = pd.read_csv('addresses_validated.csv')

df[df['status'] == 'accept'].to_csv('addresses_clean.csv',  index=False)
df[df['status'] == 'review'].to_csv('addresses_review.csv', index=False)
df[df['status'] == 'reject'].to_csv('addresses_reject.csv', index=False)

print(f"Clean:  {len(df[df['status']=='accept']):,} addresses ready for use")
print(f"Review: {len(df[df['status']=='review']):,} addresses for manual check")
print(f"Reject: {len(df[df['status']=='reject']):,} addresses to discard or fix")

الملف addresses_review.csv هو الذي يحتاج إلى تدخل بشري. وهذه أنماط المراجعة المعتادة:

• الرجوع إلى الطبقة street (درجة ثقة بين 0.65 و0.80): عُثر على الشارع دون رقم المبنى. قد يكون السبب مبنى جديدًا، أو عنوانًا ريفيًا بتغطية محدودة، أو خطأ مطبعيًا في رقم المبنى. راجع المصدر الأصلي.
• الرجوع إلى الطبقة locality (درجة ثقة بين 0.50 و0.65): تطابقت المدينة فقط. غالبًا ما يكون اسم الشارع مكتوبًا بشكل خاطئ أو غير موجود في البيانات. ابحث عن العنوان في دليل بريدي.
• درجة ثقة منخفضة لعناوين تبدو صحيحة بوضوح: تحقق من عدم تطابق الدولة أو اللغة. فعنوان هولندي يُستعلم عنه دون تقييد الدولة قد يُرمَّز إلى بلدة ألمانية أو بلجيكية تحمل اسمًا مشابهًا.

خصوصيات العناوين الأوروبية التي يجب معرفتها

تتجاهل أدلة الترميز الجغرافي المجمع المكتوبة للسوق الأمريكية فروق التنسيق التي تُربك مجموعات البيانات الأوروبية. هذه أكثر المشكلات شيوعًا:

ألمانيا، ترتيب رقم المبنى. تستخدم العناوين الألمانية التنسيق {street} {number}، مثل Berliner Straße 42. وتخزن أنظمة CRM كثيرة العناوين بالتنسيق {number} {street} لأنها صُممت وفق العرف البريطاني أو الأمريكي. إذا كانت عناوينك الألمانية تُرمَّز بدرجة ثقة منخفضة، فجرّب عكس ترتيب الرقم واسم الشارع في نص الاستعلام قبل إرساله.

فرنسا، الدوائر البلدية (arrondissements). تتضمن عناوين باريس رقم الدائرة (من 1 إلى 20) ضمن الرمز البريدي: من 75001 إلى 75020. والاستعلامات التي تحذف الدائرة وتكتفي بكلمة Paris تُرمَّز إلى مركز المدينة لا إلى الحي المحدد، فتظهر كتطابق من نوع locality بدرجة ثقة منخفضة بدلًا من تطابق من نوع address.

هولندا، تنسيق الرمز البريدي. تتبع الرموز البريدية الهولندية نمطًا صارمًا هو DDDD LL (4 أرقام، ثم مسافة، ثم حرفان كبيران). الرموز المخزنة دون مسافة (1012LG بدلًا من 1012 LG) أو بأحرف صغيرة تُرمَّز بشكل صحيح، لكن إذا كنت تتحقق من تنسيق الرمز البريدي بشكل منفصل، فوحّدها بأحرف كبيرة مع مسافة.

بلجيكا، تعدد اللغات. لبعض البلديات البلجيكية اسمان مختلفان بالفرنسية والهولندية (Liège/Luik، Gent/Gand). يتعامل الـ API مع الاسمين، لكن عدم اتساق التسمية في بياناتك (بعض السجلات بالفرنسية وبعضها بالهولندية) قد ينتج درجات ثقة متفاوتة. وحّد اللغة لكل منطقة قبل الترميز.

إسبانيا وإيطاليا، اختلاف بادئات الشوارع. "Calle" و"Carrer" و"Via" و"Viale" كلها بادئات صالحة لنوع الشارع، وقد تظهر مختصرة ("C/") في السجلات. يتعامل نظام الترميز مع الاختصارات الشائعة، لكن الاختصارات غير المألوفة القادمة من الأنظمة القديمة قد تحتاج إلى خطوة توحيد.

زيادة معدل المعالجة بالطلبات المتزامنة

السكربت التسلسلي آمن وبسيط لكنه بطيء. لمجموعات البيانات الأكبر، استبدل الحلقة التسلسلية بمنفّذ متزامن:

from concurrent.futures import ThreadPoolExecutor, as_completed

def main_concurrent(max_workers=10):
    df = pd.read_csv(INPUT_CSV, dtype=str).fillna('')
    # ... (الإعداد نفسه كما في السابق) ...

    results = [None] * len(df)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_idx = {
            executor.submit(geocode_address, row['_query'].strip()): idx
            for idx, row in df.iterrows()
        }
        for future in tqdm(as_completed(future_to_idx), total=len(df), desc='Geocoding'):
            idx = future_to_idx[future]
            results[idx] = future.result()

    # ... (كتابة المخرجات نفسها كما في السابق) ...

مع 10 عمال متزامنين، يتحدد معدل المعالجة بزمن استجابة الـ API وحد المعدل في خطتك وليس بفترة انتظار: فبمعدل 50 طلبًا في الثانية تنتهي معالجة 10,000 عنوان في أقل من 4 دقائق. ولا تتضمن النسخة المتزامنة آلية تقييد مدمجة، لذا أضف واحدة أو خفّض max_workers إذا كانت خطتك تحدد عدد الطلبات في الثانية. تحقق من حد الطلبات المتزامنة في خطة MapAtlas الخاصة بك قبل زيادة max_workers.

إذا كانت مؤسستك تجري التحقق من العناوين كعملية متكررة، مثل تنظيف ما يُستورد إلى CRM أسبوعيًا أو التحقق من عناوين التسليم كل ليلة، فراجع صفحة حلول اللوجستيات والتوصيل لترى كيف تندمج تكاملات MapAtlas في سير العمل التشغيلي.

وإذا كنت تبني ميزة الإكمال التلقائي للعناوين لمنع دخول البيانات الخاطئة إلى نظامك من الأساس (أي اكتشاف الأخطاء عند المصدر بدلًا من تنظيفها على دفعات)، فإن مقال Address Autocomplete API: كيف يرفع حقل واحد معدل إتمام الدفع يشرح التطبيق في الواجهة الأمامية.

العمل مع الإحداثيات الناتجة

تتضمن المخرجات التي جرى التحقق منها geocoded_lat وgeocoded_lng لكل عنوان مقبول. وتفتح هذه الإحداثيات إمكانيات تحليل لم تكن متاحة مع نصوص العناوين الخام:

• حساب المسافات. احسب المسافة المستقيمة بين المستودع وكل عنوان تسليم لتقدير شرائح تكلفة الشحن.
• تحليل مناطق الاستقطاب. اعرض مواقع العملاء التي جرى التحقق منها على خريطة لترى أين يتركز الطلب جغرافيًا.
• توزيع مناطق التسليم. اربط كل عنوان بمنطقة تسليم عبر اختبار ما إذا كانت إحداثياته تقع داخل مضلع المنطقة.
• اكتشاف التكرار. السجلان اللذان يحملان الإحداثيات نفسها (ضمن بضعة أمتار) مكرران على الأرجح، حتى لو اختلف تنسيق نصوص العناوين.

أما اتساق بيانات NAP (الاسم والعنوان والهاتف) وأثره على الظهور في البحث بالذكاء الاصطناعي، وهو أمر مهم خاصة لقواعد بيانات عناوين الأنشطة المحلية، فيشرحه مقال اتساق NAP للبحث بالذكاء الاصطناعي: لماذا تقضي العناوين غير المتطابقة على ظهورك في ChatGPT، ويبين لماذا تشكل العناوين المتحقق منها بالترميز الجغرافي الأساس الصحيح لترميز البيانات المنظمة.

الخلاصة

يمنحك الترميز الجغرافي المجمع باستخدام MapAtlas Geocoding API ما يلي:

• عناوين متحقق منها مع درجات ثقة، فتعرف أي السجلات يمكن الوثوق بها تلقائيًا وأيها يحتاج إلى مراجعة.
• مكونات منظمة (الشارع، رقم المبنى، الرمز البريدي، المدينة، الدولة) موحدة من أي تنسيق إدخال.
• إحداثيات جغرافية لكل عنوان مقبول، ما يتيح التحليل المكاني وتخطيط المسارات.
• دعم تنسيقات العناوين الأوروبية مدمج في الـ API، فلا تحتاج إلى معالجته في مرحلة التجهيز المسبق.

يعالج سكربت Python خمسة عناوين في الثانية بالتسلسل، وحتى 50 في الثانية مع العمال المتزامنين. لمعالجة 10,000 عنوان، خصص ما بين 4 و50 دقيقة حسب مستوى التزامن وزمن الاستجابة. والناتج ملف CSV نظيف مقسم إلى ثلاثة مستويات: قبول، ومراجعة، ورفض.

سجّل للحصول على مفتاح MapAtlas API مجاني لتبدأ. يدعم Geocoding API الاستعلامات المجمعة في جميع الخطط، راجع الأسعار لمعرفة تكلفة كل طلب وحدود الباقة المجانية الشهرية.