""" 企业微信通知模块 提供企业微信消息发送功能，支持文本和Markdown格式 """ import requests import json import time import hashlib from typing import Optional, Dict, Any, List from datetime import datetime try: from ..core.config_manager import get_config from ..core.logger import get_logger from ..core.models import Announcement from ..core.reliability import retry_on_exception, RetryConfig, safe_execute from ..storage.md_generator import AnnouncementMarkdownFormatter except ImportError: from core.config_manager import get_config from core.logger import get_logger from core.models import Announcement from core.reliability import retry_on_exception, RetryConfig, safe_execute from storage.md_generator import AnnouncementMarkdownFormatter logger = get_logger(__name__) class WeChatService: """企业微信服务""" def __init__(self): self.config = get_config().wechat_app self._access_token = None self._token_expires_at = 0 logger.info("企业微信服务初始化完成") def _get_access_token(self) -> Optional[str]: """ 获取访问令牌 Returns: Optional[str]: 访问令牌 """ current_time = time.time() # 检查令牌是否仍然有效 if self._access_token and current_time < self._token_expires_at: return self._access_token try: # 构建请求URL if self.config.use_proxy and hasattr(self.config, 'proxy_api_url'): url = f"{self.config.proxy_api_url}/cgi-bin/gettoken" else: url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" params = { "corpid": self.config.corp_id, "corpsecret": self.config.secret } logger.debug("正在获取企业微信访问令牌") response = requests.get(url, params=params, timeout=30) result = response.json() if result.get("errcode") == 0: self._access_token = result.get("access_token") # 提前5分钟过期 expires_in = result.get("expires_in", 7200) - 300 self._token_expires_at = current_time + expires_in logger.info("成功获取企业微信访问令牌") return self._access_token else: logger.error(f"获取访问令牌失败: {result}") return None except Exception as e: logger.error(f"获取访问令牌异常: {str(e)}") return None @retry_on_exception(RetryConfig(max_retries=3)) def send_text_message(self, content: str, to_user: str = "@all", to_party: str = "", to_tag: str = "") -> bool: """ 发送文本消息 Args: content: 消息内容 to_user: 接收者用户ID，多个用|分隔，@all表示全体 to_party: 接收者部门ID，多个用|分隔 to_tag: 接收者标签ID，多个用|分隔 Returns: bool: 发送是否成功 """ try: access_token = self._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/message/send" else: url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" params = {"access_token": access_token} data = { "touser": to_user, "toparty": to_party, "totag": to_tag, "msgtype": "text", "agentid": self.config.agent_id, "text": { "content": content } } logger.debug(f"发送文本消息: {content[:100]}...") response = requests.post(url, params=params, json=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 @retry_on_exception(RetryConfig(max_retries=3)) def send_markdown_message(self, content: str, to_user: str = "@all", to_party: str = "", to_tag: str = "") -> bool: """ 发送Markdown消息 Args: content: Markdown格式的消息内容 to_user: 接收者用户ID to_party: 接收者部门ID to_tag: 接收者标签ID Returns: bool: 发送是否成功 """ try: access_token = self._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/message/send" else: url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" params = {"access_token": access_token} data = { "touser": to_user, "toparty": to_party, "totag": to_tag, "msgtype": "markdown", "agentid": self.config.agent_id, "markdown": { "content": content } } logger.debug("发送Markdown消息") response = requests.post(url, params=params, json=data, timeout=30) result = response.json() if result.get("errcode") == 0: logger.info("Markdown消息发送成功") return True else: logger.error(f"Markdown消息发送失败: {result}") return False except Exception as e: logger.error(f"发送Markdown消息异常: {str(e)}") return False @retry_on_exception(RetryConfig(max_retries=3)) def send_textcard_message(self, title: str, description: str, url: str, to_user: str = "@all", to_party: str = "", to_tag: str = "", btn_txt: str = "查看详情") -> bool: """ 发送文本卡片消息 Args: title: 标题 description: 描述内容（支持HTML） url: 点击跳转的链接 to_user: 接收者用户ID to_party: 接收者部门ID to_tag: 接收者标签ID btn_txt: 按钮文字 Returns: bool: 发送是否成功 """ try: access_token = self._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_endpoint = f"{self.config.proxy_api_url}/cgi-bin/message/send" else: url_endpoint = "https://qyapi.weixin.qq.com/cgi-bin/message/send" params = {"access_token": access_token} data = { "touser": to_user, "toparty": to_party, "totag": to_tag, "msgtype": "textcard", "agentid": self.config.agent_id, "textcard": { "title": title, "description": description, "url": url, "btntxt": btn_txt }, "enable_id_trans": 0, "enable_duplicate_check": 0, "duplicate_check_interval": 1800 } logger.debug(f"发送文本卡片消息: {title}") response = requests.post(url_endpoint, params=params, json=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 send_announcement_notification(self, announcements: List[Announcement], max_count: int = 20) -> bool: """ 发送公告通知（每条公告发送一条单独的文本卡片消息） Args: announcements: 公告列表 max_count: 最大显示数量 Returns: bool: 是否至少有一条消息发送成功 """ if not announcements: logger.info("没有新公告，跳过通知") return True success_count = 0 total_count = len(announcements) logger.info(f"开始发送 {total_count} 条公告通知，每条单独发送") for i, announcement in enumerate(announcements[:max_count], 1): try: logger.debug(f"发送第 {i}/{min(total_count, max_count)} 条公告: {announcement.title[:30]}...") # 为每条公告生成单独的文本卡片 if self.send_single_announcement_notification(announcement): success_count += 1 logger.debug(f"第 {i} 条公告发送成功") else: logger.warning(f"第 {i} 条公告发送失败: {announcement.title[:30]}...") # 添加短暂延迟，避免发送过快 if i < len(announcements[:max_count]): import time time.sleep(0.5) except Exception as e: logger.error(f"发送第 {i} 条公告时发生异常: {str(e)}") continue logger.info(f"公告通知发送完成: {success_count}/{min(total_count, max_count)} 条成功") if total_count > max_count: logger.info(f"还有 {total_count - max_count} 条公告未发送（超过最大数量限制）") return success_count > 0 def send_single_announcement_notification(self, announcement: Announcement) -> bool: """ 发送单条公告的通知（文本卡片消息） Args: announcement: 单条公告 Returns: bool: 发送是否成功 """ try: # 生成单条公告的文本卡片内容 title, description, url = self._generate_single_textcard_notification(announcement) # 发送文本卡片消息 return self.send_textcard_message(title, description, url, btn_txt="查看详情") except Exception as e: logger.error(f"发送单条公告通知失败: {str(e)}") return False def _generate_single_textcard_notification(self, announcement: Announcement) -> tuple[str, str, str]: """ 生成单条公告的文本卡片内容 新格式示例： --- **北海市涠洲岛旅游区管理委员会关于办公桌的网上超市采购项目成交公告** 工程类公告 | 北海市涠洲岛旅游区管理委员会 | 2026-01-08 09:28 --- Args: announcement: 单条公告 Returns: tuple[str, str, str]: (标题, 描述HTML, URL) """ # 标题：公告标题（加粗显示，作为卡片标题） announcement_title = announcement.title if len(announcement_title) > 128: # 企业微信卡片标题限制128字符 announcement_title = announcement_title[:125] + "..." title = announcement_title # 公告类型映射（英文枚举值 -> 中文显示名称） type_mapping = { "PURCHASE": "采购公告", "RESULT": "结果公告", "CONTRACT": "合同公告", "CORRECTION": "更正公告", "PRE_ANNOUNCEMENT": "招标文件预公示", "SINGLE_SOURCE": "单一来源公示", "ELECTRONIC_MARKET": "电子卖场公示", "ACCEPTANCE": "履约验收公示", "ENGINEERING": "工程类公告", "FRAMEWORK_AGREEMENT": "框架协议征集公告", "FRAMEWORK_RESULT": "框架协议入围结果公告", "FRAMEWORK_SUMMARY": "框架协议成交结果汇总公告", "INTENTION": "采购意向公开" } # 获取公告类型的中文显示名称 announcement_type_enum = str(announcement.announcement_type).split('.')[-1] announcement_type_display = type_mapping.get(announcement_type_enum, announcement_type_enum) # 确定来源名称 source_name = announcement.purchase_name if announcement.purchase_name else announcement.source_name if len(source_name) > 25: # 限制来源名称长度 source_name = source_name[:22] + "..." # 时间格式化 if announcement.publish_date: time_str = announcement.publish_date.strftime("%Y-%m-%d %H:%M") else: time_str = "时间未知" # 生成描述：类型 | 来源 | 时间（使用默认颜色） description = f'
{announcement_type_display} | {source_name} | {time_str}
' # URL：公告详情链接 url = announcement.content_url return title, description, url def _generate_announcement_notification(self, announcements: List[Announcement], max_count: int) -> str: """ 生成公告通知内容（改进版） Args: announcements: 公告列表 max_count: 最大显示数量 Returns: str: Markdown格式的通知内容 """ if not announcements: return f"""# 🔔 广西政府采购网公告更新 **暂无新公告** --- *更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}* *点击公告标题查看详情*""" # 按日期分组 today_announcements = [] other_announcements = [] today = datetime.now().date() for announcement in announcements: if announcement.publish_date and announcement.publish_date.date() == today: today_announcements.append(announcement) else: other_announcements.append(announcement) lines = [] # 标题和概要 total_count = len(announcements) lines.append("# 🔔 广西政府采购网公告更新") lines.append("") lines.append(f"📊 **共发现 {total_count} 条新公告**") lines.append("") # 今日公告 if today_announcements: lines.append(f"## 🔥 今日公告 ({len(today_announcements)}条)") lines.append("") display_today = today_announcements[:max_count//2] for i, announcement in enumerate(display_today, 1): # 改进标题显示：保留更多字符，但确保美观 title = announcement.title if len(title) > 50: title = title[:47] + "..." # 显示时间 time_str = announcement.publish_date.strftime("%H:%M") if announcement.publish_date else "N/A" # 添加序号和更好的格式 lines.append(f"**{i}.** [{title}]({announcement.content_url})") lines.append(f" ⏰ {time_str} | 📍 {announcement.source_name}") lines.append("") if len(today_announcements) > len(display_today): lines.append(f"⚠️ 还有 {len(today_announcements) - len(display_today)} 条今日公告未显示") lines.append("") # 其他公告 if other_announcements: lines.append(f"## 📄 其他公告 ({len(other_announcements)}条)") lines.append("") remaining_slots = max_count - len(today_announcements) if today_announcements else max_count display_other = other_announcements[:remaining_slots] for i, announcement in enumerate(display_other, 1): title = announcement.title if len(title) > 45: title = title[:42] + "..." date_str = announcement.publish_date.strftime("%m-%d") if announcement.publish_date else "N/A" lines.append(f"**{i}.** [{title}]({announcement.content_url}) - {date_str}") if len(other_announcements) > len(display_other): lines.append(f"⚠️ 还有 {len(other_announcements) - len(display_other)} 条历史公告未显示") lines.append("") # 统计信息 - 改进版 lines.append("## 📈 数据统计") lines.append("") # 按来源统计 source_stats = {} for announcement in announcements: source = announcement.source_name source_stats[source] = source_stats.get(source, 0) + 1 # 按类型统计 type_stats = {} for announcement in announcements: ann_type = str(announcement.announcement_type).split('.')[-1] # 获取枚举名称 type_stats[ann_type] = type_stats.get(ann_type, 0) + 1 lines.append("**按来源统计：**") for source, count in sorted(source_stats.items()): lines.append(f"• {source}: {count}条") lines.append("") lines.append("**按类型统计：**") for ann_type, count in sorted(type_stats.items()): lines.append(f"• {ann_type}: {count}条") lines.append("") # 分割线和时间 lines.append("---") lines.append("") lines.append(f"🕒 *更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*") lines.append("💡 *点击公告标题查看详情*") return "\n".join(lines) def _generate_textcard_notification(self, announcements: List[Announcement], max_count: int) -> tuple[str, str, str]: """ 生成文本卡片格式的通知内容 Args: announcements: 公告列表 max_count: 最大显示数量 Returns: tuple[str, str, str]: (标题, 描述HTML, URL) """ # 按日期分组 today_announcements = [] other_announcements = [] today = datetime.now().date() for announcement in announcements: if announcement.publish_date and announcement.publish_date.date() == today: today_announcements.append(announcement) else: other_announcements.append(announcement) # 生成标题 total_count = len(announcements) title = f"🔔 广西政府采购网公告更新 ({total_count}条)" # 生成描述HTML html_parts = [] # 总统计 html_parts.append('
📊 发现 {total_count} 条新公告
'.format(total_count=total_count)) html_parts.append("") # 今日公告 if today_announcements: html_parts.append('
🔥 今日公告 ({count}条)
'.format(count=len(today_announcements))) display_today = today_announcements[:max_count//2] for i, announcement in enumerate(display_today, 1): # 标题处理 ann_title = announcement.title if len(ann_title) > 35: # 文本卡片标题较短 ann_title = ann_title[:32] + "..." # 时间和来源 time_str = announcement.publish_date.strftime("%H:%M") if announcement.publish_date else "N/A" source = announcement.source_name[:10] # 限制来源名称长度 html_parts.append('{i}. {title}'.format( i=i, url=announcement.content_url, title=ann_title)) html_parts.append('
⏰ {time} | 📍 {source}
'.format( time=time_str, source=source)) if len(today_announcements) > len(display_today): remaining = len(today_announcements) - len(display_today) html_parts.append('
还有 {remaining} 条今日公告...
'.format(remaining=remaining)) # 其他公告 if other_announcements: html_parts.append("") html_parts.append('
📄 其他公告 ({count}条)
'.format(count=len(other_announcements))) remaining_slots = max_count - len(today_announcements) if today_announcements else max_count display_other = other_announcements[:remaining_slots] for i, announcement in enumerate(display_other, 1): ann_title = announcement.title if len(ann_title) > 30: ann_title = ann_title[:27] + "..." date_str = announcement.publish_date.strftime("%m-%d") if announcement.publish_date else "N/A" html_parts.append('{i}. {title} ({date})'.format( i=i, url=announcement.content_url, title=ann_title, date=date_str)) if len(other_announcements) > len(display_other): remaining = len(other_announcements) - len(display_other) html_parts.append('
还有 {remaining} 条历史公告...
'.format(remaining=remaining)) # 统计信息 html_parts.append("") html_parts.append('
📈 数据统计
') # 按来源统计 source_stats = {} for announcement in announcements: source = announcement.source_name source_stats[source] = source_stats.get(source, 0) + 1 html_parts.append('
按来源: {stats}
'.format( stats=" | ".join([f"{source}:{count}" for source, count in sorted(source_stats.items())]))) # 时间戳 update_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') html_parts.append("") html_parts.append('
🕒 更新时间: {time}
'.format(time=update_time)) description = "\n".join(html_parts) # 限制描述长度（企业微信文本卡片description不超过512字符） if len(description) > 500: description = description[:497] + "..." # 生成跳转URL（可以跳转到公告列表页面或第一条公告） if announcements: url = announcements[0].content_url # 默认跳转到第一条公告 else: url = "https://zfcg.gxzf.gov.cn" # 默认跳转到网站首页 return title, description, url def send_system_notification(self, title: str, content: str, message_type: str = "text") -> bool: """ 发送系统通知 Args: title: 通知标题 content: 通知内容 message_type: 消息类型 (text/markdown) Returns: bool: 发送是否成功 """ try: if message_type == "markdown": full_content = f"# {title}\n\n{content}" return self.send_markdown_message(full_content) else: full_content = f"{title}\n\n{content}" return self.send_text_message(full_content) except Exception as e: logger.error(f"发送系统通知失败: {str(e)}") return False def send_error_notification(self, error_message: str, error_details: Optional[str] = None) -> bool: """ 发送错误通知 Args: error_message: 错误消息 error_details: 错误详情 Returns: bool: 发送是否成功 """ content = f"## ❌ 系统错误\n\n**错误信息**: {error_message}" if error_details: content += f"\n\n**错误详情**:\n```\n{error_details}\n```" content += f"\n\n*发生时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*" return self.send_markdown_message(content) def test_connection(self) -> bool: """ 测试连接 Returns: bool: 连接是否正常 """ try: token = self._get_access_token() return token is not None except Exception as e: logger.error(f"企业微信连接测试失败: {str(e)}") return False def get_service_status(self) -> Dict[str, Any]: """ 获取服务状态 Returns: Dict[str, Any]: 服务状态信息 """ return { "service": "wechat", "enabled": self.config.enabled, "corp_id": self.config.corp_id[:10] + "..." if self.config.corp_id else None, "agent_id": self.config.agent_id, "has_token": self._access_token is not None, "token_expires_at": datetime.fromtimestamp(self._token_expires_at).isoformat() if self._token_expires_at > 0 else None, "use_proxy": self.config.use_proxy, "connection_test": self.test_connection() if self.config.enabled else False } class NotificationManager: """通知管理器""" def __init__(self): self.wechat = WeChatService() self._services = { "wechat": self.wechat } def send_announcement_notification(self, announcements: List[Announcement]) -> Dict[str, bool]: """ 发送公告通知 Args: announcements: 公告列表 Returns: Dict[str, bool]: 各服务发送结果 """ results = {} # 企业微信通知 if self.wechat.config.enabled: try: results["wechat"] = self.wechat.send_announcement_notification(announcements) except Exception as e: logger.error(f"企业微信通知失败: {str(e)}") results["wechat"] = False else: results["wechat"] = None # 未启用 return results def send_system_notification(self, title: str, content: str) -> Dict[str, bool]: """ 发送系统通知 Args: title: 通知标题 content: 通知内容 Returns: Dict[str, bool]: 发送结果 """ results = {} if self.wechat.config.enabled: try: results["wechat"] = self.wechat.send_system_notification(title, content, "markdown") except Exception as e: logger.error(f"企业微信系统通知失败: {str(e)}") results["wechat"] = False else: results["wechat"] = None return results def send_error_notification(self, error_message: str, error_details: Optional[str] = None) -> Dict[str, bool]: """ 发送错误通知 Args: error_message: 错误消息 error_details: 错误详情 Returns: Dict[str, bool]: 发送结果 """ results = {} if self.wechat.config.enabled: try: results["wechat"] = self.wechat.send_error_notification(error_message, error_details) except Exception as e: logger.error(f"企业微信错误通知失败: {str(e)}") results["wechat"] = False else: results["wechat"] = None return results def get_status(self) -> Dict[str, Any]: """ 获取通知服务状态 Returns: Dict[str, Any]: 服务状态 """ return { "services": { name: service.get_service_status() for name, service in self._services.items() } } # 全局通知管理器实例 _notification_manager = None def get_notification_manager() -> NotificationManager: """ 获取通知管理器实例 Returns: NotificationManager: 通知管理器实例 """ global _notification_manager if _notification_manager is None: _notification_manager = NotificationManager() return _notification_manager def send_announcements_notification(announcements: List[Announcement]) -> bool: """ 发送公告通知 Args: announcements: 公告列表 Returns: bool: 是否至少有一个服务发送成功 """ manager = get_notification_manager() results = manager.send_announcement_notification(announcements) # 检查是否有服务发送成功 return any(result for result in results.values() if result is True) def send_system_notification(title: str, content: str) -> bool: """ 发送系统通知 Args: title: 通知标题 content: 通知内容 Returns: bool: 是否至少有一个服务发送成功 """ manager = get_notification_manager() results = manager.send_system_notification(title, content) return any(result for result in results.values() if result is True) def send_error_alert(error_message: str, error_details: Optional[str] = None) -> bool: """ 发送错误警报 Args: error_message: 错误消息 error_details: 错误详情 Returns: bool: 是否至少有一个服务发送成功 """ manager = get_notification_manager() results = manager.send_error_notification(error_message, error_details) return any(result for result in results.values() if result is True)