safesight/edge/scripts/mock_alarm_server.py

#!/usr/bin/env python3
"""
模拟告警服务器
用于测试 RK3588 Media Server 的 alarm 节点 external_api 功能

提供两个接口：
1. POST /api/getToken - 获取访问令牌
2. POST /api/putMessage - 接收告警信息

使用方法：
    python3 mock_alarm_server.py
    
默认监听：0.0.0.0:8080
"""

from flask import Flask, request, jsonify
from datetime import datetime
import json
import sys

app = Flask(__name__)

# 模拟 token 存储 - 使用纯ASCII字符，避免转义问题
mock_token = "mocktoken123456789abcdef123456789"
token_expire_time = 30 * 60  # 30分钟过期时间（秒）

# 统计信息
stats = {
    "token_requests": 0,
    "alarm_requests": 0,
    "last_alarm": None
}


@app.route('/api/getToken', methods=['POST'])
def get_token():
    """
    模拟获取 token 接口
    返回格式与 d8_1.py 中一致
    """
    stats["token_requests"] += 1
    
    # 打印请求信息
    print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 收到 Token 请求")
    
    # 简化的响应，纯ASCII，避免转义问题
    response = {
        "errorDesc": "",
        "message": "",
        "responseBody": {
            "expireTime": str(token_expire_time),
            "token": mock_token,
            "refreshToken": mock_token + "_refresh"
        },
        "retCode": "200"
    }
    
    # 手动序列化并打印原始JSON
    raw_json = json.dumps(response, ensure_ascii=True)
    print(f"  返回 JSON: {raw_json}")
    
    # 返回纯文本JSON，避免Flask jsonify的额外处理
    return raw_json, 200, {'Content-Type': 'application/json'}


@app.route('/api/putMessage', methods=['POST'])
def put_message():
    """
    模拟接收告警信息接口
    接收格式与 d8_1.py 中 send_post_request 一致
    """
    stats["alarm_requests"] += 1
    
    # 获取请求头中的 token
    token = request.headers.get('X-Access-Token', 'None')
    
    # 解析请求体
    try:
        data = request.json if request.is_json else json.loads(request.data)
    except:
        data = {}
    
    # 打印告警信息
    raw_body = json.dumps(data, ensure_ascii=False, indent=2)
    print(f"\n[%s] 收到告警信息" % datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    print(f"  %s" % raw_body, flush=True)
    
    # 更新统计
    stats["last_alarm"] = {
        "time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        "content": data.get('alarmContent', 'N/A'),
        "channel": data.get('channelNo', 'N/A')
    }
    
    # 返回成功响应
    response = {
        "responseBody": "1",
        "message": None,
        "retCode": "200",
        "errorDesc": None
    }
    
    print(f"  响应: 告警接收成功")
    return jsonify(response)


@app.route('/stats', methods=['GET'])
def get_stats():
    """查看统计信息"""
    return jsonify({
        "token_requests": stats["token_requests"],
        "alarm_requests": stats["alarm_requests"],
        "last_alarm": stats["last_alarm"]
    })


@app.route('/', methods=['GET'])
def index():
    """首页说明"""
    return """
    <h1>RK3588 模拟告警服务器</h1>
    <p>可用接口:</p>
    <ul>
        <li><b>POST /api/getToken</b> - 获取访问令牌</li>
        <li><b>POST /api/putMessage</b> - 接收告警信息</li>
        <li><b>GET /stats</b> - 查看统计信息</li>
    </ul>
    <p>当前统计:</p>
    <ul>
        <li>Token 请求次数: {token_requests}</li>
        <li>告警请求次数: {alarm_requests}</li>
    </ul>
    """.format(**stats)


def main():
    host = "0.0.0.0"
    port = 8080
    
    print("=" * 60)
    print("RK3588 模拟告警服务器")
    print("=" * 60)
    print(f"启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"监听地址: http://{host}:{port}")
    print("")
    print("接口列表:")
    print(f"  1. POST http://{host}:{port}/api/getToken")
    print(f"  2. POST http://{host}:{port}/api/putMessage")
    print(f"  3. GET  http://{host}:{port}/stats")
    print("")
    print("按 Ctrl+C 停止服务")
    print("=" * 60)
    
    try:
        app.run(host=host, port=port, debug=False, threaded=True)
    except KeyboardInterrupt:
        print("\n\n服务已停止")
        print(f"总计 Token 请求: {stats['token_requests']}")
        print(f"总计告警请求: {stats['alarm_requests']}")
        sys.exit(0)


if __name__ == '__main__':
    main()