Merge pull request #17 from Integration-Automation/dev

Dev

Author: JE-Chen

dev.toml

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

file_automation/__init__.py

@@ -6,21 +6,24 @@
 
 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
+    drive_search_all_file, drive_search_field, drive_search_file_mimetype
 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.remote.google_drive.dir.folder_manager import add_folder
+    drive_upload_dir_to_folder, drive_upload_to_folder, drive_upload_dir_to_drive, drive_upload_to_drive
+from file_automation.remote.google_drive.dir.folder_manager import drive_add_folder
 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.delete.delete_manager import delete_file
-from file_automation.remote.google_drive.download.download_file import download_file, download_file_from_folder
-
+    drive_share_file_to_anyone, drive_share_file_to_domain, drive_share_file_to_user
+from file_automation.remote.google_drive.delete.delete_manager import drive_delete_file
+from file_automation.remote.google_drive.download.download_file import drive_download_file, drive_download_file_from_folder
+from file_automation.utils.executor.action_executor import execute_action, execute_files, add_command_to_executor
+from file_automation.utils.json.json_file import read_action_json
+from file_automation.utils.file_process.get_dir_file_list import get_dir_files_as_list
 __all__ = [
     "copy_file", "rename_file", "remove_file", "copy_all_file_to_dir", "copy_specify_extension_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",
-    "delete_file", "download_file", "download_file_from_folder"
+    "unzip_all", "driver_instance", "drive_search_all_file", "drive_search_field", "drive_search_file_mimetype",
+    "drive_upload_dir_to_folder", "drive_upload_to_folder", "drive_upload_dir_to_drive", "drive_upload_to_drive",
+    "drive_add_folder", "drive_share_file_to_anyone", "drive_share_file_to_domain", "drive_share_file_to_user",
+    "drive_delete_file", "drive_download_file", "drive_download_file_from_folder", "execute_action", "execute_files",
+    "add_command_to_executor", "read_action_json", "get_dir_files_as_list"
 ]

file_automation/local/dir/dir_process.py

@@ -6,7 +6,7 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def copy_dir(dir_path: str, target_dir_path: str):
+def copy_dir(dir_path: str, target_dir_path: str) -> None:
     dir_path = Path(dir_path)
     target_dir_path = Path(target_dir_path)
     if dir_path.is_dir():
@@ -19,7 +19,7 @@ def copy_dir(dir_path: str, target_dir_path: str):
         file_automation_logger.error(f"Copy dir {dir_path} failed: {repr(DirNotExistsException)}")
 
 
-def remove_dir_tree(dir_path: str):
+def remove_dir_tree(dir_path: str) -> None:
     dir_path = Path(dir_path)
     if dir_path.is_dir():
         try:
@@ -29,7 +29,7 @@ def remove_dir_tree(dir_path: str):
             file_automation_logger.error(f"Remove dir tree {dir_path} error: {repr(error)}")
 
 
-def rename_dir(origin_dir_path, target_dir: str):
+def rename_dir(origin_dir_path, target_dir: str) -> None:
     origin_dir_path = Path(origin_dir_path)
     if origin_dir_path.exists() and origin_dir_path.is_dir():
         try:
@@ -48,7 +48,7 @@ def rename_dir(origin_dir_path, target_dir: str):
             f"target dir path: {target_dir}")
 
 
-def create_dir(dir_path: str):
+def create_dir(dir_path: str) -> None:
     dir_path = Path(dir_path)
     dir_path.mkdir(exist_ok=True)
     file_automation_logger.info(f"Create dir {dir_path}")

file_automation/local/file/file_process.py

@@ -6,7 +6,7 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def copy_file(file_path: str, target_path: str):
+def copy_file(file_path: str, target_path: str) -> None:
     file_path = Path(file_path)
     if file_path.is_file() and file_path.exists():
         try:
@@ -18,7 +18,7 @@ def copy_file(file_path: str, target_path: str):
         file_automation_logger.error(f"Copy file failed: {repr(FileNotExistsException)}")
 
 
-def copy_specify_extension_file(file_dir_path: str, target_extension: str, target_path: str):
+def copy_specify_extension_file(file_dir_path: str, target_extension: str, target_path: str) -> None:
     file_dir_path = Path(file_dir_path)
     if file_dir_path.exists() and file_dir_path.is_dir():
         for file in file_dir_path.glob(f"**/*.{target_extension}"):
