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

第四天文件操作（文本/CSV/JSON） - 异常处理机制 - 练习：日志文件分析器

news 2025/7/2 23:01:40

Python文件操作与异常处理完全指南：从入门到实战

一、为什么要学习文件操作和异常处理？

在Python编程的初级阶段，我们主要处理内存中的数据。但当我们需要：

永久保存数据
处理各种格式的数据文件
分析系统日志
构建数据处理管道

文件操作和异常处理就成为了必备的核心技能。本教程将带你系统学习这些知识点，并通过一个真实的日志分析案例巩固所学内容。

二、基础文件操作全解析

2.1 文本文件读写基础

# 写入文本文件
with open('diary.txt', 'w', encoding='utf-8') as f:
    f.write("2023-07-20 晴\n")
    f.write("今天学会了Python文件操作！\n")

# 读取文本文件
with open('diary.txt', 'r', encoding='utf-8') as f:
    content = f.read()
    print("文件内容：\n" + content)

关键点解析：

with语句自动管理文件关闭
指定编码格式避免乱码
不同模式的区别：
- ‘r’：只读（默认）
- ‘w’：写入（覆盖）
- ‘a’：追加
- ‘x’：排他创建

2.2 CSV文件处理技巧

import csv

# 写入CSV文件
data = [
    ['姓名', '年龄', '城市'],
    ['张三', 25, '北京'],
    ['李四', 30, '上海']
]

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

# 读取CSV文件
with open('people.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['姓名']} 来自 {row['城市']}")

进阶技巧：

处理带标题的CSV文件
处理不同分隔符的文件
处理包含特殊字符的字段

2.3 JSON数据操作指南

import json

# 写入JSON文件
user_data = {
    "name": "王五",
    "age": 28,
    "hobbies": ["编程", "阅读", "游泳"]
}

with open('user.json', 'w', encoding='utf-8') as f:
    json.dump(user_data, f, ensure_ascii=False, indent=2)

# 读取JSON文件
with open('user.json', 'r', encoding='utf-8') as f:
    data = json.load(f)
    print(f"{data['name']} 的年龄是 {data['age']}")

注意事项：

中文编码处理
日期等特殊类型的转换
大文件的分块处理

三、异常处理深度解析

3.1 异常处理基础结构

try:
    with open('missing.txt', 'r') as f:
        content = f.read()
except FileNotFoundError:
    print("文件不存在！")
except PermissionError:
    print("没有访问权限！")
except Exception as e:
    print(f"未知错误：{str(e)}")
else:
    print("文件读取成功！")
finally:
    print("清理完成")

3.2 常见文件操作异常

异常类型	触发场景	解决方案
FileNotFoundError	文件不存在	检查文件路径
PermissionError	没有访问权限	修改文件权限
IsADirectoryError	尝试打开目录	检查路径类型
UnicodeDecodeError	编码格式不匹配	指定正确编码

3.3 自定义异常实践

class InvalidLogFormat(Exception):
    """自定义日志格式异常"""
    def __init__(self, line_num):
        super().__init__(f"第 {line_num} 行格式错误")

def parse_log(line, line_num):
    if len(line.split('|')) != 3:
        raise InvalidLogFormat(line_num)
    # 其他解析逻辑...

try:
    parse_log("错误日志内容", 5)
except InvalidLogFormat as e:
    print(e)

四、综合实战：日志文件分析器

4.1 项目需求分析

开发一个能够：

解析多种日志格式
统计错误类型数量
生成分析报告
处理各种异常情况

的日志分析系统

4.2 完整实现代码

import re
import json
from datetime import datetime
from collections import defaultdict

class LogAnalyzer:
    LOG_PATTERN = r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(?P<level>\w+)\] (?P<message>.+)'
    
    def __init__(self, log_path):
        self.log_path = log_path
        self.stats = defaultdict(int)
        self.errors = []
        
    def analyze(self):
        try:
            with open(self.log_path, 'r', encoding='utf-8') as f:
                for line_num, line in enumerate(f, 1):
                    try:
                        self._process_line(line.strip(), line_num)
                    except Exception as e:
                        print(f"第 {line_num} 行处理失败：{str(e)}")
        except FileNotFoundError:
            print("日志文件不存在")
        except Exception as e:
            print(f"分析失败：{str(e)}")
            
    def _process_line(self, line, line_num):
        match = re.match(self.LOG_PATTERN, line)
        if not match:
            raise ValueError("无效的日志格式")
            
        data = match.groupdict()
        self.stats[data['level']] += 1
        
        if data['level'] in ['ERROR', 'FATAL']:
            self.errors.append({
                'timestamp': data['timestamp'],
                'message': data['message'],
                'line': line_num
            })
            
    def generate_report(self):
        report = {
            'summary': dict(self.stats),
            'error_count': len(self.errors),
            'errors': self.errors,
            'analysis_time': datetime.now().isoformat()
        }
        
        try:
            with open('log_report.json', 'w', encoding='utf-8') as f:
                json.dump(report, f, ensure_ascii=False, indent=2)
            print("报告生成成功！")
        except Exception as e:
            print(f"报告生成失败：{str(e)}")

# 使用示例
analyzer = LogAnalyzer('system.log')
analyzer.analyze()
analyzer.generate_report()

4.3 功能扩展建议

增加日志文件监控功能
实现邮件报警功能
支持多种日志格式配置
添加可视化图表展示

五、常见问题解答

Q1：文件操作时为什么推荐使用with语句？

自动管理文件关闭
异常安全保证
代码更简洁

Q2：如何处理大文件？

# 分块读取示例
def process_large_file(path):
    with open(path, 'r', encoding='utf-8') as f:
        while True:
            chunk = f.read(4096)  # 每次读取4KB
            if not chunk:
                break
            # 处理数据块...