import re import requests import redis import time from http import HTTPStatus import semver from smelog.factory import BoundLogger from retrieve_and_save_mobile_versions.config import Config from retrieve_and_save_mobile_versions.constants import RETRY_COUNT, RETRY_WAIT from retrieve_and_save_mobile_versions.exceptions import RequestError from retrieve_and_save_mobile_versions.redis_db import get_version_key def send_request(url: str, logger: BoundLogger) -> requests.Response: """Method to send request for specified URL with retry logic. Arguments: url (str): URL value. logger: Logger object. Raises: RequestError: Raises when request response status code is not equal 200 or response body is empty. Returns: requests.Response: Response object. """ retry_count = 0 if not url: raise RequestError("URL data was not provided.") logger.info(f"Published app version URL: {url}") while True: response = requests.get(url, headers={"Cache-Control": "no-cache"}) if response.status_code == HTTPStatus.OK: break if response.status_code == HTTPStatus.TOO_MANY_REQUESTS and retry_count < RETRY_COUNT: retry_count += 1 logger.info(f"Request retry #{retry_count} to url {url}.") time.sleep(retry_count * RETRY_WAIT) continue raise RequestError(f"Unexpected response status code (url: {url}).", status_code=response.status_code) if not response: raise RequestError(f"Unexpected response value (url: {url}).") if not response.text: raise RequestError(f"Response body is empty (url: {url}).") return response def validate_version(version_str: str, logger: BoundLogger) -> bool: """Validate string version value format. Arguments: version_str (str): Version string value. logger: Logger object. Raises: VersionValidationError: If version value format is wrong or missing one. Returns: bool: True of version value is valid. """ pattern = re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}(\-.+)?(\+.+)?$") is_valid = bool(pattern.fullmatch(version_str.strip())) if not is_valid: logger.warning(f"String `{version_str}` is not a valid version value.") return is_valid def is_newer_version(stored_version: str, parsed_version: str) -> bool: """Check is parsed version value bigger than stored one. Arguments: stored_version (str): Latest stored in DB version value. parsed_version (str): Scrapped from website version data. Raises: ValueError: If some of versions values were not provided. Returns: bool: True if parsed version is newer that stored one. """ if stored_version and parsed_version: return semver.compare(stored_version, parsed_version) < 0 raise ValueError("Version data was not provided.") def clean_cached_os_version(os: str, config: Config, redis_client: redis.Redis or None, logger: BoundLogger) -> bool: keys_to_delete = [get_version_key(os, prefix) for prefix in config.APOLLO_LATEST_VERSION_KEY_PREFIX_LIST] if redis_client is None: return False num_of_items = redis_client.delete(*keys_to_delete) logger.info(f"Redis - {os}: num of deleted keys: {num_of_items}") return bool(num_of_items) def get_android_version(text: str): """ Args: text: response.text from Play Market Returns: """ # Play Market changed their page and for now it`s not possible to get an app version by key. # The version information is now on popup window shown after the button click that has no proper link # to get! # Though there is a place where possible version is for now version = None potential_version = re.search(r"\"\d{1,3}\.\d{1,3}\.\d{1,3}\"", text) if potential_version: version = potential_version.group()[1:-1] return version