更便捷的访问共享盘方式 pyfilesystem

2024-01-07 23:16:04
class SMB:
    info_dict = {
        "host": '1.66.12.1',
        "username": "H7",
        'passwd': password,
        'port': 445,
        'direct_tcp': True,
    }


class MySMB:
    def __init__(self, smb_dict):
        self.smb_dict = smb_dict

    def __enter__(self):
        self.smb_fs = fs.smbfs.SMBFS(**self.smb_dict)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.smb_fs.close()
        return True

    def do_something(self):
        print(self.smb_fs.listdir(r"/xt")[:19])

    def download(self, smb_path, local_dir):
        os.makedirs(local_dir, exist_ok=True)
        filename = os.path.basename(smb_path)
        local_path = os.path.join(local_dir, filename)
        with open(local_path, 'wb') as f:
            self.smb_fs.download(smb_path, f)
            print('download remote file success', smb_path)
        return local_path

    def upload(self, local_path, smb_path):
        print('smb_path', smb_path,os.path.dirname(smb_path))
        self.smb_fs.makedirs(os.path.dirname(smb_path), recreate=True)
        try:
            with open(local_path, 'rb') as f:
                self.smb_fs.upload(smb_path, f)
                print('write remote file success', smb_path)
        except Exception as e:
            print('write remote file fail')

    def upload_dir(self, local_folder, remote_dir):
        for item in os.listdir(local_folder):
            local_item = os.path.join(local_folder, item)
            remote_item = os.path.join(remote_dir, item).replace("\\", "/")
            print(local_item, remote_item)
            if os.path.isdir(local_item):
                self.upload_dir(local_item, remote_item)
            elif os.path.isfile(local_item):
                print("upload file", local_item)
                self.upload(local_item, remote_item)
            else:
                print("unknown item", local_item)

    def download_dir(self, smb_folder, local_dir):
        os.makedirs(local_dir, exist_ok=True)
        print(self.smb_fs.listdir(smb_folder))
        for item in self.smb_fs.listdir(smb_folder):
            smb_item = os.path.join(smb_folder, item).replace("\\", "/")
            local_item = os.path.join(local_dir, item)
            if self.smb_fs.isdir(smb_item):
                self.download_dir(smb_item, local_item)
            elif self.smb_fs.isfile(smb_item):
                self.download_dir(smb_item, local_dir)
            else:
                print("unknown item", smb_item)

with MySMB(SMB.info_dict) as smb:
    smb.do_something()
    local_dir = r'E:\code'
    smb_dir = r"/xt"
    # smb.download_smb_folder(smb_dir, local_dir)
    smb.upload_dir(local_dir, smb_dir)

pyFileSystem api
https://docs.pyfilesystem.org/en/latest/interface.html

支持列表
https://www.pyfilesystem.org/page/index-of-filesystems/
在这里插入图片描述

文章来源:https://blog.csdn.net/nongcunqq/article/details/135427321
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。