uncloud-mravi/uncloud_pay/management/commands/import-vat-rates.py

from django.core.management.base import BaseCommand
from uncloud_pay.models import VATRate

import logging
import urllib
import csv
import sys
import io

logger = logging.getLogger(__name__)

class Command(BaseCommand):
    help = '''Imports VAT Rates. Assume vat rates of format https://github.com/kdeldycke/vat-rates/blob/master/vat_rates.csv'''
    vat_url = "https://raw.githubusercontent.com/ungleich/vat-rates/main/vat_rates.csv"


    def add_arguments(self, parser):
        parser.add_argument('--vat-url', default=self.vat_url)

    def handle(self, *args, **options):
        vat_url = options['vat_url']
        url_open = urllib.request.urlopen(vat_url)

        # map to fileio using stringIO
        csv_file = io.StringIO(url_open.read().decode('utf-8'))
        reader = csv.DictReader(csv_file)

        for row in reader:
            if row["territory_codes"] and len(row["territory_codes"].splitlines()) > 1:
                for code in row["territory_codes"].splitlines():
                    VATRate.objects.get_or_create(
                    starting_date=row["start_date"],
                    ending_date=row["stop_date"] if row["stop_date"] != "" else None,
                    territory_codes=code,
                    currency_code=row["currency_code"],
                    rate=row["rate"],
                    rate_type=row["rate_type"],
                    description=row["description"]
                )
            else:
                VATRate.objects.get_or_create(
                    starting_date=row["start_date"],
                    ending_date=row["stop_date"] if row["stop_date"] != "" else None,
                    territory_codes=row["territory_codes"],
                    currency_code=row["currency_code"],
                    rate=row["rate"],
                    rate_type=row["rate_type"],
                    description=row["description"]
                )
        logger.info('All VAT Rates have been added!')