脚本执行器编写流程

什么是脚本执行器？

脚本执行器是芒果测试平台的扩展功能，允许用户编写自定义的监控脚本，实现定时检查业务数据、系统状态等功能。当监控到异常情况时，可以自动发送通知。

快速开始

1. 安装依赖

首先确保你的环境中已安装必要的依赖：

bash

pip install mangotools

2. 了解基类结构

脚本执行器基于 MonitorBase 基类开发，你只需要继承这个基类并实现 run() 方法即可。

核心概念

MonitorBase 基类

MonitorBase 提供了以下核心功能：

MySQL 数据库连接管理：自动维护数据库连接，支持查询和重连
日志系统：提供结构化的日志输出（INFO、WARNING、ERROR、DEBUG）
通知机制：统一的异常通知发送接口
生命周期管理：自动处理连接的创建和关闭

关键方法

python

class MonitorBase(ABC):
    def __init__(self, mysql_config=None, task_id=None):
        # 初始化数据库连接和任务ID

    @abstractmethod
    def run(self):
        # 你的监控逻辑在这里实现
        pass

    def start_monitoring(self):
        # 启动监控服务
        pass

    # 日志方法
    def log(self, message, level="INFO")
    def log_error(self, message)
    def log_warning(self, message)

    # 通知方法
    def send(self, send_text, msg=None)

    # 数据库方法
    def execute_query(self, sql, params=None)
    def execute_query_many(self, sql, params=None)

编写你的第一个监控脚本

步骤1：创建脚本文件

创建一个新的 Python 文件，比如 my_monitor.py

步骤2：导入和继承

python

# -*- coding: utf-8 -*-
from mangotools.monitoring import MonitorBase

class MyMonitor(MonitorBase):
    def __init__(self, mysql_config=None):
        super().__init__(mysql_config)
        # 初始化你的监控参数
        self.check_interval = 60  # 检查间隔（秒）

步骤3：实现监控逻辑

python

def run(self):
    """
    核心监控逻辑
    这里实现你的业务监控代码
    """
    self.log("监控服务启动，开始检查业务状态...")

    while True:
        try:
            # ===== 你的监控逻辑 =====

            # 示例1：检查数据库状态
            result = self.execute_query("SELECT COUNT(*) as count FROM your_table WHERE status = 'error'")
            if result and result['count'] > 0:
                self.send(f"发现 {result['count']} 条错误数据，请及时处理！")

            # 示例2：检查业务指标
            # 这里可以添加更多的检查逻辑...

            # ===== 异常处理 =====
        except Exception as e:
            self.log_error(f"监控过程中发生错误: {e}")
            self.send(f"监控脚本异常: {str(e)}")

        # 等待下次检查
        time.sleep(self.check_interval)

步骤4：配置数据库连接

python

# MySQL配置示例
MYSQL_CONFIG = {
    'host': 'localhost',      # 数据库地址
    'port': 3306,            # 数据库端口
    'user': 'your_username',  # 用户名
    'password': 'your_password', # 密码
    'database': 'your_database' # 数据库名
}

步骤5：启动脚本

python

if __name__ == '__main__':
    # 创建监控实例
    monitor = MyMonitor(MYSQL_CONFIG)

    # 启动监控服务
    monitor.start_monitoring()

完整示例：爬虫监控脚本

python

# -*- coding: utf-8 -*-
import random
import time
from mangotools.monitoring import MonitorBase

# 任务状态映射
status = {
    0: '待开始',
    1: '成功',
    2: '失败',
    3: '进行中',
}

# 任务类型映射
task_map_fields = {
    '601': '新闻爬虫-601',
    '602': '数据同步-602',
}

# MySQL配置（请填写你的数据库信息）
MYSQL_CONFIG = {
    'host': '',
    'port': 3306,
    'user': '',
    'password': '',
    'database': ''
}

