Python文件操作逻辑与解耦优化

逐行读取与写入：

def file_fetch(name):    with open("data", "r", encoding="utf-8") as fi:        for line in fi:            if line.find(name) > -1:                yield line

通过使用yield语句，逐行读取文件内容，不仅减少内存占用，还能在处理大文件时保持高效。

批量处理：

def file_change(*args):    with open("data", "r", encoding="utf-8") as f1, open("data1", "w", encoding="utf-8") as f2:        for line in f1:            if line in args[1]:                f2.write(args[1][line])            else:                f2.write(line)

将修改内容存入一个集合args[1]，然后在处理每一行时直接判断是否存在于集合中，大幅减少判断时间。

使用缓冲流：

import iodef file_delete(res):    with open("data", "r", encoding="utf-8") as f, open("data1", "w", encoding="utf-8") as f1:        for line in f:            if line != res[0]:                f1.write(line)

使用io.StringIO缓冲流可以显著提升读写性能，特别是在处理大量数据时。

文件不存在处理：

try:    with open(filepath, 'r') as f:        content = f.read()except FileNotFoundError:    # 处理文件未找到情况    pass

异常捕获与日志记录：

import loggingdef handle_exception(ex, context):    logging.error(f"文件操作异常：{str(ex)}", extra={'context': context})

定期检查和处理日志，及时发现和解决问题。

清晰的变量命名：

def file_fetch(name):    matched_lines = []    with open("data", "r", encoding="utf-8") as fi:        for line in fi:            if name in line:                matched_lines.append(line)    return matched_lines

通过命名清晰，代码更易理解。

合理的注释：

# 查询文件中包含指定字符串的所有行def file_fetch(name):    with open("data", "r", encoding="utf-8") as fi:        return [line for line in fi if name in line]

注释简洁明了，直接说明函数功能。

单元测试：

import unittestclass TestFileOperations(unittest.TestCase):    def setUp(self):        # 初始化测试文件        self.filename = "data"    def test_fetch(self):        self.assertEqual(len(file_fetch("小明")), 1)        self.assertEqual(file_fetch("小明")[0], "user:Marry\n")    def test_change(self):        # 初始化测试数据        original = ["user:Marry\n", "user_name:Marry,user_password:123456,user_add:上海"]        modified = ["user:Marry\n", "user_name:Marry3,user_password:1234563,user_add:上海3"]        # 调用修改函数        file_change("小明", modified)        # 验证修改结果        self.assertEqual(len(file_fetch("小明")), 2)        self.assertEqual(file_fetch("小明")[1], "user_name:Marry3,user_password:1234563,user_add:上海3")    def test_delete(self):        # 调用删除函数        file_delete(["Marry3"])        # 验证删除结果        self.assertEqual(len(file_fetch("小明")), 1)        self.assertEqual(file_fetch("小明")[0], "user:Marry\n")    def test_add(self):        # 调用添加函数        file_add(["\nuser:Marry3\n", "user_name:Marry3,user_password:1234563,user_add:上海3"])        # 验证添加结果        self.assertEqual(len(file_fetch("小明")), 3)unittest.main()

集成测试：

import pytest@pytest.fixturedef test_file():    # 初始化测试文件    with open("data", "w", encoding="utf-8") as f:        f.write("test content\n")    yield    os.remove("data")@pytest.mark.parametrize("test_input,expected", [    ("test", ["test\n"])])def test_fetch_multiple(test_file, test_input, expected):    result = file_fetch(test_input)    assert sorted(result) == sorted(expected)

通过单元测试和集成测试，确保各项功能正常运行。