1
|
|
|
#!/usr/bin/env python3 |
2
|
|
|
# -*- coding: utf-8 -*- |
3
|
|
|
|
4
|
|
|
# pylint: disable=C0111,C0326,C0103 |
5
|
|
|
|
6
|
|
|
"""Tools to manage saved device databases.""" |
7
|
|
|
|
8
|
|
|
import json |
9
|
|
|
import os |
10
|
|
|
import time |
11
|
|
|
|
12
|
|
|
import requests |
13
|
|
|
|
14
|
|
|
from . import ansi |
15
|
|
|
|
16
|
|
|
|
17
|
|
|
DEVICELIST_URL = "https://tclota.birth-online.de/json_lastupdates.php" |
18
|
|
|
DEVICELIST_FILE = "prds.json" |
19
|
|
|
DEVICELIST_CACHE_SECONDS = 86400 |
20
|
|
|
|
21
|
|
|
|
22
|
|
|
def _load_local_devicelist(): |
23
|
|
|
"""Load local devicelist and return decoded JSON (or None) and need_download status.""" |
24
|
|
|
need_download = True |
25
|
|
|
try: |
26
|
|
|
filestat = os.stat(DEVICELIST_FILE) |
27
|
|
|
filemtime = filestat.st_mtime |
28
|
|
|
if filemtime > time.time() - DEVICELIST_CACHE_SECONDS: |
29
|
|
|
need_download = False |
30
|
|
|
with open(DEVICELIST_FILE, "rt") as dlfile: |
31
|
|
|
return json.load(dlfile), need_download |
32
|
|
|
except (FileNotFoundError, json.decoder.JSONDecodeError): |
|
|
|
|
33
|
|
|
return None, True |
34
|
|
|
|
35
|
|
|
def _download_devicelist(doit: bool): |
36
|
|
|
"""Download device list if doit is set. Or do nothing.""" |
37
|
|
|
if doit: |
38
|
|
|
prds_json = requests.get(DEVICELIST_URL).text |
39
|
|
|
with open(DEVICELIST_FILE, "wt") as dlfile: |
40
|
|
|
dlfile.write(prds_json) |
41
|
|
|
|
42
|
|
|
def _load_devicelist_with_diff(output_diff: bool, old_prds: dict = {}) -> dict: |
|
|
|
|
43
|
|
|
"""Load local devicelist and output diff if requested.""" |
44
|
|
|
with open(DEVICELIST_FILE, "rt") as dlfile: |
45
|
|
|
prds = json.load(dlfile) |
46
|
|
|
|
47
|
|
|
if old_prds and output_diff: |
48
|
|
|
print_prd_diff(old_prds, prds) |
49
|
|
|
|
50
|
|
|
return prds |
51
|
|
|
|
52
|
|
|
def get_devicelist(force: bool=False, output_diff: bool=True, local: bool=False) -> dict: |
53
|
|
|
"""Return device list from saved database.""" |
54
|
|
|
old_prds, need_download = _load_local_devicelist() |
55
|
|
|
|
56
|
|
|
if local: |
57
|
|
|
return old_prds |
58
|
|
|
|
59
|
|
|
_download_devicelist(need_download or force) |
60
|
|
|
|
61
|
|
|
return _load_devicelist_with_diff(output_diff, old_prds) |
62
|
|
|
|
63
|
|
|
|
64
|
|
|
def print_versions_diff(old_data: dict, new_data: dict): |
65
|
|
|
"""Print version changes between old and new databases.""" |
66
|
|
|
prd = new_data["curef"] |
67
|
|
|
if new_data["last_full"] != old_data["last_full"] and new_data["last_ota"] != old_data["last_ota"]: |
68
|
|
|
print("> {}: {} ⇨ {} (OTA: {} ⇨ {})".format( |
69
|
|
|
prd, |
70
|
|
|
ansi.CYAN_DARK + str(old_data["last_full"]) + ansi.RESET, |
71
|
|
|
ansi.CYAN + str(new_data["last_full"]) + ansi.RESET, |
72
|
|
|
ansi.YELLOW_DARK + str(old_data["last_ota"]) + ansi.RESET, |
73
|
|
|
ansi.YELLOW + str(new_data["last_ota"]) + ansi.RESET |
74
|
|
|
)) |
75
|
|
|
elif new_data["last_full"] != old_data["last_full"]: |
76
|
|
|
print("> {}: {} ⇨ {} (FULL)".format(prd, ansi.CYAN_DARK + str(old_data["last_full"]) + ansi.RESET, ansi.CYAN + str(new_data["last_full"]) + ansi.RESET)) |
77
|
|
|
elif new_data["last_ota"] != old_data["last_ota"]: |
78
|
|
|
print("> {}: {} ⇨ {} (OTA)".format(prd, ansi.YELLOW_DARK + str(old_data["last_ota"]) + ansi.RESET, ansi.YELLOW + str(new_data["last_ota"]) + ansi.RESET)) |
79
|
|
|
|
80
|
|
|
def _print_removed_prds(prds_data: dict, removed_prds: list): |
81
|
|
|
"""Print details of selected PRDs as removed.""" |
82
|
|
|
for prd in removed_prds: |
83
|
|
|
print("> Removed device {} (was at {} / OTA: {}).".format(ansi.RED + prd + ansi.RESET, prds_data[prd]["last_full"], prds_data[prd]["last_ota"])) |
84
|
|
|
|
85
|
|
|
def _print_added_prds(prds_data: dict, added_prds: list): |
86
|
|
|
"""Print details of selected PRDs as added.""" |
87
|
|
|
for prd in added_prds: |
88
|
|
|
print("> New device {} ({} / OTA: {}).".format(ansi.GREEN + prd + ansi.RESET, prds_data[prd]["last_full"], prds_data[prd]["last_ota"])) |
89
|
|
|
|
90
|
|
|
def _print_changed_prds(old_prds: dict, new_prds: dict, skip_prds: list): |
91
|
|
|
"""Print details of changed PRDs.""" |
92
|
|
|
for prd, pdata in new_prds.items(): |
93
|
|
|
if prd in skip_prds: |
94
|
|
|
continue |
95
|
|
|
odata = old_prds[prd] |
96
|
|
|
print_versions_diff(odata, pdata) |
97
|
|
|
|
98
|
|
|
def print_prd_diff(old_prds: dict, new_prds: dict): |
99
|
|
|
"""Print PRD changes between old and new databases.""" |
100
|
|
|
added_prds = [prd for prd in new_prds if prd not in old_prds] |
101
|
|
|
removed_prds = [prd for prd in old_prds if prd not in new_prds] |
102
|
|
|
_print_removed_prds(old_prds, removed_prds) |
103
|
|
|
_print_added_prds(new_prds, added_prds) |
104
|
|
|
_print_changed_prds(old_prds, new_prds, added_prds) |
105
|
|
|
|
This check looks for calls to members that are non-existent. These calls will fail.
The member could have been renamed or removed.