使用 Python 连续读取日志以在 GTK window 中显示
Continously read from log with Python to display in GTK window
我正在尝试使用 Python 3.9 和 PyGObject 编写一个小型应用程序 GUI (GTK) 应用程序，它从日志文件中读取数据并在 GUI 中显示数据 window。
在这种特殊情况下，我试图不断地从 systemd Journal 中读取数据，这可以通过例如journalctl -f 通过终端。
对于 CLI 应用程序，我看到可以使用类似这样的方法将日志文件的输出通过管道传输到 python 脚本的 STOUT：
p = Popen(["journalctl", "--user-unit=appToMonitor"], stdout=PIPE)
with p.stdout:
for line in iter(p.stdout.readline, b''):
print(line, end=''),
p.wait()
用于 GUI 应用程序的 OFC 我需要以某种方式连接到 GTK 主循环，因此上述解决方案不起作用。我做了一些谷歌搜索，据我所知，一个解决方案是使用 gobject.timeout_add() 定期轮询日志文件以进行更改。
这是从系统日志文件中获取数据的最佳方式吗？还是有其他解决方案可以用来避免轮询？如果 gobject.timeout_add() 是要走的路，我如何确保只添加自上次从日志文件读取后添加的日志行？
没有必要使用gobject.timeout_add()；您可以只使用专门用于线程的 threading 模块，并且比尝试使用 GTK 更简单。
使用 `threading` 调用函数
使用 threading 的好处在于，即使它不是 Gtk 模块组的一部分，它仍然不会阻塞 Gtk 的主循环。这是一个使用 threading:
的简单超时线程的简单示例
import threading
def function():
print("Hello!")
# Create a timer
timer = threading.Timer(
1, # Interval (in seconds; can be a float) after which to call the function
function # Function to call after time interval
)
timer.start() # Start the timer
这会在 1 秒后调用 function() 一次，然后退出。
但是，在您的情况下，您想多次调用该函数，以反复检查日志的状态。为此，您可以在 function() 中重新创建计时器，然后再次 运行 ：
import threading
def function():
global timer
print("Hello!")
# Recreate and start the timer
timer = threading.Timer(1, function)
timer.start()
timer = threading.Timer(1, function)
timer.start()
现在要检查日志中是否添加了任何新行，程序需要读取日志，然后将最近读取的行与之前读取的行进行比较。
读取行，并使用 `difflib` 进行比较
首先，要完全比较这些行，需要将它们存储在两个列表中：一个包含最近读取的一组行，另一个包含前一组行。下面是读取日志输出、将行存储在列表中然后打印列表的代码示例：
from subprocess import PIPE, Popen
# The list storing the lines of the log
current_lines = []
# Read the log
p = Popen(["journalctl", "--user-unit=appToMonitor"], stdout=PIPE)
with p.stdout:
for line in iter(p.stdout.readline, b''):
current_lines.append(line.decode("utf-8")) # Add the lines of the log to the list
p.wait()
print(current_lines)
注意一定要用line.decode("uft-8")，因为p.stdout.readline的输出是以字节为单位的。使用您的程序使用的任何编码； utf-8只是一个普通的编码。
然后您可以使用 difflib 模块来比较两个列表。这是一个示例程序，它比较两个行列表，并打印第二个列表中而不是第一个列表中的任何行：
import difflib
# Two lists to compare
last_lines = ["Here is the first line\n"]
current_lines = ["Here is the first line\n", "and here is the second line"]
# Iterate through all the lines, and check for new ones
for line in difflib.ndiff(last_lines, current_lines):
# Print the line only if it was not in last_lines
if line[0] == "+": # difflib inserts a "+" for every addition
print("Line added:", line.replace("+ ", "")) # Remove the "+" from the line and print it
很好，但是我们如何将所有这些都放在一个程序中呢？
结局：将所有这些概念放入一个程序中，该程序每秒从日志中读取输出，并将任何新行添加到 window 中的文本小部件。这是一个简单的 Gtk 应用程序，它就是这样做的：
import difflib
import gi
import threading
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from subprocess import PIPE, Popen
class App(Gtk.Window):
"""The application window."""
def __init__(self):
Gtk.Window.__init__(self)
self.connect("delete-event", self.quit)
# The list of lines read from the log file
self.current_lines = []
# The list of previously read lines to compare to the current one
self.last_lines = []
# The box to hold the widgets in the window
self.box = Gtk.VBox()
self.add(self.box)
# The text widget to output the log file to
self.text = Gtk.TextView()
self.text.set_editable(False)
self.box.pack_start(self.text, True, True, 0)
# A button to demonstrate non-blocking
self.button = Gtk.Button.new_with_label("Click")
self.box.pack_end(self.button, True, True, 0)
# Add a timer thread
self.timer = threading.Timer(0.1, self.read_log)
self.timer.start()
self.show_all()
def quit(self, *args):
"""Quit."""
# Stop the timer, in case it is still waiting when the window is closed
self.timer.cancel()
Gtk.main_quit()
def read_log(self):
"""Read the log."""
# Read the log
self.current_lines = []
p = Popen(["journalctl", "--user-unit=appToMonitor"], stdout=PIPE)
with p.stdout:
for line in iter(p.stdout.readline, b''):
self.current_lines.append(line.decode("utf-8"))
p.wait()
# Compare the log with the previous reading
for d in difflib.ndiff(self.last_lines, self.current_lines):
# Check if this is a new line, and if so, add it to the TextView
if d[0] == "+":
self.text.set_editable(True)
self.text.get_buffer().insert_at_cursor(d.replace("+ ", ""))
self.text.set_editable(False)
self.last_lines = self.current_lines
# Reset the timer
self.timer = threading.Timer(1, self.read_log)
self.timer.start()
if __name__ == "__main__":
app = App()
Gtk.main()
该按钮表明 threading 不会在等待时阻止执行；您可以随意单击它，即使程序正在读取日志！
我正在尝试使用 Python 3.9 和 PyGObject 编写一个小型应用程序 GUI (GTK) 应用程序，它从日志文件中读取数据并在 GUI 中显示数据 window。
在这种特殊情况下，我试图不断地从 systemd Journal 中读取数据，这可以通过例如journalctl -f 通过终端。
对于 CLI 应用程序，我看到可以使用类似这样的方法将日志文件的输出通过管道传输到 python 脚本的 STOUT：
p = Popen(["journalctl", "--user-unit=appToMonitor"], stdout=PIPE)
with p.stdout:
for line in iter(p.stdout.readline, b''):
print(line, end=''),
p.wait()
用于 GUI 应用程序的 OFC 我需要以某种方式连接到 GTK 主循环，因此上述解决方案不起作用。我做了一些谷歌搜索，据我所知，一个解决方案是使用 gobject.timeout_add() 定期轮询日志文件以进行更改。
这是从系统日志文件中获取数据的最佳方式吗？还是有其他解决方案可以用来避免轮询？如果 gobject.timeout_add() 是要走的路，我如何确保只添加自上次从日志文件读取后添加的日志行？
没有必要使用gobject.timeout_add()；您可以只使用专门用于线程的 threading 模块，并且比尝试使用 GTK 更简单。
使用 `threading` 调用函数
使用 threading 的好处在于，即使它不是 Gtk 模块组的一部分，它仍然不会阻塞 Gtk 的主循环。这是一个使用 threading:
import threading
def function():
print("Hello!")
# Create a timer
timer = threading.Timer(
1, # Interval (in seconds; can be a float) after which to call the function
function # Function to call after time interval
)
timer.start() # Start the timer
这会在 1 秒后调用 function() 一次，然后退出。
但是，在您的情况下，您想多次调用该函数，以反复检查日志的状态。为此，您可以在 function() 中重新创建计时器，然后再次 运行 ：
import threading
def function():
global timer
print("Hello!")
# Recreate and start the timer
timer = threading.Timer(1, function)
timer.start()
timer = threading.Timer(1, function)
timer.start()
现在要检查日志中是否添加了任何新行，程序需要读取日志，然后将最近读取的行与之前读取的行进行比较。
读取行，并使用 `difflib` 进行比较
首先，要完全比较这些行，需要将它们存储在两个列表中：一个包含最近读取的一组行，另一个包含前一组行。下面是读取日志输出、将行存储在列表中然后打印列表的代码示例：
from subprocess import PIPE, Popen
# The list storing the lines of the log
current_lines = []
# Read the log
p = Popen(["journalctl", "--user-unit=appToMonitor"], stdout=PIPE)
with p.stdout:
for line in iter(p.stdout.readline, b''):
current_lines.append(line.decode("utf-8")) # Add the lines of the log to the list
p.wait()
print(current_lines)
注意一定要用line.decode("uft-8")，因为p.stdout.readline的输出是以字节为单位的。使用您的程序使用的任何编码； utf-8只是一个普通的编码。
然后您可以使用 difflib 模块来比较两个列表。这是一个示例程序，它比较两个行列表，并打印第二个列表中而不是第一个列表中的任何行：
import difflib
# Two lists to compare
last_lines = ["Here is the first line\n"]
current_lines = ["Here is the first line\n", "and here is the second line"]
# Iterate through all the lines, and check for new ones
for line in difflib.ndiff(last_lines, current_lines):
# Print the line only if it was not in last_lines
if line[0] == "+": # difflib inserts a "+" for every addition
print("Line added:", line.replace("+ ", "")) # Remove the "+" from the line and print it
很好，但是我们如何将所有这些都放在一个程序中呢？
结局：将所有这些概念放入一个程序中，该程序每秒从日志中读取输出，并将任何新行添加到 window 中的文本小部件。这是一个简单的 Gtk 应用程序，它就是这样做的：
import difflib
import gi
import threading
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from subprocess import PIPE, Popen
class App(Gtk.Window):
"""The application window."""
def __init__(self):
Gtk.Window.__init__(self)
self.connect("delete-event", self.quit)
# The list of lines read from the log file
self.current_lines = []
# The list of previously read lines to compare to the current one
self.last_lines = []
# The box to hold the widgets in the window
self.box = Gtk.VBox()
self.add(self.box)
# The text widget to output the log file to
self.text = Gtk.TextView()
self.text.set_editable(False)
self.box.pack_start(self.text, True, True, 0)
# A button to demonstrate non-blocking
self.button = Gtk.Button.new_with_label("Click")
self.box.pack_end(self.button, True, True, 0)
# Add a timer thread
self.timer = threading.Timer(0.1, self.read_log)
self.timer.start()
self.show_all()
def quit(self, *args):
"""Quit."""
# Stop the timer, in case it is still waiting when the window is closed
self.timer.cancel()
Gtk.main_quit()
def read_log(self):
"""Read the log."""
# Read the log
self.current_lines = []
p = Popen(["journalctl", "--user-unit=appToMonitor"], stdout=PIPE)
with p.stdout:
for line in iter(p.stdout.readline, b''):
self.current_lines.append(line.decode("utf-8"))
p.wait()
# Compare the log with the previous reading
for d in difflib.ndiff(self.last_lines, self.current_lines):
# Check if this is a new line, and if so, add it to the TextView
if d[0] == "+":
self.text.set_editable(True)
self.text.get_buffer().insert_at_cursor(d.replace("+ ", ""))
self.text.set_editable(False)
self.last_lines = self.current_lines
# Reset the timer
self.timer = threading.Timer(1, self.read_log)
self.timer.start()
if __name__ == "__main__":
app = App()
Gtk.main()
该按钮表明 threading 不会在等待时阻止执行；您可以随意单击它，即使程序正在读取日志！