功能概述
实现一个 FileUploader 类，用于将本地文件夹及其子文件上传到 Linux 服务器的指定目录，并支持：
- 冲突处理策略：
- 覆盖：直接覆盖远程文件。
- 跳过：跳过已存在的远程文件。
- 重命名：避免冲突，为文件生成唯一名称。
- 日志功能：
- 记录上传成功的文件（
upload_success.log）。 - 记录上传失败的文件和原因（
upload_failures.log）。
- 记录上传成功的文件（
- 目录校验和自动创建：
- 自动检测远程目录是否存在，不存在时会创建。
代码实现
import os
import logging
import paramiko
class FileUploader:
def __init__(self, hostname: str, port: int, username: str, password: str):
"""
初始化SSH连接参数
:param hostname: Linux服务器地址
:param port: SSH端口
:param username: 登录用户名
:param password: 登录密码
"""
self.hostname = hostname
self.port = port
self.username = username
self.password = password
self.sftp: paramiko.SFTPClient | None = None
self.ssh_client: paramiko.SSHClient | None = None
self._init_logging()
def _init_logging(self) -> None:
"""初始化日志记录"""
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
self.success_logger = logging.getLogger("success_logger")
self.failure_logger = logging.getLogger("failure_logger")
success_handler = logging.FileHandler("upload_success.log", mode="w", encoding="utf-8")
failure_handler = logging.FileHandler("upload_failures.log", mode="w", encoding="utf-8")
success_handler.setFormatter(logging.Formatter("%(asctime)s - %(message)s"))
failure_handler.setFormatter(logging.Formatter("%(asctime)s - %(message)s"))
self.success_logger.addHandler(success_handler)
self.failure_logger.addHandler(failure_handler)
def connect(self) -> None:
"""建立SSH和SFTP连接"""
try:
self.ssh_client = paramiko.SSHClient()
self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh_client.connect(
hostname=self.hostname,
port=self.port,
username=self.username,
password=self.password
)
self.sftp = self.ssh_client.open_sftp()
print("连接成功")
except Exception as e:
raise Exception(f"连接失败: {e}")
def upload_directory(
self,
local_dir: str,
remote_dir: str,
conflict_strategy: str = "overwrite"
) -> None:
"""
上传本地文件夹及其子文件到远程文件夹。
:param local_dir: 本地文件夹路径
:param remote_dir: 远程目标文件夹路径
:param conflict_strategy: 冲突处理策略，可选值：
- "overwrite": 覆盖文件（默认）
- "skip": 跳过已存在文件
- "rename": 重命名冲突文件
"""
if not os.path.exists(local_dir):
raise FileNotFoundError(f"本地目录不存在: {local_dir}")
self._ensure_remote_dir(remote_dir)
for root, _, files in os.walk(local_dir):
relative_path = os.path.relpath(root, local_dir)
remote_path = os.path.join(remote_dir, relative_path).replace("\\", "/")
self._ensure_remote_dir(remote_path)
for file in files:
local_file = os.path.join(root, file)
remote_file = os.path.join(remote_path, file).replace("\\", "/")
try:
if self._check_file_exists(remote_file):
if conflict_strategy == "skip":
print(f"跳过已存在文件: {remote_file}")
continue
elif conflict_strategy == "rename":
remote_file = self._get_unique_filename(remote_file)
print(f"重命名并上传: {local_file} -> {remote_file}")
elif conflict_strategy == "overwrite":
print(f"覆盖文件: {remote_file}")
self.sftp.put(local_file, remote_file)
print(f"已上传: {local_file} -> {remote_file}")
self.success_logger.info(f"{local_file} -> {remote_file}")
except Exception as e:
print(f"上传失败: {local_file} -> {remote_file}，原因: {e}")
self.failure_logger.error(f"{local_file} -> {remote_file}，原因: {e}")
def _ensure_remote_dir(self, remote_dir: str) -> None:
"""确保远程目录存在，不存在则创建"""
try:
self.sftp.stat(remote_dir)
except FileNotFoundError:
self.sftp.mkdir(remote_dir)
print(f"已创建远程目录: {remote_dir}")
def _check_file_exists(self, remote_file: str) -> bool:
"""检查远程文件是否存在"""
try:
self.sftp.stat(remote_file)
return True
except FileNotFoundError:
return False
def _get_unique_filename(self, remote_file: str) -> str:
"""生成唯一的文件名以避免冲突"""
base, ext = os.path.splitext(remote_file)
counter = 1
while self._check_file_exists(remote_file):
remote_file = f"{base}_{counter}{ext}"
counter += 1
return remote_file
def close(self) -> None:
"""关闭SFTP和SSH连接"""
if self.sftp:
self.sftp.close()
if self.ssh_client:
self.ssh_client.close()
print("连接已关闭")
if __name__ == "__main__":
uploader = FileUploader(hostname="192.168.1.1", port=22, username="user", password="password")
try:
uploader.connect()
uploader.upload_directory(
local_dir="C:/local_folder",
remote_dir="/remote_folder",
conflict_strategy="rename"
)
finally:
uploader.close()
功能说明
-
参数类型说明：
hostname(str): 服务器地址。port(int): SSH端口号。username(str): 用户名。password(str): 密码。local_dir(str): 本地目录路径。remote_dir(str): 远程目录路径。conflict_strategy(str): 冲突策略，可选"overwrite"、"skip"或"rename"。
-
日志功能：
- 成功日志：记录成功上传的文件，保存为
upload_success.log。 - 失败日志：记录失败文件及原因，保存为
upload_failures.log。
- 成功日志：记录成功上传的文件，保存为
-
冲突处理：
- 覆盖：直接覆盖远程文件。
- 跳过：跳过已存在的文件。
- 重命名：为冲突文件自动生成唯一名称。
运行结果示例
-
成功日志 (
upload_success.log)：2024-11-19 15:00:00 - C:/local_folder/file1.txt -> /remote_folder/file1.txt -
失败日志 (
upload_failures.log)：2024-11-19 15:01:00 - C:/local_folder/file3.txt -> /remote_folder/file3.txt，原因: Permission denied
适用场景
- 定时备份本地文件到远程服务器。
- 上传大规模文件夹时自动处理冲突。
- 记录上传操作的成功和失败，便于问题排查。