97 lines
No EOL
2.4 KiB
Python
97 lines
No EOL
2.4 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
def ftp_upload(
|
|
host: str,
|
|
filepath: Path,
|
|
storage_path: str,
|
|
filename: str,
|
|
overwrite: bool = False,
|
|
timeout: float = 30,
|
|
user: str = "",
|
|
passwd: str = "",
|
|
):
|
|
from ftplib import FTP, error_perm
|
|
|
|
if not os.path.isfile(filepath):
|
|
raise FileNotFoundError(f"Local file not found: {filepath}")
|
|
|
|
remote_name = filename or os.path.basename(filepath)
|
|
|
|
ftp = FTP(host, timeout=timeout)
|
|
ftp.login(user=user, passwd=passwd)
|
|
|
|
try:
|
|
# Change to target directory
|
|
ftp.cwd(storage_path)
|
|
|
|
# Check if file exists remotely
|
|
existing = ftp.nlst()
|
|
if remote_name in existing:
|
|
if not overwrite:
|
|
raise FileExistsError(f"Remote file already exists: {remote_name}")
|
|
# delete existing file
|
|
try:
|
|
ftp.delete(str(remote_name))
|
|
except error_perm as e:
|
|
raise RuntimeError(f"Failed to delete remote file: {e}")
|
|
|
|
# Upload new file
|
|
with open(filepath, 'rb') as f:
|
|
ftp.storbinary(f'STOR {remote_name}', f)
|
|
|
|
finally:
|
|
ftp.quit()
|
|
|
|
def ftp_wipe(
|
|
host: str,
|
|
storage_path: str,
|
|
timeout: int = 30,
|
|
user: str = "",
|
|
passwd: str = "",
|
|
):
|
|
from ftplib import FTP, error_perm
|
|
|
|
ftp = FTP(host, timeout=timeout)
|
|
ftp.login(user=user, passwd=passwd)
|
|
|
|
try:
|
|
ftp.cwd(storage_path)
|
|
existing = ftp.nlst()
|
|
for f in existing:
|
|
try:
|
|
ftp.delete(filename=f)
|
|
except error_perm as e:
|
|
raise RuntimeError(f"Failed to delete remote file: {f}")
|
|
except error_perm as e:
|
|
raise RuntimeError("e")
|
|
finally:
|
|
ftp.quit()
|
|
|
|
def ftp_get_filesize(
|
|
host: str,
|
|
storage_path: str,
|
|
filename: str,
|
|
timeout: float = 30,
|
|
user: str = "",
|
|
passwd: str = "",
|
|
) -> int:
|
|
from ftplib import FTP, error_perm
|
|
|
|
ftp = FTP(host, timeout=timeout)
|
|
ftp.login(user=user, passwd=passwd)
|
|
|
|
try:
|
|
ftp.cwd(storage_path)
|
|
try:
|
|
size = ftp.size(filename)
|
|
except error_perm as e:
|
|
raise FileNotFoundError(f"Remote file not found or SIZE not supported: {filename}") from e
|
|
|
|
if size is None:
|
|
raise RuntimeError("FTP server returned no size information")
|
|
|
|
return size
|
|
|
|
finally:
|
|
ftp.quit() |