#!/usr/bin/env python3
|
"""
|
在线状态监控功能使用示例
|
"""
|
|
import asyncio
|
import time
|
from config import settings
|
from app.services.ecloud_client import ecloud_client
|
from app.services.email_service import email_service
|
from app.services.sms_service import sms_service
|
from app.workers.online_status_worker import online_status_worker
|
|
|
async def test_online_status_monitoring():
|
"""测试在线状态监控功能"""
|
print("=== 在线状态监控功能测试 ===\n")
|
|
# 1. 测试查询在线微信列表
|
print("1. 测试查询在线微信列表...")
|
online_list = ecloud_client.query_online_wechat_list()
|
if online_list is not None:
|
print(f" 查询成功，在线微信数量: {len(online_list)}")
|
for i, wechat in enumerate(online_list):
|
print(f" 微信{i+1}: wcId={wechat.get('wcId')}, wId={wechat.get('wId')}")
|
else:
|
print(" 查询失败")
|
print()
|
|
# 2. 测试邮件服务
|
print("2. 测试邮件服务...")
|
if settings.email_enabled:
|
# 测试连接
|
email_connected = email_service.test_connection()
|
print(f" 邮件服务连接: {'成功' if email_connected else '失败'}")
|
|
if email_connected:
|
# 发送测试邮件
|
test_subject = "在线状态监控测试邮件"
|
test_content = f"这是一封测试邮件，发送时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
|
email_sent = email_service.send_notification(test_subject, test_content)
|
print(f" 测试邮件发送: {'成功' if email_sent else '失败'}")
|
else:
|
print(" 邮件服务已禁用")
|
print()
|
|
# 3. 测试短信服务
|
print("3. 测试短信服务...")
|
if settings.sms_enabled:
|
# 发送测试短信
|
test_content = f"【测试】在线状态监控测试短信，时间: {time.strftime('%H:%M:%S')}"
|
sms_sent = sms_service.send_notification(test_content)
|
print(f" 测试短信发送: {'成功' if sms_sent else '失败'}")
|
else:
|
print(" 短信服务已禁用")
|
print()
|
|
# 4. 测试w_id动态更新
|
print("4. 测试w_id动态更新...")
|
current_w_id = settings.get_current_w_id()
|
print(f" 当前配置的w_id: {current_w_id}")
|
|
if online_list and len(online_list) > 0:
|
online_w_id = online_list[0].get("wId")
|
if online_w_id != current_w_id:
|
print(f" 检测到w_id变化: {current_w_id} -> {online_w_id}")
|
success = settings.update_ecloud_w_id(online_w_id)
|
print(f" w_id更新: {'成功' if success else '失败'}")
|
else:
|
print(" w_id无变化，无需更新")
|
else:
|
print(" 无在线微信，无法测试w_id更新")
|
print()
|
|
# 5. 显示工作进程状态
|
print("5. 在线状态监控工作进程状态...")
|
status = online_status_worker.get_status()
|
print(f" 运行状态: {'运行中' if status['running'] else '已停止'}")
|
print(f" 功能启用: {'是' if status['enabled'] else '否'}")
|
print(f" 检测间隔: {status['check_interval_minutes']}分钟")
|
print(f" 通知冷却: {status['notification_cooldown_minutes']}分钟")
|
if status['last_notification_time']:
|
last_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(status['last_notification_time']))
|
print(f" 上次通知: {last_time}")
|
else:
|
print(" 上次通知: 无")
|
print()
|
|
print("=== 测试完成 ===")
|
|
|
def show_configuration():
|
"""显示当前配置"""
|
print("=== 当前配置信息 ===\n")
|
|
print("在线状态监控配置:")
|
print(f" 启用状态: {'是' if settings.online_status_enabled else '否'}")
|
print(f" 检测间隔: {settings.online_status_check_interval}分钟")
|
print(f" 通知消息: {settings.online_status_notification_message}")
|
print()
|
|
print("邮件通知配置:")
|
print(f" 启用状态: {'是' if settings.email_enabled else '否'}")
|
if settings.email_enabled:
|
print(f" SMTP服务器: {settings.email_smtp_server}:{settings.email_smtp_port}")
|
print(f" 发件人: {settings.email_from_email}")
|
print(f" 收件人: {', '.join(settings.email_to_emails)}")
|
print()
|
|
print("短信通知配置:")
|
print(f" 启用状态: {'是' if settings.sms_enabled else '否'}")
|
if settings.sms_enabled:
|
print(f" API地址: {settings.sms_api_url}")
|
print(f" 用户名: {settings.sms_username}")
|
print(f" 接收号码: {', '.join(settings.sms_phone_numbers)}")
|
print()
|
|
print("E云管家配置:")
|
print(f" API地址: {settings.ecloud_base_url}")
|
print(f" 当前w_id: {settings.ecloud_w_id}")
|
print()
|
|
|
if __name__ == "__main__":
|
print("在线状态监控功能使用示例\n")
|
|
# 显示配置信息
|
show_configuration()
|
|
# 运行测试
|
asyncio.run(test_online_status_monitoring())
|
|
print("\n注意事项:")
|
print("1. 请确保已正确配置config.json中的相关参数")
|
print("2. 邮件和短信测试会实际发送消息，请谨慎使用")
|
print("3. 建议在生产环境中禁用测试功能")
|
