79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
import json
|
|
import os
|
|
from urllib import request
|
|
|
|
import pandas as pd
|
|
from django.core.management.base import BaseCommand
|
|
from pflaenzli_django.settings import BASE_DIR
|
|
from pflaenzli.models import Plz
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Get the zip code index from post and compile it to a dataframe pickle'
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("--force", action="store_true", required=False)
|
|
|
|
def handle(self, *args, **options):
|
|
self.parse_data(*self.download_geojson(api='v2', data='v2', force=options["force"]), force=options["force"])
|
|
|
|
def download_geojson(self, api, data, force=False):
|
|
file = f'plz_verzeichnis_{data}.json'
|
|
if os.path.exists(file) and not force:
|
|
self.stdout.write('File already downloaded.')
|
|
self.stdout.write(self.style.SUCCESS(
|
|
'Skipping...\n'))
|
|
exists = True
|
|
else:
|
|
self.stdout.write('Downloading geojson...')
|
|
url = f'https://swisspost.opendatasoft.com/api/{api}/catalog/datasets/plz_verzeichnis_{data}/exports/geojson'
|
|
request.urlretrieve(url, file)
|
|
self.stdout.write(self.style.SUCCESS('Done!\n'))
|
|
exists = False
|
|
|
|
return file, exists
|
|
|
|
def parse_data(self, file, exists, force=False):
|
|
if exists and not force:
|
|
self.stdout.write(self.style.WARNING(
|
|
'Nothing was done, if you want to redownload the PLZ index, use the --force option.\n'))
|
|
return
|
|
|
|
self.stdout.write('Opening file...')
|
|
# Load the GeoJSON data for the zip codes
|
|
with open(file, encoding='UTF-8') as f:
|
|
full_data = json.load(f)
|
|
self.stdout.write(self.style.SUCCESS('Done!\n'))
|
|
|
|
self.stdout.write('Deleting existing data...')
|
|
Plz.objects.all().delete()
|
|
self.stdout.write(self.style.SUCCESS('Done!\n'))
|
|
|
|
self.stdout.write('Parsing file and add new data...')
|
|
|
|
for plz_entry in full_data['features']:
|
|
plz_entry = plz_entry['properties']
|
|
try:
|
|
plz = plz_entry['postleitzahl']
|
|
except (AttributeError, TypeError, ValueError):
|
|
continue
|
|
|
|
try:
|
|
lat = plz_entry['geo_point_2d']['lat']
|
|
lon = plz_entry['geo_point_2d']['lon']
|
|
except (AttributeError, TypeError):
|
|
continue
|
|
|
|
if plz is None or lat is None or lon is None:
|
|
continue
|
|
|
|
try:
|
|
name = plz_entry['ortbez27']
|
|
except (KeyError, AttributeError, TypeError):
|
|
name = None
|
|
|
|
plz, _ = Plz.objects.get_or_create(plz=int(plz), lat=lat, lon=lon, name=name)
|
|
|
|
self.stdout.write(self.style.SUCCESS('Wrote PLZ data to the databse successfully\n'))
|
|
self.stdout.write(self.style.SUCCESS('Done!'))
|