#!/usr/bin/env python3 """NAS 客户端 - 支持 SMB、SFTP、WebDAV 协议""" import os from pathlib import Path from typing import List, Dict, Any, Optional from abc import ABC, abstractmethod import logging logger = logging.getLogger(__name__) class NASClient(ABC): """NAS 客户端基类""" @abstractmethod def connect(self) -> bool: pass @abstractmethod def disconnect(self): pass @abstractmethod def list_files(self, path: str) -> List[Dict[str, Any]]: pass @abstractmethod def move_file(self, src: str, dst: str) -> bool: pass @abstractmethod def create_dir(self, path: str) -> bool: pass @abstractmethod def file_exists(self, path: str) -> bool: pass class SMBClient(NASClient): """SMB/CIFS 协议客户端（适用于群晖、威联通、极空间等）""" def __init__(self, host: str, username: str, password: str, share: str, port: int = 445): self.host = host self.username = username self.password = password self.share = share self.port = port self.session = None def connect(self) -> bool: try: import smbclient # 注册 SMB 会话 smbclient.register_session( self.host, username=self.username, password=self.password, port=self.port ) # 测试连接 test_path = f"//{self.host}/{self.share}" smbclient.listdir(test_path) logger.info(f"SMB 连接成功：{self.host}:{self.port}/{self.share}") return True except Exception as e: logger.error(f"SMB 连接失败：{e}") return False def disconnect(self): try: import smbclient smbclient.delete_session(self.host, port=self.port) except: pass def list_files(self, path: str) -> List[Dict[str, Any]]: import smbclient from datetime import datetime files = [] full_path = f"//{self.host}/{self.share}/{path.lstrip('/')}" try: for entry in smbclient.scandir(full_path): try: stat_info = entry.stat() files.append({ 'name': entry.name, 'path': f"{path}/{entry.name}".replace('//', '/'), 'size': stat_info.st_size, 'is_dir': entry.is_dir(), 'modified': datetime.fromtimestamp(stat_info.st_mtime), 'created': datetime.fromtimestamp(stat_info.st_ctime) }) except Exception as e: logger.debug(f"跳过文件 {entry.name}: {e}") except Exception as e: logger.error(f"列出目录失败 {full_path}: {e}") return files def move_file(self, src: str, dst: str) -> bool: import shutil try: src_full = f"//{self.host}/{self.share}/{src.lstrip('/')}" dst_full = f"//{self.host}/{self.share}/{dst.lstrip('/')}" # 确保目标目录存在 dst_dir = str(Path(dst).parent) self.create_dir(dst_dir) # 移动文件 shutil.move(src_full, dst_full) logger.debug(f"移动文件：{src} -> {dst}") return True except Exception as e: logger.error(f"移动文件失败 {src} -> {dst}: {e}") return False def create_dir(self, path: str) -> bool: import smbclient try: full_path = f"//{self.host}/{self.share}/{path.lstrip('/')}" smbclient.makedirs(full_path, exist_ok=True) return True except Exception as e: logger.debug(f"创建目录失败 {path}: {e}") return True # 目录可能已存在 def file_exists(self, path: str) -> bool: import smbclient try: full_path = f"//{self.host}/{self.share}/{path.lstrip('/')}" smbclient.stat(full_path) return True except: return False class SFTPClient(NASClient): """SFTP 协议客户端（适用于支持 SSH 的 NAS）""" def __init__(self, host: str, username: str, password: str, port: int = 22, key_file: Optional[str] = None): self.host = host self.username = username self.password = password self.port = port self.key_file = key_file self.client = None self.sftp = None def connect(self) -> bool: try: import paramiko self.client = paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 优先使用密钥认证 if self.key_file and os.path.exists(self.key_file): self.client.connect( hostname=self.host, port=self.port, username=self.username, key_filename=self.key_file ) else: self.client.connect( hostname=self.host, port=self.port, username=self.username, password=self.password ) self.sftp = self.client.open_sftp() logger.info(f"SFTP 连接成功：{self.host}:{self.port}") return True except Exception as e: logger.error(f"SFTP 连接失败：{e}") return False def disconnect(self): try: if self.sftp: self.sftp.close() if self.client: self.client.close() except: pass def list_files(self, path: str) -> List[Dict[str, Any]]: from datetime import datetime files = [] try: for entry in self.sftp.listdir_attr(path): files.append({ 'name': entry.filename, 'path': f"{path}/{entry.filename}", 'size': entry.st_size, 'is_dir': entry.st_mode & 0o40000 != 0, 'modified': datetime.fromtimestamp(entry.st_mtime), 'created': datetime.fromtimestamp(entry.st_ctime) }) except Exception as e: logger.error(f"列出目录失败 {path}: {e}") return files def move_file(self, src: str, dst: str) -> bool: try: # 确保目标目录存在 dst_dir = str(Path(dst).parent) self.create_dir(dst_dir) self.sftp.rename(src, dst) logger.debug(f"移动文件：{src} -> {dst}") return True except Exception as e: logger.error(f"移动文件失败 {src} -> {dst}: {e}") return False def create_dir(self, path: str) -> bool: try: self.sftp.makedirs(path) return True except: return True def file_exists(self, path: str) -> bool: try: self.sftp.stat(path) return True except: return False class WebDAVClient(NASClient): """WebDAV 协议客户端（适用于群晖 Drive 等）""" def __init__(self, host: str, username: str, password: str, port: int = 5005, use_ssl: bool = False): self.host = host self.username = username self.password = password self.port = port self.use_ssl = use_ssl self.client = None self.base_url = None def connect(self) -> bool: try: from webdav3.client import Client protocol = "https" if self.use_ssl else "http" self.base_url = f"{protocol}://{self.host}:{self.port}" options = { 'webdav_hostname': self.base_url, 'webdav_login': self.username, 'webdav_password': self.password, 'disable_check': True } self.client = Client(options) # 测试连接 self.client.list() logger.info(f"WebDAV 连接成功：{self.base_url}") return True except Exception as e: logger.error(f"WebDAV 连接失败：{e}") return False def disconnect(self): self.client = None def list_files(self, path: str) -> List[Dict[str, Any]]: from datetime import datetime files = [] try: items = self.client.list(path, info=True) for item in items: if isinstance(item, dict): files.append({ 'name': item.get('display_name', item.get('path', '')), 'path': item.get('path', ''), 'size': item.get('size', 0), 'is_dir': item.get('isdir', False), 'modified': datetime.fromtimestamp(item.get('modified', 0)) }) except Exception as e: logger.error(f"列出目录失败 {path}: {e}") return files def move_file(self, src: str, dst: str) -> bool: try: # 确保目标目录存在 dst_dir = str(Path(dst).parent) self.create_dir(dst_dir) self.client.move(remote_src=src, remote_dst=dst) logger.debug(f"移动文件：{src} -> {dst}") return True except Exception as e: logger.error(f"移动文件失败 {src} -> {dst}: {e}") return False def create_dir(self, path: str) -> bool: try: self.client.makedirs(path) return True except: return True def file_exists(self, path: str) -> bool: return self.client.check(path) def create_nas_client(config: Dict) -> Optional[NASClient]: """根据配置创建 NAS 客户端""" protocol = config.get('protocol', 'smb').lower() if protocol == 'smb': return SMBClient( host=config['host'], username=config['username'], password=config['password'], share=config['share'], port=config.get('port', 445) ) elif protocol == 'sftp': return SFTPClient( host=config['host'], username=config['username'], password=config.get('password'), port=config.get('port', 22), key_file=config.get('key_file') ) elif protocol == 'webdav': return WebDAVClient( host=config['host'], username=config['username'], password=config['password'], port=config.get('port', 5005), use_ssl=config.get('use_ssl', False) ) else: raise ValueError(f"不支持的协议：{protocol}")