Python处理TXT文件：从基础读写到高级应用

1. Python与TXT文件操作全指南

作为一名长期使用Python处理文本数据的开发者，我经常需要与TXT文件打交道。无论是数据清洗、日志分析还是简单的文本处理，TXT文件都是最基础也最常用的格式之一。今天我就来分享Python处理TXT文件的完整方案，从基础读写到高级应用，涵盖实际开发中的各种场景。

TXT文件看似简单，但在实际应用中却有许多需要注意的细节。比如编码问题、大文件处理、性能优化等，这些都是新手容易踩坑的地方。通过本文，你将掌握Python处理TXT文件的核心技巧，并能应对各种实际需求。

2. 基础读写操作

2.1 文件打开模式详解

Python提供了多种文件打开模式，理解它们的区别至关重要：

# 'r' - 只读模式（默认） with open('file.txt', 'r') as f: content = f.read() # 'w' - 写入模式（会覆盖原有内容） with open('file.txt', 'w') as f: f.write('新内容') # 'a' - 追加模式 with open('file.txt', 'a') as f: f.write('\n追加内容') # 'r+' - 读写模式 with open('file.txt', 'r+') as f: content = f.read() f.write('新增内容') # 'x' - 独占创建模式（文件存在则报错） try: with open('new_file.txt', 'x') as f: f.write('全新文件') except FileExistsError: print("文件已存在")

注意：始终使用with语句管理文件对象，它能确保文件正确关闭，即使在发生异常时也是如此。

2.2 编码问题处理

编码问题是TXT文件处理中最常见的坑之一。Python3默认使用UTF-8编码，但实际工作中会遇到各种编码格式：

# 尝试不同编码读取 encodings = ['utf-8', 'gbk', 'gb2312', 'iso-8859-1'] for enc in encodings: try: with open('file.txt', 'r', encoding=enc) as f: print(f.read()) break except UnicodeDecodeError: continue

对于不确定编码的文件，可以使用chardet库自动检测：

import chardet def detect_encoding(file_path): with open(file_path, 'rb') as f: result = chardet.detect(f.read()) return result['encoding'] encoding = detect_encoding('file.txt') with open('file.txt', 'r', encoding=encoding) as f: content = f.read()

3. 高级文件操作技巧

3.1 大文件处理策略

处理大型TXT文件时，内存效率至关重要。以下是几种高效处理方法：

1. 逐行读取：

with open('large_file.txt', 'r') as f: for line in f: process_line(line) # 处理每一行

2. 按块读取：

def read_in_chunks(file_object, chunk_size=1024): while True: data = file_object.read(chunk_size) if not data: break yield data with open('very_large_file.txt', 'r') as f: for chunk in read_in_chunks(f): process_chunk(chunk)

3. 使用内存映射：

import mmap with open('huge_file.txt', 'r') as f: with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm: # 像操作字符串一样操作文件内容 if b'search_term' in mm: print("找到搜索词")

3.2 常见文本处理模式

1. 统计词频：

from collections import Counter import re def word_count(file_path): with open(file_path, 'r') as f: words = re.findall(r'\w+', f.read().lower()) return Counter(words) word_counts = word_count('document.txt') print(word_counts.most_common(10))

2. 日志文件分析：

def analyze_logs(log_file): error_count = 0 with open(log_file, 'r') as f: for line in f: if 'ERROR' in line: error_count += 1 # 提取错误详情 timestamp = line.split()[0] message = ' '.join(line.split()[2:]) print(f"{timestamp} - {message}") print(f"总错误数: {error_count}") analyze_logs('app.log')

3. CSV转TXT：

import csv def csv_to_txt(csv_file, txt_file): with open(csv_file, 'r') as csv_f, open(txt_file, 'w') as txt_f: reader = csv.reader(csv_f) for row in reader: txt_f.write(' | '.join(row) + '\n') csv_to_txt('data.csv', 'output.txt')

4. 实际应用场景

4.1 配置文件处理

许多应用使用TXT格式的配置文件，Python可以方便地读写：

def read_config(config_file): config = {} with open(config_file, 'r') as f: for line in f: line = line.strip() if line and not line.startswith('#'): key, value = line.split('=', 1) config[key.strip()] = value.strip() return config def write_config(config, config_file): with open(config_file, 'w') as f: for key, value in config.items(): f.write(f"{key}={value}\n") # 使用示例 config = read_config('settings.txt') config['timeout'] = '30' write_config(config, 'settings_updated.txt')

