-
-
-
-
-
- Volume
@@ -77,7 +77,7 @@ >Volume
@@ -122,7 +122,7 @@Hot
@@ -139,7 +139,7 @@ >Cold
@@ -161,7 +161,7 @@ >{{ getTooltipForStirTime(getTypeForRecipeListAtIndex(i)) }}
@@ -301,7 +301,7 @@ > dict: - return {"accept":"application/json", "authorization": "Basic cGFraW46YWRtaW4xMjM="} - -def get_headers_for_blob() -> dict: - return {"authorization": "Basic cGFraW46YWRtaW4xMjM=", "Content-Type": "application/octet-stream"} - -def get_all_releases_url() -> str: - return f"{source}/api/v1/repos/pakin/taobin_recipe_manager/releases" - -def get_download_url(tag_id: int, asset_id: int) -> str: - return f"{source}/api/v1/repos/pakin/taobin_recipe_manager/releases/{tag_id}/assets/{asset_id}" - -# Flags -update_available_now = False - -# Logs -def save_to_log_file(msg: str): - with open("./patches/.updater.log", "a") as f: - f.write(f"{datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=7)))} : {msg}\n") - -# Redis -redis_client = redis.StrictRedis(host="localhost", port=6379, db=0) -MAIN_CHANNEL = "updater.noti" -pubsub = redis_client.pubsub() -pubsub.subscribe(MAIN_CHANNEL) - -# Settings -settings = {} - -# ------------------------------------------------------------------------------------------------ - -def test_path_from_settings(): - # from settings - path_list = settings["path"].keys() - - for path in path_list: - logger.debug(f"Checking {path}") - if not os.path.exists(settings["path"][path]["path"]): - logger.warning(f"Deps.fallback: {settings['path'][path]["path"]}") - # retry with fallback - if "fallback_paths" in settings['path'][path].keys(): - for fallback_path in settings['path'][path]["fallback_paths"]: - - # check kind of fallback - path_kind = fallback_path.split(":")[0] - path_real = fallback_path.split(":")[1] - - if os.path.exists(path_real): - logger.info(f"Deps.fallback[{path_kind}]: {path_real} = ok") - break - else: - logger.warning(f"Deps.fallback: skipped {path}") - else: - # path ok( - logger.info(f"Deps: {path} = ok") - - -def find_latest_version(rels: list[dict]) -> str: - major = 0 - minor = 0 - patch = 0 - for rel in rels: - if rel["tag_name"].startswith("v"): - version = rel["tag_name"][1:].split(".") - if len(version) == 3: - if int(version[0]) > major: - major = int(version[0]) - minor = int(version[1]) - patch = int(version[2]) - elif int(version[0]) == major: - if int(version[1]) > minor: - minor = int(version[1]) - patch = int(version[2]) - elif int(version[1]) == minor: - if int(version[2]) > patch: - patch = int(version[2]) - - return f"{major}.{minor}.{patch}" - -def get_version(rels: list[dict], tag: str) -> dict: - # assert that tag is valid - is_in_correct_format = tag.startswith("v") and len(tag[1:].split(".")) == 3 - if is_in_correct_format: - for rel in rels: - if rel["tag_name"] == tag: - return rel - -# -def check_version_job() -> dict: - logger.info("Checking version") - result = requests.get(get_all_releases_url(), timeout=10, headers=get_headers()) - if result.status_code == 200: - # connect to server ok! - releases = result.json() - if len(releases) > 0: - # find the latest {major}.{minor}.{patch} - expected_version = find_latest_version(releases) - # get data from list - # version data - latest_version_data = get_version(releases, f"v{expected_version}") - - - latest_id = latest_version_data["id"] - latest_asset_id = latest_version_data["assets"][0]["id"] - - # get download url - download_url = get_download_url(latest_id, latest_asset_id) - # download - logger.info(f"Found version {expected_version} from source") - - dl_result = requests.get(download_url, timeout=10, headers=get_headers()) - if dl_result.status_code == 200: - # download ok! - # logger.debug(f"data: {dl_result.json()}") - - dl_content = dl_result.json() - - dl_link = dl_content["browser_download_url"] - # download link is not usable without tailscale vpn - # cut uri, start at /attachments - dl_link = f"{source}{dl_link[dl_link.find("/attachments"):]}" - # logger.info(f"Download link: {dl_link}") - - is_already_latest = False - old_version = "" - # check dl and latest - if os.path.exists("./patches/latest.version"): - with open("./patches/latest.version", "r") as f: - latest_version = f.read() - - if latest_version == None or latest_version == "": - logger.warning("No version found on running machine. Starting overwrite ...") - save_to_log_file("Warning: version.overwrite caused by missing version on latest.version.") - old_version = "