一、代碼實現

import logging
import os
import threading
import timefrom watchdog.events import FileSystemEventHandler
from watchdog.observers import Observerlogger = logging.getLogger(__name__)class LogWatcher(FileSystemEventHandler):def __init__(self, log_file, on_modified_callback=None):""""初始化 LogWatcher 類的實例。參數：- log_file：日志文件的路徑- on_modified_callback：可選的回調函數，在文件修改時調用屬性：- log_file：日志文件的路徑- file_object：日志文件對象- on_modified_callback：文件修改回調函數- last_line：最后一行文本- observer：觀察者對象- match_string：需要匹配的字符串- stop_watching：停止監視的標志"""self.log_file = log_fileself.file_object = open(log_file, 'rb')self.on_modified_callback = on_modified_callbackself.last_line = self.get_last_line()  # 初始化時獲取最后一行文本self.observer = Observer()self.observer.schedule(self, ".", recursive=False)self.match_string = Noneself.stop_watching = Falsedef start(self):"""啟動觀察者對象，開始監視文件變化。"""self.observer.start()def stop(self):"""停止觀察者對象，結束監視文件變化。"""self.observer.stop()self.observer.join()self.file_object.close()def get_last_line(self):"""獲取日志文件的最后一行文本。它通過將文件指針移動到文件末尾，然后逐個字符向前搜索，直到找到換行符為止。返回值：- 最后一行文本，如果文件為空則返回None"""# 將文件指針移動到文件末尾self.file_object.seek(0, os.SEEK_END)# 獲取當前文件指針的位置（此時指針在最后一行的末尾）position = self.file_object.tell()try:# 嘗試向前移動兩個字節new_position = max(position - 2, 0)self.file_object.seek(new_position, os.SEEK_SET)except OSError as e:# 如果發生錯誤，可能是文件太小，返回Nonereturn None# 逐個字符向前搜索，確保文件指針最終停在當前行的第一個字符處while True:# read(1)讀取的是指針位置的下一個字符，每次調用read(1)都會讀取一個字符，并將指針向后移動一個字符的位置。char = self.file_object.read(1).decode('utf-8', errors='ignore')if char == '\n':breakif new_position == 0:# 如果已經到達文件開頭，跳出循環break# 嘗試向前移動一個字節位置，確保不越界到文件開頭new_position = max(new_position - 1, 0)# 將文件指針移動到新的位置self.file_object.seek(new_position, os.SEEK_SET)# last_line = self.file_object.readline().decode('utf-8', errors='ignore').strip()last_line = self.file_object.read(position - new_position).decode('utf-8', errors='ignore').strip()# 輸出調試信息logger.debug(f'Reading line: {last_line}')return last_linedef on_modified(self, event):"""on_modified方法是FileSystemEventHandler的回調方法，當日志文件發生變化時，都會調用這個方法。參數：- event：文件變化事件對象"""# 注意，這里一個要用絕對路徑比較，不能直接使用 event.src_path == self.log_file，# event.src_path == self.log_file 的值為false# if event.src_path == self.log_file:if os.path.abspath(event.src_path) == os.path.abspath(self.log_file):# 在文件發生變化時，實時獲取最后一行文本self.last_line = self.get_last_line()# 用戶可在外部傳入一個回調方法，在文本發生變化時執行該事件if self.on_modified_callback:self.on_modified_callback()# 調用基類的同名方法，以便執行基類的默認行為super(LogWatcher, self).on_modified(event)def tail_last_line_and_match(self, match_string=None, max_match_seconds=10):"""實時監控日志文件的變化，并實時獲取最后一行文本。如果匹配到指定的字符串，停止監視。參數：- match_string：需要匹配的字符串"""self.match_string = match_stringself.start()end_time = time.time() + max_match_secondstry:while not self.stop_watching and time.time() <= end_time:if self.match_string and self.match_string in self.last_line:self.stop_watching = Trueexcept KeyboardInterrupt:passself.stop_watching = True  # 停止監視循環def write_logs(log_file):"""在新線程中寫入日志"""for i in range(10):with open(log_file, 'a') as file:file.write(f'New log entry {i}\n')time.sleep(1)  # 每秒寫入一次日志if __name__ == '__main__':import logginglogging.basicConfig(level=logging.DEBUG)log_file = 'demo.log'# 創建日志文件并寫入示例日志with open(log_file, 'w') as file:file.write('This is the first line of the log.\n')file.write('This is the second line of the log.\n')log_watcher = LogWatcher(log_file)# 啟動新線程寫入日志write_thread = threading.Thread(target=write_logs, args=(log_file,))write_thread.start()# 啟動實時監控日志文件變化，并打印最后一行文本，直到匹配到指定字符串或超時才停止監視log_watcher.tail_last_line_and_match(match_string='New log entry 9', max_match_seconds=20)# 等待寫入線程結束write_thread.join()

三、Demo驗證

運行代碼，控制臺的輸出結果：

DEBUG:__main__:Reading line: This is the second line of the log.
DEBUG:__main__:Reading line: New log entry 0
DEBUG:__main__:Reading line: New log entry 1
DEBUG:__main__:Reading line: New log entry 2
DEBUG:__main__:Reading line: New log entry 3
DEBUG:__main__:Reading line: New log entry 4
DEBUG:__main__:Reading line: New log entry 5
DEBUG:__main__:Reading line: New log entry 6
DEBUG:__main__:Reading line: New log entry 7
DEBUG:__main__:Reading line: New log entry 8
DEBUG:__main__:Reading line: New log entry 9Process finished with exit code 0

歡迎技術交流：