diff --git a/README.md b/README.md index 3d28054..41b5b9d 100644 --- a/README.md +++ b/README.md @@ -16,10 +16,8 @@ lp uses ``JSON`` to run commands. For example, command 1 would be called "1", co **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 7d364af..149cd61 100644 --- a/lp.py +++ b/lp.py @@ -1,93 +1,133 @@ #!/usr/bin/env python3 -import sys, os, requests, subprocess, platform +import argparse +import json +import logging +from pathlib import Path +import platform +import requests +import subprocess as sp +from typing import Any, List -# Define global variables -arg1 = None -arg2 = None -# Define command execution function -def exec_cmd(command): - stream = os.popen(command) - output = stream.read() - output +logging.basicConfig( + level=logging.INFO, + format="[%(asctime)s %(levelname)-.4s] %(message)s", + datefmt="%H:%M:%S", +) +log = logging.getLogger(__name__) - return output -# Define check if cmd exists function -def command_exists(command): +def exec_cmd(command: str) -> int: try: - subprocess.check_output(['which', command]) - return True - except subprocess.CalledProcessError: - return False + return sp.run(command.split()).returncode + except Exception as e: + log.error(e) + return -1 -# Define fetch json file function -def fetch_json_data(url): + +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.RequestExceptions as e: - print(f"[WARN] Error fetching JSON data {e}, Now exiting.") - sys.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 - - # 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 - - # Check value of arg1 and adjust argument parsing - if arg1 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.") + except requests.exceptions.RequestException as e: + log.error("Error fetching JSON data: %s, Now exiting.", e) + exit(1) + + +def read_preset(path: str) -> Any: + try: + with open(path) as f: + return json.load(f) + except Exception as e: + log.error("Error reading JSON data: %s", e) + 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) + + +def run_preset(preset: List[str]): + 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") + parser.add_argument( + "-a", + "--async", + action="store_true", + dest="is_async", + help="run preset asynchronously", + ) + return parser.parse_args() + + +def main(): + if platform.system() != "Linux": + log.error("lp can only be run on linux-based systems. Now exiting.") + exit(1) + + args = parse_args() + if Path(args.url).exists(): + preset = read_preset(args.url) + else: + preset = fetch_preset(args.url) + + 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 preset) + + "\nSome presets require root. Errors may occur if lp is not run as root" + ) + + 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") + if args.is_async: + run_preset_async(preset) else: - print(f"[WARN] Invalid option: {arg1}, Now Exiting") + run_preset(preset) + log.info("lp has finished running the preset. 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.") +if __name__ == "__main__": + main()