@@ -32,7 +32,7 @@ def copy_specify_extension_file(file_dir_path: str, target_extension: str, targe
             f"Copy specify extension file failed: {repr(FileNotExistsException)}")
 
 
-def copy_all_file_to_dir(dir_path: str, target_dir_path: str):
+def copy_all_file_to_dir(dir_path: str, target_dir_path: str) -> None:
     dir_path = Path(dir_path)
     target_dir_path = Path(target_dir_path)
     if dir_path.is_dir() and target_dir_path.is_dir():
@@ -54,7 +54,7 @@ def copy_all_file_to_dir(dir_path: str, target_dir_path: str):
         print(repr(DirNotExistsException), file=sys.stderr)
 
 
-def rename_file(origin_file_path, target_name: str, file_extension=None):
+def rename_file(origin_file_path, target_name: str, file_extension=None) -> None:
     origin_file_path = Path(origin_file_path)
     if origin_file_path.exists() and origin_file_path.is_dir():
         if file_extension is None:
@@ -81,8 +81,13 @@ def rename_file(origin_file_path, target_name: str, file_extension=None):
             f"Rename file failed, error: {repr(DirNotExistsException)}")
 
 
-def remove_file(file_path: str):
+def remove_file(file_path: str) -> None:
     file_path = Path(file_path)
     if file_path.exists() and file_path.is_file():
         file_path.unlink(missing_ok=True)
         file_automation_logger.info(f"Remove file, file path: {file_path}")
+
+
+def create_file(file_path: str, content: str) -> None:
+    with open(file_path, "w+") as file:
+        file.write(content)

file_automation/local/zip/zip_process.py

@@ -7,12 +7,12 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def zip_dir(dir_we_want_to_zip: str, zip_name: str):
+def zip_dir(dir_we_want_to_zip: str, zip_name: str) -> None:
     make_archive(root_dir=dir_we_want_to_zip, base_name=zip_name, format="zip")
     file_automation_logger.info(f"Dir to zip: {dir_we_want_to_zip}, zip file name: {zip_name}")
 
 
-def zip_file(zip_file_path: str, file: [str, List[str]]):
+def zip_file(zip_file_path: str, file: [str, List[str]]) -> None:
     current_zip = zipfile.ZipFile(zip_file_path, mode="w")
     if isinstance(file, str):
         file_name = Path(file)
@@ -34,7 +34,7 @@ def zip_file(zip_file_path: str, file: [str, List[str]]):
     current_zip.close()
 
 
-def read_zip_file(zip_file_path: str, file_name: str, password: [str, None] = None):
+def read_zip_file(zip_file_path: str, file_name: str, password: [str, None] = None) -> None:
     current_zip = zipfile.ZipFile(zip_file_path, mode="r")
     with current_zip.open(name=file_name, mode="r", pwd=password, force_zip64=True) as read_file:
         data = read_file.read()
