Работа с постоянно изменяющимися файлами, такими как журналы событий, требует особого подхода, чтобы обеспечить эффективность и избежать проблем с производительностью и потерей данных. Вот несколько стратегий и приемов:
Основные стратегии:
import time
def tail(filename):
with open(filename, 'r') as f:
f.seek(0, 2) # Перейти в конец файла
while True:
line = f.readline()
if line:
yield line
else:
time.sleep(0.1) # Небольшая задержка для предотвращения чрезмерной нагрузки на процессор
if __name__ == "__main__":
for line in tail('my_log_file.log'):
print(line.strip())
# Пример с watchdog (требуется установка: pip install watchdog)
import time
import sys
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class LogFileHandler(FileSystemEventHandler):
def __init__(self, filename):
self.filename = filename
self.last_position = 0
with open(filename, 'r') as f:
f.seek(0,2) #начальная позиция - конец файла.
self.last_position = f.tell()
super().__init__()
def on_modified(self, event):
if event.src_path == self.filename:
self.read_new_lines()
def read_new_lines(self):
try:
with open(self.filename, 'r') as f:
f.seek(self.last_position)
for line in f:
print(line.strip())
self.last_position = f.tell()
except FileNotFoundError:
print(f"Файл {self.filename} не найден.")
except Exception as e:
print(f"Ошибка при чтении файла: {e}")
if __name__ == "__main__":
log_file = "my_log_file.log" # Замените на имя вашего файла логов
event_handler = LogFileHandler(log_file)
observer = Observer()
observer.schedule(event_handler, path='.', recursive=False) #следим за текущим каталогом
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
Дополнительные соображения:
В заключение, выбор конкретного подхода зависит от требований вашего проекта, размера и формата логов, а также от доступных ресурсов. Важно тщательно протестировать выбранное решение, чтобы убедиться, что оно работает надежно и эффективно.