class CrawlerMonitor(MonitorBase):
    def __init__(self, mysql_config=None):
        super().__init__(mysql_config)
        self.target_types = ['601', '701', '702', '703', '704', '705', '706', '3002']
        self.TIMEOUT_MINUTES = 3

    def run(self):
        """监控爬虫任务状态"""
        target_types_msg = [task_map_fields.get(type_code, f'未知类型-{type_code}')
                           for type_code in self.target_types]

        self.log_warning(
            f"爬虫监控服务启动，监控类型：{target_types_msg}，"
            f"检查任务是否在{self.TIMEOUT_MINUTES}分钟内未被处理"
        )

        while True:
            try:
                # 检查超时任务
                self.check_timeout_tasks()

                # 检查失败任务
                self.check_failed_tasks()

                # 其他业务检查逻辑...

            except Exception as e:
                self.log_error(f"监控循环异常: {e}")
                self.send(f"爬虫监控服务异常: {str(e)}")

            # 每分钟检查一次
            time.sleep(60)

    def check_timeout_tasks(self):
        """检查超时的任务"""
        timeout_seconds = self.TIMEOUT_MINUTES * 60

        sql = """
        SELECT id, task_type, status, create_time
        FROM crawler_tasks
        WHERE task_type IN %s
        AND status IN (0, 3)  # 待开始或进行中
        AND TIMESTAMPDIFF(SECOND, create_time, NOW()) > %s
        """

        tasks = self.execute_query_many(sql, (self.target_types, timeout_seconds))

        if tasks:
            for task in tasks:
                task_type_name = task_map_fields.get(str(task['task_type']), f'未知类型-{task["task_type"]}')
                self.send(
                    f"爬虫任务超时告警：任务ID {task['id']} ({task_type_name}) "
                    f"已超过{self.TIMEOUT_MINUTES}分钟未完成"
                )

    def check_failed_tasks(self):
        """检查失败的任务"""
        sql = """
        SELECT COUNT(*) as failed_count
        FROM crawler_tasks
        WHERE task_type IN %s
        AND status = 2  # 失败状态
        AND update_time >= DATE_SUB(NOW(), INTERVAL 1 HOUR)  # 最近1小时
        """

        result = self.execute_query(sql, (self.target_types,))
        if result and result['failed_count'] > 5:  # 失败任务超过5个
            self.send(f"爬虫任务失败过多：最近1小时内失败任务数 {result['failed_count']}")

if __name__ == '__main__':
    monitor = CrawlerMonitor(MYSQL_CONFIG)
    monitor.start_monitoring()

集成到芒果测试平台

脚本执行器只能通过芒果测试平台进行管理和运行，请按照以下步骤操作：

1. 进入脚本运行器模块

在左侧导航栏中找到并点击 脚本运行器 → 脚本运行器，进入脚本管理页面。

2. 创建新脚本任务

点击页面右上角的新增按钮，打开新增脚本对话框。
图片走丢了

3. 填写脚本配置信息

在弹出的对话框中，按照以下要求填写信息：

项目/产品：选择脚本所属的项目和产品（必填）
任务名称：输入任务的名称，便于识别和管理（必填）
任务描述：填写任务的详细描述信息（可选）
是否通知：开启后，脚本执行过程中会发送通知
通知组：如果开启了通知，选择相应的通知组
脚本内容：将编写好的Python脚本代码粘贴到文本框中（必填）

填写完成后，点击提交按钮保存配置。

脚本启动后会在后台持续运行，您可以随时查看运行状态和日志信息。

注意事项

重要提醒

数据库配置：确保MySQL配置正确，否则无法连接数据库
异常处理：在监控逻辑中要做好异常捕获，避免脚本崩溃
检查频率：根据业务需求合理设置检查间隔，避免过于频繁的数据库查询
通知频率：避免在循环中频繁发送通知，可以设置告警间隔
日志记录：使用提供的日志方法记录重要信息，便于排查问题

常见问题

Q: 脚本启动后没有反应？

A: 检查数据库连接配置是否正确，请先手动在本地运行脚本，检查是否正常。

Q: 如何修改监控逻辑？

A: 编辑 run() 方法中的代码逻辑，重启脚本即可生效。

Q: 支持哪些数据库操作？

A: 支持基本的查询操作，包括单条查询 execute_query() 和批量查询 execute_query_many()。

进阶用法

自定义代码

python

def __init__(self, mysql_config=None):
    super().__init__(mysql_config)
    # 自定义初始化逻辑
    self.load_config_from_file()
    self.setup_additional_connections()

现在你可以基于这个文档开始编写自己的监控脚本了！如果遇到问题，请查看日志输出或连续作者寻求帮助。

脚本执行器编写流程 ​

快速开始 ​

1. 安装依赖 ​

2. 了解基类结构 ​

核心概念 ​

MonitorBase 基类 ​

关键方法 ​

编写你的第一个监控脚本 ​

步骤1：创建脚本文件 ​

步骤2：导入和继承 ​

步骤3：实现监控逻辑 ​

步骤4：配置数据库连接 ​

步骤5：启动脚本 ​

完整示例：爬虫监控脚本 ​

集成到芒果测试平台 ​

1. 进入脚本运行器模块 ​

2. 创建新脚本任务 ​

3. 填写脚本配置信息 ​

注意事项 ​

常见问题 ​

Q: 脚本启动后没有反应？ ​

Q: 如何修改监控逻辑？ ​

Q: 支持哪些数据库操作？ ​

进阶用法 ​

自定义代码 ​

脚本执行器编写流程

快速开始

1. 安装依赖

2. 了解基类结构

核心概念

MonitorBase 基类

关键方法

编写你的第一个监控脚本

步骤1：创建脚本文件

步骤2：导入和继承

步骤3：实现监控逻辑

步骤4：配置数据库连接

步骤5：启动脚本

完整示例：爬虫监控脚本

集成到芒果测试平台

1. 进入脚本运行器模块

2. 创建新脚本任务

3. 填写脚本配置信息

注意事项

常见问题

Q: 脚本启动后没有反应？

Q: 如何修改监控逻辑？

Q: 支持哪些数据库操作？

进阶用法

自定义代码