From bcb985c32c6254611dc3104d3295875391097e95 Mon Sep 17 00:00:00 2001 From: Iam54r1n4 Date: Sun, 26 Jan 2025 08:12:56 +0000 Subject: [PATCH] Replace " with ' ODC --- core/cli_api.py | 148 ++++++++++++++++++++++++------------------------ 1 file changed, 75 insertions(+), 73 deletions(-) diff --git a/core/cli_api.py b/core/cli_api.py index 59397fd..47150dc 100644 --- a/core/cli_api.py +++ b/core/cli_api.py @@ -44,27 +44,27 @@ class Command(Enum): class HysteriaError(Exception): - """Base class for Hysteria-related exceptions.""" + '''Base class for Hysteria-related exceptions.''' pass class CommandExecutionError(HysteriaError): - """Raised when a command execution fails.""" + '''Raised when a command execution fails.''' pass class InvalidInputError(HysteriaError): - """Raised when the provided input is invalid.""" + '''Raised when the provided input is invalid.''' pass class PasswordGenerationError(HysteriaError): - """Raised when password generation fails.""" + '''Raised when password generation fails.''' pass class ScriptNotFoundError(HysteriaError): - """Raised when a required script is not found.""" + '''Raised when a required script is not found.''' pass # region Utils @@ -84,9 +84,9 @@ def run_cmd(command: list[str]) -> str | None: return result except subprocess.CalledProcessError as e: if DEBUG: - raise CommandExecutionError(f"Command execution failed: {e}\nOutput: {e.output.decode()}") + raise CommandExecutionError(f'Command execution failed: {e}\nOutput: {e.output.decode()}') else: - return None + return None return None @@ -98,7 +98,7 @@ def generate_password() -> str: try: return subprocess.check_output(['pwgen', '-s', '32', '1'], shell=False).decode().strip() except subprocess.CalledProcessError as e: - raise PasswordGenerationError(f"Failed to generate password: {e}") + raise PasswordGenerationError(f'Failed to generate password: {e}') # endregion @@ -108,43 +108,43 @@ def generate_password() -> str: def install_hysteria2(port: int, sni: str): - """ + ''' Installs Hysteria2 on the given port and uses the provided or default SNI value. - """ + ''' run_cmd(['bash', Command.INSTALL_HYSTERIA2.value, str(port), sni]) def uninstall_hysteria2(): - """Uninstalls Hysteria2.""" + '''Uninstalls Hysteria2.''' run_cmd(['bash', Command.UNINSTALL_HYSTERIA2.value]) def update_hysteria2(): - """Updates Hysteria2.""" + '''Updates Hysteria2.''' run_cmd(['bash', Command.UPDATE_HYSTERIA2.value]) def restart_hysteria2(): - """Restarts Hysteria2.""" + '''Restarts Hysteria2.''' run_cmd(['bash', Command.RESTART_HYSTERIA2.value]) def change_hysteria2_port(port: int): - """ + ''' Changes the port for Hysteria2. - """ + ''' run_cmd(['bash', Command.CHANGE_PORT_HYSTERIA2.value, str(port)]) def change_hysteria2_sni(sni: str): - """ + ''' Changes the SNI for Hysteria2. - """ + ''' run_cmd(['bash', Command.CHANGE_SNI_HYSTERIA2.value, sni]) def backup_hysteria(): - """Backups Hysteria configuration.""" + '''Backups Hysteria configuration.''' run_cmd(['bash', Command.BACKUP_HYSTERIA.value]) # endregion @@ -153,25 +153,25 @@ def backup_hysteria(): def list_users() -> dict | None: - """ + ''' Lists all users. - """ + ''' if res := run_cmd(['bash', Command.LIST_USERS.value]): return json.loads(res) def get_user(username: str) -> dict | None: - """ + ''' Retrieves information about a specific user. - """ + ''' if res := run_cmd(['bash', Command.GET_USER.value, '-u', str(username)]): return json.loads(res) def add_user(username: str, traffic_limit: int, expiration_days: int, password: str | None, creation_date: str | None): - """ + ''' Adds a new user with the given parameters. - """ + ''' if not password: password = generate_password() if not creation_date: @@ -180,9 +180,9 @@ def add_user(username: str, traffic_limit: int, expiration_days: int, password: def edit_user(username: str, new_username: str | None, new_traffic_limit: int | None, new_expiration_days: int | None, renew_password: bool, renew_creation_date: bool, blocked: bool): - """ + ''' Edits an existing user's details. - """ + ''' if not username: raise InvalidInputError('Error: username is required') if not any([new_username, new_traffic_limit, new_expiration_days, renew_password, renew_creation_date, blocked is not None]): @@ -194,11 +194,11 @@ def edit_user(username: str, new_username: str | None, new_traffic_limit: int | if renew_password: password = generate_password() else: - password = "" + password = '' if renew_creation_date: creation_date = datetime.now().strftime('%Y-%m-%d') else: - creation_date = "" + creation_date = '' command_args = [ 'bash', Command.EDIT_USER.value, @@ -214,23 +214,24 @@ def edit_user(username: str, new_username: str | None, new_traffic_limit: int | def reset_user(username: str): - """ + ''' Resets a user's configuration. - """ + ''' run_cmd(['bash', Command.RESET_USER.value, username]) def remove_user(username: str): - """ + ''' Removes a user by username. - """ + ''' run_cmd(['bash', Command.REMOVE_USER.value, username]) +# TODO: it's better to return json def show_user_uri(username: str, qrcode: bool, ipv: int, all: bool, singbox: bool, normalsub: bool) -> str | None: - """ + ''' Displays the URI for a user, with options for QR code and other formats. - """ + ''' command_args = ['bash', Command.SHOW_USER_URI.value, '-u', username] if qrcode: command_args.append('-qr') @@ -250,19 +251,20 @@ def show_user_uri(username: str, qrcode: bool, ipv: int, all: bool, singbox: boo def traffic_status(): - """Fetches traffic status.""" + '''Fetches traffic status.''' traffic.traffic_status() +# TODO: it's better to return json def server_info() -> str | None: - """Retrieves server information.""" + '''Retrieves server information.''' return run_cmd(['bash', Command.SERVER_INFO.value]) def manage_obfs(remove: bool, generate: bool): - """ + ''' Manages 'obfs' in Hysteria2 configuration. - """ + ''' if remove and generate: raise InvalidInputError('Error: You cannot use both --remove and --generate at the same time') elif remove: @@ -270,51 +272,51 @@ def manage_obfs(remove: bool, generate: bool): elif generate: run_cmd(['bash', Command.MANAGE_OBFS.value, '--generate']) else: - raise InvalidInputError("Error: Please specify either --remove or --generate.") + raise InvalidInputError('Error: Please specify either --remove or --generate.') def ip_address(edit: bool, ipv4: str, ipv6: str): - """ + ''' Manages IP address configuration with edit options. - """ + ''' if edit: if ipv4: run_cmd(['bash', Command.IP_ADD.value, 'edit', '-4', ipv4]) if ipv6: run_cmd(['bash', Command.IP_ADD.value, 'edit', '-6', ipv6]) if not ipv4 and not ipv6: - raise InvalidInputError("Error: --edit requires at least one of --ipv4 or --ipv6.") + raise InvalidInputError('Error: --edit requires at least one of --ipv4 or --ipv6.') else: run_cmd(['bash', Command.IP_ADD.value, 'add']) def update_geo(country: str): - """ + ''' Updates geographic data files based on the specified country. - """ + ''' script_path = Command.UPDATE_GEO.value try: subprocess.run(['python3', script_path, country.lower()], check=True) except subprocess.CalledProcessError as e: - raise CommandExecutionError(f"Failed to update geo files: {e}") + raise CommandExecutionError(f'Failed to update geo files: {e}') except FileNotFoundError: - raise ScriptNotFoundError(f"Script not found: {script_path}") + raise ScriptNotFoundError(f'Script not found: {script_path}') except Exception as e: - raise HysteriaError(f"An unexpected error occurred: {e}") + raise HysteriaError(f'An unexpected error occurred: {e}') def masquerade(remove: bool, enable: str): - """ + ''' Configures masquerade settings. - """ + ''' if remove and enable: - raise InvalidInputError("Error: You cannot use both --remove and --enable at the same time.") + raise InvalidInputError('Error: You cannot use both --remove and --enable at the same time.') if remove: run_cmd(['bash', Command.MASQUERADE_SCRIPT.value, '2']) elif enable: run_cmd(['bash', Command.MASQUERADE_SCRIPT.value, '1', enable]) else: - raise InvalidInputError("Error: Please specify either --remove or --enable.") + raise InvalidInputError('Error: Please specify either --remove or --enable.') # endregion @@ -322,33 +324,33 @@ def masquerade(remove: bool, enable: str): def install_tcp_brutal(): - """Installs TCP Brutal.""" + '''Installs TCP Brutal.''' run_cmd(['bash', Command.INSTALL_TCP_BRUTAL.value]) def install_warp(): - """Installs WARP.""" + '''Installs WARP.''' run_cmd(['bash', Command.INSTALL_WARP.value]) def uninstall_warp(): - """Uninstalls WARP.""" + '''Uninstalls WARP.''' run_cmd(['bash', Command.UNINSTALL_WARP.value]) def configure_warp(all: bool, popular_sites: bool, domestic_sites: bool, block_adult_sites: bool, warp_option: str, warp_key: str): - """ + ''' Configures WARP with various options. - """ + ''' if warp_option == 'warp plus' and not warp_key: - raise InvalidInputError("Error: WARP Plus key is required when 'warp plus' is selected.") + raise InvalidInputError('Error: WARP Plus key is required when \'warp plus\' is selected.') options = { - "all": 'true' if all else 'false', - "popular_sites": 'true' if popular_sites else 'false', - "domestic_sites": 'true' if domestic_sites else 'false', - "block_adult_sites": 'true' if block_adult_sites else 'false', - "warp_option": warp_option or '', - "warp_key": warp_key or '' + 'all': 'true' if all else 'false', + 'popular_sites': 'true' if popular_sites else 'false', + 'domestic_sites': 'true' if domestic_sites else 'false', + 'block_adult_sites': 'true' if block_adult_sites else 'false', + 'warp_option': warp_option or '', + 'warp_key': warp_key or '' } cmd_args = [ 'bash', Command.CONFIGURE_WARP.value, @@ -364,17 +366,17 @@ def configure_warp(all: bool, popular_sites: bool, domestic_sites: bool, block_a def warp_status() -> str | None: - """Checks the status of WARP.""" + '''Checks the status of WARP.''' return run_cmd(['bash', Command.STATUS_WARP.value]) def telegram(action: str, token: str, adminid: str): - """ + ''' Manages the Telegram bot with start/stop actions. - """ + ''' if action == 'start': if not token or not adminid: - raise InvalidInputError("Error: Both --token and --adminid are required for the start action.") + raise InvalidInputError('Error: Both --token and --adminid are required for the start action.') admin_ids = f'{adminid}' run_cmd(['bash', Command.INSTALL_TELEGRAMBOT.value, 'start', token, admin_ids]) elif action == 'stop': @@ -382,24 +384,24 @@ def telegram(action: str, token: str, adminid: str): def singbox(action: str, domain: str, port: int): - """ + ''' Manages Singbox with start/stop actions. - """ + ''' if action == 'start': if not domain or not port: - raise InvalidInputError("Error: Both --domain and --port are required for the start action.") + raise InvalidInputError('Error: Both --domain and --port are required for the start action.') run_cmd(['bash', Command.INSTALL_SINGBOX.value, 'start', domain, str(port)]) elif action == 'stop': run_cmd(['bash', Command.INSTALL_SINGBOX.value, 'stop']) def normalsub(action: str, domain: str, port: int): - """ + ''' Manages Normalsub with start/stop actions. - """ + ''' if action == 'start': if not domain or not port: - raise InvalidInputError("Error: Both --domain and --port are required for the start action.") + raise InvalidInputError('Error: Both --domain and --port are required for the start action.') run_cmd(['bash', Command.INSTALL_NORMALSUB.value, 'start', domain, str(port)]) elif action == 'stop': run_cmd(['bash', Command.INSTALL_NORMALSUB.value, 'stop'])