Geokodierung Wie binde ich in 10 Minuten 250.000 Adressen an Koordinaten?

• Hintergrund

• Batch-Geokodierung

• Auswählen eines Paket-Geocoder-Anbieters

• Python-Servicehandbuch

• Beschreibung der vollständigen Chargenklasse

• Ergebnisanalyse

Hintergrund

1. Webanwendung für Geokodierung, die an der Universität verwendet wurde;
2. Schreiben Sie ein Skript, das auf der REST-API des Geocoders basiert.

Batch-Geokodierung

1. Vorbereiten des Datensatzes, damit der Dienst ihn fehlerfrei akzeptiert;
2. Einstellen der Parameter des Arbeitsergebnisses (Auswahl der Spalten, Trennzeichen ...);
3. Laden Sie eine Datei in die Cloud hoch.
4. Warten auf den Abschluss der Verarbeitung;
5. Laden Sie die fertige Datei herunter.

Auswählen eines Batch-Geokodierungsdienstanbieters

• Google Maps
• HIER Technologien;
• MapBox
• TomTom
• ESRI.

• Die Anzahl der Anfragen an den Geokodierungsdienst pro Monat ist kostenlos.
• Begrenzung der Anzahl der Transaktionen pro Tag;
• Verfügbarkeit des Batch-Geokodierungsdienstes;
• Möglichkeit, einen Paket-Geocoder in einem kostenlosen Plan zu verwenden.

Google Maps

HIER Technologien

Mapbox

Tomtom

Yandex Technologies

ESRI

Was soll man am Ende wählen?

Python-Servicehandbuch

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()

• "Aktion": "Ausführen" - Start der Adressverarbeitung;
• “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

Beschreibung der vollständigen Chargenklasse

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)

Ergebnisanalyse