Merge pull request #12 from carlossilva2/dev

feat(Refactor-&-Remote-Store): Removed Registry Action from Core; Imp…

Author: carlossilva2

.gitignore

@@ -27,3 +27,5 @@ wheels/
 *.egg-info/
 .installed.cfg
 *.egg
+
+regutils.py

Tasker/__init__.py

@@ -1,42 +1,29 @@
 import chalk
 
-from Tasker.parser import Parser
-
-from Tasker.operations import Command
-from Tasker.operations import Echo
-from Tasker.operations import Input
-from Tasker.operations import Copy
-from Tasker.operations import Delete
-from Tasker.operations import Request
-from Tasker.operations import Registry
-from Tasker.operations import Move
-
-from Tasker.cli import get_logger
-from Tasker.cli import get_args
-
-from Tasker.inspector import inspect
-
 from Tasker.__version__ import __version__
+from Tasker.cli import get_args, get_logger
+from Tasker.inspector import inspect
+from Tasker.operations import Command, Copy, Delete, Echo, Input, Move, Request
+from Tasker.parser import Parser
 
 __all__ = [
- #Colorizer
+ # Colorizer
  "chalk",
- #Parser
+ # Parser
  "Parser",
- #Operations
+ # Operations
  "Command",
  "Echo",
  "Input",
  "Copy",
  "Delete",
  "Request",
- "Registry",
  "Move",
- #CLI
+ # CLI
  "get_logger",
  "get_args",
- #Inspector
+ # Inspector
  "inspect",
- #Version
- "__version__"
-]
+ # Version
+ "__version__",
+]

Tasker/__main__.py

@@ -60,6 +60,24 @@ def main() -> None:
  )
  else:
  logger.error("Path and Name flags are required when creating Aliases")
+ elif args.action == "install":
+ if args.Extension is None:
+ logger.error("Missing Extension flag `-e`. Check Help for command syntax")
+ sys.exit(1)
+ Parser.install_remote_extension(args.Extension, logger)
+ elif args.action == "uninstall":
+ if args.Extension is None:
+ logger.error("Missing Extension flag `-e`. Check Help for command syntax")
+ sys.exit(1)
+ Parser.uninstall_extension(args.Extension, logger)
+ elif args.action == "remote-search" or args.action == "remote-list":
+ if args.action == "remote-search":
+ if args.Extension is None:
+ logger.error("Missing Extension flag `-e`. Check Help for command syntax")
+ sys.exit(1)
+ Parser.search_remote(args.Extension, logger)
+ else:
+ Parser.list_remote(logger)
  else:
  logger.error("Check Help for command syntax")
 

Tasker/__version__.py

@@ -1 +1 @@
-__version__ = "0.5.3"
+__version__ = "0.6.0"

Tasker/cli.py

@@ -30,7 +30,18 @@ def get_args(version: str, logger: logging.Logger) -> argparse.Namespace:
  parser.add_argument(
  "action",
  help="What action should Tasker perform",
- choices=["list", "execute", "create", "edit", "extension", "alias"],
+ choices=[
+ "list",
+ "execute",
+ "create",
+ "edit",
+ "extension",
+ "alias",
+ "install",
+ "uninstall",
+ "remote-list",
+ "remote-search",
+ ],
  )
 
  options = parser.add_argument_group("parameters")
@@ -42,6 +53,14 @@ def get_args(version: str, logger: logging.Logger) -> argparse.Namespace:
  required=False,
  help=f"Instruction Set name flag. Use {chalk.green('`tasker list`')} for a list with all InstructionSets",
  )
