使用restapi来控制mikrotik的功能

我写了个脚本来监控家里电视机的在线时长，如果时间达到我设定的阈值，则出发告警。

我现在想在告警的基础上，可以调用routeos的接口来启用我提前配置好的限速策略。最终实现，比如我家娃娃看电视半小时了，则自动给电视机断网。

curl -k -u admin:password http://192.168.1.1/rest/queue/simple

curl -k -u admin:password -X PATCH http://192.168.1.1/rest/queue/simple/*1 --data '{"disabled":"false"}' -H "content-type: application/json"

curl -k -u admin:password -X PATCH http://192.168.1.1/rest/queue/simple/*1 --data '{"disabled":"false"}' -H "content-type: application/json"

我们可以通过上面这种接口来实现我们的需求，更多api能力参考：

https://help.mikrotik.com/docs/display/ROS/REST+API

我的完整监控脚本

import subprocess
import time
import requests
import json
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedelta
from requests.auth import HTTPBasicAuth
import logging
import os

# 配置日志记录
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("TVMonitor.log"),
        logging.StreamHandler()
    ]
)

# 电视的 IP 地址
IP_ADDRESS = "192.168.1.100"

# 发送消息的 API URL
API_URL = "https://wechat.api/send_message"

# Mikrotik REST API 配置
MIKROTIK_IP = "192.168.1..1"
MIKROTIK_PORT = 80
MIKROTIK_USERNAME = os.getenv('MIKROTIK_USERNAME', 'admin')
MIKROTIK_PASSWORD = os.getenv('MIKROTIK_PASSWORD', 'admin')
QUEUE_NAME = "TVLimit"  # 需要操作的队列名称
WX_ID="weixin_id"

# 阈值时间配置（单位：秒）
QUEUE_ENABLE_DELAY = 1800  # 30 分钟
QUEUE_DURATION = 3 * 3600  # 3 小时

# 创建调度器
scheduler = BlockingScheduler()

