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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
#!/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()
geo = town.main_geodata
town_dct = {
"model": "ishtar_common.town",
"fields": {
"name": town.name,
"surface": town.surface,
"numero_insee": town.numero_insee,
"notice": town.notice,
"year": town.year,
"cached_label": town.cached_label,
"main_geodata": [
"ishtar_common",
"town",
town.numero_insee
],
"geodata": [
["ishtar_common", "town", town.numero_insee]
],
"children": [
t.numero_insee
for t in town.children.filter(numero_insee__isnull=False).all()
]
}
}
geo_dct = {
"model": "ishtar_common.geovectordata",
"fields": {
"name": geo.name,
"source_content_type": [
"ishtar_common",
"town"
],
"source": town.numero_insee,
"data_type": [
"town-limit"
],
"provider": geo.provider.txt_idx,
"comment": geo.comment,
"cached_x": geo.cached_x,
"cached_y": geo.cached_y,
"spatial_reference_system": None,
"multi_polygon": geo.multi_polygon.wkt
}
}
result.append(geo_dct)
result.append(town_dct)
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")
|
