handle_nas.py 

import os
import datetime
from smb.SMBConnection import SMBConnection


class Nas_CloudDisk_Handler:
    def __init__(self, username, password, server_ip, port):
        """
        初始化 NAS 云盘处理类
        :param username: 连接 SMB 服务器的用户名
        :param password: 连接 SMB 服务器的密码
        :param server_ip: 连接的 SMB 服务器的 IP 地址
        :param port: SMB 协议使用的端口号，默认为 445
        """
        self.username = username
        self.password = password
        self.server_ip = server_ip
        self.port = port
        self.conn = None  # 初始化连接对象为 None
        self.status = False  # 初始化连接状态为未连接（False）

    def connect_nas(self):
        """
        建立 SMB 服务器连接
        """
        try:
            # 创建 SMB 连接对象，传入用户名、密码，本地标识和远程标识为空，使用 NTLMv2 认证
            self.conn = SMBConnection(self.username, self.password, my_name='', remote_name='', use_ntlm_v2=True)
            # 使用创建的连接对象连接到指定的服务器 IP 和端口
            self.conn.connect(self.server_ip, self.port)
            # 获取连接的认证结果，更新连接状态
            self.status = self.conn.auth_result
        except Exception as e:
            # 捕获并打印具体的异常信息
            print(f"连接过程中出现错误: {e}")
            # 重新抛出异常
            raise e
        finally:
            # 如果连接对象已成功创建且认证失败，关闭连接
            if self.conn and not self.status:
                try:
                    # 关闭连接对象
                    self.conn.close()
                except Exception as close_error:
                    print(f"关闭连接时出现错误: {close_error}")
                    raise close_error
                print("连接对象已成功创建且认证失败，已经关闭连接")
                # 抛出异常，表明连接未成功建立
                raise Exception

    def get_all_share_directory(self):
        """
        获取 SMB 服务器下的所有共享目录
        :return: 共享目录列表
        """
        # 初始化一个空列表，用于存储共享目录名
        share_list = []
        # 获取服务器上的所有共享目录信息
        share_name = self.conn.listShares()
        for i in share_name:
            # 将每个共享目录的名称添加到列表中
            share_list.append(i.name)
        # 返回包含所有共享目录名的列表
        return share_list

    def get_all_file_name(self, service_name, path):
        """
        获取文件夹内所有文件名
        :param service_name: 共享服务名
        :param path: 文件夹路径
        :return: 文件列表
        """
        # 初始化一个空列表，用于存储文件名
        all_file_list = []
        # 获取指定共享服务和路径下的文件信息
        all_file_name = self.conn.listPath(service_name, path)
        for i in all_file_name:
            # 排除以点开头的隐藏文件
            if i.filename[0] != '.':
                # 将符合条件的文件名添加到列表中
                all_file_list.append(i.filename)
        # 返回包含所有文件名的列表
        return all_file_list

    def create_folder(self, service_name, path):
        """
        创建文件夹
        :param service_name: 共享服务名
        :param path: 文件夹路径
        """
        try:
            # 在指定共享服务下创建文件夹
            self.conn.createDirectory(service_name, path)
        except Exception as e:
            print(f"创建文件夹失败，请检查：{e}")
            raise e
        else:
            print("创建成功！")

    def upload_files(self, service_name, local_path, nas_path, file_name):
        """
        上传文件
        :param service_name: 共享服务名
        :param local_path: 本地文件路径
        :param nas_path: NAS 云盘路径
        :param file_name: 文件名
        """
        # 打开本地文件
        local_obj = open(os.path.join(local_path, file_name), 'rb')
        try:
            # 将本地文件上传到指定的 NAS 云盘路径下
            self.conn.storeFile(service_name, os.path.join(nas_path, file_name), local_obj)
        except Exception as e:
            print(f"上传文件失败，原因是{e}")
            raise e
        else:
            print("文件上传成功")
        finally:
            # 无论上传成功与否，都关闭本地文件对象
            local_obj.close()

    def download_files(self, service_name, local_path, nas_path, file_name):
        """
        下载文件
        :param service_name: 共享服务名
        :param local_path: 本地保存路径
        :param nas_path: NAS 云盘文件路径
        :param file_name: 文件名
        """
        # 以二进制写入模式打开本地文件，准备写入下载的内容
        local_obj = open(os.path.join(local_path, file_name), 'wb')
        try:
            # 从 NAS 云盘下载指定文件到本地
            self.conn.retrieveFile(service_name, os.path.join(nas_path, file_name), local_obj)
        except Exception as e:
            print(f"文件下载失败，原因是{e}")
            raise e
        else:
            print("文件下载成功")
        finally:
            # 无论下载成功与否，都关闭本地文件对象
            local_obj.close()

    def get_file_update_time(self, service_name, nas_path):
        """
        获取文件最后更新时间
        :param service_name: 共享服务名
        :param nas_path: NAS 云盘文件路径
        :return: 文件最后更新时间字符串
        """
        try:
            # 获取文件属性信息
            file_attribute_info = self.conn.getAttributes(service_name, nas_path)
        except Exception as e:
            print(f"获取文件更新时间失败，原因是{e}")
            raise e
        else:
            # 获取文件的最后写入时间戳
            timestamp = file_attribute_info.last_write_time
            # 将时间戳转换为 datetime 对象
            dt_object = datetime.datetime.fromtimestamp(timestamp)
            # 将 datetime 对象格式化为字符串，你可以根据需要调整格式
            file_update_time = dt_object.strftime('%Y-%m-%d %H:%M:%S')
            # 返回格式化后的最后更新时间
            return file_update_time

    def delete_file(self, service_name, file_name, nas_path):
        """
        删除文件
        :param service_name: 共享服务名
        :param file_name: 文件名
        :param nas_path: NAS 云盘文件路径
        """
        try:
            # 删除指定文件
            self.conn.deleteFiles(service_name, os.path.join(nas_path, file_name))
        except Exception as e:
            print(f"文件删除失败，原因是{e}")
            raise e
        else:
            print("文件删除成功")

    def delete_folder(self, service_name, nas_path):
        """
        删除文件夹
        :param service_name: 共享服务名
        :param nas_path: NAS 云盘文件夹路径
        """
        try:
            # 删除指定文件夹
            self.conn.deleteDirectory(service_name, nas_path)
        except Exception as e:
            print(f"文件夹删除失败，原因是{e}")
            raise e
        else:
            print("文件夹删除成功")

    def close_connection(self):
        """
        关闭 SMB 连接
        """
        try:
            # 关闭连接对象
            self.conn.close()
        except Exception as e:
            print(f"关闭连接时出现错误: {e}")
            raise e

