OnmyojiAutoScript 通知推送功能优化实践

概述

OnmyojiAutoScript（阴阳师自动化脚本）的通知推送功能是其核心特性之一，能够在脚本运行过程中及时向用户推送关键信息，如任务完成状态、错误报警、异常情况等。本文将从技术实现角度深入分析通知推送模块的架构设计、优化策略以及最佳实践。

通知推送系统架构

核心组件

classDiagram
    class Notifier {
        -config_name: str
        -enable: bool
        -config: dict
        -provider_name: str
        -notifier: Provider
        -required: list[str]
        +__init__(config: str, enable: bool)
        +push(**kwargs) bool
    }
    
    class Config {
        -model: ConfigModel
        -config_name: str
        +notifier: Notifier
        +get_next() Function
        +task_call(task: str) bool
    }
    
    class Script {
        -config: Config
        -device: Device
        +run(command: str) bool
        +loop()
    }
    
    Script --> Config : 使用
    Config --> Notifier : 包含
    Notifier --> onepush.Provider : 依赖

配置结构

通知推送系统采用YAML配置格式，支持多种推送提供商：

provider: "custom"  # 推送提供商
method: "post"      # 请求方法
url: "https://your-webhook-url"  # 推送地址
datatype: "json"    # 数据格式
data:               # 自定义数据
  title: "推送标题"
  content: "推送内容"

错误处理与通知触发机制

异常类型与通知策略

OnmyojiAutoScript定义了多种异常类型，每种异常都会触发相应的通知：

异常类型	触发条件	通知内容	处理方式
GameStuckError	游戏卡死或点击过多	GameStuckError or GameTooManyClickError	重启游戏
GamePageUnknownError	游戏页面未知	GamePageUnknownError	检查服务器状态
ScriptError	脚本开发错误	ScriptError	退出脚本
RequestHumanTakeover	需要人工干预	RequestHumanTakeover	退出脚本
Exception	未捕获异常	Exception occured	退出脚本

通知推送实现代码

def push(self, **kwargs) -> bool:
    if not self.enable:
        return False
    
    # 更新配置
    kwargs["title"] = f"{self.config_name} {kwargs['title']}"
    self.config.update(kwargs)
    
    # 参数预检查
    for key in self.required:
        if key not in self.config:
            logger.warning(f"Notifier {self.notifier} require param '{key}' but not provided")
    
    # 自定义提供商特殊处理
    if isinstance(self.notifier, Custom):
        if "method" not in self.config or self.config["method"] == "post":
            self.config["datatype"] = "json"
        if not ("data" in self.config or isinstance(self.config["data"], dict)):
            self.config["data"] = {}
        if "title" in kwargs:
            self.config["data"]["title"] = kwargs["title"]
        if "content" in kwargs:
            self.config["data"]["content"] = kwargs["content"]
    
    # GoCqHttp特殊处理
    if self.provider_name.lower() == "gocqhttp":
        access_token = self.config.get("access_token")
        if access_token:
            self.config["token"] = access_token
    
    try:
        resp = self.notifier.notify(**self.config)
        # 响应状态检查
        if isinstance(resp, Response):
            if resp.status_code != 200:
                logger.warning("Push notify failed!")
                logger.warning(f"HTTP Code:{resp.status_code}")
                return False
            else:
                if self.provider_name.lower() == "gocqhttp":
                    return_data: dict = resp.json()
                    if return_data["status"] == "failed":
                        logger.warning("Push notify failed!")
                        logger.warning(f"Return message:{return_data['wording']}")
                        return False
    except SMTPResponseException:
        logger.warning("Appear SMTPResponseException")
        pass
    except OnePushException:
        logger.exception("Push notify failed")
        return False
    except Exception as e:
        logger.exception(e)
        return False
    
    logger.info("Push notify success")
    return True

性能优化策略

1. 懒加载机制

通知推送器采用懒加载设计，只有在真正需要推送时才初始化：

@cached_property
def notifier(self):
    notifier = Notifier(self.model.script.error.notify_config, 
                       enable=self.model.script.error.notify_enable)
    notifier.config_name = self.config_name.upper()
    logger.info(f'Notifier: {notifier.config_name}')
    return notifier

2. 错误重试机制

系统实现了智能的错误重试机制：

# 检查失败次数
failed = self.failure_record[task] if task in self.failure_record else 0
failed = 0 if success else failed + 1
self.failure_record[task] = failed

if failed >= 3:
    logger.critical(f"Task `{task}` failed 3 or more times.")
    logger.critical("Possible reason #1: You haven't used it correctly.")
    logger.critical("Possible reason #2: There is a problem with this task.")
    logger.critical('Request human takeover')
    exit(1)

3. 异步处理优化

对于高频率的通知推送，建议采用异步处理：

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncNotifier(Notifier):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.executor = ThreadPoolExecutor(max_workers=3)
    
    async def async_push(self, **kwargs):
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(self.executor, self.push, **kwargs)

配置最佳实践

1. 多提供商配置示例

