| """
|
| file_manager.py
|
|
|
| This module provides the FileManager class, a high-level interface for managing
|
| files and directories inside a specified media folder. It centralizes common
|
| operations such as reading, writing, listing, copying, moving, and deleting files,
|
| ensuring that all actions are safely constrained within the defined root directory.
|
| The class is designed to simplify file handling logic in larger applications by
|
| offering a consistent, validated, and extensible API for filesystem interactions.
|
| """
|
|
|
| from pathlib import Path
|
|
|
| class FileManager:
|
| """
|
| FileManager is a utility class that encapsulates common filesystem operations
|
| within a defined media directory. It ensures that all file manipulations remain
|
| inside the configured root folder and provides helper methods to interact with
|
| files and subdirectories in a structured, safe, and predictable manner.
|
|
|
| Typical use cases include managing uploaded media, performing batch operations
|
| on directory contents, and abstracting filesystem complexity behind a clean API.
|
| """
|
|
|
| def __init__(self, media_folder: Path):
|
| """
|
| Initialize the FileManager with a specific root directory for all file
|
| operations.
|
|
|
| Parameters
|
| ----------
|
| media_folder : Path
|
| The base directory where all file and folder operations will be performed.
|
| It must be a valid filesystem path. If the directory does not exist, the
|
| instance will attempt to create it automatically.
|
|
|
| Raises
|
| ------
|
| ValueError
|
| If the provided media_folder path is not a valid directory or cannot be
|
| created.
|
| """
|
| self.media_folder = Path(media_folder)
|
|
|
| if not self.media_folder.exists():
|
| try:
|
| self.media_folder.mkdir(parents=True)
|
| except Exception as exc:
|
| raise ValueError(
|
| f"Unable to create media folder at: {self.media_folder}"
|
| ) from exc
|
|
|
| if not self.media_folder.is_dir():
|
| raise ValueError(f"Media folder is not a directory: {self.media_folder}")
|
|
|
|
|
| def upload_file(self, file_handler, destination: Path):
|
| """
|
| Upload a file to a target destination inside the media folder.
|
|
|
| This method takes a file-like object and writes its contents to the
|
| specified destination within the media folder. The method ensures the path
|
| remains inside the media folder and creates necessary directories. It
|
| returns a structured response indicating whether the operation succeeded
|
| or failed, along with any relevant error message.
|
|
|
| Parameters
|
| ----------
|
| file_handler : file-like object
|
| A file-like object opened in binary mode that provides a `.read()` method.
|
| destination : Path
|
| The relative or absolute path (within the media folder) where the file
|
| should be saved.
|
|
|
| Returns
|
| -------
|
| dict
|
| A dictionary with:
|
| - "operation_success" (bool): True if the file was saved successfully.
|
| - "error" (str): An empty string on success, or the error message on failure.
|
|
|
| Raises
|
| ------
|
| None
|
| Any exceptions are captured and returned inside the result dictionary.
|
| """
|
| try:
|
| destination = self.media_folder / destination
|
| destination = destination.resolve()
|
|
|
|
|
| if self.media_folder not in destination.parents:
|
| return {
|
| "operation_success": False,
|
| "error": "Destination path is outside the media folder."
|
| }
|
|
|
|
|
| destination.parent.mkdir(parents=True, exist_ok=True)
|
|
|
| with open(destination, "wb") as f:
|
| f.write(file_handler.read())
|
|
|
| return {"operation_success": True, "error": ""}
|
|
|
| except Exception as exc:
|
| return {
|
| "operation_success": False,
|
| "error": str(exc)
|
| }
|
|
|
| def get_file(self, file_path: Path):
|
| """
|
| Retrieve a file inside the media folder and return an open file handler.
|
|
|
| This method receives a path pointing to a file expected to be located
|
| inside the media folder. If the file exists and is valid, the method
|
| returns a file handler opened in binary read mode. If the file does not
|
| exist or the resolved path escapes the media folder, the method returns
|
| None.
|
|
|
| Parameters
|
| ----------
|
| file_path : Path
|
| The relative or absolute path to the file within the media folder.
|
|
|
| Returns
|
| -------
|
| file object or None
|
| A file handler opened in 'rb' mode if the file exists and is accessible.
|
| Returns None if the file does not exist or the path is invalid.
|
|
|
| Raises
|
| ------
|
| None
|
| All exceptions are handled internally and result in returning None.
|
| """
|
| try:
|
| target = self.media_folder / file_path
|
| target = target.resolve()
|
|
|
|
|
| if self.media_folder not in target.parents:
|
| return None
|
|
|
| if not target.exists() or not target.is_file():
|
| return None
|
|
|
| return open(target, "rb")
|
|
|
| except Exception:
|
| return None
|
|
|
| def compute_sha1(self, file_handler):
|
| """
|
| Compute the SHA-1 checksum of a file handler.
|
|
|
| This method calculates the SHA-1 hash of the content provided by a
|
| file-like object opened in binary mode. The method preserves the
|
| current file pointer position by saving it, rewinding to the start
|
| of the file, computing the hash, and restoring the file pointer to
|
| its original position.
|
|
|
| Parameters
|
| ----------
|
| file_handler : file-like object
|
| A file-like object opened in binary mode, providing `.read()` and
|
| `.seek()` methods.
|
|
|
| Returns
|
| -------
|
| str
|
| The hexadecimal SHA-1 checksum of the file content.
|
|
|
| Raises
|
| ------
|
| ValueError
|
| If the provided file handler is invalid or does not support the
|
| required read/seek operations.
|
| """
|
| import hashlib
|
|
|
| if not hasattr(file_handler, "read") or not hasattr(file_handler, "seek"):
|
| raise ValueError("The provided file handler does not support read/seek operations.")
|
|
|
| original_pos = file_handler.tell()
|
|
|
| try:
|
| file_handler.seek(0)
|
| sha1 = hashlib.sha1()
|
|
|
| for chunk in iter(lambda: file_handler.read(8192), b""):
|
| sha1.update(chunk)
|
|
|
| file_handler.seek(original_pos)
|
| return sha1.hexdigest()
|
|
|
| except Exception as exc:
|
| try:
|
| file_handler.seek(original_pos)
|
| except Exception:
|
| pass
|
| raise ValueError(f"Unable to compute SHA-1 checksum: {exc}") from exc
|
|
|
