331 lines
10 KiB
Python
331 lines
10 KiB
Python
"""
|
|
企业微信菜单管理器
|
|
负责创建和管理企业微信应用菜单
|
|
"""
|
|
|
|
import json
|
|
import requests
|
|
from typing import Dict, Any, Optional
|
|
|
|
try:
|
|
from ..core.config_manager import get_config
|
|
from ..core.logger import get_logger
|
|
from ..notification.wechat import WeChatService
|
|
except ImportError:
|
|
try:
|
|
from core.config_manager import get_config
|
|
from core.logger import get_logger
|
|
from notification.wechat import WeChatService
|
|
except ImportError as e:
|
|
raise ImportError(f"菜单管理器导入失败: {e}")
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class WeChatMenuManager:
|
|
"""企业微信菜单管理器"""
|
|
|
|
def __init__(self):
|
|
self.config = get_config().wechat_app
|
|
self.wechat_service = WeChatService()
|
|
|
|
# 菜单配置
|
|
self.menu_data = {
|
|
"button": [
|
|
{
|
|
"name": "监控操作",
|
|
"sub_button": [
|
|
{
|
|
"type": "click",
|
|
"name": "立即搜索",
|
|
"key": "crawl_now"
|
|
},
|
|
{
|
|
"type": "click",
|
|
"name": "今日统计",
|
|
"key": "today_stats"
|
|
},
|
|
{
|
|
"type": "click",
|
|
"name": "关键词搜索",
|
|
"key": "keyword_search"
|
|
},
|
|
{
|
|
"type": "click",
|
|
"name": "最新公告",
|
|
"key": "latest_news"
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"name": "系统管理",
|
|
"sub_button": [
|
|
{
|
|
"type": "click",
|
|
"name": "关键词管理",
|
|
"key": "keyword_manage"
|
|
},
|
|
{
|
|
"type": "click",
|
|
"name": "系统状态",
|
|
"key": "system_status"
|
|
},
|
|
{
|
|
"type": "click",
|
|
"name": "清理缓存",
|
|
"key": "clear_cache"
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"name": "帮助",
|
|
"sub_button": [
|
|
{
|
|
"type": "click",
|
|
"name": "使用说明",
|
|
"key": "help_guide"
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
|
|
logger.info("企业微信菜单管理器初始化完成")
|
|
|
|
def create_menu(self) -> bool:
|
|
"""
|
|
创建菜单
|
|
|
|
Returns:
|
|
bool: 创建是否成功
|
|
"""
|
|
try:
|
|
logger.info("开始创建企业微信菜单")
|
|
|
|
# 获取访问令牌
|
|
access_token = self.wechat_service._get_access_token()
|
|
if not access_token:
|
|
logger.error("获取访问令牌失败，无法创建菜单")
|
|
return False
|
|
|
|
# 构建请求URL
|
|
if self.config.use_proxy and hasattr(self.config, 'proxy_api_url'):
|
|
url = f"{self.config.proxy_api_url}/cgi-bin/menu/create"
|
|
else:
|
|
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/create"
|
|
|
|
params = {
|
|
"access_token": access_token,
|
|
"agentid": self.config.agent_id
|
|
}
|
|
|
|
# 发送创建菜单请求
|
|
response = requests.post(url, params=params, json=self.menu_data, timeout=30)
|
|
result = response.json()
|
|
|
|
if result.get("errcode") == 0:
|
|
logger.info("企业微信菜单创建成功")
|
|
return True
|
|
else:
|
|
logger.error(f"企业微信菜单创建失败: {result}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"创建菜单异常: {str(e)}")
|
|
return False
|
|
|
|
def delete_menu(self) -> bool:
|
|
"""
|
|
删除菜单
|
|
|
|
Returns:
|
|
bool: 删除是否成功
|
|
"""
|
|
try:
|
|
logger.info("开始删除企业微信菜单")
|
|
|
|
# 获取访问令牌
|
|
access_token = self.wechat_service._get_access_token()
|
|
if not access_token:
|
|
logger.error("获取访问令牌失败，无法删除菜单")
|
|
return False
|
|
|
|
# 构建请求URL
|
|
if self.config.use_proxy and hasattr(self.config, 'proxy_api_url'):
|
|
url = f"{self.config.proxy_api_url}/cgi-bin/menu/delete"
|
|
else:
|
|
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/delete"
|
|
|
|
params = {
|
|
"access_token": access_token,
|
|
"agentid": self.config.agent_id
|
|
}
|
|
|
|
# 发送删除菜单请求
|
|
response = requests.get(url, params=params, timeout=30)
|
|
result = response.json()
|
|
|
|
if result.get("errcode") == 0:
|
|
logger.info("企业微信菜单删除成功")
|
|
return True
|
|
else:
|
|
logger.error(f"企业微信菜单删除失败: {result}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"删除菜单异常: {str(e)}")
|
|
return False
|
|
|
|
def get_menu(self) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
获取当前菜单
|
|
|
|
Returns:
|
|
Optional[Dict[str, Any]]: 菜单信息，失败返回None
|
|
"""
|
|
try:
|
|
logger.info("开始获取企业微信菜单")
|
|
|
|
# 获取访问令牌
|
|
access_token = self.wechat_service._get_access_token()
|
|
if not access_token:
|
|
logger.error("获取访问令牌失败，无法获取菜单")
|
|
return None
|
|
|
|
# 构建请求URL
|
|
if self.config.use_proxy and hasattr(self.config, 'proxy_api_url'):
|
|
url = f"{self.config.proxy_api_url}/cgi-bin/menu/get"
|
|
else:
|
|
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/get"
|
|
|
|
params = {
|
|
"access_token": access_token,
|
|
"agentid": self.config.agent_id
|
|
}
|
|
|
|
# 发送获取菜单请求
|
|
response = requests.get(url, params=params, timeout=30)
|
|
result = response.json()
|
|
|
|
if result.get("errcode") == 0:
|
|
logger.info("企业微信菜单获取成功")
|
|
return result
|
|
else:
|
|
logger.error(f"企业微信菜单获取失败: {result}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取菜单异常: {str(e)}")
|
|
return None
|
|
|
|
def update_menu(self, menu_data: Dict[str, Any]) -> bool:
|
|
"""
|
|
更新菜单
|
|
|
|
Args:
|
|
menu_data: 新的菜单数据
|
|
|
|
Returns:
|
|
bool: 更新是否成功
|
|
"""
|
|
try:
|
|
logger.info("开始更新企业微信菜单")
|
|
|
|
# 先删除旧菜单
|
|
if not self.delete_menu():
|
|
logger.warning("删除旧菜单失败，继续创建新菜单")
|
|
|
|
# 更新菜单配置
|
|
self.menu_data = menu_data
|
|
|
|
# 创建新菜单
|
|
return self.create_menu()
|
|
|
|
except Exception as e:
|
|
logger.error(f"更新菜单异常: {str(e)}")
|
|
return False
|
|
|
|
def get_menu_info(self) -> Dict[str, Any]:
|
|
"""
|
|
获取菜单信息（用于调试）
|
|
|
|
Returns:
|
|
Dict[str, Any]: 菜单信息
|
|
"""
|
|
return {
|
|
"menu_data": self.menu_data,
|
|
"menu_structure": self._analyze_menu_structure()
|
|
}
|
|
|
|
def _analyze_menu_structure(self) -> Dict[str, Any]:
|
|
"""分析菜单结构"""
|
|
try:
|
|
buttons = self.menu_data.get("button", [])
|
|
structure = {
|
|
"total_buttons": len(buttons),
|
|
"buttons": []
|
|
}
|
|
|
|
for i, button in enumerate(buttons):
|
|
button_info = {
|
|
"index": i,
|
|
"name": button.get("name", ""),
|
|
"type": button.get("type", "menu"),
|
|
}
|
|
|
|
if "sub_button" in button:
|
|
button_info["sub_buttons"] = len(button["sub_button"])
|
|
button_info["sub_button_list"] = [
|
|
{
|
|
"name": sub.get("name", ""),
|
|
"type": sub.get("type", ""),
|
|
"key": sub.get("key", "")
|
|
}
|
|
for sub in button["sub_button"]
|
|
]
|
|
else:
|
|
button_info["key"] = button.get("key", "")
|
|
|
|
structure["buttons"].append(button_info)
|
|
|
|
return structure
|
|
|
|
except Exception as e:
|
|
logger.error(f"分析菜单结构异常: {str(e)}")
|
|
return {"error": str(e)}
|
|
|
|
def test_menu_operations(self) -> Dict[str, bool]:
|
|
"""
|
|
测试菜单操作
|
|
|
|
Returns:
|
|
Dict[str, bool]: 测试结果
|
|
"""
|
|
results = {
|
|
"create_menu": False,
|
|
"get_menu": False,
|
|
"delete_menu": False
|
|
}
|
|
|
|
try:
|
|
# 测试获取菜单
|
|
menu_info = self.get_menu()
|
|
results["get_menu"] = menu_info is not None
|
|
|
|
# 测试创建菜单（如果没有菜单的话）
|
|
if not menu_info:
|
|
results["create_menu"] = self.create_menu()
|
|
else:
|
|
results["create_menu"] = True # 已经有菜单了
|
|
|
|
# 不测试删除，避免影响现有菜单
|
|
results["delete_menu"] = True
|
|
|
|
logger.info(f"菜单操作测试完成: {results}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"菜单操作测试异常: {str(e)}")
|
|
|
|
return results
|