Source code for ferrosoft.apps.ferromaps.management.commands.store_address_dataset

import csv
from decimal import Decimal
from django.core.management.base import BaseCommand

from ferrosoft.apps.ferrobase.models import CoordinateDecimalField
from ferrosoft.apps.ferromaps.models import GeocodedAddress
from django.db import transaction




class Command(BaseCommand):
    help = "Load address data from a transformed TSV file into the Address model"



    def add_arguments(self, parser):
        parser.add_argument(
            "file",
            type=str,
            nargs="?",
            default="./houses.tsv",
            help="The path to the TSV file to load (default: ./houses.tsv)",
        )




    def handle(self, *args, **options):
        file_path = options["file"]
        batch_size = 10000
        objects = []

        with open(file_path, "r", encoding="utf-8") as file:
            reader = csv.DictReader(file, dialect="excel-tab")
            for iteration, row in enumerate(reader, start=1):
                address_label = f"{row['street']} {row['house_number']}, {row['postal_code']} {row['city']}"
                longitude = CoordinateDecimalField.round(Decimal(row["x"]))
                latitude = CoordinateDecimalField.round(Decimal(row["y"]))

                objects.append(
                    GeocodedAddress(
                        address_label=address_label,
                        postal_code=row["postal_code"][
                            : min(20, len(row["postal_code"]))
                        ],
                        city=row["city"],
                        longitude=longitude,
                        latitude=latitude,
                        country=row["country"],
                        street=row["street"],
                        house_no=row["house_number"][
                            : min(20, len(row["house_number"]))
                        ],
                    )
                )

                if len(objects) >= batch_size:
                    with transaction.atomic():
                        print(".", end="", flush=True)
                        GeocodedAddress.objects.bulk_create(objects)
                    objects.clear()

            if objects:
                with transaction.atomic():
                    GeocodedAddress.objects.bulk_create(objects)

        self.stdout.write(self.style.SUCCESS("\nSuccessfully loaded address data."))