def ping_ip(ip):
    """Ping 指定 IP 地址，返回是否通畅。"""
    try:
        response = subprocess.run(['ping', '-c', '1', ip], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        if response.returncode == 0:
            logging.info(f"成功 ping 通 {ip}")
            return True
        else:
            logging.warning(f"无法 ping 通 {ip}")
            return False
    except Exception as e:
        logging.error(f"Ping {ip} 时出错: {e}")
        return False

def call_api(url, data):
    """调用发送消息的 API。"""
    try:
        response = requests.post(url, json=data, timeout=10)
        if response.status_code == 200:
            logging.info(f"API 响应: {response.status_code}, {response.text}")
        else:
            logging.warning(f"API 响应: {response.status_code}, {response.text}")
    except Exception as e:
        logging.error(f"调用 API 时出错: {e}")

def format_duration(seconds):
    """格式化持续时间为 'X小时Y分钟'。"""
    hours = seconds // 3600
    minutes = (seconds % 3600) // 60
    return f"{hours}小时{minutes}分钟"

def get_queue_id():
    """通过队列名称获取队列的 ID。"""
    url = f"http://{MIKROTIK_IP}:{MIKROTIK_PORT}/rest/queue/simple"
    try:
        response = requests.get(url, auth=HTTPBasicAuth(MIKROTIK_USERNAME, MIKROTIK_PASSWORD), verify=False, timeout=10)
        response.raise_for_status()
        queues = response.json()
        for queue in queues:
            if queue.get('name') == QUEUE_NAME:
                queue_id = queue.get('.id') or queue.get('id')
                logging.info(f"找到队列 {QUEUE_NAME} 的 ID: {queue_id}")
                return queue_id
        logging.error(f"未找到名为 {QUEUE_NAME} 的队列。")
        return None
    except requests.exceptions.RequestException as e:
        logging.error(f"获取队列 ID 时出错: {e}")
        return None

def enable_queue():
    """调用 Mikrotik REST API 启用限速队列。"""
    queue_id = get_queue_id()
    if not queue_id:
        return
    url = f"http://{MIKROTIK_IP}:{MIKROTIK_PORT}/rest/queue/simple/{queue_id}"
    data = {"disabled": False}
    try:
        response = requests.patch(
            url,
            auth=HTTPBasicAuth(MIKROTIK_USERNAME, MIKROTIK_PASSWORD),
            headers={"Content-Type": "application/json"},
            data=json.dumps(data),
            verify=False,
            timeout=10
        )
        response.raise_for_status()
        logging.info(f"已启用队列 {QUEUE_NAME}。响应: {response.status_code}")
    except requests.exceptions.RequestException as e:
        logging.error(f"启用队列时出错: {e}")

def disable_queue_func():
    """调用 Mikrotik REST API 停用限速队列。"""
    queue_id = get_queue_id()
    if not queue_id:
        return
    url = f"http://{MIKROTIK_IP}:{MIKROTIK_PORT}/rest/queue/simple/{queue_id}"
    data = {"disabled": True}
    try:
        response = requests.patch(
            url,
            auth=HTTPBasicAuth(MIKROTIK_USERNAME, MIKROTIK_PASSWORD),
            headers={"Content-Type": "application/json"},
            data=json.dumps(data),
            verify=False,
            timeout=10
        )
        response.raise_for_status()
        logging.info(f"已停用队列 {QUEUE_NAME}。响应: {response.status_code}")
    except requests.exceptions.RequestException as e:
        logging.error(f"停用队列时出错: {e}")

class TVChecker:
    def __init__(self, scheduler):
        self.ip_alive = False
        self.start_time = None
        self.queue_enabled = False
        self.scheduler = scheduler

    def check(self):
        """检查电视的状态并执行相应的操作。"""
        is_up = ping_ip(IP_ADDRESS)
        current_time = time.time()

        if is_up and not self.ip_alive:
            # 电视刚开机
            self.ip_alive = True
            self.start_time = current_time
            call_api(API_URL, {"userid": WX_ID, "message": "电视开机了"})
            logging.info("电视开机，发送通知。")

        elif is_up and self.ip_alive:
            # 电视持续开机，检查是否超过阈值时间
            if not self.queue_enabled and (current_time - self.start_time) > QUEUE_ENABLE_DELAY:
                message = "电视已开机超过30分钟，启用限速队列。"
                logging.info(message)
                call_api(API_URL, {"userid": WX_ID, "message": message})
                enable_queue()
                self.queue_enabled = True
                # 调度3小时后停用队列
                run_time = datetime.now() + timedelta(seconds=QUEUE_DURATION)
                self.scheduler.add_job(disable_queue_func, 'date', run_date=run_time, id='disable_queue_job')
                message = f"已安排在 {run_time} 停用限速队列。"
                logging.info(message)
                call_api(API_URL, {"userid": WX_ID, "message": message})

        elif not is_up and self.ip_alive:
            # 电视关机
            self.ip_alive = False
            if self.start_time:
                alive_duration = format_duration(int(current_time - self.start_time))
            else:
                alive_duration = "未知时长"
            message = f"电视关机了，开机时长{alive_duration}"
            call_api(API_URL, {"userid": WX_ID, "message": message})
            logging.info(f"电视关机，发送通知。开机时长: {alive_duration}")
            # 如果限速队列已启用，立即停用
            if self.queue_enabled:
                logging.info("电视关机时，限速队列已启用，立即停用。")
                # 移除已调度的停用任务，以防重复执行
                try:
                    self.scheduler.remove_job('disable_queue_job')
                    logging.info("已移除停用队列的调度任务。")
                except Exception as e:
                    logging.warning(f"移除停用任务时出错: {e}")
                disable_queue_func()
            self.start_time = None

def main():
    """主函数，设置调度任务并启动调度器。"""
    checker = TVChecker(scheduler)
    # 每60秒检查一次电视状态
    scheduler.add_job(checker.check, 'interval', seconds=60, id='tv_check_job')
    try:
        logging.info("启动调度器，开始监控电视状态...")
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        logging.info("调度器已停止。")

if __name__ == "__main__":
    # 测试启用和停用队列
    enable_queue()
    time.sleep(10)
    disable_queue_func()
    # 主函数
    main()