""" 企业微信菜单管理器 负责创建和管理企业微信应用菜单 """ 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