Dev

dev.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "automation_file_dev"
-version = "0.0.6"
+version = "0.0.7"
 authors = [
     { name = "JE-Chen", email = "zenmailman@gmail.com" },
 ]

file_automation/__init__.py

@@ -17,8 +17,8 @@
 
 __all__ = [
     "copy_file", "rename_file", "remove_file", "copy_all_file_to_dir", "copy_specify_extension_file",
-    "copy_dir", "create_dir", "copy_specify_extension_file", "remove_dir_tree",
-    "zip_dir", "zip_file", "zip_info", "zip_file_info", "set_zip_password", "unzip_file", "read_zip_file",
+    "copy_dir", "create_dir", "remove_dir_tree", "zip_dir", "zip_file", "zip_info",
+    "zip_file_info", "set_zip_password", "unzip_file", "read_zip_file",
     "unzip_all", "driver_instance", "search_all_file", "search_field", "search_file_mimetype",
     "upload_dir_to_folder", "upload_to_folder", "upload_dir_to_drive", "upload_to_drive",
     "add_folder", "share_file_to_anyone", "share_file_to_domain", "share_file_to_user",

file_automation/utils/callback/callback_function_executor.py

@@ -0,0 +1,95 @@
+import typing
+from sys import stderr
+
+from file_automation.local.dir.dir_process import copy_dir, create_dir, remove_dir_tree
+from file_automation.local.file.file_process import copy_file, remove_file, rename_file, copy_specify_extension_file, \
+    copy_all_file_to_dir
+from file_automation.local.zip.zip_process import zip_dir, zip_file, zip_info, zip_file_info, set_zip_password, \
+    read_zip_file, unzip_file, unzip_all
+from file_automation.remote.google_drive.delete.delete_manager import delete_file
+from file_automation.remote.google_drive.dir.folder_manager import add_folder
+from file_automation.remote.google_drive.download.download_file import download_file, download_file_from_folder
+from file_automation.remote.google_drive.driver_instance import driver_instance
+from file_automation.remote.google_drive.search.search_drive import \
+    search_all_file, search_field, search_file_mimetype
+from file_automation.remote.google_drive.share.share_file import \
+    share_file_to_anyone, share_file_to_domain, share_file_to_user
+from file_automation.remote.google_drive.upload.upload_to_driver import \
+    upload_dir_to_folder, upload_to_folder, upload_dir_to_drive, upload_to_drive
+from file_automation.utils.exception.exception_tags import get_bad_trigger_function, get_bad_trigger_method
+from file_automation.utils.exception.exceptions import CallbackExecutorException
+
+
+class CallbackFunctionExecutor(object):
+
+    def __init__(self):
+        self.event_dict: dict = {
+            "copy_file": copy_file,
+            "rename_file": rename_file,
+            "remove_file": remove_file,
+            "copy_all_file_to_dir": copy_all_file_to_dir,
+            "copy_specify_extension_file": copy_specify_extension_file,
+            "copy_dir": copy_dir,
+            "create_dir": create_dir,
+            "remove_dir_tree": remove_dir_tree,
+            "zip_dir": zip_dir,
+            "zip_file": zip_file,
+            "zip_info": zip_info,
+            "zip_file_info": zip_file_info,
+            "set_zip_password": set_zip_password,
+            "unzip_file": unzip_file,
+            "read_zip_file": read_zip_file,
+            "unzip_all": unzip_all,
+            "driver_instance": driver_instance,
+            "search_all_file": search_all_file,
+            "search_field": search_field,
+            "search_file_mimetype": search_file_mimetype,
+            "upload_dir_to_folder": upload_dir_to_folder,
+            "upload_to_folder": upload_to_folder,
+            "upload_dir_to_drive": upload_dir_to_drive,
+            "upload_to_drive": upload_to_drive,
+            "add_folder": add_folder,
+            "share_file_to_anyone": share_file_to_anyone,
+            "share_file_to_domain": share_file_to_domain,
+            "share_file_to_user": share_file_to_user,
+            "delete_file": delete_file,
+            "download_file": download_file,
+            "download_file_from_folder": download_file_from_folder
+        }
+
+    def callback_function(
+            self,
+            trigger_function_name: str,
+            callback_function: typing.Callable,
+            callback_function_param: [dict, None] = None,
+            callback_param_method: str = "kwargs",
+            **kwargs
+    ) -> typing.Any:
+        """
+        :param trigger_function_name: what function we want to trigger only accept function in event_dict
+        :param callback_function: what function we want to callback
+        :param callback_function_param: callback function's param only accept dict 
+        :param callback_param_method: what type param will use on callback function only accept kwargs and args
+        :param kwargs: trigger_function's param
+        :return: trigger_function_name return value
+        """
+        try:
+            if trigger_function_name not in self.event_dict.keys():
+                raise CallbackExecutorException(get_bad_trigger_function)
+            execute_return_value = self.event_dict.get(trigger_function_name)(**kwargs)
+            if callback_function_param is not None:
+                if callback_param_method not in ["kwargs", "args"]:
+                    raise CallbackExecutorException(get_bad_trigger_method)
+                if callback_param_method == "kwargs":
+                    callback_function(**callback_function_param)
+                else:
+                    callback_function(*callback_function_param)
+            else:
+                callback_function()
+            return execute_return_value
+        except Exception as error:
+            print(repr(error), file=stderr)
+
+
+callback_executor = CallbackFunctionExecutor()
+

file_automation/utils/exception/exception_tags.py

@@ -1 +1,14 @@
 token_is_exist: str = "token file is exists"
+# Callback executor
+get_bad_trigger_method: str = "get bad trigger method, only accept kwargs and args"
+get_bad_trigger_function: str = "get bad trigger function only accept function in event_dict"
+# add command
+add_command_exception: str = "command value type should be as method or function"
+# executor
+executor_list_error: str = "executor receive wrong data list is none or wrong type"
+# json tag
+cant_execute_action_error: str = "cant execute action"
+cant_generate_json_report: str = "can't generate json report"
+cant_find_json_error: str = "cant find json file"
+cant_save_json_error: str = "cant save json file"
+action_is_null_error: str = "json action is null"

file_automation/utils/exception/exceptions.py

@@ -1,10 +1,30 @@
-class FileNotExistsException(Exception):
+class FileAutomationException(Exception):
     pass
 
 
-class DirNotExistsException(Exception):
+class FileNotExistsException(FileAutomationException):
     pass
 
 
-class ZIPGetWrongFileException(Exception):
+class DirNotExistsException(FileAutomationException):
+    pass
+
+
+class ZIPGetWrongFileException(FileAutomationException):
+    pass
+
+
+class CallbackExecutorException(FileAutomationException):
+    pass
+
+
+class ExecuteActionException(FileAutomationException):
+    pass
+
+
+class AddCommandException(FileAutomationException):
+    pass
+
+
+class JsonActionException(FileAutomationException):
     pass

file_automation/utils/executor/action_executor.py

@@ -0,0 +1,142 @@
+import builtins
+import sys
+import types
+from inspect import getmembers, isbuiltin
+
+from file_automation.local.dir.dir_process import copy_dir, create_dir, remove_dir_tree
+from file_automation.local.file.file_process import copy_file, remove_file, rename_file, copy_specify_extension_file, \
+    copy_all_file_to_dir
+from file_automation.local.zip.zip_process import zip_dir, zip_file, zip_info, zip_file_info, set_zip_password, \
+    read_zip_file, unzip_file, unzip_all
+from file_automation.remote.google_drive.delete.delete_manager import delete_file
+from file_automation.remote.google_drive.dir.folder_manager import add_folder
+from file_automation.remote.google_drive.download.download_file import download_file, download_file_from_folder
+from file_automation.remote.google_drive.driver_instance import driver_instance
+from file_automation.remote.google_drive.search.search_drive import \
+    search_all_file, search_field, search_file_mimetype
+from file_automation.remote.google_drive.share.share_file import \
+    share_file_to_anyone, share_file_to_domain, share_file_to_user
+from file_automation.remote.google_drive.upload.upload_to_driver import \
+    upload_dir_to_folder, upload_to_folder, upload_dir_to_drive, upload_to_drive
+from file_automation.utils.exception.exception_tags import add_command_exception, executor_list_error, \
+    action_is_null_error, cant_execute_action_error
+from file_automation.utils.exception.exceptions import ExecuteActionException, AddCommandException
+from file_automation.utils.json.json_file import read_action_json
+
+
+class Executor(object):
+
+    def __init__(self):
+        self.event_dict: dict = {
+            "copy_file": copy_file,
+            "rename_file": rename_file,
+            "remove_file": remove_file,
+            "copy_all_file_to_dir": copy_all_file_to_dir,
+            "copy_specify_extension_file": copy_specify_extension_file,
+            "copy_dir": copy_dir,
+            "create_dir": create_dir,
+            "remove_dir_tree": remove_dir_tree,
+            "zip_dir": zip_dir,
+            "zip_file": zip_file,
+            "zip_info": zip_info,
+            "zip_file_info": zip_file_info,
+            "set_zip_password": set_zip_password,
+            "unzip_file": unzip_file,
+            "read_zip_file": read_zip_file,
+            "unzip_all": unzip_all,
+            "driver_instance": driver_instance,
+            "search_all_file": search_all_file,
+            "search_field": search_field,
+            "search_file_mimetype": search_file_mimetype,
+            "upload_dir_to_folder": upload_dir_to_folder,
+            "upload_to_folder": upload_to_folder,
+            "upload_dir_to_drive": upload_dir_to_drive,
+            "upload_to_drive": upload_to_drive,
+            "add_folder": add_folder,
+            "share_file_to_anyone": share_file_to_anyone,
+            "share_file_to_domain": share_file_to_domain,
+            "share_file_to_user": share_file_to_user,
+            "delete_file": delete_file,
+            "download_file": download_file,
+            "download_file_from_folder": download_file_from_folder
+        }
+        # get all builtin function and add to event dict
+        for function in getmembers(builtins, isbuiltin):
+            self.event_dict.update({str(function[0]): function[1]})
+
+    def _execute_event(self, action: list):
+        event = self.event_dict.get(action[0])
+        if len(action) == 2:
+            if isinstance(action[1], dict):
+                return event(**action[1])
+            else:
+                return event(*action[1])
+        elif len(action) == 1:
+            return event()
+        else:
+            raise ExecuteActionException(cant_execute_action_error + " " + str(action))
+
+    def execute_action(self, action_list: [list, dict]) -> dict:
+        """
+        use to execute all action on action list(action file or program list)
+        :param action_list the list include action
+        for loop the list and execute action
+        """
+        if isinstance(action_list, dict):
+            action_list: list = action_list.get("auto_control", None)
+            if action_list is None:
+                raise ExecuteActionException(executor_list_error)
+        execute_record_dict = dict()
+        try:
+            if len(action_list) > 0 or isinstance(action_list, list):
+                pass
+            else:
+                raise ExecuteActionException(action_is_null_error)
+        except Exception as error:
+            print(repr(error), file=sys.stderr, flush=True)
+        for action in action_list:
+            try:
+                event_response = self._execute_event(action)
+                execute_record = "execute: " + str(action)
+                execute_record_dict.update({execute_record: event_response})
+            except Exception as error:
+                print(repr(error), file=sys.stderr, flush=True)
+                print(action, file=sys.stderr, flush=True)
+                execute_record = "execute: " + str(action)
+                execute_record_dict.update({execute_record: repr(error)})
+        for key, value in execute_record_dict.items():
+            print(key, flush=True)
+            print(value, flush=True)
+        return execute_record_dict
+
+    def execute_files(self, execute_files_list: list) -> list:
+        """
+        :param execute_files_list: list include execute files path
+        :return: every execute detail as list
+        """
+        execute_detail_list: list = list()
+        for file in execute_files_list:
+            execute_detail_list.append(self.execute_action(read_action_json(file)))
+        return execute_detail_list
+
+
+executor = Executor()
+
+
+def add_command_to_executor(command_dict: dict):
+    """
+    :param command_dict: dict include command we want to add to event_dict
+    """
+    for command_name, command in command_dict.items():
+        if isinstance(command, (types.MethodType, types.FunctionType)):
+            executor.event_dict.update({command_name: command})
+        else:
+            raise AddCommandException(add_command_exception)
+
+
+def execute_action(action_list: list) -> dict:
+    return executor.execute_action(action_list)
+
+
+def execute_files(execute_files_list: list) -> list:
+    return executor.execute_files(execute_files_list)

file_automation/utils/json/json_file.py

@@ -0,0 +1,41 @@
+import json
+from pathlib import Path
+from threading import Lock
+
+from file_automation.utils.exception.exception_tags import cant_find_json_error
+from file_automation.utils.exception.exceptions import JsonActionException
+
+_lock = Lock()
+
+
+def read_action_json(json_file_path: str) -> list:
+    """
+    use to read action file
+    :param json_file_path json file's path to read
+    """
+    _lock.acquire()
+    try:
+        file_path = Path(json_file_path)
+        if file_path.exists() and file_path.is_file():
+            with open(json_file_path) as read_file:
+                return json.loads(read_file.read())
+    except JsonActionException:
+        raise JsonActionException(cant_find_json_error)
+    finally:
+        _lock.release()
+
+
+def write_action_json(json_save_path: str, action_json: list) -> None:
+    """
+    use to save action file
+    :param json_save_path  json save path
+    :param action_json the json str include action to write
+    """
+    _lock.acquire()
+    try:
+        with open(json_save_path, "w+") as file_to_write:
+            file_to_write.write(json.dumps(action_json, indent=4))
+    except AutoControlJsonActionException:
+        raise AutoControlJsonActionException(cant_save_json_error)
+    finally:
+        _lock.release()