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: |
33
|
|
|
return None, True |
34
|
|
|
|
35
|
|
|
|
36
|
|
|
def get_devicelist(force: bool=False, output_diff: bool=True, local: bool=False) -> dict: |
37
|
|
|
"""Return device list from saved database.""" |
38
|
|
|
old_prds, need_download = load_local_devicelist() |
39
|
|
|
|
40
|
|
|
if local: |
41
|
|
|
return old_prds |
42
|
|
|
|
43
|
|
|
if need_download or force: |
44
|
|
|
prds_json = requests.get(DEVICELIST_URL).text |
45
|
|
|
with open(DEVICELIST_FILE, "wt") as dlfile: |
46
|
|
|
dlfile.write(prds_json) |
47
|
|
|
|
48
|
|
|
with open(DEVICELIST_FILE, "rt") as dlfile: |
49
|
|
|
prds = json.load(dlfile) |
50
|
|
|
|
51
|
|
|
if old_prds and output_diff: |
52
|
|
|
print_prd_diff(old_prds, prds) |
53
|
|
|
|
54
|
|
|
return prds |
55
|
|
|
|
56
|
|
|
|
57
|
|
|
def print_versions_diff(old_data: dict, new_data: dict): |
58
|
|
|
"""Print version changes between old and new databases.""" |
59
|
|
|
prd = new_data["curef"] |
60
|
|
|
if new_data["last_full"] != old_data["last_full"] and new_data["last_ota"] != old_data["last_ota"]: |
61
|
|
|
print("> {}: {} ⇨ {} (OTA: {} ⇨ {})".format( |
62
|
|
|
prd, |
63
|
|
|
ansi.CYAN_DARK + str(old_data["last_full"]) + ansi.RESET, |
64
|
|
|
ansi.CYAN + str(new_data["last_full"]) + ansi.RESET, |
65
|
|
|
ansi.YELLOW_DARK + str(old_data["last_ota"]) + ansi.RESET, |
66
|
|
|
ansi.YELLOW + str(new_data["last_ota"]) + ansi.RESET |
67
|
|
|
)) |
68
|
|
|
elif new_data["last_full"] != old_data["last_full"]: |
69
|
|
|
print("> {}: {} ⇨ {} (FULL)".format(prd, ansi.CYAN_DARK + str(old_data["last_full"]) + ansi.RESET, ansi.CYAN + str(new_data["last_full"]) + ansi.RESET)) |
70
|
|
|
elif new_data["last_ota"] != old_data["last_ota"]: |
71
|
|
|
print("> {}: {} ⇨ {} (OTA)".format(prd, ansi.YELLOW_DARK + str(old_data["last_ota"]) + ansi.RESET, ansi.YELLOW + str(new_data["last_ota"]) + ansi.RESET)) |
72
|
|
|
|
73
|
|
|
def print_removed_prds(prds_data: dict, removed_prds: list): |
74
|
|
|
for prd in removed_prds: |
75
|
|
|
print("> Removed device {} (was at {} / OTA: {}).".format(ansi.RED + prd + ansi.RESET, prds_data[prd]["last_full"], prds_data[prd]["last_ota"])) |
76
|
|
|
|
77
|
|
|
def print_added_prds(prds_data: dict, added_prds: list): |
78
|
|
|
for prd in added_prds: |
79
|
|
|
print("> New device {} ({} / OTA: {}).".format(ansi.GREEN + prd + ansi.RESET, prds_data[prd]["last_full"], prds_data[prd]["last_ota"])) |
80
|
|
|
|
81
|
|
|
def print_prd_diff(old_prds: dict, new_prds: dict): |
82
|
|
|
"""Print PRD changes between old and new databases.""" |
83
|
|
|
added_prds = [prd for prd in new_prds if prd not in old_prds] |
84
|
|
|
removed_prds = [prd for prd in old_prds if prd not in new_prds] |
85
|
|
|
print_removed_prds(old_prds, removed_prds) |
86
|
|
|
print_added_prds(new_prds, added_prds) |
87
|
|
|
for prd, pdata in new_prds.items(): |
88
|
|
|
if prd in added_prds: |
89
|
|
|
continue |
90
|
|
|
odata = old_prds[prd] |
91
|
|
|
print_versions_diff(odata, pdata) |
92
|
|
|
|