4.2 数据清洗与转换

TXT文件常用于数据交换，常需要清洗和转换：

def clean_data(input_file, output_file): with open(input_file, 'r') as infile, open(output_file, 'w') as outfile: for line in infile: # 移除特殊字符 cleaned = ''.join(c for c in line if c.isalnum() or c in ' .,\n') # 标准化空格 cleaned = ' '.join(cleaned.split()) outfile.write(cleaned + '\n') clean_data('raw_data.txt', 'cleaned_data.txt')

4.3 批量文件处理

自动化处理多个TXT文件：

import os from pathlib import Path def batch_process(input_dir, output_dir, process_func): Path(output_dir).mkdir(exist_ok=True) for filename in os.listdir(input_dir): if filename.endswith('.txt'): input_path = os.path.join(input_dir, filename) output_path = os.path.join(output_dir, filename) with open(input_path, 'r') as infile, open(output_path, 'w') as outfile: processed = process_func(infile.read()) outfile.write(processed) # 示例处理函数：转换为大写 def to_uppercase(text): return text.upper() batch_process('input_files', 'output_files', to_uppercase)

5. 性能优化与错误处理

5.1 提高IO性能的技巧

1. 缓冲设置：

# 使用更大的缓冲区（单位：字节） with open('large.txt', 'r', buffering=8192) as f: for line in f: process(line)

2. 使用生成器处理大文件：

def process_lines(file_path): with open(file_path, 'r') as f: for line in f: yield process_line(line) for processed in process_lines('big_file.txt'): save_result(processed)

3. 并行处理：

from multiprocessing import Pool def process_line_parallel(line): return process_line(line) def parallel_file_process(file_path, num_processes=4): with open(file_path, 'r') as f: lines = f.readlines() with Pool(num_processes) as p: results = p.map(process_line_parallel, lines) return results

5.2 常见错误与解决方案

1. 文件不存在错误：

try: with open('nonexistent.txt', 'r') as f: content = f.read() except FileNotFoundError: print("文件不存在，请检查路径") # 可选：创建新文件 open('nonexistent.txt', 'w').close()

2. 权限问题：

try: with open('/root/restricted.txt', 'w') as f: f.write('test') except PermissionError: print("没有写入权限，请使用sudo或更改文件权限")

3. 处理损坏文件：

def safe_read(file_path): try: with open(file_path, 'r') as f: return f.read() except UnicodeDecodeError: # 尝试二进制读取 with open(file_path, 'rb') as f: return f.read().decode('utf-8', errors='replace') except Exception as e: print(f"读取文件出错: {e}") return None

6. 实用工具函数库

6.1 常用文本处理函数

def count_lines(file_path): """高效统计文件行数""" with open(file_path, 'r') as f: return sum(1 for _ in f) def search_in_file(file_path, search_term): """在文件中搜索关键词""" matches = [] with open(file_path, 'r') as f: for line_num, line in enumerate(f, 1): if search_term in line: matches.append((line_num, line.strip())) return matches def compare_files(file1, file2): """比较两个文件内容是否相同""" with open(file1, 'r') as f1, open(file2, 'r') as f2: return f1.read() == f2.read()

6.2 文件差异比较

import difflib def file_diff(file1, file2): with open(file1, 'r') as f1, open(file2, 'r') as f2: diff = difflib.unified_diff( f1.readlines(), f2.readlines(), fromfile=file1, tofile=file2, ) return ''.join(diff) # 生成差异报告 diff_report = file_diff('version1.txt', 'version2.txt') with open('diff_report.txt', 'w') as f: f.write(diff_report)

6.3 文件合并与分割

def merge_files(file_list, output_file): """合并多个文件""" with open(output_file, 'w') as outfile: for fname in file_list: with open(fname, 'r') as infile: outfile.write(infile.read()) outfile.write('\n') # 文件间添加空行 def split_file(input_file, lines_per_file=1000): """分割大文件为多个小文件""" with open(input_file, 'r') as infile: file_count = 0 while True: lines = [] for _ in range(lines_per_file): line = infile.readline() if not line: break lines.append(line) if not lines: break file_count += 1 with open(f'split_{file_count}.txt', 'w') as outfile: outfile.writelines(lines) return file_count

7. 实际项目应用案例

7.1 日志分析系统

构建一个简单的日志分析系统：

