Geocodificación ¿Cómo vincular 250 mil direcciones a coordenadas en 10 minutos?

• Antecedentes

• Geocodificación por lotes

• Elegir un proveedor de geocodificador de paquetes

• Guía de servicio de Python

• Descripción completa de la clase de lote

• Análisis de resultados

Antecedentes

1. Aplicación web para geocodificación, que se utilizó en la universidad;
2. Escriba un script basado en la API REST del geocodificador.

Geocodificación por lotes

1. Preparar el conjunto de datos para que el servicio lo acepte sin errores;
2. Establecer los parámetros del resultado del trabajo (selección de columnas, separador ...);
3. Sube un archivo a la nube;
4. Esperando a que se complete el procesamiento;
5. Descargue el archivo terminado.

Elegir un proveedor de servicios de geocodificación por lotes

• mapas de Google
• AQUÍ Tecnologías;
• MapBox
• TomTom
• ESRI

• El número de solicitudes al servicio de geocodificación por mes es gratuito;
• Límite en la cantidad de transacciones por día;
• Disponibilidad de servicio de geocodificación por lotes;
• Posibilidad de utilizar un geocodificador de paquetes en un plan gratuito.

Mapas de Google

Tecnologías AQUÍ

Mapbox

Tomtom

Yandex Technologies

ESRI

¿Qué elegir al final?

Guía de servicio de Python

import requests
import json
import time
import zipfile
import io
from bs4 import BeautifulSoup

class Batch:

    SERVICE_URL = "https://batch.geocoder.ls.hereapi.com/6.2/jobs"
    jobId = None

    def __init__(self, apikey="your_api_key"):
        self.apikey = apikey

   recId; searchText
   1; -, . , 6
   2; ,  1,  -., 72
   3; 425 W Randolph St Chicago IL 60606
   4; , DJ106 20-30, Sibiu 557260
   5; 200 S Mathilda Ave Sunnyvale CA 94086
  

def start(self, filename, indelim=";", outdelim=";"):
        
        file = open(filename, 'rb')

        params = {
            "action": "run",
            "apiKey": self.apikey,
            "politicalview":"RUS",
            "gen": 9,
            "maxresults": "1",
            "header": "true",
            "indelim": indelim,
            "outdelim": outdelim,
            "outcols": "displayLatitude,displayLongitude,locationLabel,houseNumber,street,district,city,postalCode,county,state,country",
            "outputcombined": "true",
        }

        response = requests.post(self.SERVICE_URL, params=params, data=file)
        self.__stats (response)
        file.close()

• “Acción”: “ejecutar”: inicio del procesamiento de la dirección;
• “politicalView”: “RUS” – . ( );
• “gen”: 9 – ( );
• “maxresults”: 1 – ;
• “header”: true – ;
• “indelim”: “;” – , ;
• “outdelim”: “;” – ;
• “outcols”: “” – , ;
• “outcombined”: true – .

    def status (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        statusUrl = self.SERVICE_URL + "/" + self.jobId
        
        params = {
            "action": "status",
            "apiKey": self.apikey,
        }
        
        response = requests.get(statusUrl, params=params)
        self.__stats (response)

    def result (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        print("Requesting result data ...")
        
        resultUrl = self.SERVICE_URL + "/" + self.jobId + "/result"
        
        params = {
            "apiKey": self.apikey
        }
        
        response = requests.get(resultUrl, params=params, stream=True)
        
        if (response.ok):    
            zipResult = zipfile.ZipFile(io.BytesIO(response.content))
            zipResult.extractall()
            print("File saved successfully")
        
        else:
            print("Error")
            print(response.text)

    def __stats (self, response):
        if (response.ok):
            parsedXMLResponse = BeautifulSoup(response.text, "lxml")

            self.jobId = parsedXMLResponse.find('requestid').get_text()
            
            for stat in parsedXMLResponse.find('response').findChildren():
                if(len(stat.findChildren()) == 0):
                    print("{name}: {data}".format(name=stat.name, data=stat.get_text()))

        else:
            print(response.text)

>>> from geocoder import Batch
>>> service = Batch(apikey="   REST API")
>>> service.start("big_data_addresses.csv", indelim=";", outdelim=";")

requestid: "  Id "
status: accepted
totalcount: 0
validcount: 0
invalidcount: 0
processedcount: 0
pendingcount: 0
successcount: 0
errorcount: 0

>>> service.status()

requestid: "  Id "
status: completed
jobstarted: 2020-04-27T10:09:58.000Z
jobfinished: 2020-04-27T10:17:18.000Z
totalcount: 249999
validcount: 249999
invalidcount: 0
processedcount: 249999
pendingcount: 0
successcount: 249978
errorcount: 21

>>> service.result()
Requesting result data ...
File saved successfully

Descripción completa de la clase de lote

import requests
import json
import time
import zipfile
import io
from bs4 import BeautifulSoup

class Batch:

    SERVICE_URL = "https://batch.geocoder.ls.hereapi.com/6.2/jobs"
    jobId = None

    def __init__(self, apikey="   REST API "):
        self.apikey = apikey
        
            
    def start(self, filename, indelim=";", outdelim=";"):
        
        file = open(filename, 'rb')

        params = {
            "action": "run",
            "apiKey": self.apikey,
            "politicalview":"RUS",
            "gen": 9,
            "maxresults": "1",
            "header": "true",
            "indelim": indelim,
            "outdelim": outdelim,
            "outcols": "displayLatitude,displayLongitude,locationLabel,houseNumber,street,district,city,postalCode,county,state,country",
            "outputcombined": "true",
        }

        response = requests.post(self.SERVICE_URL, params=params, data=file)
        self.__stats (response)
        file.close()
    

    def status (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        statusUrl = self.SERVICE_URL + "/" + self.jobId
        
        params = {
            "action": "status",
            "apiKey": self.apikey,
        }
        
        response = requests.get(statusUrl, params=params)
        self.__stats (response)
        

    def result (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        print("Requesting result data ...")
        
        resultUrl = self.SERVICE_URL + "/" + self.jobId + "/result"
        
        params = {
            "apiKey": self.apikey
        }
        
        response = requests.get(resultUrl, params=params, stream=True)
        
        if (response.ok):    
            zipResult = zipfile.ZipFile(io.BytesIO(response.content))
            zipResult.extractall()
            print("File saved successfully")
        
        else:
            print("Error")
            print(response.text)
    

    
    def __stats (self, response):
        if (response.ok):
            parsedXMLResponse = BeautifulSoup(response.text, "lxml")

            self.jobId = parsedXMLResponse.find('requestid').get_text()
            
            for stat in parsedXMLResponse.find('response').findChildren():
                if(len(stat.findChildren()) == 0):
                    print("{name}: {data}".format(name=stat.name, data=stat.get_text()))

        else:
            print(response.text)

Análisis de resultados