博客

# -*- coding = utf-8 -*-
# @Time : 2022/6/23 12:13
# @Author : Suage
# @File : scan_files.py
import re
from ftplib import FTP


class ScanFiles:
    __SIZE_UNITS = list(f'{_}B'.strip() for _ in ' KMGTP')

    def __init__(self, host: str, port: int, username: str, password: str):
        """初始化FTP连接并登陆"""
        self.f = FTP()
        self.f.connect(host=host, port=port)
        self.f.login(user=username, passwd=password)

    @staticmethod
    def convert_size(size: int):
        """文件尺寸单位转换"""
        c = 0
        while size > 1024:
            size /= 1024
            c += 1
        return f"{size:.2f} {ScanFiles.__SIZE_UNITS[c]}"

    @staticmethod
    def analyze_ftp_line(line):
        """解析FTP文件系统信息行"""
        res = re.search(
            r'(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s(.*)',
            line)
        is_dir: bool = res.group(1).startswith('d')
        size: int = int(res.group(5))
        over_path: str = res.group(9)
        return {'is_dir': is_dir,
                'size': size,
                'item': over_path, }

    @staticmethod
    def default_callback(data: dict):
        """默认回调函数"""
        if data['is_dir']: return  # 开启此行不显示文件夹
        print(f'{data["path"]} Sizes:{ScanFiles.convert_size(data["size"])}')

    def walk(self, p, callback=None):
        """
        遍历文件
        :param p: 起始路径
        :param callback: 回调函数, 传入类型为dict(keys: is_dir, size, item, path)
        :return:
        """
        if callback is None: callback = ScanFiles.default_callback
        lis = list()
        self.f.dir(p, lis.append)
        for line in lis:
            al = ScanFiles.analyze_ftp_line(line)
            al['path'] = f'{p}/{al["item"]}'.replace('\\', '/').replace('//', '/')
            callback(al)
            if al['is_dir']: self.walk(al['path'], callback)

    def quit(self):
        """与FTP服务器断开连接"""
        self.f.quit()


if __name__ == '__main__':
    sf = ScanFiles(
        host='127.0.0.1',
        port=2121,
        username='root',
        password='passwd')
    try:
        sf.walk('./')
    finally:
        sf.quit()