Passed
Push — master ( 2f9af8...4400c0 )
by Fabio
01:37 queued 15s
created

scripts.updatedata._update_countries_data()   B

Complexity

Conditions 5

Size

Total Lines 56
Code Lines 43

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 43
dl 0
loc 56
rs 8.3813
c 0
b 0
f 0
cc 5
nop 0

How to fix   Long Method   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

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