"""Implementation of the FTPFS server to be used by the download module.""" from pathlib import Path from fs.ftpfs import FTPFS __all__ = ["FTPFSServer"] class FTPFSServer: """Implementation of the FTPFS server to be used by the download module. This class establishes a connection to an FTP server and provides methods to interact with the server, check its availability, and download files. Attributes: host (str): The FTP server hostname or IP address. user (str): The username for authentication. acct (str): The account name for authentication. tls (bool, optional): Indicates whether TLS encryption should be used. Defaults to True. Methods: __init__(self, host: str, user: str, acct: str, tls: bool = True) -> None: Initializes the FTPFSServer with the provided host, user, acct, and tls settings. connect(self) -> None: Establishes a connection to the FTP server using the specified credentials. is_alive(self) -> bool: Checks if the FTP server connection is alive by attempting to list the root directory. is_available(self, ftp_file_path: str) -> bool: Checks if a file or directory specified by ftp_file_path exists on the server. download(self, ftp_file_path: str, save_path: Path) -> None: Downloads a file from the FTP server identified by ftp_file_path and saves it to the local filesystem at the specified save_path. """ def __init__(self, host: str, user: str, acct: str, tls: bool = True) -> None: """Initialize the FTPFSServer. Args: host (str): The FTP server hostname or IP address. user (str): The username for authentication. acct (str): The account name for authentication. tls (bool, optional): Indicates whether TLS encryption should be used. Defaults to True. """ self.host = host self.user = user self.acct = acct self.tls = tls def connect(self) -> None: """Establish a connection to the FTP server using the specified credentials.""" self.fs = FTPFS( host=self.host, user=self.user, acct=self.acct, tls=self.tls, ) return def is_alive(self) -> bool: """Check if the FTP server connection is alive by attempting to list the root directory. Returns: bool: True if the connection is alive, False otherwise. """ try: # Attempt to list the root directory self.fs.listdir("/") return True except Exception: return False def is_available(self, ftp_file_path: str) -> bool: """Check if a file or directory specified by ftp_file_path exists on the server. Args: ftp_file_path (str): The path of the file or directory on the server. Returns: bool: True if the file or directory exists, False otherwise. """ if not self.is_alive(): self.connect() return self.fs.exists(ftp_file_path) def download(self, ftp_file_path: str, save_path: str) -> None: """Download a file from the FTP server and save it to the local filesystem. Args: ftp_file_path (str): The path of the file on the FTP server. save_path (str): The path where the file should be saved on the local filesystem. """ # If not connected, connect to the server if not self.is_alive(): self.connect() # Get the file extension from the file path fname = Path(ftp_file_path).name # Open the file in binary mode with open(Path(save_path) / fname, "wb") as f: # Copy the file from the FTP server to the local file self.fs.download(ftp_file_path, f) def listdir(self, ftp_dir_path: str) -> list: """List the contents of a directory on the FTP server. Args: ftp_dir_path (str): The path of the directory on the FTP server. Returns: list: A list of the contents of the directory. """ # If not connected, connect to the server if not self.is_alive(): self.connect() # List the contents of the directory return self.fs.listdir(ftp_dir_path) def close(self) -> None: """Close the connection to the FTP server.""" if self.is_alive(): self.fs.close() return