Folder-Organizer documentation¶
Add your content using reStructuredText syntax. See the
reStructuredText
documentation for details.
Main program¶
- async clean_folder.main.log_work(path: AsyncPath) dict¶
Recursively logs the contents of a directory, including files and subdirectories, and counts the occurrences of each file extension.
- Parameters:
path (AsyncPath): The directory path to log and analyze.
- Returns:
- dict: A nested dictionary where:
Keys are directory names or file extensions.
Values are either subdirectories’ dictionaries or counts of file extensions.
- Behavior:
Recursively iterates through the directory and its subdirectories.
Prints the contents of each directory.
Counts the occurrences of each file extension in the directory.
Returns a nested dictionary representing the directory structure and file extension counts.
- Notes:
This function is useful for analyzing and logging the structure of a directory tree.
It prints the contents of each directory to the console for visibility.
The returned dictionary can be used for further analysis or reporting.
- Examples:
>>> await log_work(AsyncPath("path/to/directory")) # Logs the contents of "path/to/directory" and returns a dictionary with file extension counts.
- async clean_folder.main.main(path: AsyncPath)¶
The main function that orchestrates file processing, including reading folders, unzipping archives, moving files, and logging results.
- Parameters:
path (AsyncPath): The root directory path where the operations will be performed.
- Behavior:
Reads and processes files in the specified directory using read_folder.
Unzips archives in the directory using unzip.
Prints a list of moved files and deleted folders.
Logs the results of file extensions and their counts using log_work.
Prints a summary of file extensions and their counts.
- Notes:
This function serves as the entry point for the file processing workflow.
It relies on other helper functions (read_folder, unzip, log_work) to perform specific tasks.
The results are printed to the console in a formatted manner.
- Examples:
>>> await main(AsyncPath("path/to/directory")) # Processes files, unzips archives, moves files, and logs results.
- async clean_folder.main.read_folder(path: AsyncPath, base_dir: AsyncPath, extensions: Dict)¶
Recursively reads and processes files and directories in a specified folder. Files are moved to corresponding directories based on their extensions, and empty directories are deleted.
- Parameters:
path (AsyncPath): The directory path to read and process. base_dir (AsyncPath): The base directory where files will be moved. extensions (Dict): A dictionary mapping file extensions to target directory names.
- Behavior:
Recursively iterates through the directory and its subdirectories.
Moves files to directories based on their extensions (as defined in extensions).
Deletes empty directories after processing.
Logs errors if file movement or directory deletion fails.
- Notes:
This function is useful for organizing files into specific directories based on their types.
It ensures that empty directories are cleaned up after processing.
Errors during file movement or directory deletion are logged for debugging.
- Examples:
>>> await read_folder(AsyncPath("path/to/source"), AsyncPath("path/to/destination"), {".txt": "text_files", ".jpg": "images"}) # Moves .txt files to "text_files" and .jpg files to "images", then deletes empty directories.
additional module¶
- async clean_folder.additional.async_rmtree(filename: AsyncPath)¶
Asynchronously removes a directory tree (including all files and subdirectories). Handles read-only files or directories by changing their permissions and retrying the removal.
- Parameters:
filename (AsyncPath): The path to the directory to be removed.
- Returns:
AsyncPath: The path of the removed directory.
- Behavior:
Uses shutil.rmtree to remove the directory tree.
Handles read-only files or directories by calling handle_remove_readonly to change permissions and retry the removal.
Runs the synchronous shutil.rmtree operation in a thread pool using asyncio.to_thread.
- Notes:
This function is useful for asynchronous environments where blocking I/O operations (like directory removal) need to be offloaded to a thread pool.
The handle_remove_readonly_sync function is a bridge between synchronous shutil.rmtree and the asynchronous handle_remove_readonly function.
- Examples:
>>> await async_rmtree(AsyncPath("path/to/directory")) # Removes the directory tree at "path/to/directory" and returns the path.
- async clean_folder.additional.handle_remove_readonly(func, path, exc)¶
Handles the removal of read-only files or directories by changing their permissions and retrying the operation. This function is designed to be used as an error handler for file or directory removal operations.
- Parameters:
func (callable): The function that caused the error (e.g., os.rmdir, os.remove). path (str): The path to the file or directory that caused the error. exc (tuple): A tuple containing the exception type, value, and traceback.
- Behavior:
If the error is due to a permission issue (errno.EACCES), changes the file or directory permissions to 0777 (read, write, and execute for all users) and retries the operation.
If the error is not related to permissions, re-raises the exception.
- Notes:
This function is typically used with shutil.rmtree or similar functions to handle read-only files or directories during deletion.
Uses asyncio.to_thread to run synchronous I/O operations in a thread pool.
- Examples:
>>> await handle_remove_readonly(os.remove, "readonly_file.txt", (PermissionError, PermissionError(13, 'Permission denied'), None)) # Changes permissions of "readonly_file.txt" and retries the removal.
- async clean_folder.additional.move_file(element: AsyncPath, exist_path: AsyncPath, path_to_move: AsyncPath) str¶
Moves a file from the source directory to the target directory, ensuring that the file name is normalized and unique in the target directory. If a file with the same name already exists, the function renames the file by incrementing a numeric suffix.
- Parameters:
element (AsyncPath): The path to the file to be moved. exist_path (AsyncPath): The source directory path where the file currently resides. path_to_move (AsyncPath): The target directory path where the file will be moved.
- Returns:
- str: A message indicating the successful move of the file, including the target directory
and the new file name.
- Raises:
OSError: If an error occurs during the file move operation.
- Examples:
>>> await move_file(AsyncPath("source/file.txt"), AsyncPath("source"), AsyncPath("target")) "target <-- moved target/file.txt"
>>> await move_file(AsyncPath("source/document_1.pdf"), AsyncPath("source"), AsyncPath("target")) "target <-- moved target/document_2.pdf"
- async clean_folder.additional.normalize(file_name: str) str¶
Normalizes a file name by replacing Cyrillic characters with their Latin equivalents and replacing all non-alphanumeric characters with an underscore “_”.
- Parameters:
file_name (str): The input string containing the file name to be normalized.
- Returns:
- str: The normalized string where:
Cyrillic characters are replaced with their Latin equivalents.
Non-alphanumeric characters are replaced with “_”.
- Examples:
>>> await normalize("файл_з_кирилицею.txt") "fail_z_kiryliceiu.txt"
>>> await normalize("file_with_123_and_!@#.txt") "file_with_123_and____.txt"
>>> await normalize("мій_документ.pdf") "mii_dokument.pdf"
- async clean_folder.additional.re_name(_name: str) str¶
Renames a file by incrementing a numeric suffix or adding a new one if it doesn’t exist. The function assumes that the numeric suffix is separated by an underscore “_”.
- Parameters:
_name (str): The input file name, including its extension.
- Returns:
str: The renamed file name with an incremented numeric suffix or a new suffix added.
- Examples:
>>> await re_name("file_1.txt") "file_2.txt"
>>> await re_name("document.pdf") "document_1.pdf"
>>> await re_name("data_10.csv") "data_11.csv"
>>> await re_name("image.png") "image_1.png"
- async clean_folder.additional.unzip(path: AsyncPath, archive_dir)¶
Extracts files from archives found in a specified directory and organizes them into subdirectories. Each archive is extracted into a folder named after the archive (without its extension), and the original archive file is deleted after extraction.
- Parameters:
path (AsyncPath): The directory path to search for archives. archive_dir (str): The suffix of the subdirectories containing the archives.
- Behavior:
Iterates through all subdirectories in the specified path that end with archive_dir.
For each file in these subdirectories, checks if it is an archive.
Extracts the archive into a folder named after the archive (without its extension).
Deletes the original archive file after successful extraction.
Logs an error if extraction fails.
- Notes:
Uses patoolib for archive extraction.
Skips files without extensions.
Creates necessary directories if they do not exist.
- Examples:
>>> await unzip(AsyncPath("path/to/archives"), "archive_dir") # Archives in "path/to/archives/archive_dir" are extracted into subfolders.