pflaenz.li/pflaenzli/pflaenzli/management/commands/getplzindex.py

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!'))