+ options.add_argument(
+ "-e",
+ "--Extension",
+ type=str,
+ metavar="",
+ required=False,
+ help="Usable only on `install`",
+ )
  options.add_argument(
  "-f",
  "--File",

Tasker/common.py

@@ -1,4 +1,6 @@
 import os.path as Path
+import subprocess
+import sys
 from hashlib import md5
 from os import listdir
 from time import time
@@ -73,3 +75,14 @@ def reset(self) -> None:
  self.start_time = 0.0
  self.end_time = 0.0
  self.ellapsed_time = ""
+
+
+def pip(package: str):
+ return subprocess.check_call([sys.executable, "-m", "pip", "install", package])
+
+
+def pip_freeze() -> list[str]:
+ return [
+ _.decode().split("==")[0]
+ for _ in subprocess.check_output([sys.executable, "-m", "pip", "freeze"]).split()
+ ]

Tasker/operations.py

@@ -6,25 +6,13 @@
 import chalk
 import requests
 
-from Tasker.regutils import backup
-
 from .common import alias, get_file_name, ref
 from .inspector import implements
 from .types import OperationType as Operation
 from .types import ParserType as Parser
-from .types import Registry as RegType
 from .types import Task
 
-try:
- from .regutils import (
- create_key,
- get_key_value,
- get_type,
- parse_input,
- set_key_value,
- )
-except Exception:
- not_imported = True
+# from Tasker.regutils import backup
 
 
 @implements(Operation)
@@ -322,69 +310,6 @@ def get_state(self) -> bool:
  return self.__internal_state
 
 
-@implements(Operation)
-class Registry(Operation):
- "Registry Action"
-
- __annotations__ = {
- "name": "Registry Action",
- "intent": "Change Windows Registry values/keys",
- }
-
- def __init__(self, ctx: Parser, task: RegType, logger: Logger) -> None:
- if ctx.system != "Windows":
- ctx.abort(f"Operation not available on {ctx.system} Systems!")
- self.regex = r"\w>"
- self.context = ctx # Parser Context
- self.task = task # Current assigned Task
- self.logger = logger
- self.affected_files: list[str] = []
- self.__internal_state = True # Faulty execution flag
- self._type = "registry"
- ref(self)
- alias(self)
-
- def execute(self) -> None:
- self.path = ">".join([self.task["start_key"], self.task["key"]])
- if self.task["function"] == "get":
- self.value = get_key_value(self.path, self.logger)
- elif self.task["function"] == "set":
- if self.task["value"] == "" or self.task["value"] == None:
- self.context.abort("No value was provided")
- self.safeguard = get_key_value(self.path, self.logger)
- parsed = parse_input(self.path)
- values = parsed[1]
- self.v = values[-1]
- values.pop()
- self.correct_path = ">".join([self.task["start_key"], *values])
- set_key_value(
- self.correct_path, self.v, get_type(self.task["type"]), self.task["value"]
- )
- self.value = get_key_value(self.path, self.logger)
- elif self.task["function"] == "create":
- create_key(self.path, self.task["value"])
- elif self.task["function"] == "backup":
- raise NotImplementedError(
- "Feature under development. Refer to future versions"
- )
- backup(self.task["start_key"], self.task["key"], self.task["rename"])
-
- def rollback(self) -> None:
- if self.task["function"] == "set":
- set_key_value(
- self.correct_path, self.v, get_type(self.task["type"]), self.safeguard
- )
- self.logger.warn(f"Rolled back \"{self.task['name']}\" task")
-
- def set_state(self, state: bool) -> None:
- "Sets the state for the Internal Fault flag"
- self.__internal_state = state
-
- def get_state(self) -> bool:
- "Returns the of the Internal Fault flag"
- return self.__internal_state
-
-
 @implements(Operation)
 class Input(Operation):
  "Input Action"

Tasker/parser.py

@@ -5,14 +5,15 @@
 import sys
 from hashlib import md5
 from importlib.machinery import SourceFileLoader as importer
-from logging import Logger
+from logging import WARNING, Logger, getLogger
 from time import time
 from typing import List, Literal, Union
 from webbrowser import open as FileOpener
 
 import chalk
+from requests import get
 
-from .common import Timer
+from .common import Timer, pip, pip_freeze
 from .inspector import implements
 from .operations import *
 from .types import (
@@ -25,7 +26,6 @@
  OP_INPUT,
  OP_INSTRUCTION,
  OP_MOVE,
- OP_REGISTRY,
  OP_REQUEST,
  OP_TASK,
  OP_ZIP,
@@ -146,10 +146,6 @@ def __execute(self, task: Task) -> bool:
  r = Request(self, task, self.logger)
  self.__operation_stack.append(r)
  r.execute()
- elif task["operation"] == "registry":
- reg = Registry(self, task, self.logger)
- self.__operation_stack.append(reg)
- reg.execute()
  elif task["operation"] == "custom":
  ex = next(
  (e for e in self.extensions if e["summon"] == task["extension_name"]),
@@ -468,3 +464,94 @@ def add_alias(alias: Alias, logger: Logger) -> None:
  sys.exit(1)
  settings["alias"].append(alias)
  json.dump(settings, open(f"{root}/.tasker/config.json", "w"), indent=4)
+
+ @staticmethod
+ def install_remote_extension(extension: str, logger: Logger) -> bool:
+ Parser.do_config()
+ root = Path.expanduser("~")
+ # Remove DEBUG WARNINGS
+ getLogger("requests").setLevel(WARNING)
+ getLogger("urllib3").setLevel(WARNING)
+ # Get Context file
+ context = get(
+ "https://raw.githubusercontent.com/carlossilva2/pyTasker-actions/main/context.json"
+ ).json()
+ # Check if extension exists on remote
+ if extension not in context.keys():
+ logger.error("Extension does not exist. Check spelling")
+ return False
+ # Check if extension is allowed on Current OS
+ os = platform.system()
+ if context[extension]["platform"] != os and context[extension]["platform"] != "*":
+ logger.error(f"Extension cannot be installed in '{os}'")
+ return False
+ # Check for extension dependencies
+ existing_modules = pip_freeze()
+ for dep in context[extension]["dependencies"]:
+ if dep not in existing_modules:
+ pip(dep)
+ # Retrieve Template
+ template = get(context[extension]["extension"]).text
+ # Install Extension locally
+ _n = md5(f"{time()}_{extension}".encode("UTF-8")).hexdigest()[:10]
+ f_name = f"extension_{_n}.py"
+ with open(f"{root}/.tasker/Templates/{f_name}", "w") as ex:
+ ex.write(template)
+ ex.close()
+ j: Settings = json.load(open(f"{root}/.tasker/config.json"))
+ j["extensions"].append(
+ {
+ "name": extension,
+ "file": f_name,
+ "path": f"{root}/.tasker/Templates/{f_name}",
+ }
+ )
+ json.dump(j, open(f"{root}/.tasker/config.json", "w"), indent=4)
+ return True
+
+ @staticmethod
+ def uninstall_extension(extension: str, logger: Logger) -> None:
+ Parser.do_config()
+ root = Path.expanduser("~")
+ j: Settings = json.load(open(f"{root}/.tasker/config.json"))
+ if extension not in [_["name"] for _ in j["extensions"]]:
+ logger.error(f"{extension} is not installed")
+ sys.exit(1)
+ index = None
+ for i, ex in enumerate(j["extensions"]):
+ if ex["name"] == extension:
+ index = i
+ os.remove(ex["path"])
+ break
+ j["extensions"].pop(index)
+ json.dump(j, open(f"{root}/.tasker/config.json", "w"), indent=4)
+ logger.debug("Extension removed successfully")
+
+ @staticmethod
+ def search_remote(extension: str, logger: Logger) -> None:
+ Parser.do_config()
+ # Remove DEBUG WARNINGS
+ getLogger("requests").setLevel(WARNING)
+ getLogger("urllib3").setLevel(WARNING)
+ # Get Context file
+ context = get(
+ "https://raw.githubusercontent.com/carlossilva2/pyTasker-actions/main/context.json"
+ ).json()
+ # Check if extension exists on remote
+ if extension not in context.keys():
+ logger.error("Extension does not exist. Check spelling")
+ sys.exit(1)
+ logger.debug(f"{extension}=={context[extension]['version']}")
+
+ @staticmethod
+ def list_remote(logger: Logger) -> None:
+ Parser.do_config()
+ # Remove DEBUG WARNINGS
+ getLogger("requests").setLevel(WARNING)
+ getLogger("urllib3").setLevel(WARNING)
+ # Get Context file
+ context = get(
+ "https://raw.githubusercontent.com/carlossilva2/pyTasker-actions/main/context.json"
+ ).json()
+ for extension in context.keys():
+ logger.debug(f"{extension}=={context[extension]['version']}")