Passed
Push — master ( 30b883...611766 )
by Fabio
01:15 queued 12s
created

scripts.updatedata._slugify_names()   A

Complexity

Conditions 1

Size

Total Lines 2
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

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