Python使用with语句高效实现数据清洗

f = open('data.txt', 'r')
try:
    content = f.read()
    # 处理数据...
finally:
    f.close()

with open('data.txt', 'r') as f:
    content = f.read()
    # 处理数据...
# 文件在此处自动关闭，即使发生异常也会安全关闭

# 同时读取原始数据和配置文件
with open('raw_logs.txt', 'r') as source, open('config.json', 'r') as config:
    data = source.read()
    settings = config.read()

import re

log_line = "[2023-10-27 10:05:23] ERROR: User 'user@example.com' failed to login from IP 192.168.1.1."

# 编译正则模式（预编译能提升性能）
date_pattern = re.compile(r'\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]')
email_pattern = re.compile(r"([\w\.-]+@[\w\.-]+)")

# 查找匹配
date_match = date_pattern.search(log_line)
email_match = email_pattern.search(log_line)

if date_match:
    print(f"发现时间: {date_match.group(1)}") # group(1) 获取括号内捕获的内容

if email_match:
    print(f"发现邮箱: {email_match.group(1)}")

ip_list = ['192.168.1.1', '10.0.0.1', '192.168.1.1', '172.16.0.5']
unique_ips = []
for ip in ip_list:
    if ip not in unique_ips: # 每次都要遍历列表，效率极低
        unique_ips.append(ip)

ip_list = ['192.168.1.1', '10.0.0.1', '192.168.1.1', '172.16.0.5']

# 直接转换为集合，自动去重
unique_ips = {ip for ip in ip_list}
print(unique_ips)  # 输出: {'10.0.0.1', '172.16.0.5', '192.168.1.1'}
print(len(unique_ips)) # 输出: 3

emails = ['admin@company.com', 'user@gmail.com', 'ceo@company.com', 'admin@company.com']

# 集合推导 + 条件过滤
company_emails = {email for email in emails if email.endswith('@company.com')}

print(company_emails) 
# 输出: {'admin@company.com', 'ceo@company.com'}

import re

def clean_log_data(file_path):
    # 1. 使用 with 安全打开文件
    with open(file_path, 'r', encoding='utf-8') as f:
        lines = f.readlines()

    # 定义正则：匹配包含 "ERROR" 的行，并提取其中的邮箱
    # 这是一个复合模式：先匹配 ERROR，再贪婪匹配直到行尾，最后提取邮箱
    error_email_pattern = re.compile(r"ERROR.*?([\w\.-]+@[\w\.-]+)")

    # 2. 集合推导 + 正则提取
    # 遍历每一行，如果匹配到模式，则提取邮箱，并直接存入集合
    # 这里使用了集合推导的逻辑，但在循环内部处理复杂的正则匹配
    # 或者我们可以先生成一个生成器，再传给 set()
    
    def extract_emails_generator(lines):
        for line in lines:
            match = error_email_pattern.search(line)
            if match:
                yield match.group(1)

    # 3. 生成集合并自动去重
    # 将生成器传递给 set()，这是处理大数据流的最佳内存优化方式
    bad_emails = set(extract_emails_generator(lines))

    return bad_emails

# 模拟数据写入
sample_data = """
[INFO] System started
[ERROR] Connection failed for user 'bad.actor@spam.com'
[WARN] High memory usage
[ERROR] Login failed for user 'duplicate@spam.com'
[INFO] Task completed
[ERROR] Timeout for user 'bad.actor@spam.com'
"""

# 写入临时文件
with open('temp_log.txt', 'w') as f:
    f.write(sample_data)

# 执行清洗
result = clean_log_data('temp_log.txt')
print(f"发现 {len(result)} 个异常邮箱:")
for email in result:
    print(f" - {email}")

def clean_large_log(file_path):
    pattern = re.compile(r"ERROR.*?([\w\.-]+@[\w\.-]+)")
    bad_emails = set()
    
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:  # 逐行读取，内存友好
            match = pattern.search(line)
            if match:
                bad_emails.add(match.group(1)) # 直接添加到集合
                
    return bad_emails