file | 某文件夹【解耦合】下的文件查找功能实现及功能单元测试

文件查找工具

概要

• 梳理业务主逻辑：

1. 查看存放被采集JSON数据的文件夹内的文件列表【所有 包含文件夹下的文件夹下的文件】这是本节内容聚焦的点
2. 和MySQL内记录的信息做比对，判断哪些文件是新文件，需要采集的。
3. 读取JSON文件，执行ETL操作(读取->写入CSV->写入MySQL（目的地库））。
4. 将被处理的JSON信息，记录到MySQL数据库（元数据库）

思路

读取某个文件中的文件名--------------------set()

读取数据库中已经处理过的文件名称-----set()
做处理 – 得到未处理过的文件名-----------set()-set()

如何处理？

• 思路1：集合相减得到的结果是在第一个集合且不在第二个集合中的元素
• 思路2: for循环

OS模块 — 学习版

• 耦合

和固定路径绑定在一起，若是将这部分代码给其他人，他们需要创建对应文件夹才可以使用

# 学习OS 接口模块import os

# 输出文件下的子文件  - 不包括子文件下的文件 仅返回子文件夹的名称
files = os.listdir(r'E:\pythonProject\ETL\day04_商品数据采集\04_数据\采集JSON')print(files)# 输出文件下的子文件 包括子文件夹呢？# 递归调用defread_dir(dir):
    results =[]
    files = os.listdir(dir)forfilein files:# 判断是否是目录if os.path.isdir(file):
            results += read_dir(dir+'/'+file)else:
            results.append(dir+'/'+file)return results

• 调用

print(read_dir(os.getcwd()))

['E:\\pythonProject\\pythonetl\\learning/learning_os.py','E:\\pythonProject\\pythonetl\\learning/learning_time.py','E:\\pythonProject\\pythonetl\\learning/learning_unittest.py','E:\\pythonProject\\pythonetl\\learning/learn_logging.py','E:\\pythonProject\\pythonetl\\learning/log_t1.py','E:\\pythonProject\\pythonetl\\learning/mycode.py']

当前路径E:\pythonProject\pythonetl\learning\learning_os.py,在learning_os.py中执行以下代码

os.getcwd()

print('getcwd',os.getcwd())

getcwd E:\pythonProject\pythonetl\learning

os.path.dirname(os.getcwd())

print('dirname',os.path.dirname(os.getcwd()))

dirname E:\pythonProject\pythonetl

os.path.dirname() 和 os.path.basename()

stra =['E:\pythonProject\pythonetl']for i in stra:print(os.path.dirname(i))print(os.path.basename(i))

E:\pythonProject
pythonetl

OS模块 — 实战版

• 在until文件夹下创建filr_until.py文件

import os

defget_dir_files_list(path="./",recursive=False):# 判断文件夹下面，有哪些文件# :param path:被判断的文件夹的路径，默认当前路径# :param recursive:是否递归读取，默认不递归# :return:1ist对象，list里面存储的是文件的路径#os.listdir这个API返回的是你给定的path下面有哪些`文件和文件夹`、
    dir_names = os.listdir(path)# print('dir_names',dir_names)# if not dir_names:#     return os.path.basename(path)
    files =[]#定义一个list,用来记录文件for dir_name in dir_names:
        absolute_path =f"{path}/{dir_name}"ifnot os.path.isdir(absolute_path):# print('not a dir')#如果进来这个if,表明这个是：文件
            files.append(absolute_path)else:# print('is a dir')#表明是文件夹if recursive:#如果recursive是True,表明要进到文件夹里面继续找文件
                files += get_dir_files_list(absolute_path,recursive)return files

单元测试

• test文件夹下的test_file_until.py文件

import os.path
from unittest import TestCase
from util.file_util import get_dir_files_list

classTestFileUtil(TestCase):# 确定手动创建的测试目录的绝对路径defsetUp(self)->None:# 测试前需要提前执行的代码  例如连接数据库
        self.project_root_path = os.path.dirname(os.getcwd())print('解耦合获得路径：',self.project_root_path)passdeftest_myfunc(self):# 要以test开头运行# 测试get_dir_files_list 函数# 该以哪一个作为我们的测试目录# 解耦合  ---'''
        请在工程根目录的test文件夹内建立：
        test_dir /
            inner1 /
                iner2 /
                    innner3/
                    5
                3
                4
            1
            2
        的目录结构用于进行此方法的单元测试
        不递归结果应该是1和2
        递归结果应该是1, 2, 3, 4, 5
        '''# 测试没有开启递归调用的代码
        result1 = get_dir_files_list(
            path = self.project_root_path +'/'+'test_dir',
            recursive=False)

        predicted_result=['1','2']# self.assertEqual(results, predicted_result)
        result_1 =[]for p in predicted_result:
            result_1.append(self.project_root_path +'/'+'test_dir'+'/'+p)# 排除顺序对结果的影响
        result1.sort()
        result_1.sort()# 这里断言 函数获得的结果和预期的结果路径是一致的
        self.assertEqual(result1, result_1)# 测试开启递归调用的代码  不包括inner3这种情况
        result2 = get_dir_files_list(
            path = self.project_root_path +'/'+'test_dir',
            recursive=True)
        predicted_result =['1','2','inner1/3','inner1/4','inner1/inner2/5',]# 若是  'inner1/4','inner1/3'  则会出现错误   因此 我们要加上一道保险  调整顺序# 使用sort()函数  没有返回值
        result_2 =[]for p in predicted_result:
            result_2.append(self.project_root_path +'/'+'test_dir'+'/'+p)

        result2.sort()
        result_2.sort()
        self.assertEqual(result2, result_2)# 测试开启递归调用的代码  测试空文件夹  针对 inner3这种情况'''
        实际上输出  不应该出现  'E:\\pythonProject\\pythonetl/test_dir/inner1/inner2/inner3'这种情况
        因为我们遍历的是可以用的文件   而非目录     所以想办法排除目录
        '''
        result3 = get_dir_files_list(
            path = self.project_root_path +'/'+'test_dir',
            recursive=True)
        predicted_result =['1','2','inner1/3','inner1/4','inner1/inner2/5','inner1/inner2/inner3']## 若是  'inner1/4','inner1/3'  则会出现错误   因此 我们要加上一道保险  调整顺序# 使用sort()函数  没有返回值
        result_3 =[]for p in predicted_result:
            result_3.append(self.project_root_path +'/'+'test_dir'+'/'+p)#if os.path.isdir(result_3[-1]):ifnot os.listdir(result_3[-1]):
                    result_3.pop(-1)

        result3.sort()
        result_3.sort()
        self.assertEqual(result3, result_3)deftearDown(self)->None:# 收尾工作pass

解耦合

在软件开发中，解耦合（Decoupling）是指减少或去除系统中各组件之间的相互依赖关系，以提高系统的灵活性和可维护性。

解耦合的目标是创建松散耦合的系统，其中每个组件或模块都可以独立于其他组件进行开发、测试和维护。

例如：

我的一个系统的文件路径是绝对路径， 若是我将该系统给另一个人，他会因为文件夹是否存在的问题而出现报错的风险，因此我们要做解耦合的措施