当前位置：首页 > news >正文

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

news 来源：原创 2024/9/22 6:50:21

文件查找工具

概要
思路
OS模块 --- 学习版
- os.getcwd()
- os.path.dirname(os.getcwd())
- os.path.dirname() 和 os.path.basename()
OS模块 — 实战版
- - 单元测试
  - - 解耦合

概要

梳理业务主逻辑：

查看存放被采集JSON数据的文件夹内的文件列表【所有包含文件夹下的文件夹下的文件】
这是本节内容聚焦的点
和MySQL内记录的信息做比对，判断哪些文件是新文件，需要采集的。
读取JSON文件，执行ETL操作(读取->写入CSV->写入MySQL（目的地库））。
将被处理的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)# 输出文件下的子文件 包括子文件夹呢？
# 递归调用
def read_dir(dir):results =[]files = os.listdir(dir)for file in 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 osdef get_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}"if not 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_listclass TestFileUtil(TestCase):# 确定手动创建的测试目录的绝对路径def setUp(self) -> None: # 测试前需要提前执行的代码  例如连接数据库self.project_root_path = os.path.dirname(os.getcwd())print('解耦合获得路径：',self.project_root_path)passdef test_myfunc(self): # 要以test开头运行# 测试get_dir_files_list 函数# 该以哪一个作为我们的测试目录# 解耦合  ---'''请在工程根目录的test文件夹内建立：test_dir /inner1 /iner2 /innner3/53412的目录结构用于进行此方法的单元测试不递归结果应该是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]):if not os.listdir(result_3[-1]):result_3.pop(-1)result3.sort()result_3.sort()self.assertEqual(result3, result_3)def tearDown(self)-> None: # 收尾工作pass