MacOS下Bark通知

本笔记分享一个在 macOS 下自动读取指定目录最新日志文件，并实时检索关键词、发送通知的 Python 脚本。适用于需要自动化监控日志并即时处理异常或关键信息的场景。

1. 主要功能

• 自动查找最新日志文件在指定目录下，自动定位最新生成的日志文件（如 better-genshin-impact_xxx.log）。
• 实时监控日志内容使用 tail -f 持续读取日志新增内容，实现实时监控。
• 检索关键词并通知支持正则表达式匹配关键词（如 error、结束），一旦检测到，立即通过 HTTP 接口发送通知。
• 详细日志记录支持将所有操作和事件写入本地日志文件，并在控制台输出，便于追踪和调试。

2. 实现代码

import logging
import subprocess
import re
import os
import time
from datetime import datetime

def setup_logger():
    """
    此函数用于设置日志记录器
    :return: 配置好的日志记录器对象
    """
    # 获取当前时间，格式为 yyyy_mm_dd_hh_mm
    timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M")
    log_filename = f'/Users/cao/Desktop/webhook/webhook_{timestamp}.log'  # 修改文件路径
    logger = logging.getLogger('log_monitor')
    logger.setLevel(logging.DEBUG)
    # 创建文件处理器，将日志输出到用户目录下的动态命名的日志文件
    file_handler = logging.FileHandler(log_filename)
    file_handler.setLevel(logging.DEBUG)
    # 创建控制台处理器，将日志输出到控制台
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    # 定义日志格式
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)
    # 将处理器添加到日志记录器
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger


def main():
    logger = setup_logger()
    log_dir = "/Users/cao/Desktop/"  # 假设日志文件在根目录下
    latest_log_file = find_latest_log_file(log_dir)
    if latest_log_file is None:
        logger.error("未找到符合模式的日志文件")
        return
    try:
        # 使用 subprocess.Popen 调用 tail -f 命令
        tail_process = subprocess.Popen(['tail', '-f', latest_log_file], stdout=subprocess.PIPE, universal_newlines=True)
        for line in tail_process.stdout:
            logger.debug(f"读取到新行: {line.strip()}")  # 记录读取到新行
            if re.search(r'error|结束', line, re.IGNORECASE):  # 检查是否有特定关键词
                message = line.strip()
                logger.info(f"检测到关键词，消息内容: {message}")  # 记录检测到关键词的情况
                # 使用 curl 发送通知消息，需要确保系统中已经安装 curl
                curl_command = f'http://bjv.xxxxxx.com:xxx/xxxx/{message}'
                try:
                    subprocess.run(["curl", curl_command], check=True)
                    logger.info(f"消息发送成功: {message}")  # 记录消息发送成功
                except subprocess.CalledProcessError as e:
                    logger.error(f"发送消息时出错: {e}")  # 记录发送消息出错
    except Exception as e:
        logger.error(f"打开文件时出现异常: {e}")  # 记录打开文件出错

def find_latest_log_file(log_dir):
    """
    此函数用于查找最新的日志文件
    :param log_dir: 日志文件所在的目录
    :return: 最新日志文件的路径
    """
    log_files = []
    for file in os.listdir(log_dir):
        if file.startswith("better-genshin-impact") and file.endswith(".log"):
            log_files.append(os.path.join(log_dir, file))
    if not log_files:
        return None
    log_files.sort(key=lambda x: os.path.getmtime(x), reverse=True)
    return log_files[0]


if __name__ == "__main__":
    main()

3. 使用说明

1. 环境准备

• 需安装 Python 3
• 系统中需已安装 curl命令

2. 参数配置

• 修改 log_dir 为实际日志文件存放目录
• 如需变更关键词，请调整 re.search(r'error|结束', line, re.IGNORECASE) 部分

3. 运行方式

• 直接运行脚本即可自动监控日志并推送通知

4. 适用场景

• 日志实时监控和异常自动告警
• 自动化运维、定制化消息推送
• 需对关键事件进行及时响应的开发/测试环境

‍