class LogAnalyzer: def __init__(self, log_file): self.log_file = log_file self.stats = { 'errors': 0, 'warnings': 0, 'info': 0, 'patterns': {} } def analyze(self): with open(self.log_file, 'r') as f: for line in f: self._process_line(line) return self.stats def _process_line(self, line): if 'ERROR' in line: self.stats['errors'] += 1 self._track_pattern(line, 'ERROR') elif 'WARNING' in line: self.stats['warnings'] += 1 self._track_pattern(line, 'WARNING') else: self.stats['info'] += 1 def _track_pattern(self, line, level): # 提取错误消息中的第一个单词作为模式 parts = line.split() if len(parts) > 2: pattern = parts[2] if pattern not in self.stats['patterns']: self.stats['patterns'][pattern] = {'count': 0, 'level': level} self.stats['patterns'][pattern]['count'] += 1 # 使用示例 analyzer = LogAnalyzer('server.log') results = analyzer.analyze() print(f"错误总数: {results['errors']}") print("常见错误模式:") for pattern, data in results['patterns'].items(): print(f"{pattern}: {data['count']}次")

7.2 文本数据清洗管道

构建一个可扩展的文本清洗管道：

class TextCleaner: def __init__(self): self.pipeline = [] def add_step(self, func): self.pipeline.append(func) return self # 支持链式调用 def clean(self, text): for step in self.pipeline: text = step(text) return text @staticmethod def remove_special_chars(text): import string return ''.join(c for c in text if c in string.printable) @staticmethod def normalize_whitespace(text): import re return re.sub(r'\s+', ' ', text).strip() @staticmethod def to_lowercase(text): return text.lower() # 使用示例 cleaner = (TextCleaner() .add_step(TextCleaner.remove_special_chars) .add_step(TextCleaner.normalize_whitespace) .add_step(TextCleaner.to_lowercase)) with open('dirty_text.txt', 'r') as f: dirty_text = f.read() clean_text = cleaner.clean(dirty_text) with open('clean_text.txt', 'w') as f: f.write(clean_text)

7.3 自动化报告生成

结合文本模板生成报告：

class ReportGenerator: def __init__(self, template_file): with open(template_file, 'r') as f: self.template = f.read() def generate(self, output_file, **kwargs): report = self.template for key, value in kwargs.items(): report = report.replace(f'{{{{{key}}}}}', str(value)) with open(output_file, 'w') as f: f.write(report) # 模板文件内容示例： # 报告日期: {{date}} # 总销售额: ${{sales}} # 热门产品: {{product}} # 使用示例 generator = ReportGenerator('report_template.txt') generator.generate( 'sales_report.txt', date='2023-07-15', sales=125000, product='Python编程书籍' )

8. 性能对比与最佳实践

8.1 不同读取方式性能对比

我们比较几种常见的文件读取方法：

import timeit def test_read_all(): with open('medium_file.txt', 'r') as f: content = f.read() def test_line_by_line(): with open('medium_file.txt', 'r') as f: for line in f: pass def test_readlines(): with open('medium_file.txt', 'r') as f: lines = f.readlines() # 测试结果（1MB文件，1000行） print("一次性读取:", timeit.timeit(test_read_all, number=100)) print("逐行读取:", timeit.timeit(test_line_by_line, number=100)) print("readlines:", timeit.timeit(test_readlines, number=100))

典型结果：

• 一次性读取：最快，但内存消耗大
• 逐行读取：内存效率高，适合大文件
• readlines：介于两者之间

8.2 最佳实践总结

1. 文件路径处理：

   • 使用os.path或pathlib处理路径，确保跨平台兼容性

   from pathlib import Path file_path = Path('data') / 'subdir' / 'file.txt'

2. 异常处理：

   • 始终处理可能的IOError异常
   • 考虑文件锁定情况（特别是多进程/线程场景）

3. 资源管理：

   • 优先使用with语句
   • 对于长期打开的文件，考虑定期flush()

4. 性能敏感场景：

   • 大文件使用生成器逐行处理
   • 频繁读写考虑内存映射(mmap)
   • 批量操作使用缓冲

5. 代码可维护性：

   • 将文件操作封装成函数或类
   • 添加适当的日志记录
   • 编写单元测试覆盖边界情况

9. 扩展应用：与其他格式转换

9.1 JSON与TXT互转

