summaryrefslogtreecommitdiff
path: root/ishtar_common/management/commands/dump_towns.py
blob: aa5270aef39d30154ce34cdea7504bc9c940aadb (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (C) 2025  Étienne Loks  <etienne.loks_AT_peacefrogsDOTnet>

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

# See the file COPYING for details.

import datetime
import json
import sys

from django.core.management.base import BaseCommand
from django.db import transaction

from ishtar_common.utils import BColors, get_progress
from ishtar_common.models import Town


class Command(BaseCommand):
    help = 'Export towns fixtures'

    def add_arguments(self, parser):
        parser.add_argument(
            '--query', type=str, default="", dest='query',
            help='Limit exported towns to query. Query is a valid Django query dict with '
            ' " for quotes. Example: --query=\'{"numero_insee__startswith": "35"}')
        parser.add_argument(
            '--quiet', dest='quiet', action='store_true',
            help='Quiet output')

    @transaction.atomic
    def handle(self, *args, **options):
        quiet = options['quiet']
        query = options["query"]
        q = Town.objects.filter(main_geodata__isnull=False,
                                main_geodata__multi_polygon__isnull=False)
        if query:
            try:
                query = json.loads(query)
            except json.JSONDecodeError:
                sys.stdout.write(f"\n{BColors.FAIL}Bad query{BColors.ENDC}\n")
                return
            q = q.filter(**query)
        nb_lines = q.count()
        started = datetime.datetime.now()
        result = []
        for idx, town in enumerate(q.all()):
            if not quiet:
                sys.stdout.write(
                    get_progress("processing town", idx, nb_lines, started)
                )
                sys.stdout.flush()
            result += town.geodata_export
        today = datetime.date.today()
        result_file = f"ishtar-towns-{today.strftime('%Y-%m-%d')}.json"
        with open(result_file, "w") as r:
            r.write(json.dumps(result, indent=4))
        if quiet:
            return
        sys.stdout.write(f"\n{BColors.WARNING}{result_file}{BColors.ENDC}\n")