# Discord Webhook 配置
provider: "custom"
method: "post"
url: "https://discord.com/api/webhooks/your-webhook"
datatype: "json"
data:
  username: "OnmyojiBot"
  avatar_url: "https://example.com/avatar.png"
  embeds:
    - title: "{{title}}"
      description: "{{content}}"
      color: 5814783

# 即时通讯 Bot 配置  
provider: "im_bot"
token: "your-bot-token"
chat_id: "your-chat-id"
parse_mode: "HTML"

# Server酱配置
provider: "serverchan"
sckey: "your-sckey"

2. 消息模板优化

# 自定义消息模板函数
def format_notification(task_name, error_type, config_name):
    """格式化通知消息"""
    templates = {
        "GameStuckError": f"🚨 游戏卡死报警 - {task_name}\n"
                         f"配置: {config_name}\n"
                         f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
                         f"建议: 检查游戏网络连接或重启游戏",
        
        "ScriptError": f"🐛 脚本错误 - {task_name}\n"
                      f"配置: {config_name}\n"
                      f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
                      f"建议: 联系开发者或查看错误日志",
    }
    return templates.get(error_type, f"{error_type} - {task_name}")

监控与日志分析

1. 推送成功率监控

class NotificationMonitor:
    def __init__(self):
        self.success_count = 0
        self.failure_count = 0
        self.last_notification = None
    
    def record_success(self):
        self.success_count += 1
        self.last_notification = datetime.now()
    
    def record_failure(self):
        self.failure_count += 1
    
    @property
    def success_rate(self):
        total = self.success_count + self.failure_count
        return self.success_count / total if total > 0 else 0
    
    def get_stats(self):
        return {
            "success_count": self.success_count,
            "failure_count": self.failure_count,
            "success_rate": f"{self.success_rate:.2%}",
            "last_notification": self.last_notification
        }

2. 推送延迟分析

import time
from dataclasses import dataclass

@dataclass
class PushMetrics:
    start_time: float
    end_time: float = 0
    success: bool = False
    
    @property
    def duration(self):
        return self.end_time - self.start_time if self.end_time else 0

class TimedNotifier(Notifier):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.metrics = []
    
    def push(self, **kwargs):
        metric = PushMetrics(start_time=time.time())
        try:
            result = super().push(**kwargs)
            metric.end_time = time.time()
            metric.success = result
            self.metrics.append(metric)
            return result
        except Exception as e:
            metric.end_time = time.time()
            self.metrics.append(metric)
            raise e
    
    def get_performance_stats(self):
        if not self.metrics:
            return {}
        
        durations = [m.duration for m in self.metrics if m.success]
        return {
            "total_pushes": len(self.metrics),
            "successful_pushes": sum(1 for m in self.metrics if m.success),
            "avg_duration": sum(durations) / len(durations) if durations else 0,
            "max_duration": max(durations) if durations else 0,
            "min_duration": min(durations) if durations else 0
        }

安全性与可靠性保障

1. 敏感信息处理

from module.handler.sensitive_info import handle_sensitive_image, handle_sensitive_logs

def save_error_log(self):
    """保存错误日志和截图，处理敏感信息"""
    if self.config.script.error.save_error:
        folder = f'./log/error/{int(time.time() * 1000)}'
        os.makedirs(folder, exist_ok=True)
        
        # 处理敏感截图
        for data in self.device.screenshot_deque:
            image = handle_sensitive_image(data['image'])
            save_image(image, f'{folder}/{data["time"].strftime("%Y-%m-%d_%H-%M-%S-%f")}.png')
        
        # 处理敏感日志
        with open(logger.log_file, 'r', encoding='utf-8') as f:
            lines = handle_sensitive_logs(f.readlines())
        
        with open(f'{folder}/log.txt', 'w', encoding='utf-8') as f:
            f.writelines(lines)

2. 配置验证机制

def validate_notification_config(config_str):
    """验证通知配置有效性"""
    try:
        config = yaml.safe_load(config_str)
        if not config:
            return False, "配置为空"
        
        provider = config.get("provider")
        if not provider:
            return False, "未指定推送提供商"
        
        # 获取提供商必填参数
        notifier = get_notifier(provider)
        required_params = notifier.params["required"]
        
        missing_params = []
        for param in required_params:
            if param not in config:
                missing_params.append(param)
        
        if missing_params:
            return False, f"缺少必填参数: {', '.join(missing_params)}"
        
        return True, "配置有效"
    except Exception as e:
        return False, f"配置解析失败: {str(e)}"

总结与展望

OnmyojiAutoScript的通知推送系统通过精心的架构设计和优化策略，实现了高效、可靠的消息推送功能。系统具有以下特点：

1. 多提供商支持：集成多种推送服务，满足不同用户需求
2. 智能错误处理：根据异常类型提供针对性的通知和处理方案
3. 性能优化：懒加载、异步处理等机制确保系统高效运行
4. 安全可靠：敏感信息处理和配置验证保障用户数据安全

未来可进一步优化的方向包括：

• 支持更多推送提供商和消息格式
• 实现消息优先级和去重机制
• 添加推送历史查询和统计分析功能
• 支持消息模板自定义和国际化

通过持续的优化和改进，OnmyojiAutoScript的通知推送功能将为用户提供更加完善和便捷的自动化脚本体验。