import json def json_to_txt(json_file, txt_file): with open(json_file, 'r') as jf, open(txt_file, 'w') as tf: data = json.load(jf) if isinstance(data, dict): for key, value in data.items(): tf.write(f"{key}: {value}\n") elif isinstance(data, list): for item in data: tf.write(f"{item}\n") def txt_to_json(txt_file, json_file, format='list'): data = [] with open(txt_file, 'r') as tf: for line in tf: line = line.strip() if line: if format == 'dict' and ':' in line: key, value = line.split(':', 1) data.append({key.strip(): value.strip()}) else: data.append(line) with open(json_file, 'w') as jf: json.dump(data, jf, indent=2)

9.2 CSV与TXT互转

import csv def csv_to_txt(csv_file, txt_file, delimiter=','): with open(csv_file, 'r') as cf, open(txt_file, 'w') as tf: reader = csv.reader(cf, delimiter=delimiter) for row in reader: tf.write(' | '.join(row) + '\n') def txt_to_csv(txt_file, csv_file, delimiter='|'): with open(txt_file, 'r') as tf, open(csv_file, 'w') as cf: writer = csv.writer(cf) for line in tf: row = [col.strip() for col in line.split(delimiter)] writer.writerow(row)

9.3 XML与TXT转换

from xml.etree import ElementTree as ET def xml_to_txt(xml_file, txt_file): tree = ET.parse(xml_file) root = tree.getroot() with open(txt_file, 'w') as tf: for elem in root.iter(): if elem.text and elem.text.strip(): tf.write(f"{elem.tag}: {elem.text.strip()}\n") def txt_to_xml(txt_file, xml_file, root_tag='data'): root = ET.Element(root_tag) with open(txt_file, 'r') as tf: for line in tf: if ':' in line: tag, text = line.split(':', 1) elem = ET.SubElement(root, tag.strip()) elem.text = text.strip() tree = ET.ElementTree(root) tree.write(xml_file, encoding='utf-8', xml_declaration=True)

10. 调试与测试技巧

10.1 文件操作单元测试

使用unittest测试文件操作：

import unittest import os from tempfile import NamedTemporaryFile class TestFileOperations(unittest.TestCase): def setUp(self): self.test_file = NamedTemporaryFile(delete=False, mode='w+') self.test_file.write("Line 1\nLine 2\nLine 3") self.test_file.close() def tearDown(self): os.unlink(self.test_file.name) def test_read_lines(self): with open(self.test_file.name, 'r') as f: lines = f.readlines() self.assertEqual(len(lines), 3) self.assertEqual(lines[0].strip(), "Line 1") def test_write_file(self): test_content = "Test content" with open(self.test_file.name, 'w') as f: f.write(test_content) with open(self.test_file.name, 'r') as f: content = f.read() self.assertEqual(content, test_content) if __name__ == '__main__': unittest.main()

10.2 使用日志调试文件操作

import logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', filename='file_ops.log' ) def safe_file_operation(file_path): try: logging.info(f"尝试打开文件: {file_path}") with open(file_path, 'r') as f: content = f.read() logging.debug(f"读取到 {len(content)} 字节数据") return content except Exception as e: logging.error(f"文件操作失败: {str(e)}") return None # 使用示例 content = safe_file_operation('important.txt')

10.3 性能分析

使用cProfile分析文件操作性能：

import cProfile def process_large_file(): with open('large_file.txt', 'r') as f: for line in f: # 模拟处理 _ = line.upper() # 性能分析 cProfile.run('process_large_file()', sort='cumtime')

11. 安全注意事项

11.1 文件操作安全最佳实践

1. 路径安全：

from pathlib import Path def safe_join(base, *paths): """安全拼接路径，防止目录遍历攻击""" base_path = Path(base).resolve() try: full_path = base_path.joinpath(*paths).resolve() # 确保最终路径仍在基础目录内 full_path.relative_to(base_path) return str(full_path) except ValueError: raise ValueError("非法路径访问尝试")

2. 文件权限检查：

import os def check_permissions(file_path): """检查文件权限是否安全""" if not os.path.exists(file_path): raise FileNotFoundError(f"文件不存在: {file_path}") mode = os.stat(file_path).st_mode if mode & 0o777 == 0o777: raise PermissionError(f"文件权限过于宽松: {oct(mode)}") return True

3. 安全文件写入：

def atomic_write(file_path, content): """原子写入文件，避免写入过程中断导致文件损坏""" import tempfile dirname = os.path.dirname(file_path) with tempfile.NamedTemporaryFile( mode='w', dir=dirname, delete=False ) as tmp_file: tmp_file.write(content) tmp_path = tmp_file.name # 原子重命名 os.replace(tmp_path, file_path)

