校园作业：编写Python脚本获取systemd单元指定时段日志信息

嘿，这个校园作业的需求其实挺实用的——用Python提取systemd journal里指定单元的日志对吧？我给你整理了一套入门级的实现方案，一步步来很容易上手：

1. 先搞定依赖 #

首先得装一个官方的Python绑定库，用来直接和systemd journal交互，叫systemd-python。用pip安装就行：

pip install systemd-python

注意：部分Linux发行版（比如Debian/Ubuntu）可能需要先装系统级的开发依赖，不然pip编译会失败：

sudo apt-get install libsystemd-dev

2. 核心代码实现 #

我写了一个示例脚本，专门针对你的需求——获取sshd.service最近24小时和30分钟的日志，你可以直接复用或者修改：

from systemd import journal
from datetime import datetime, timedelta

def fetch_unit_logs(unit_name, time_range_type="hours", time_value=24):
    # 初始化journal读取器
    journal_reader = journal.Reader()
    # 过滤出指定单元的日志
    journal_reader.add_match(_SYSTEMD_UNIT=unit_name)
    # 跳到日志末尾，从最新的开始往前读
    journal_reader.seek_tail()
    journal_reader.get_previous()

    # 计算时间范围的起始点
    end_time = datetime.now()
    if time_range_type == "minutes":
        start_time = end_time - timedelta(minutes=time_value)
    else:
        start_time = end_time - timedelta(hours=time_value)
    
    # 转换成systemd journal用的微秒级时间戳
    start_timestamp = int(start_time.timestamp() * 1000000)

    collected_logs = []
    # 遍历日志直到超出设定的时间范围
    while True:
        try:
            log_entry = journal_reader.get_previous()
            # 检查当前日志是否在时间范围内
            if log_entry["__REALTIME_TIMESTAMP"] < start_timestamp:
                break
            # 提取关键信息整理成字典，方便后续处理
            formatted_log = {
                "时间": datetime.fromtimestamp(log_entry["__REALTIME_TIMESTAMP"] / 1000000),
                "消息": log_entry.get("MESSAGE", "无日志消息"),
                "优先级": log_entry.get("PRIORITY", "未知")
            }
            collected_logs.append(formatted_log)
        except StopIteration:
            # 没有更多日志了，退出循环
            break
    
    return collected_logs

# 测试脚本
if __name__ == "__main__":
    # 获取最近24小时的sshd日志
    print("=== 最近24小时的sshd服务日志 ===")
    sshd_logs_24h = fetch_unit_logs("sshd.service")
    for log in sshd_logs_24h:
        print(f"{log['时间']}: {log['消息']}")
    
    # 获取最近30分钟的sshd日志
    print("\n=== 最近30分钟的sshd服务日志 ===")
    sshd_logs_30m = fetch_unit_logs("sshd.service", time_range_type="minutes", time_value=30)
    for log in sshd_logs_30m:
        print(f"{log['时间']}: {log['消息']}")

3. 关键细节解释 #

• add_match方法：这是过滤指定单元的核心，通过_SYSTEMD_UNIT这个字段精准定位到sshd.service的日志，不会拿到其他服务的内容。
• 时间处理：systemd journal用__REALTIME_TIMESTAMP存储微秒级的时间戳，我们把它转换成Python的datetime对象，看起来更直观。
• 日志读取顺序：用seek_tail()跳到日志末尾再往前读，这样最先拿到的是最新的日志，符合你要“近期消息”的需求。

4. 权限问题要注意 #

• 默认情况下，读取systemd journal需要root权限，所以运行脚本的时候记得加sudo：sudo python3 your_script.py。
• 如果不想每次都用root，可以把当前用户加入systemd-journal组，之后重新登录就有权限了：

  sudo usermod -aG systemd-journal $USER
  

内容的提问来源于stack exchange，提问作者Robin K