From 25afba1c35ade1fdfc58f10860b51758d979ef85 Mon Sep 17 00:00:00 2001 From: su55y Date: Thu, 4 Jan 2024 23:30:42 +0200 Subject: [PATCH 1/8] basic refactoring --- lp.py | 125 ++++++++++++++++++++++++++-------------------------------- 1 file changed, 55 insertions(+), 70 deletions(-) diff --git a/lp.py b/lp.py index 7d364af..5814d71 100644 --- a/lp.py +++ b/lp.py @@ -1,93 +1,78 @@ #!/usr/bin/env python3 -import sys, os, requests, subprocess, platform +import platform +import requests +import subprocess +import sys -# Define global variables -arg1 = None -arg2 = None # Define command execution function def exec_cmd(command): - stream = os.popen(command) - output = stream.read() - output + try: + return subprocess.run(command.split()).returncode + except Exception as e: + print(e) + return -1 - return output # Define check if cmd exists function -def command_exists(command): - try: - subprocess.check_output(['which', command]) - return True - except subprocess.CalledProcessError: - return False +# def command_exists(command): +# try: +# subprocess.check_output(["which", command]) +# return True +# except subprocess.CalledProcessError: +# return False + # Define fetch json file function def fetch_json_data(url): try: response = requests.get(url) - response.raise_for_status() # Raise an HTTPError for bad responses + response.raise_for_status() # Raise an HTTPError for bad responses return response.json() - except requests.exceptions.RequestExceptions as e: - print(f"[WARN] Error fetching JSON data {e}, Now exiting.") - sys.exit(1) + except requests.exceptions.RequestException as e: + print(f"[WARN] Error fetching JSON data: {e}, Now exiting.") + exit(1) -# Define extract values function -def extract_values(json_data): - values_list= [] - for item in json_data: - for key, value in item.items(): - values_list.append(value) - - return values_list - -def main(): # Define main function - global arg1, arg2 # Declare arg1 + arg2 as global variables +def main(): # Define main function + if platform.system() != "Linux": # Check if running on Linux, if not, exit. + print("[WARN] lp can only be run on linux-based systems. Now exiting.") + exit(1) # Check if the correct number of arguments has been defined by the user - if len(sys.argv) < 2: - print("[WARN] lp usage: {} [arg1] [arg2]".format(sys.argv[0])) - - # Extract the arguments - arg1 = sys.argv[1] if len(sys.argv) > 1 else None - arg2 = sys.argv[2] if len(sys.argv) > 1 else None + if len(sys.argv) != 2: + print("[WARN] lp usage: {} [PRESET_URL]".format(sys.argv[0])) + exit(1) # Check value of arg1 and adjust argument parsing - if arg1 in ["h", "help", "?"]: + if any(sys.argv[1] == help_opt for help_opt in ["-h", "--help"]): print("Help coming soon idk") - elif arg1 in ["r", "run"]: - if len(sys.argv) > 2: # Check if arg2 exists - arg2 = sys.argv[2] # begin script after getting arg2 again - json_data = fetch_json_data(arg2) - - if json_data is not None: - values_list = extract_values(json_data) - print("[INFO] Selected Preset:", values_list) - print("[INFO] Some presets require root. Errors may occur if lp is not run as root") - index = 0 - continueBoolean = input("Would you like to continue? [y/n]") - if continueBoolean in ["y", "Y"]: - print("[INFO] This may take some time depending on the preset. Please wait") - while index < len(values_list): - current_entry = values_list[index] - # exec script - exec_cmd(current_entry) - # Increment the index for the next iteration - index += 1 - elif continueBoolean in ["n", "N"]: - print(f"[WARN] {continueBoolean} was selected. Aborting preset and exiting.") - else: - print("[WARN] Please select a valid option. Aborting preset and exiting.") - print("[INFO] lp has finished running the preset. Now exiting.") - else: - print(f"[WARN] arg2 is required for {arg1} to properly function. Now exiting.") - else: - print(f"[WARN] Invalid option: {arg1}, Now Exiting") - -if platform.system() == "Linux": # Check if running on Linux, if not, exit. - if __name__ == "__main__": - main() -else: - print("[WARN] lp can only be run on linux-based systems. Now exiting.") + exit(0) + + # Extract the arguments + url = sys.argv[1] + + # Extract values list + values_list = list(fetch_json_data(url).values()) + + print( + f"[INFO] Selected Preset: {values_list}\n" + "[INFO] Some presets require root. Errors may occur if lp is not run as root" + ) + continueBoolean = input("Would you like to continue? [y/n]") + if not any(continueBoolean == y for y in ["y", "Y"]): + print(f"[WARN] {continueBoolean} was selected. Aborting preset and exiting.") + exit(0) + + print("[INFO] This may take some time depending on the preset. Please wait") + for cmd in values_list: + print(f"{cmd!r} running...") + status = exec_cmd(cmd) + if status != 0: + print(f"[WARN] cmd {cmd!r} failed. Exit status {status}") + print("[INFO] lp has finished running the preset. Now exiting.") + +if __name__ == "__main__": + main() From fbeb05971cef35e92f389888f99354ad034ab622 Mon Sep 17 00:00:00 2001 From: su55y Date: Thu, 4 Jan 2024 23:32:37 +0200 Subject: [PATCH 2/8] fixed readme --- README.md | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 3d28054..ca7d85a 100644 --- a/README.md +++ b/README.md @@ -15,11 +15,9 @@ I'll add a better and faster installation script later on. lp uses ``JSON`` to run commands. For example, command 1 would be called "1", command 2 would be called "2", etc. **Example:** - Make sure to add ``-y`` if you're using apt. ``` -[ - { - "1": "apt-get install btop -y", - "2": "apt-get install htop -y"" - } -] +{ + "1": "apt-get install btop -y", + "2": "apt-get install htop -y" +} ``` Remember, if you're using github to host your preset file, you'll need the raw file url. From a50351e4efbc1de1c8ca2d28f703d28bd84419b1 Mon Sep 17 00:00:00 2001 From: su55y Date: Thu, 4 Jan 2024 23:44:10 +0200 Subject: [PATCH 3/8] change json to list --- README.md | 8 ++++---- lp.py | 17 ++++++++++++++--- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index ca7d85a..41b5b9d 100644 --- a/README.md +++ b/README.md @@ -15,9 +15,9 @@ I'll add a better and faster installation script later on. lp uses ``JSON`` to run commands. For example, command 1 would be called "1", command 2 would be called "2", etc. **Example:** - Make sure to add ``-y`` if you're using apt. ``` -{ - "1": "apt-get install btop -y", - "2": "apt-get install htop -y" -} +[ + "apt-get install btop -y", + "apt-get install htop -y" +] ``` Remember, if you're using github to host your preset file, you'll need the raw file url. diff --git a/lp.py b/lp.py index 5814d71..6e30b84 100644 --- a/lp.py +++ b/lp.py @@ -35,6 +35,11 @@ def fetch_json_data(url): exit(1) +# json validation func +def validate_json_data(data): + return isinstance(data, list) and all(isinstance(v, str) for v in data) + + def main(): # Define main function if platform.system() != "Linux": # Check if running on Linux, if not, exit. print("[WARN] lp can only be run on linux-based systems. Now exiting.") @@ -54,11 +59,17 @@ def main(): # Define main function url = sys.argv[1] # Extract values list - values_list = list(fetch_json_data(url).values()) + values_list = fetch_json_data(url) + + # Validate data + if not validate_json_data(values_list): + print("[WARN] Invalid json format, should be list[str]") + exit(1) print( - f"[INFO] Selected Preset: {values_list}\n" - "[INFO] Some presets require root. Errors may occur if lp is not run as root" + "[INFO] Selected Preset:\n" + + "\n".join(f" {v!r}" for v in values_list) + + "\n[INFO] Some presets require root. Errors may occur if lp is not run as root" ) continueBoolean = input("Would you like to continue? [y/n]") if not any(continueBoolean == y for y in ["y", "Y"]): From c0a46829d1f1a583b0e361d0db5ff96e5334533f Mon Sep 17 00:00:00 2001 From: su55y Date: Thu, 4 Jan 2024 23:50:17 +0200 Subject: [PATCH 4/8] std args parsing, accept local preset --- lp.py | 39 ++++++++++++++++++++++++--------------- 1 file changed, 24 insertions(+), 15 deletions(-) diff --git a/lp.py b/lp.py index 6e30b84..d5e6527 100644 --- a/lp.py +++ b/lp.py @@ -1,13 +1,15 @@ #!/usr/bin/env python3 +import argparse +import json +from pathlib import Path import platform import requests import subprocess -import sys # Define command execution function -def exec_cmd(command): +def exec_cmd(command: str) -> int: try: return subprocess.run(command.split()).returncode except Exception as e: @@ -35,31 +37,38 @@ def fetch_json_data(url): exit(1) +# read preset from local json file +def read_json_data(path): + try: + with open(path) as f: + return json.load(f) + except: + exit(1) + + # json validation func def validate_json_data(data): return isinstance(data, list) and all(isinstance(v, str) for v in data) +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument("url", metavar="PRESET_URL") + return parser.parse_args() + + def main(): # Define main function if platform.system() != "Linux": # Check if running on Linux, if not, exit. print("[WARN] lp can only be run on linux-based systems. Now exiting.") exit(1) - # Check if the correct number of arguments has been defined by the user - if len(sys.argv) != 2: - print("[WARN] lp usage: {} [PRESET_URL]".format(sys.argv[0])) - exit(1) - - # Check value of arg1 and adjust argument parsing - if any(sys.argv[1] == help_opt for help_opt in ["-h", "--help"]): - print("Help coming soon idk") - exit(0) - - # Extract the arguments - url = sys.argv[1] + args = parse_args() # Extract values list - values_list = fetch_json_data(url) + if Path(args.url).exists(): + values_list = read_json_data(args.url) + else: + values_list = fetch_json_data(args.url) # Validate data if not validate_json_data(values_list): From db046c0cc652af705a92ee1af373374a641d4179 Mon Sep 17 00:00:00 2001 From: su55y Date: Fri, 5 Jan 2024 22:59:18 +0200 Subject: [PATCH 5/8] logging added --- lp.py | 48 ++++++++++++++++++++++++++++-------------------- 1 file changed, 28 insertions(+), 20 deletions(-) diff --git a/lp.py b/lp.py index d5e6527..3394bc1 100644 --- a/lp.py +++ b/lp.py @@ -2,10 +2,18 @@ import argparse import json +import logging from pathlib import Path -import platform -import requests -import subprocess +import sys, requests, subprocess, platform +from typing import Any + + +logging.basicConfig( + level=logging.INFO, + format="[%(asctime)s %(levelname)-.4s] %(message)s", + datefmt="%H:%M:%S", +) +log = logging.getLogger(__name__) # Define command execution function @@ -13,7 +21,7 @@ def exec_cmd(command: str) -> int: try: return subprocess.run(command.split()).returncode except Exception as e: - print(e) + log.error(e) return -1 @@ -27,18 +35,18 @@ def exec_cmd(command: str) -> int: # Define fetch json file function -def fetch_json_data(url): +def fetch_json_data(url) -> Any: try: response = requests.get(url) response.raise_for_status() # Raise an HTTPError for bad responses return response.json() except requests.exceptions.RequestException as e: - print(f"[WARN] Error fetching JSON data: {e}, Now exiting.") - exit(1) + log.error("Error fetching JSON data: %s, Now exiting.", e) + sys.exit(1) # read preset from local json file -def read_json_data(path): +def read_json_data(path: Path) -> Any: try: with open(path) as f: return json.load(f) @@ -47,11 +55,11 @@ def read_json_data(path): # json validation func -def validate_json_data(data): +def validate_json_data(data) -> bool: return isinstance(data, list) and all(isinstance(v, str) for v in data) -def parse_args(): +def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("url", metavar="PRESET_URL") return parser.parse_args() @@ -59,7 +67,7 @@ def parse_args(): def main(): # Define main function if platform.system() != "Linux": # Check if running on Linux, if not, exit. - print("[WARN] lp can only be run on linux-based systems. Now exiting.") + log.error("lp can only be run on linux-based systems. Now exiting.") exit(1) args = parse_args() @@ -72,26 +80,26 @@ def main(): # Define main function # Validate data if not validate_json_data(values_list): - print("[WARN] Invalid json format, should be list[str]") + log.error("Invalid json format, should be list[str]") exit(1) - print( - "[INFO] Selected Preset:\n" + log.info( + "Selected Preset:\n" + "\n".join(f" {v!r}" for v in values_list) - + "\n[INFO] Some presets require root. Errors may occur if lp is not run as root" + + "\nSome presets require root. Errors may occur if lp is not run as root" ) continueBoolean = input("Would you like to continue? [y/n]") if not any(continueBoolean == y for y in ["y", "Y"]): - print(f"[WARN] {continueBoolean} was selected. Aborting preset and exiting.") + log.info(f"{continueBoolean!r} was selected. Aborting preset and exiting.") exit(0) - print("[INFO] This may take some time depending on the preset. Please wait") + log.info("This may take some time depending on the preset. Please wait") for cmd in values_list: - print(f"{cmd!r} running...") + log.info(f"{cmd!r} running...") status = exec_cmd(cmd) if status != 0: - print(f"[WARN] cmd {cmd!r} failed. Exit status {status}") - print("[INFO] lp has finished running the preset. Now exiting.") + log.warning(f"cmd {cmd!r} failed. Exit status {status}") + log.info("lp has finished running the preset. Now exiting.") if __name__ == "__main__": From 2d1bde58d793400f98180b4fe0705d3ec722f2e2 Mon Sep 17 00:00:00 2001 From: su55y Date: Fri, 5 Jan 2024 23:01:59 +0200 Subject: [PATCH 6/8] naming, type hints --- lp.py | 68 ++++++++++++++++++++++++++--------------------------------- 1 file changed, 30 insertions(+), 38 deletions(-) diff --git a/lp.py b/lp.py index 3394bc1..777bf0c 100644 --- a/lp.py +++ b/lp.py @@ -4,8 +4,10 @@ import json import logging from pathlib import Path -import sys, requests, subprocess, platform -from typing import Any +import platform +import requests +import subprocess as sp +from typing import Any, List logging.basicConfig( @@ -16,46 +18,34 @@ log = logging.getLogger(__name__) -# Define command execution function def exec_cmd(command: str) -> int: try: - return subprocess.run(command.split()).returncode + return sp.run(command.split()).returncode except Exception as e: log.error(e) return -1 -# Define check if cmd exists function -# def command_exists(command): -# try: -# subprocess.check_output(["which", command]) -# return True -# except subprocess.CalledProcessError: -# return False - - -# Define fetch json file function -def fetch_json_data(url) -> Any: +def fetch_preset(url: str) -> Any: try: response = requests.get(url) response.raise_for_status() # Raise an HTTPError for bad responses return response.json() except requests.exceptions.RequestException as e: log.error("Error fetching JSON data: %s, Now exiting.", e) - sys.exit(1) + exit(1) -# read preset from local json file -def read_json_data(path: Path) -> Any: +def read_preset(path: str) -> Any: try: with open(path) as f: return json.load(f) - except: + except Exception as e: + log.error("Error reading JSON data: %s", e) exit(1) -# json validation func -def validate_json_data(data) -> bool: +def validate_json_data(data: Any) -> bool: return isinstance(data, list) and all(isinstance(v, str) for v in data) @@ -65,40 +55,42 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def main(): # Define main function - if platform.system() != "Linux": # Check if running on Linux, if not, exit. +def run_preset(preset: List[str]): + for i, cmd in enumerate(preset): + log.info(f"[{i+1}/{len(preset)}] {cmd!r} running...") + status = exec_cmd(cmd) + if status != 0: + log.warning(f"cmd {cmd!r} failed. Exit status {status}") + + +def main(): + if platform.system() != "Linux": log.error("lp can only be run on linux-based systems. Now exiting.") exit(1) args = parse_args() - - # Extract values list if Path(args.url).exists(): - values_list = read_json_data(args.url) + preset = read_preset(args.url) else: - values_list = fetch_json_data(args.url) + preset = fetch_preset(args.url) - # Validate data - if not validate_json_data(values_list): + if not validate_json_data(preset): log.error("Invalid json format, should be list[str]") exit(1) log.info( "Selected Preset:\n" - + "\n".join(f" {v!r}" for v in values_list) + + "\n".join(f" {v!r}" for v in preset) + "\nSome presets require root. Errors may occur if lp is not run as root" ) - continueBoolean = input("Would you like to continue? [y/n]") - if not any(continueBoolean == y for y in ["y", "Y"]): - log.info(f"{continueBoolean!r} was selected. Aborting preset and exiting.") + + resp = input("Would you like to continue? [y/n]") + if not any(resp == y for y in ["y", "Y"]): + log.info(f"{resp!r} was selected. Aborting preset and exiting.") exit(0) log.info("This may take some time depending on the preset. Please wait") - for cmd in values_list: - log.info(f"{cmd!r} running...") - status = exec_cmd(cmd) - if status != 0: - log.warning(f"cmd {cmd!r} failed. Exit status {status}") + run_preset(preset) log.info("lp has finished running the preset. Now exiting.") From f38b695961a8e7a966522e6e11f172b1c34591e7 Mon Sep 17 00:00:00 2001 From: su55y Date: Sun, 7 Jan 2024 23:53:11 +0200 Subject: [PATCH 7/8] fmt --- lp.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/lp.py b/lp.py index 777bf0c..1ec651e 100644 --- a/lp.py +++ b/lp.py @@ -29,7 +29,7 @@ def exec_cmd(command: str) -> int: def fetch_preset(url: str) -> Any: try: response = requests.get(url) - response.raise_for_status() # Raise an HTTPError for bad responses + response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: log.error("Error fetching JSON data: %s, Now exiting.", e) @@ -49,20 +49,20 @@ def validate_json_data(data: Any) -> bool: return isinstance(data, list) and all(isinstance(v, str) for v in data) -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser() - parser.add_argument("url", metavar="PRESET_URL") - return parser.parse_args() - - def run_preset(preset: List[str]): - for i, cmd in enumerate(preset): - log.info(f"[{i+1}/{len(preset)}] {cmd!r} running...") + for i, cmd in enumerate(preset, start=1): + log.info(f"[{i}/{len(preset)}] {cmd!r} running...") status = exec_cmd(cmd) if status != 0: log.warning(f"cmd {cmd!r} failed. Exit status {status}") +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("url", metavar="PRESET_URL") + return parser.parse_args() + + def main(): if platform.system() != "Linux": log.error("lp can only be run on linux-based systems. Now exiting.") From ae25081eef116016193248c869c08a652f02fa42 Mon Sep 17 00:00:00 2001 From: su55y Date: Mon, 8 Jan 2024 00:11:36 +0200 Subject: [PATCH 8/8] async implementation --- lp.py | 37 ++++++++++++++++++++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/lp.py b/lp.py index 1ec651e..149cd61 100644 --- a/lp.py +++ b/lp.py @@ -45,6 +45,31 @@ def read_preset(path: str) -> Any: exit(1) +def run_preset_async(preset: List[str]): + import asyncio + + async def exec_cmd_async(command: str, i: int): + try: + process = await asyncio.create_subprocess_shell(command) + return i, await process.wait() + except Exception as e: + log.error(e) + return i, -1 + + async def run_(): + task_ids = set(range(1, len(preset) + 1)) + tasks = [exec_cmd_async(cmd, i) for i, cmd in enumerate(preset, start=1)] + for future in asyncio.as_completed(tasks): + i, returncode = await future + task_ids.discard(i) + left = len(preset) - len(task_ids) + log.info( + f"[{left}/{len(preset)}] {preset[i-1]!r} done with return code {returncode}" + ) + + asyncio.run(run_()) + + def validate_json_data(data: Any) -> bool: return isinstance(data, list) and all(isinstance(v, str) for v in data) @@ -60,6 +85,13 @@ def run_preset(preset: List[str]): def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("url", metavar="PRESET_URL") + parser.add_argument( + "-a", + "--async", + action="store_true", + dest="is_async", + help="run preset asynchronously", + ) return parser.parse_args() @@ -90,7 +122,10 @@ def main(): exit(0) log.info("This may take some time depending on the preset. Please wait") - run_preset(preset) + if args.is_async: + run_preset_async(preset) + else: + run_preset(preset) log.info("lp has finished running the preset. Now exiting.")