11.2 处理敏感数据

处理包含敏感信息的TXT文件：

def redact_sensitive(input_file, output_file, sensitive_words): """模糊处理敏感信息""" with open(input_file, 'r') as infile, open(output_file, 'w') as outfile: for line in infile: for word in sensitive_words: line = line.replace(word, '***REDACTED***') outfile.write(line) # 使用示例 sensitive = ['password', 'credit card', 'SSN'] redact_sensitive('log.txt', 'redacted_log.txt', sensitive)

12. 高级主题：自定义文件处理类

12.1 实现一个高级文件处理器

class TextFileProcessor: def __init__(self, file_path, encoding='utf-8'): self.file_path = file_path self.encoding = encoding self._line_count = None @property def line_count(self): """缓存行数统计结果""" if self._line_count is None: self._line_count = self._count_lines() return self._line_count def _count_lines(self): with open(self.file_path, 'r', encoding=self.encoding) as f: return sum(1 for _ in f) def search(self, pattern, ignore_case=False): """在文件中搜索模式""" import re flags = re.IGNORECASE if ignore_case else 0 regex = re.compile(pattern, flags) matches = [] with open(self.file_path, 'r', encoding=self.encoding) as f: for line_num, line in enumerate(f, 1): if regex.search(line): matches.append((line_num, line.strip())) return matches def transform_lines(self, transform_func, output_file=None): """应用转换函数到每一行""" output_file = output_file or self.file_path temp_file = f"{output_file}.tmp" with open(self.file_path, 'r', encoding=self.encoding) as infile, \ open(temp_file, 'w', encoding=self.encoding) as outfile: for line in infile: outfile.write(transform_func(line)) os.replace(temp_file, output_file) return True # 使用示例 processor = TextFileProcessor('data.txt') print(f"文件行数: {processor.line_count}") # 搜索包含"error"的行 errors = processor.search(r'error', ignore_case=True) for line_num, line in errors: print(f"Line {line_num}: {line}") # 转换所有行为大写 processor.transform_lines(str.upper, 'data_uppercase.txt')

12.2 实现文件差异分析器

class FileDiffer: def __init__(self, file1, file2): self.file1 = file1 self.file2 = file2 def compute_diff(self, context=3): """计算并返回文件差异""" from difflib import ndiff with open(self.file1, 'r') as f1, open(self.file2, 'r') as f2: diff = ndiff( f1.readlines(), f2.readlines() ) return list(diff) def unified_diff(self, output_file=None, context=3): """生成unified diff格式差异""" from difflib import unified_diff with open(self.file1, 'r') as f1, open(self.file2, 'r') as f2: diff = unified_diff( f1.readlines(), f2.readlines(), fromfile=self.file1, tofile=self.file2, n=context ) if output_file: with open(output_file, 'w') as f: f.writelines(diff) return True else: return list(diff) def html_diff(self, output_file): """生成HTML格式的差异报告""" from difflib import HtmlDiff with open(self.file1, 'r') as f1, open(self.file2, 'r') as f2: differ = HtmlDiff() html = differ.make_file( f1.readlines(), f2.readlines(), fromdesc=self.file1, todesc=self.file2 ) with open(output_file, 'w') as f: f.write(html) return True # 使用示例 differ = FileDiffer('version1.txt', 'version2.txt') diff = differ.compute_diff() for line in diff: print(line, end='') # 生成HTML差异报告 differ.html_diff('diff_report.html')

13. 性能优化深入探讨

13.1 内存映射高级用法

对于超大文件，内存映射(mmap)是最佳选择：

