Remove legacy test script and enhance main application with WeChat server and menu management features. Added commands for starting the WeChat callback server and managing WeChat menus, along with necessary imports and error handling. Updated requirements to include Flask for the WeChat server functionality.
This commit is contained in:
@@ -0,0 +1,310 @@
|
||||
"""
|
||||
企业微信菜单管理器
|
||||
负责创建和管理企业微信应用菜单
|
||||
"""
|
||||
|
||||
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_summary"
|
||||
},
|
||||
{
|
||||
"type": "click",
|
||||
"name": "自定义爬取",
|
||||
"key": "custom_crawl"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "帮助",
|
||||
"sub_button": [
|
||||
{
|
||||
"type": "click",
|
||||
"name": "使用说明",
|
||||
"key": "help_guide"
|
||||
},
|
||||
{
|
||||
"type": "click",
|
||||
"name": "联系我们",
|
||||
"key": "contact_us"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user