run.py

from common.handle_nas import Nas_CloudDisk_Handler

if __name__ == '__main__':
    nas_method = Nas_CloudDisk_Handler('username', 'password', '127.0.0.1', 445)
    # 建立 SMB 服务器连接
    nas_method.connect_nas()

    # 获取 SMB 服务器下的所有共享目录
    share_list = nas_method.get_all_share_directory()
    print(share_list)

    # 获取文件夹内所有文件名
    path1 = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹'
    all_file_list = nas_method.get_all_file_name(share_list[6], path1)
    print(all_file_list)

    # 创建文件夹
    path2 = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹\新建文件夹2'
    # nas_method.create_folder(share_list[6], path2)

    # 上传文件
    local_path = r'C:\Users\admin\Desktop'
    nas_path = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹\新建文件夹2'
    file_name = '永远的图片2.jpg'
    # nas_method.upload_files(share_list[6], local_path, nas_path, file_name)

    # 下载文件
    local_path1 = r'D:\Program Files\JetBrains\pythonProject\Nas云盘上传下载'
    nas_path1 = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹\新建文件夹2'
    file_name1 = '永远的图片2.jpg'
    # nas_method.download_files(share_list[6], local_path1, nas_path1, file_name1)

    # 获取文件最后更新时间
    nas_path2 = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹\新建文件夹2'
    formatted_time = nas_method.get_file_update_time(share_list[6], nas_path2)
    print(formatted_time)

    # 删除文件
    file_name2 = '永远的图片2.jpg'
    nas_path3 = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹\新建文件夹2'
    # nas_method.delete_file(share_list[6], file_name2, nas_path3)

    # 删除文件夹
    nas_path4 = r'\一层目录\二层目录\三层目录\四层目录\新建文件夹\新建文件夹2'
    # nas_method.delete_folder(share_list[6], nas_path4)

    nas_method.close_connection()