import mmap import os class MappedFile: def __init__(self, file_path, access=mmap.ACCESS_READ): self.file_path = file_path self.access = access self._file = None self._mmap = None def __enter__(self): self._file = open(self.file_path, 'r' if self.access == mmap.ACCESS_READ else 'r+') self._mmap = mmap.mmap( self._file.fileno(), 0, # 映射整个文件 access=self.access ) return self def __exit__(self, exc_type, exc_val, exc_tb): if self._mmap: self._mmap.close() if self._file: self._file.close() def find_all(self, pattern): """查找所有匹配模式的位置""" result = [] offset = 0 pattern = pattern.encode('utf-8') while True: pos = self._mmap.find(pattern, offset) if pos == -1: break result.append(pos) offset = pos + 1 return result def replace_all(self, old, new): """替换所有匹配的字符串""" if self.access == mmap.ACCESS_READ: raise ValueError("文件以只读模式打开") old = old.encode('utf-8') new = new.encode('utf-8') if len(old) != len(new): raise ValueError("替换字符串长度必须相同") positions = self.find_all(old.decode('utf-8')) for pos in positions: self._mmap[pos:pos+len(old)] = new return len(positions) # 使用示例 with MappedFile('large_file.txt') as mf: print(f"文件大小: {len(mf._mmap)} 字节") positions = mf.find_all('important') print(f"找到 {len(positions)} 处匹配") # 替换内容 with MappedFile('editable.txt', mmap.ACCESS_WRITE) as mf: count = mf.replace_all('old', 'new') print(f"替换了 {count} 处")

13.2 多进程处理大文件

利用多核CPU并行处理大文件：

from multiprocessing import Pool, cpu_count import os class ParallelFileProcessor: def __init__(self, file_path, num_processes=None): self.file_path = file_path self.num_processes = num_processes or cpu_count() def _split_file(self, chunk_size): """将文件分割为多个块""" file_size = os.path.getsize(self.file_path) chunks = [] with open(self.file_path, 'rb') as f: start = 0 while start < file_size: end = min(start + chunk_size, file_size) f.seek(end) # 确保在行边界结束 while end < file_size and f.read(1) != b'\n': end += 1 f.seek(end) chunks.append((start, end)) start = end + 1 return chunks def _process_chunk(self, start_end): """处理单个文件块""" start, end = start_end results = [] with open(self.file_path, 'rb') as f: f.seek(start) while f.tell() < end: line = f.readline() if not line: break # 在这里添加实际的处理逻辑 processed = line.decode('utf-8').upper().strip() results.append(processed) return results def process(self, chunk_size=1024*1024): """并行处理文件""" chunks = self._split_file(chunk_size) with Pool(self.num_processes) as pool: results = pool.map(self._process_chunk, chunks) # 合并结果 return [item for sublist in results for item in sublist] # 使用示例 processor = ParallelFileProcessor('very_large.txt') results = processor.process() print(f"处理了 {len(results)} 行数据")

14. 实际项目：构建日志分析工具

14.1 设计日志分析工具

import re from collections import defaultdict, Counter from datetime import datetime class LogAnalyzerPro: LOG_PATTERNS = { 'apache': r'^(\S+) (\S+) (\S+) \[([^\]]+)\] "(\S+) (\S+) (\S+)" (\d+) (\d+)', 'nginx': r'^(\S+) - (\S+) \[([^\]]+)\] "(\S+) (\S+) (\S+)" (\d+) (\d+) "([^"]*)" "([^"]*)"', 'syslog': r'^(\w{3} \d{2} \d{2}:\d{2}:\d{2}) (\S+) (\S+)\[(\d+)\]: (.*)' } def __init__(self, log_file, log_type='auto'): self.log_file = log_file self.log_type = self._detect_log_type() if log_type == 'auto' else log_type self.stats = { 'total_lines': 0, 'parsed_lines': 0, 'errors': defaultdict(int), 'requests': Counter(), 'status_codes': Counter(), 'sections': Counter(), 'hourly': defaultdict(int) } def _detect_log_type(self): """自动检测日志类型""" with open(self.log_file, 'r') as f: sample = f.readline() for log_type, pattern in self.LOG_PATTERNS.items(): if re.match(pattern, sample): return log_type return 'unknown' def analyze(self): """分析日志文件""" pattern = self.LOG_PATTERNS.get(self.log_type) if not pattern: raise ValueError(f"不支持的日志类型: {self.log_type}") with open(self.log_file, 'r') as f: for line in f: self.stats['total_lines'] += 1 match = re.match(pattern, line) if not match: self.stats['errors']['parse_failed'] += 1 continue self.stats['parsed_lines'] += 1 self._process_match(match) return self.stats def _process_match(self, match): """处理匹配的日志行""" if self.log_type == 'apache': ip, _, user, date, method, path, _, status, size = match.groups() self._process_apache_log(ip, date, method, path, status) elif self.log_type == 'nginx': ip, _, date, method, path, _, status, size, _, _ = match.groups() self._process_nginx_log(ip, date, method, path, status) elif self.log_type == 'syslog': date, host, app, pid, message = match.groups() self._process_syslog(date, host, app, message) def _process_apache_log(self, ip, date