51源码之家seo关键词排名优化官网

实现Python使用SFTP批量上传和下载一个目录下的所有文件的程序，要求参数所有参数写到ini配置文件中，多个文件同时操作时实现异步多线程，异常处理和详细的日志信息输出，需要考虑线程安全和程序的安全性。

实现SFTP批量上传下载的Python程序，满足配置文件、多线程、异常处理和日志记录等要求：

import os
import configparser
import logging
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import paramiko
from stat import S_ISDIRclass SFTPConfig:def __init__(self, config_file='sftp_config.ini'):self.config = configparser.ConfigParser()self.config.read(config_file)# SFTP连接配置self.host = self.config.get('SFTP', 'Host', fallback='localhost')self.port = self.config.getint('SFTP', 'Port', fallback=22)self.username = self.config.get('SFTP', 'Username', fallback='user')self.password = self.config.get('SFTP', 'Password', fallback='pass')self.keyfile = self.config.get('SFTP', 'KeyFile', fallback=None)# 目录配置self.local_dir = self.config.get('PATHS', 'LocalDir', fallback='./local')self.remote_dir = self.config.get('PATHS', 'RemoteDir', fallback='./remote')# 线程配置self.max_workers = self.config.getint('THREADING', 'MaxWorkers', fallback=5)# 日志配置self.log_file = self.config.get('LOGGING', 'LogFile', fallback='sftp_transfer.log')self.log_level = self.config.get('LOGGING', 'LogLevel', fallback='INFO')class SFTPClient:def __init__(self, config):self.config = configself._setup_logging()self.transport = Noneself.sftp = Noneself.lock = threading.Lock()self._connect()def _setup_logging(self):log_format = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s'logging.basicConfig(level=self.config.log_level,format=log_format,handlers=[logging.FileHandler(self.config.log_file),logging.StreamHandler()])self.logger = logging.getLogger(__name__)def _connect(self):try:self.transport = paramiko.Transport((self.config.host, self.config.port))if self.config.keyfile:private_key = paramiko.RSAKey.from_private_key_file(self.config.keyfile)self.transport.connect(username=self.config.username, pkey=private_key)else:self.transport.connect(username=self.config.username, password=self.config.password)self.sftp = paramiko.SFTPClient.from_transport(self.transport)self.logger.info("SFTP connection established successfully")except Exception as e:self.logger.error(f"Connection failed: {str(e)}")raisedef _ensure_remote_dir(self, remote_path):try:self.sftp.stat(remote_path)except IOError:self.sftp.mkdir(remote_path)self.logger.info(f"Created remote directory: {remote_path}")def _put_file(self, local_path, remote_path):try:with self.lock:self.sftp.put(local_path, remote_path)self.logger.info(f"Uploaded: {local_path} -> {remote_path}")return Trueexcept Exception as e:self.logger.error(f"Upload failed {local_path}: {str(e)}")return Falsedef _get_file(self, remote_path, local_path):try:with self.lock:self.sftp.get(remote_path, local_path)self.logger.info(f"Downloaded: {remote_path} -> {local_path}")return Trueexcept Exception as e:self.logger.error(f"Download failed {remote_path}: {str(e)}")return Falsedef _process_local_dir(self, operation='upload'):with ThreadPoolExecutor(max_workers=self.config.max_workers) as executor:futures = []for root, dirs, files in os.walk(self.config.local_dir):rel_path = os.path.relpath(root, self.config.local_dir)remote_path = os.path.join(self.config.remote_dir, rel_path).replace('\\', '/')if operation == 'upload':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('\\', '/')if operation == 'upload':futures.append(executor.submit(self._put_file, local_file, remote_file))elif operation == 'download':os.makedirs(os.path.dirname(local_file), exist_ok=True)futures.append(executor.submit(self._get_file, remote_file, local_file))for future in as_completed(futures):future.result()def _process_remote_dir(self, operation='download'):def _walk_remote(remote_path):entries = []for entry in self.sftp.listdir_attr(remote_path):remote_entry = f"{remote_path}/{entry.filename}"if S_ISDIR(entry.st_mode):entries.extend(_walk_remote(remote_entry))else:entries.append(remote_entry)return entrieswith ThreadPoolExecutor(max_workers=self.config.max_workers) as executor:remote_files = _walk_remote(self.config.remote_dir)futures = []for remote_file in remote_files:rel_path = os.path.relpath(remote_file, self.config.remote_dir)local_file = os.path.join(self.config.local_dir, rel_path).replace('/', os.sep)if operation == 'download':os.makedirs(os.path.dirname(local_file), exist_ok=True)futures.append(executor.submit(self._get_file, remote_file, local_file))elif operation == 'upload':futures.append(executor.submit(self._put_file, local_file, remote_file))for future in as_completed(futures):future.result()def upload(self):self.logger.info("Starting upload process")self._process_local_dir(operation='upload')def download(self):self.logger.info("Starting download process")self._process_remote_dir(operation='download')def close(self):if self.sftp:self.sftp.close()if self.transport:self.transport.close()self.logger.info("SFTP connection closed")if __name__ == "__main__":try:config = SFTPConfig()client = SFTPClient(config)# 示例操作（实际应根据配置添加操作选择）client.upload()# client.download()except Exception as e:logging.error(f"Main process failed: {str(e)}")finally:client.close()

配置文件示例（sftp_config.ini）：

[SFTP]
Host = sftp.example.com
Port = 22
Username = your_username
Password = your_password
; KeyFile = /path/to/private_key[PATHS]
LocalDir = ./local_files
RemoteDir = /remote/files[THREADING]
MaxWorkers = 8[LOGGING]
LogFile = sftp_transfer.log
LogLevel = DEBUG

程序特点说明：

1. 配置管理：使用独立的SFTPConfig类管理所有配置参数
2. 多线程安全：使用ThreadPoolExecutor管理线程池，使用Lock保证SFTP操作原子性
3. 连接管理：支持密码和密钥两种认证方式
4. 目录处理：自动创建必要的远程目录结构，保持本地和远程目录一致性
5. 异常处理：所有文件操作都有try-catch块，错误信息会记录到日志
6. 日志系统：包含时间戳、线程名称、操作类型和结果信息
7. 递归处理：支持目录结构的递归上传/下载

注意事项：

1. 建议使用SSH密钥认证（配置KeyFile参数）
2. 大文件传输时适当调整线程数量
3. Windows路径需要使用双反斜杠或原始字符串
4. 确保本地和远程目录存在且有适当权限
5. 传输过程中不要修改目录结构

扩展建议：

1. 添加断点续传功能
2. 增加文件校验（MD5/SHA校验）
3. 支持通配符文件过滤
4. 添加进度条显示
5. 支持配置文件加密