Windows下Bark通知

本笔记记录了如何在 Windows 环境下，自动读取指定目录下最新的日志文件，实时检索关键词并通过 curl 发送通知。适合日常运维和自动化监控场景。

1. 前置条件

• 已安装 Python
• 已安装 curl 工具

2. 功能说明

1. 自动查找指定目录下最新的日志文件（支持动态命名）。
2. 实时监控日志内容，检索关键词（如 error 或 结束）。
3. 匹配到关键词后，自动通过 curl 发送通知到指定接口。
4. 日志操作过程均有详细记录，便于溯源和排查。

3. 核心代码示例

import logging
import os
import re
import subprocess
import time
import urllib.parse
from datetime import datetime
from tailer import follow

def setup_logger():
    """
    此函数用于设置日志记录器
    :return: 配置好的日志记录器对象
    """
    # 获取当前时间，格式为 yyyy_mm_dd_hh_mm
    timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M")
    log_filename = f'C:\\games\\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 = "C:\\games\\BetterGI\\log"  # 假设日志文件在根目录下
    latest_log_file = find_latest_log_file(log_dir)
    if latest_log_file is None:
        logger.error("未找到符合模式的日志文件")
        return
    try:
        # 使用 tailer 库中的 follow 函数
        with open(latest_log_file, 'r', encoding='utf-8') as log_file:
            for line in follow(log_file):
                logger.debug(f"读取到新行: {line.strip()}")  # 记录读取到新行
                if re.search(r'error|结束', line, re.IGNORECASE):  # 检查是否有特定关键词
                    message = line.strip()
                    encoded_message = urllib.parse.quote(message)
                    curl_command = ['curl', f'http://xxxx.com:xx/xxxx/{encoded_message}']
                    print(f"正在发送的 curl 命令: {curl_command}")# 打印 curl 命令
                    try:
                        subprocess.run(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()

4. 关键点说明

• 日志文件自动定位：根据文件名和修改时间，自动找到最新日志。
• 关键词监控：可自定义监控的关键词，支持正则表达式。
• 通知发送：通过 curl 工具，将消息推送到指定接口，支持 URL 编码。
• 异常处理与日志记录：所有操作均有日志记录，便于排查故障。

5. 使用建议

• 可根据实际需求调整监控目录和关键词。
• 建议将脚本设置为后台运行，确保实时监控。
• 若需扩展通知方式，可增加不同的接口或自定义通知逻辑。

6. 常见问题

• 日志文件未找到：请确保目录路径正确且日志文件命名符合要求。
• curl 未安装：需提前安装 curl 工具，或用其他方式替换。
• Python 环境问题：建议使用 Python 3.7 及以上版本。

7. 参考资料

• Python 官方文档
• tailer 库文档
• curl 官方文档