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
73
74
75
|
#!/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 json
import os
import sys
from django.conf import settings
from django.core.management.base import BaseCommand
from django.db import transaction
from ishtar_common.utils import BColors, get_log_time
from ishtar_common.models import Town
class Command(BaseCommand):
help = 'Import towns fixtures'
def add_arguments(self, parser):
parser.add_argument('towns_file')
parser.add_argument(
'--limit', type=str, default="", dest='limit',
help='Limit import town to town code begining with the string provided.'
'Example: --limit=35')
parser.add_argument(
'--quiet', dest='quiet', action='store_true',
help='Quiet output')
@transaction.atomic
def handle(self, *args, **options):
log_path = os.sep.join([settings.ROOT_PATH, "logs"])
if not os.path.exists(log_path):
os.mkdir(log_path, mode=0o770)
quiet = options['quiet']
self.limit = options['limit']
towns_file = options['towns_file']
with open(towns_file, "r") as t:
src = json.loads(t.read())
log_filename = f"load_towns-{get_log_time().replace(':', '')}.csv"
log_path = os.sep.join([log_path, log_filename])
town_created, geo_created, geo_updated, nb_rels = Town.geodata_import(
src, limit=self.limit, log_path=log_path, verbose=not quiet
)
if quiet:
return
sys.stdout.write(BColors.OKGREEN)
if town_created:
sys.stdout.write(f'\n* {town_created} town created')
if geo_created:
sys.stdout.write(f'\n* {geo_created} geo created')
if geo_updated:
sys.stdout.write(f'\n* {geo_updated} geo updated')
if nb_rels:
sys.stdout.write(f'\n* {nb_rels} relations updated')
sys.stdout.write(BColors.ENDC + "\n")
sys.stdout.flush()
|
