Passed
Push — master ( d3d16b...37303e )
by Fabio
01:02
created

scripts.updatedata.main()   A

Complexity

Conditions 1

Size

Total Lines 3
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 3
dl 0
loc 3
rs 10
c 0
b 0
f 0
cc 1
nop 0
1
import fsutil
2
from benedict import benedict
3
from slugify import slugify
4
5
6
def _update_countries_data():
7
    # https://www.anagrafenazionale.interno.it/il-progetto/strumenti-di-lavoro/tabelle-decodifica/
8
    data_url = "https://www.anagrafenazionale.interno.it/wp-content/uploads/2021/03/tabella_2_statiesteri.xlsx"
9
    data = benedict.from_xls(data_url)
10
    data.standardize()
11
    # print(data.dump())
12
13
    def map_item(item):
14
        if not item:
15
            return None
16
        code = item.get_str("codat").upper()
17
        if not code:
18
            return None
19
        assert len(code) == 4, f"Invalid code: '{code}'"
20
21
        name = item.get_str("denominazione").title()
22
        assert name != "", f"Invalid name: '{name}'"
23
        name_alt = item.get_str("denominazioneistat").title()
24
        name_alt_en = item.get_str("denominazioneistat_en").title()
25
        name_slugs = sorted(
26
            set(
27
                filter(
28
                    bool,
29
                    [
30
                        slugify(name),
31
                        slugify(name_alt),
32
                        slugify(name_alt_en),
33
                    ],
34
                )
35
            )
36
        )
37
        province = "EE"
38
39
        date_created = item.get_datetime("datainiziovalidita")
40
        date_deleted = item.get_datetime("datafinevalidita")
41
        date_deleted_raw = item.get_str("datafinevalidita")
42
        if "9999" in date_deleted_raw:
43
            date_deleted = ""
44
45
        return {
46
            "active": False if date_deleted else True,
47
            "code": code,
48
            "date_created": date_created,
49
            "date_deleted": date_deleted,
50
            "name": name,
51
            "name_alt": name_alt,
52
            "name_alt_en": name_alt_en,
53
            "name_slugs": name_slugs,
54
            "province": province,
55
        }
56
57
    output_data = list(
58
        filter(bool, [map_item(benedict(item)) for item in data["values"]])
59
    )
60
    output_data = sorted(output_data, key=lambda item: item["name"])
61
    output_path = "../codicefiscale/data/countries.json"
62
    output_abspath = fsutil.join_path(__file__, output_path)
63
    fsutil.write_file_json(output_abspath, output_data, indent=4, sort_keys=True)
64
65
66
def _update_municipalities_data():
67
    # https://www.anagrafenazionale.interno.it/il-progetto/strumenti-di-lavoro/tabelle-decodifica/
68
    data_url = "https://www.anagrafenazionale.interno.it/wp-content/uploads/ANPR_archivio_comuni.csv"
69
    data = benedict.from_csv(data_url)
70
    data.standardize()
71
72
    def map_item(item):
73
74
        status = item.get("stato", "").upper()
75
        assert len(status) == 1 and status in ["A", "C"], f"Invalid status: '{status}'"
76
        active = status == "A"
77
78
        code = item.get_str("codcatastale").upper()
79
        assert code == "ND" or len(code) == 4, f"Invalid code: '{code}'"
80
81
        name = item.get_str("denominazione_it").title()
82
        assert name != "", f"Invalid name: {name}"
83
84
        name_trans = item.get_str("denomtraslitterata").title()
85
        name_alt = item.get_str("altradenominazione").title()
86
        name_alt_trans = item.get_str("altradenomtraslitterata").title()
87
        name_slugs = sorted(
88
            set(
89
                filter(
90
                    bool,
91
                    [
92
                        slugify(name),
93
                        slugify(name_trans),
94
                        slugify(name_alt),
95
                        slugify(name_alt_trans),
96
                    ],
97
                )
98
            )
99
        )
100
        province = item.get("siglaprovincia", "").upper()
101
        assert len(province) == 2, f"Invalid province: '{province}'"
102
103
        date_created = item.get_datetime("dataistituzione")
104
        date_deleted = item.get_datetime("datacessazione")
105
        date_deleted_raw = item.get_str("datacessazione")
106
        if "9999" in date_deleted_raw:
107
            date_deleted = ""
108
109
        return {
110
            "active": active,
111
            "code": code,
112
            "date_created": date_created,
113
            "date_deleted": date_deleted,
114
            "name": name,
115
            "name_trans": name_trans,
116
            "name_alt": name_alt,
117
            "name_alt_trans": name_alt_trans,
118
            "name_slugs": name_slugs,
119
            "province": province,
120
        }
121
122
    output_data = list(
123
        filter(bool, [map_item(benedict(item)) for item in data["values"]])
124
    )
125
    output_data = sorted(output_data, key=lambda item: item["name"])
126
    output_path = "../codicefiscale/data/municipalities.json"
127
    output_abspath = fsutil.join_path(__file__, output_path)
128
    fsutil.write_file_json(output_abspath, output_data, indent=4, sort_keys=True)
129
130
131
def main():
132
    _update_countries_data()
133
    _update_municipalities_data()
134
135
136
if __name__ == "__main__":
137
    main()
138