"""Utilities.""" import os import re import subprocess import sys from pathlib import Path from typing import Any, Dict, List import requests from loguru import logger from requests import Response, Session from src.config import RevancedConfig from src.downloader.sources import APK_MIRROR_APK_CHECK from src.downloader.utils import status_code_200 from src.exceptions import ScrapingError default_build = [ "youtube", "youtube_music", ] possible_archs = ["armeabi-v7a", "x86", "x86_64", "arm64-v8a"] request_header = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (HTML, like Gecko)" " Chrome/96.0.4664.93 Safari/537.36", "Authorization": "Basic YXBpLWFwa3VwZGF0ZXI6cm01cmNmcnVVakt5MDRzTXB5TVBKWFc4", "Content-Type": "application/json", } bs4_parser = "html.parser" changelog_file = "changelog.md" request_timeout = 60 session = Session() session.headers["User-Agent"] = request_header["User-Agent"] def update_changelog(name: str, response: Dict[str, str]) -> None: """The function `update_changelog` updates the changelog file. Parameters ---------- name : str A string representing the name of the change or update. response : Dict[str, str] The `response` parameter is a dictionary that contains information about the changes made. The keys in the dictionary represent the type of change (e.g., "bug fix", "feature", "documentation"), and the values represent the specific changes made for each type. """ parent_repo = get_parent_repo() change_log = format_changelog(name, response, parent_repo) write_to_file(change_log) def format_changelog(name: str, response: Dict[str, str], parent_repo: str) -> str: """The `format_changelog` returns formatted changelog string. Parameters ---------- name : str The `name` parameter is a string that represents the name of the changelog. It is used to create a collapsible section in the formatted changelog. response : Dict[str, str] The `response` parameter is a dictionary that contains information about a release. It has the following keys: parent_repo : str The `parent_repo` parameter is a string that represents the URL or name of the parent repository. It is used to generate a footer in the formatted changelog, indicating that the changelogs were generated by a specific tool or script. Returns ------- a formatted changelog as a string. """ collapse_start = f"\n
👀 {name} \n\n" release_version = f"**Release Version** - [{response['tag_name']}]({response['html_url']})
" change_log = f"**Changelog** -
{response['body']}" publish_time = f"**Published at** -
{response['published_at']}" footer = f"
Change logs generated by [Docker Py Revanced]({parent_repo})\n" collapse_end = "
" return f"{collapse_start}{release_version}{change_log}{publish_time}{footer}{collapse_end}" def write_to_file(change_log: str) -> None: """The function `write_to_file` writes a given changelog string to a file. Parameters ---------- change_log : str A string representing the changelog that you want to write to the file. """ with Path(changelog_file).open("a", encoding="utf_8") as file1: file1.write(change_log) def get_parent_repo() -> str: """The `get_parent_repo()` function returns the URL of the parent repository. Returns ------- the URL of the parent repository, which is "https://github.com/nikhilbadyal/docker-py-revanced". """ return "https://github.com/nikhilbadyal/docker-py-revanced" def handle_request_response(response: Response, url: str) -> None: """The function handles the response of a GET request and raises an exception if the response code is not 200. Parameters ---------- response : Response The parameter `response` is of type `Response`, which is likely referring to a response object from an HTTP request. This object typically contains information about the response received from the server, such as the status code, headers, and response body. url: str The url on which request was made """ response_code = response.status_code if response_code != status_code_200: msg = f"Unable to downloaded assets. Reason - {response.text}" raise ScrapingError(msg, url=url) def slugify(string: str) -> str: """The `slugify` function converts a string to a slug format. Parameters ---------- string : str The `string` parameter is a string that you want to convert to a slug format. Returns ------- The function `slugify` returns a modified version of the input string in slug format. """ # Convert to lowercase modified_string = string.lower() # Remove special characters modified_string = re.sub(r"[^\w\s-]", "", modified_string) # Replace spaces with dashes modified_string = re.sub(r"\s+", "-", modified_string) # Remove consecutive dashes modified_string = re.sub(r"-+", "-", modified_string) # Remove leading and trailing dashes return modified_string.strip("-") def _check_version(output: str) -> None: """Check version.""" if "Runtime Environment" not in output: raise subprocess.CalledProcessError(-1, "java -version") if "17" not in output and "20" not in output: raise subprocess.CalledProcessError(-1, "java -version") def check_java() -> None: """The function `check_java` checks if Java version 17 or higher is installed. Returns ------- The function `check_java` does not return any value. """ try: jd = subprocess.check_output(["java", "-version"], stderr=subprocess.STDOUT).decode("utf-8") jd = jd[1:-1] _check_version(jd) logger.debug("Cool!! Java is available") except subprocess.CalledProcessError: logger.error("Java>= 17 must be installed") sys.exit(-1) def extra_downloads(config: RevancedConfig) -> None: """The function `extra_downloads` downloads extra files specified. Parameters ---------- config : RevancedConfig The `config` parameter is an instance of the `RevancedConfig` class. It is used to provide configuration settings for the download process. """ from src.app import APP try: for extra in config.extra_download_files: url, file_name = extra.split("@") file_name_without_extension, file_extension = os.path.splitext(file_name) new_file_name = f"{file_name_without_extension}-output{file_extension}" APP.download( url, config, assets_filter=f".*{file_extension}", file_name=new_file_name, ) except (ValueError, IndexError): logger.info("Unable to download extra file. Provide input in url@name.apk format.") def delete_old_changelog() -> None: """The function `delete_old_changelog` deleted old changelog file.""" Path(changelog_file).unlink(missing_ok=True) def apkmirror_status_check(package_name: str) -> Any: """The `apkmirror_status_check` function checks if an app exists on APKMirror. Parameters ---------- package_name : str The `package_name` parameter is a string that represents the name of the app package to check on APKMirror. Returns ------- the response from the APKMirror API as a JSON object. """ body = {"pnames": [package_name]} response = requests.post(APK_MIRROR_APK_CHECK, json=body, headers=request_header, timeout=60) return response.json() def contains_any_word(string: str, words: List[str]) -> bool: """Checks if a string contains any word.""" return any(word in string for word in words)