@@ -45,7 +45,8 @@ def read_zip_file(zip_file_path: str, file_name: str, password: [str, None] = No
     return data
 
 
-def unzip_file(zip_file_path: str, extract_member, extract_path: [str, None] = None, password: [str, None] = None):
+def unzip_file(
+        zip_file_path: str, extract_member, extract_path: [str, None] = None, password: [str, None] = None) -> None:
     current_zip = zipfile.ZipFile(zip_file_path, mode="r")
     current_zip.extract(member=extract_member, path=extract_path, pwd=password)
     file_automation_logger.info(
@@ -59,7 +60,7 @@ def unzip_file(zip_file_path: str, extract_member, extract_path: [str, None] = N
 
 def unzip_all(
         zip_file_path: str, extract_member: [str, None] = None,
-        extract_path: [str, None] = None, password: [str, None] = None):
+        extract_path: [str, None] = None, password: [str, None] = None) -> None:
     current_zip = zipfile.ZipFile(zip_file_path, mode="r")
     current_zip.extractall(members=extract_member, path=extract_path, pwd=password)
     file_automation_logger.info(
@@ -71,7 +72,7 @@ def unzip_all(
     current_zip.close()
 
 
-def zip_info(zip_file_path: str):
+def zip_info(zip_file_path: str) -> None:
     current_zip = zipfile.ZipFile(zip_file_path, mode="r")
     info_list = current_zip.infolist()
     current_zip.close()
@@ -81,7 +82,7 @@ def zip_info(zip_file_path: str):
     return info_list
 
 
-def zip_file_info(zip_file_path: str):
+def zip_file_info(zip_file_path: str) -> None:
     current_zip = zipfile.ZipFile(zip_file_path, mode="r")
     name_list = current_zip.namelist()
     current_zip.close()
@@ -91,7 +92,7 @@ def zip_file_info(zip_file_path: str):
     return name_list
 
 
-def set_zip_password(zip_file_path: str, password: bytes):
+def set_zip_password(zip_file_path: str, password: bytes) -> None:
     current_zip = zipfile.ZipFile(zip_file_path)
     current_zip.setpassword(pwd=password)
     current_zip.close()

file_automation/remote/google_drive/delete/delete_manager.py

@@ -1,12 +1,12 @@
-from typing import Union
+from typing import Union, Dict
 
 from googleapiclient.errors import HttpError
 
 from file_automation.remote.google_drive.driver_instance import driver_instance
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def delete_file(file_id: str) -> Union[dict, None]:
+def drive_delete_file(file_id: str) -> Union[Dict[str, str], None]:
     try:
         file = driver_instance.service.files().delete(fileId=file_id).execute()
         file_automation_logger.info(f"Delete drive file: {file_id}")

file_automation/remote/google_drive/dir/folder_manager.py

@@ -6,7 +6,7 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def add_folder(folder_name: str) -> Union[dict, None]:
+def drive_add_folder(folder_name: str) -> Union[dict, None]:
     try:
         file_metadata = {
             "name": folder_name,

file_automation/remote/google_drive/download/download_file.py

@@ -9,7 +9,7 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def download_file(file_id: str, file_name: str) -> BytesIO:
+def drive_download_file(file_id: str, file_name: str) -> BytesIO:
     try:
         request = driver_instance.service.files().get_media(fileId=file_id)
         file = io.BytesIO()
@@ -34,7 +34,7 @@ def download_file(file_id: str, file_name: str) -> BytesIO:
     return file
 
 
-def download_file_from_folder(folder_name: str) -> Union[dict, None]:
+def drive_download_file_from_folder(folder_name: str) -> Union[dict, None]:
     try:
         files = dict()
         response = driver_instance.service.files().list(
@@ -46,7 +46,7 @@ def download_file_from_folder(folder_name: str) -> Union[dict, None]:
             q=f"'{folder_id}' in parents"
         ).execute()
         for file in response.get("files", []):
-            download_file(file.get("id"), file.get("name"))
+            drive_download_file(file.get("id"), file.get("name"))
             files.update({file.get("name"): file.get("id")})
         file_automation_logger.info(
             f"Download all file on {folder_name} done."

file_automation/remote/google_drive/search/search_drive.py

@@ -6,7 +6,7 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def search_all_file() -> Union[dict, None]:
+def drive_search_all_file() -> Union[dict, None]:
     try:
         item = dict()
         response = driver_instance.service.files().list().execute()
@@ -24,7 +24,7 @@ def search_all_file() -> Union[dict, None]:
         return None
 
 
-def search_file_mimetype(mime_type: str) -> Union[dict, None]:
+def drive_search_file_mimetype(mime_type: str) -> Union[dict, None]:
     try:
         files = dict()
         page_token = None
@@ -51,7 +51,7 @@ def search_file_mimetype(mime_type: str) -> Union[dict, None]:
         return None
 
 
-def search_field(field_pattern: str) -> Union[dict, None]:
+def drive_search_field(field_pattern: str) -> Union[dict, None]:
     try:
         files = dict()
         response = driver_instance.service.files().list(fields=field_pattern).execute()

file_automation/remote/google_drive/share/share_file.py

@@ -6,7 +6,7 @@
 from file_automation.utils.logging.loggin_instance import file_automation_logger
 
 
-def share_file_to_user(
+def drive_share_file_to_user(
         file_id: str, user: str, user_role: str = "writer") -> Union[dict, None]:
     try:
         service = driver_instance.service
@@ -32,7 +32,7 @@ def share_file_to_user(
         return None
 
 
-def share_file_to_anyone(file_id: str, share_role: str = "reader") -> Union[dict, None]:
+def drive_share_file_to_anyone(file_id: str, share_role: str = "reader") -> Union[dict, None]:
     try:
         service = driver_instance.service
         user_permission = {
@@ -55,7 +55,7 @@ def share_file_to_anyone(file_id: str, share_role: str = "reader") -> Union[dict
         return None
 
 
-def share_file_to_domain(
+def drive_share_file_to_domain(
         file_id: str, domain: str, domain_role: str = "reader") -> Union[dict, None]:
     try:
         service = driver_instance.service