Python中提升文件操作速度的七个秘诀

# 使用with语句安全地打开并读取文件
filename = 'example.txt'

with open(filename, mode='r', encoding='utf-8') as file:
    content = file.read()

print(content)

import os

directory = 'path/to/directory'
batch_size = 1000  # 每批处理的文件数量

files = os.listdir(directory)

for i in range(0, len(files), batch_size):
    batch = files[i:i + batch_size]
    
    for filename in batch:
        filepath = os.path.join(directory, filename)
        
        with open(filepath, mode='r', encoding='utf-8') as file:
            content = file.read()
            
        # 处理文件内容
        print(content)

buffer_size = 4096  # 缓冲区大小

with open('large_file.txt', mode='r', encoding='utf-8', buffering=buffer_size) as file:
    while True:
        chunk = file.read(buffer_size)
        
        if not chunk:
            break
        
        # 处理数据块
        print(chunk)

with open('large_binary_file.bin', mode='rb', buffering=4096) as file:
    while True:
        chunk = file.read(4096)
        
        if not chunk:
            break
        
        # 处理二进制数据块
        print(chunk)

import concurrent.futures

def process_file(filepath):
    with open(filepath, mode='r', encoding='utf-8') as file:
        content = file.read()
        
    # 处理文件内容
    print(content)

directory = 'path/to/directory'
files = os.listdir(directory)

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    executor.map(process_file, [os.path.join(directory, f) for f in files])

import pickle

data = {'name': 'Alice', 'age': 30, 'city': 'New York'}

# 将对象序列化并写入文件
with open('data.pickle', 'wb') as file:
    pickle.dump(data, file)

# 从文件中读取并反序列化对象
with open('data.pickle', 'rb') as file:
    loaded_data = pickle.load(file)

print(loaded_data)

import csv

# 写入CSV文件
data = [
    ['Name', 'Age', 'City'],
    ['Alice', 30, 'New York'],
    ['Bob', 25, 'Los Angeles']
]

with open('data.csv', mode='w', newline='', encoding='utf-8') as file:
    writer = csv.writer(file)
    writer.writerows(data)

# 读取CSV文件
with open('data.csv', mode='r', encoding='utf-8') as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)

[ERROR] - User Alice tried to access unauthorized resource.
[WARNING] - Disk space is running low.
[ERROR] - Database connection failed.
[INFO] - User Bob logged in successfully.
...

import os

# 定义错误类型计数器
error_counts = {}

# 设置缓冲区大小
buffer_size = 4096

# 日志文件路径
log_file_path = 'path/to/logfile.log'

# 使用with语句安全地打开文件
with open(log_file_path, mode='r', encoding='utf-8', buffering=buffer_size) as log_file:
    while True:
        chunk = log_file.read(buffer_size)
        
        if not chunk:
            break
        
        # 分割数据块中的每一行
        lines = chunk.splitlines()
        
        for line in lines:
            # 提取错误类型
            error_type = line.split(']')[0].strip('[')
            
            # 更新计数器
            if error_type in error_counts:
                error_counts[error_type] += 1
            else:
                error_counts[error_type] = 1

# 输出结果
for error_type, count in error_counts.items():
    print(f"{error_type}: {count}")