Refactor the cron job functionality to transition from a crawling to a searching model. Update related documentation and scripts to reflect the new terminology and functionality, including changes in the README and script comments. Enhance WeChat menu and message handling to support the new search features, ensuring clarity in user interactions and logging.
This commit is contained in:
+7
-7
@@ -1,11 +1,11 @@
|
||||
# 定时爬取脚本使用说明
|
||||
# 定时搜索脚本使用说明
|
||||
|
||||
## 脚本功能
|
||||
|
||||
`cron_crawl.py` 是一个用于定时爬取广西政府采购网公告的脚本，具有以下功能：
|
||||
`cron_crawl.py` 是一个用于定时搜索广西政府采购网公告的脚本，具有以下功能：
|
||||
|
||||
- ✅ 从 `gx_gp_monitor/config/config.yaml` 读取关键词配置
|
||||
- ✅ 爬取最新的公告（数据库中没有的）
|
||||
- ✅ 搜索最新的公告（数据库中没有的）
|
||||
- ✅ 自动筛选匹配关键词的公告
|
||||
- ✅ 将筛选结果保存到数据库
|
||||
- ✅ 以卡片形式发送企业微信通知
|
||||
@@ -16,7 +16,7 @@
|
||||
project/
|
||||
├── venv/ # Python虚拟环境
|
||||
├── gx_gp_monitor/
|
||||
│ ├── cron_crawl.py # 定时爬取主脚本
|
||||
│ ├── cron_crawl.py # 定时搜索主脚本
|
||||
│ └── config/config.yaml # 配置文件
|
||||
├── run_cron_crawl.sh # 启动脚本（自动激活虚拟环境）
|
||||
├── CRON_README.md # 使用说明
|
||||
@@ -62,8 +62,8 @@ crontab -e
|
||||
### 3. 脚本输出示例
|
||||
|
||||
```
|
||||
=== 定时爬取任务完成 ===
|
||||
总共爬取: 1300 条公告
|
||||
=== 定时搜索任务完成 ===
|
||||
总共搜索: 1300 条公告
|
||||
新增公告: 5 条
|
||||
关键词筛选: 2 条
|
||||
保存到数据库: 2 条
|
||||
@@ -135,7 +135,7 @@ tail -f /home/v6ole/pyproject/GX-gp-notify/logs/cron.log
|
||||
|
||||
4. **时区设置**：确保系统时区设置正确，影响定时任务执行时间
|
||||
|
||||
5. **资源消耗**：爬取过程中会消耗一定的CPU和网络资源，建议在非高峰期运行
|
||||
5. **资源消耗**：搜索过程中会消耗一定的CPU和网络资源，建议在非高峰期运行
|
||||
|
||||
## 故障排除
|
||||
|
||||
|
||||
+73
-24
@@ -6,7 +6,7 @@
|
||||
- 企业微信应用消息推送
|
||||
- 回调服务器处理用户交互
|
||||
- 应用菜单管理
|
||||
- 实时爬取和自定义搜索
|
||||
- 实时搜索和智能筛选
|
||||
|
||||
## 配置要求
|
||||
|
||||
@@ -43,7 +43,7 @@ wechat_app:
|
||||
|
||||
系统会在以下情况下自动推送消息到企业微信：
|
||||
|
||||
- **爬取完成通知**：每次执行爬取后，推送筛选结果
|
||||
- **搜索完成通知**：每次执行搜索后，推送筛选结果
|
||||
- **关键词匹配通知**：发现匹配关键词的公告时推送
|
||||
- **系统状态通知**：系统启动、错误等重要事件
|
||||
|
||||
@@ -51,36 +51,85 @@ wechat_app:
|
||||
|
||||
### 2. 应用菜单
|
||||
|
||||
系统提供以下菜单功能：
|
||||
系统提供全新的菜单结构，包含三大功能模块：
|
||||
|
||||
#### 操作菜单
|
||||
- **立即爬取**：执行一次完整的公告爬取
|
||||
- **今日总结**：查看今日公告统计（开发中）
|
||||
- **自定义爬取**：输入关键词进行搜索
|
||||
#### 🚀 监控操作
|
||||
- **立即搜索**：执行一次完整的公告搜索，获取最新数据
|
||||
- **今日统计**：查看今日公告统计信息和数据概览
|
||||
- **关键词搜索**：输入关键词搜索相关公告
|
||||
- **最新公告**：浏览最近发布的10条公告
|
||||
|
||||
#### 帮助菜单
|
||||
- **使用说明**：显示详细使用指南
|
||||
- **联系我们**：联系信息
|
||||
#### ⚙️ 系统管理
|
||||
- **关键词管理**：查看当前系统监控的关键词
|
||||
- **系统状态**：查看各组件运行状态和系统信息
|
||||
- **清理缓存**：清理系统缓存，提升性能
|
||||
|
||||
#### ❓ 帮助
|
||||
- **使用说明**：查看详细功能介绍和使用指南
|
||||
|
||||
### 3. 文本交互
|
||||
|
||||
支持以下文本命令：
|
||||
支持丰富的文本命令交互：
|
||||
|
||||
#### 🔍 搜索功能
|
||||
```
|
||||
# 关键词搜索
|
||||
直接发送关键词，如：大化、信息化
|
||||
# 直接关键词搜索
|
||||
大化 # 搜索"大化"相关公告
|
||||
信息化 政府采购 # 搜索多个关键词
|
||||
|
||||
# 手动爬取
|
||||
爬取 [关键词] # 如：爬取 大化
|
||||
|
||||
# 获取帮助
|
||||
帮助 # 显示使用说明
|
||||
help
|
||||
|
||||
# 查看总结
|
||||
总结 # 查看今日统计
|
||||
# 关键词命令
|
||||
关键词 大化 # 明确指定关键词搜索
|
||||
搜索 政府采购 # 搜索政府采购相关公告
|
||||
```
|
||||
|
||||
#### 📊 数据查询
|
||||
```
|
||||
# 统计信息
|
||||
总结 # 查看今日统计
|
||||
统计 # 查看系统统计信息
|
||||
|
||||
# 最新数据
|
||||
最新公告 # 查看最新发布的公告
|
||||
最新 # 同上
|
||||
```
|
||||
|
||||
#### 🛠️ 系统管理
|
||||
```
|
||||
# 系统状态
|
||||
系统状态 # 查看系统运行状态
|
||||
状态 # 同上
|
||||
|
||||
# 缓存管理
|
||||
清理缓存 # 清理系统缓存
|
||||
清理 # 同上
|
||||
|
||||
# 关键词管理
|
||||
查看关键词 # 查看当前监控关键词
|
||||
添加关键词 [关键词] # 添加新关键词（开发中）
|
||||
删除关键词 [关键词] # 删除关键词（开发中）
|
||||
```
|
||||
|
||||
#### ℹ️ 信息查询
|
||||
```
|
||||
# 帮助信息
|
||||
帮助 # 显示详细使用说明
|
||||
help # 同上
|
||||
|
||||
# 系统信息
|
||||
关于系统 # 查看系统详细信息
|
||||
关于 # 同上
|
||||
|
||||
# 联系方式
|
||||
联系我们 # 查看联系信息
|
||||
联系 # 同上
|
||||
```
|
||||
|
||||
#### 💡 智能交互
|
||||
- **自动识别**：直接发送关键词即可搜索
|
||||
- **多关键词**：用空格分隔多个关键词
|
||||
- **模糊匹配**：支持标题和内容搜索
|
||||
- **实时反馈**：所有操作都有即时响应
|
||||
|
||||
## 部署和运行
|
||||
|
||||
### 1. 安装依赖
|
||||
@@ -144,10 +193,10 @@ python gx_gp_monitor/main.py wechat-menu --action get
|
||||
python gx_gp_monitor/main.py wechat-menu --action test
|
||||
```
|
||||
|
||||
### 5. 正常爬取
|
||||
### 5. 正常搜索
|
||||
|
||||
```bash
|
||||
# 执行爬取（会自动推送消息）
|
||||
# 执行搜索（会自动推送消息）
|
||||
python gx_gp_monitor/main.py crawl
|
||||
```
|
||||
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
创建企业微信菜单脚本
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# 添加项目路径
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
try:
|
||||
from gx_gp_monitor.wechat.menu_manager import WeChatMenuManager
|
||||
from gx_gp_monitor.core.config_manager import load_config
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("🚀 开始创建企业微信菜单...")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# 加载配置
|
||||
load_config()
|
||||
print("✅ 配置加载成功")
|
||||
|
||||
# 创建菜单管理器
|
||||
menu_manager = WeChatMenuManager()
|
||||
print("✅ 菜单管理器初始化成功")
|
||||
|
||||
# 显示当前菜单结构
|
||||
menu_info = menu_manager.get_menu_info()
|
||||
print("\n📋 菜单结构预览：")
|
||||
for button in menu_info['menu_structure']['buttons']:
|
||||
sub_count = len(button.get('sub_buttons', [])) if isinstance(button.get('sub_buttons'), list) else button.get('sub_buttons', 0)
|
||||
print(f"• {button['name']} ({sub_count}个子菜单)")
|
||||
if isinstance(button.get('sub_buttons'), list):
|
||||
for sub_btn in button['sub_buttons']:
|
||||
print(f" - {sub_btn['name']}")
|
||||
|
||||
print("\n🔄 正在创建菜单到企业微信...")
|
||||
|
||||
# 创建菜单
|
||||
result = menu_manager.create_menu()
|
||||
|
||||
if result:
|
||||
print("✅ 企业微信菜单创建成功！")
|
||||
print("\n💡 菜单功能说明：")
|
||||
print("• 🚀 监控操作：搜索、统计、筛选等核心功能")
|
||||
print("• ⚙️ 系统管理：状态监控、维护操作")
|
||||
print("• ❓ 帮助：使用说明和联系方式")
|
||||
print("\n🎯 现在可以在企业微信中查看和使用新菜单了！")
|
||||
else:
|
||||
print("❌ 企业微信菜单创建失败")
|
||||
print("💡 可能的原因：")
|
||||
print("• 企业微信配置错误")
|
||||
print("• 网络连接问题")
|
||||
print("• API权限不足")
|
||||
print("• 请检查日志文件获取详细错误信息")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 创建菜单失败: {str(e)}")
|
||||
print("请检查配置和网络连接")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
except ImportError as e:
|
||||
print(f"❌ 导入失败: {e}", file=sys.stderr)
|
||||
print("请确保已安装所有依赖: pip install -r gx_gp_monitor/requirements.txt", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"❌ 脚本执行失败: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
Binary file not shown.
+11
-11
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
定时爬取脚本
|
||||
执行爬取、筛选关键词、保存到数据库并发送企业微信卡片通知
|
||||
定时搜索脚本
|
||||
执行搜索、筛选关键词、保存到数据库并发送企业微信卡片通知
|
||||
"""
|
||||
|
||||
import sys
|
||||
@@ -23,7 +23,7 @@ try:
|
||||
logger = get_logger(__name__)
|
||||
|
||||
def main():
|
||||
"""主函数：执行定时爬取任务"""
|
||||
"""主函数：执行定时搜索任务"""
|
||||
try:
|
||||
# 加载配置
|
||||
config = load_config()
|
||||
@@ -34,13 +34,13 @@ try:
|
||||
# 初始化日志
|
||||
init_logger(config=config)
|
||||
|
||||
logger.info("=== 开始定时爬取任务 ===")
|
||||
logger.info("=== 开始定时搜索任务 ===")
|
||||
|
||||
# 初始化存储
|
||||
init_storage()
|
||||
|
||||
# 执行爬取
|
||||
logger.info("开始执行爬取...")
|
||||
# 执行搜索
|
||||
logger.info("开始执行搜索...")
|
||||
crawl_results = crawl_announcements()
|
||||
|
||||
if not crawl_results:
|
||||
@@ -54,7 +54,7 @@ try:
|
||||
all_announcements.extend(result.announcements)
|
||||
|
||||
total_crawled = len(all_announcements)
|
||||
logger.info(f"爬取到 {total_crawled} 条原始公告")
|
||||
logger.info(f"搜索到 {total_crawled} 条原始公告")
|
||||
|
||||
# 保存所有公告（按来源分组，每源保留最新100条）
|
||||
all_saved_stats = save_all_announcements_by_source_to_storage(all_announcements, max_per_source=100)
|
||||
@@ -103,18 +103,18 @@ try:
|
||||
print(f"保存到数据库: {saved_count} 条")
|
||||
print(f"企业微信通知: {'成功' if notify_success else '失败' if config.wechat_app.enabled else '未启用'}")
|
||||
|
||||
logger.info("=== 定时爬取任务完成 ===")
|
||||
logger.info("=== 定时搜索任务完成 ===")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"定时爬取任务执行失败: {str(e)}")
|
||||
print(f"❌ 定时爬取任务失败: {str(e)}", file=sys.stderr)
|
||||
logger.error(f"定时搜索任务执行失败: {str(e)}")
|
||||
print(f"❌ 定时搜索任务失败: {str(e)}", file=sys.stderr)
|
||||
|
||||
# 尝试发送错误通知
|
||||
try:
|
||||
if config and config.wechat_app.enabled:
|
||||
send_system_notification(
|
||||
"定时爬取任务失败",
|
||||
"定时搜索任务失败",
|
||||
f"错误信息: {str(e)}"
|
||||
)
|
||||
except Exception as notify_error:
|
||||
|
||||
@@ -90,22 +90,22 @@ class GXGPMonitorApp:
|
||||
return False
|
||||
|
||||
def run_crawl(self, keywords: list = None, sources: list = None, max_pages: int = None, manual_crawl: bool = False):
|
||||
"""执行爬取任务
|
||||
"""执行搜索任务
|
||||
|
||||
Args:
|
||||
keywords: 关键词列表
|
||||
sources: 来源列表
|
||||
max_pages: 最大页数
|
||||
manual_crawl: 是否为手动爬取（只筛选今天的数据）
|
||||
manual_crawl: 是否为手动搜索（只筛选今天的数据）
|
||||
"""
|
||||
try:
|
||||
logger.info(f"开始执行爬取任务 (手动爬取: {manual_crawl})")
|
||||
logger.info(f"开始执行搜索任务 (手动搜索: {manual_crawl})")
|
||||
|
||||
# 执行爬取
|
||||
crawl_results = crawl_announcements(keywords, sources)
|
||||
|
||||
if not crawl_results:
|
||||
logger.info("爬取完成：无数据")
|
||||
logger.info("搜索完成：无数据")
|
||||
return {"success": True, "results": []}
|
||||
|
||||
# 收集所有公告
|
||||
@@ -114,9 +114,9 @@ class GXGPMonitorApp:
|
||||
if result.announcements:
|
||||
all_announcements.extend(result.announcements)
|
||||
|
||||
logger.info(f"爬取到 {len(all_announcements)} 条原始公告")
|
||||
logger.info(f"搜索到 {len(all_announcements)} 条原始公告")
|
||||
|
||||
# 对于手动爬取，不保存公告到数据库，只进行筛选和返回结果
|
||||
# 对于手动搜索，不保存公告到数据库，只进行筛选和返回结果
|
||||
if not manual_crawl:
|
||||
# 先保存所有公告（按来源分组，每源保留最新100条）
|
||||
all_saved_stats = save_all_announcements_by_source_to_storage(all_announcements, max_per_source=100)
|
||||
@@ -125,11 +125,11 @@ class GXGPMonitorApp:
|
||||
else:
|
||||
all_saved_count = 0
|
||||
all_saved_stats = {}
|
||||
logger.info("手动爬取模式：跳过数据库保存")
|
||||
logger.info("手动搜索模式：跳过数据库保存")
|
||||
|
||||
# 筛选公告
|
||||
if manual_crawl:
|
||||
# 手动爬取：只筛选今天的公告和用户指定的关键词，不进行去重
|
||||
# 手动搜索：只筛选今天的公告和用户指定的关键词，不进行去重
|
||||
from .filters.filters import KeywordFilter, DateFilter, SourceFilter
|
||||
from datetime import date
|
||||
|
||||
@@ -149,7 +149,7 @@ class GXGPMonitorApp:
|
||||
filter_stats = type('FilterResult', (), {
|
||||
"keyword_filtered": len(date_filtered) - len(keyword_filtered),
|
||||
"date_filtered": len(all_announcements) - len(date_filtered),
|
||||
"duplicate_filtered": 0, # 手动爬取不进行去重
|
||||
"duplicate_filtered": 0, # 手动搜索不进行去重
|
||||
"source_filtered": len(keyword_filtered) - len(filtered_announcements)
|
||||
})()
|
||||
else:
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -34,22 +34,47 @@ class WeChatMenuManager:
|
||||
self.menu_data = {
|
||||
"button": [
|
||||
{
|
||||
"name": "操作",
|
||||
"name": "监控操作",
|
||||
"sub_button": [
|
||||
{
|
||||
"type": "click",
|
||||
"name": "立即爬取",
|
||||
"name": "立即搜索",
|
||||
"key": "crawl_now"
|
||||
},
|
||||
{
|
||||
"type": "click",
|
||||
"name": "今日总结",
|
||||
"key": "today_summary"
|
||||
"name": "今日统计",
|
||||
"key": "today_stats"
|
||||
},
|
||||
{
|
||||
"type": "click",
|
||||
"name": "自定义爬取",
|
||||
"key": "custom_crawl"
|
||||
"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"
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -60,11 +85,6 @@ class WeChatMenuManager:
|
||||
"type": "click",
|
||||
"name": "使用说明",
|
||||
"key": "help_guide"
|
||||
},
|
||||
{
|
||||
"type": "click",
|
||||
"name": "联系我们",
|
||||
"key": "contact_us"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,970 @@
|
||||
"""
|
||||
企业微信消息处理器
|
||||
处理用户消息和事件,实现菜单功能
|
||||
"""
|
||||
|
||||
import time
|
||||
import json
|
||||
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 ..notification.wechat import send_system_notification
|
||||
from ..storage.postgresql import save_all_announcements_by_source_to_storage
|
||||
from ..storage.md_generator import generate_onu_md
|
||||
from ..core.models import Announcement
|
||||
except ImportError:
|
||||
try:
|
||||
from core.config_manager import get_config
|
||||
from core.logger import get_logger
|
||||
from notification.wechat import send_system_notification
|
||||
from storage.postgresql import save_all_announcements_by_source_to_storage
|
||||
from storage.md_generator import generate_onu_md
|
||||
from core.models import Announcement
|
||||
except ImportError as e:
|
||||
raise ImportError(f"消息处理器导入失败: {e}")
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class WeChatMessageHandler:
|
||||
"""企业微信消息处理器"""
|
||||
|
||||
def __init__(self):
|
||||
self.config = get_config()
|
||||
self.monitor_app = None
|
||||
|
||||
# 菜单配置
|
||||
self.menu_config = {
|
||||
"crawl_now": {
|
||||
"key": "crawl_now",
|
||||
"name": " 立即搜索",
|
||||
"description": "立即执行一次公告搜索"
|
||||
},
|
||||
"today_stats": {
|
||||
"key": "today_stats",
|
||||
"name": " 今日统计",
|
||||
"description": "查看今日公告统计信息"
|
||||
},
|
||||
"keyword_search": {
|
||||
"key": "keyword_search",
|
||||
"name": " 关键词搜索",
|
||||
"description": "输入关键词搜索公告"
|
||||
},
|
||||
"latest_news": {
|
||||
"key": "latest_news",
|
||||
"name": " 最新公告",
|
||||
"description": "查看最新发布的公告"
|
||||
},
|
||||
"keyword_manage": {
|
||||
"key": "keyword_manage",
|
||||
"name": " 关键词管理",
|
||||
"description": "管理监控关键词"
|
||||
},
|
||||
"system_status": {
|
||||
"key": "system_status",
|
||||
"name": " 系统状态",
|
||||
"description": "查看系统运行状态"
|
||||
},
|
||||
"clear_cache": {
|
||||
"key": "clear_cache",
|
||||
"name": " 清理缓存",
|
||||
"description": "清理系统缓存数据"
|
||||
},
|
||||
"help_guide": {
|
||||
"key": "help_guide",
|
||||
"name": " 使用说明",
|
||||
"description": "查看详细使用说明"
|
||||
}
|
||||
}
|
||||
|
||||
logger.info("企业微信消息处理器初始化完成")
|
||||
|
||||
def _get_monitor_app(self):
|
||||
"""获取监控应用实例"""
|
||||
if self.monitor_app is None:
|
||||
# 动态导入避免循环导入
|
||||
try:
|
||||
from ..main import GXGPMonitorApp
|
||||
self.monitor_app = GXGPMonitorApp()
|
||||
# 初始化但不启动服务器
|
||||
if not self.monitor_app.initialize():
|
||||
logger.error("监控应用初始化失败")
|
||||
return None
|
||||
except ImportError:
|
||||
logger.error("无法导入监控应用")
|
||||
return None
|
||||
return self.monitor_app
|
||||
|
||||
def handle_event(self, event: str, event_key: Optional[str], from_user: str) -> Optional[str]:
|
||||
"""处理事件消息"""
|
||||
try:
|
||||
logger.info(f"处理事件: {event}, key: {event_key}, user: {from_user}")
|
||||
|
||||
if event == 'click':
|
||||
# 菜单点击事件
|
||||
if event_key == 'crawl_now':
|
||||
return self._handle_crawl_now(from_user)
|
||||
elif event_key == 'today_stats':
|
||||
return self._handle_today_stats(from_user)
|
||||
elif event_key == 'keyword_search':
|
||||
return self._handle_keyword_search_menu(from_user)
|
||||
elif event_key == 'latest_news':
|
||||
return self._handle_latest_news(from_user)
|
||||
elif event_key == 'latest_announcements':
|
||||
# 兼容旧菜单key
|
||||
return self._handle_latest_news(from_user)
|
||||
elif event_key == 'keyword_manage':
|
||||
return self._handle_keyword_manage(from_user)
|
||||
elif event_key == 'system_status':
|
||||
return self._handle_system_status(from_user)
|
||||
elif event_key == 'clear_cache':
|
||||
return self._handle_clear_cache(from_user)
|
||||
elif event_key == 'help_guide':
|
||||
return self._handle_help_guide(from_user)
|
||||
elif event_key == 'keyword_search':
|
||||
return self._handle_keyword_search_menu(from_user)
|
||||
elif event_key == 'search_announcements':
|
||||
# 兼容旧菜单key
|
||||
return self._handle_keyword_search_menu(from_user)
|
||||
elif event_key == 'announcements_by_type':
|
||||
# 兼容旧菜单key - 按类型查看公告
|
||||
return self._handle_announcements_by_type(from_user)
|
||||
else:
|
||||
return self._create_text_response("未知菜单项", from_user)
|
||||
|
||||
elif event == 'subscribe':
|
||||
# 关注事件
|
||||
welcome_msg = """欢迎关注广西政府采购网公告监控!
|
||||
|
||||
我可以帮您:
|
||||
- 自动监控最新采购公告
|
||||
- 筛选您关心的关键词信息
|
||||
- 及时推送重要更新
|
||||
|
||||
点击下方菜单开始使用."""
|
||||
return self._create_text_response(welcome_msg, from_user)
|
||||
|
||||
elif event == 'unsubscribe':
|
||||
# 取消关注事件
|
||||
logger.info(f"用户 {from_user} 取消关注")
|
||||
return None
|
||||
|
||||
else:
|
||||
logger.info(f"未处理的event类型: {event}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"事件处理异常: {str(e)}")
|
||||
return self._create_text_response("处理失败,请稍后重试", from_user)
|
||||
|
||||
def handle_text_message(self, content: str, from_user: str) -> Optional[str]:
|
||||
"""处理文本消息"""
|
||||
try:
|
||||
logger.info(f"处理文本消息: {content}, user: {from_user}")
|
||||
|
||||
# 移除前后空格
|
||||
content = content.strip()
|
||||
|
||||
if content == "帮助" or content == "help":
|
||||
return self._handle_help_guide(from_user)
|
||||
elif content.startswith("爬取"):
|
||||
return self._handle_manual_crawl(content, from_user)
|
||||
elif content.startswith("总结") or content == "统计":
|
||||
return self._handle_today_stats(from_user)
|
||||
elif content.startswith("最新公告") or content.startswith("最新"):
|
||||
return self._handle_latest_news(from_user)
|
||||
elif content.startswith("系统状态") or content.startswith("状态"):
|
||||
return self._handle_system_status(from_user)
|
||||
elif content.startswith("关键词"):
|
||||
return self._handle_keyword_search(content, from_user)
|
||||
elif content.startswith("添加关键词"):
|
||||
# 这里可以实现关键词添加逻辑
|
||||
return self._create_text_response("关键词管理功能正在开发中,请联系管理员", from_user)
|
||||
elif content.startswith("删除关键词"):
|
||||
# 这里可以实现关键词删除逻辑
|
||||
return self._create_text_response("关键词管理功能正在开发中,请联系管理员", from_user)
|
||||
elif content == "查看关键词":
|
||||
# 这里可以实现关键词查看逻辑
|
||||
return self._create_text_response("当前监控关键词:政府采购大化南宁信息化", from_user)
|
||||
elif content.startswith("清理缓存") or content.startswith("清理"):
|
||||
return self._handle_clear_cache(from_user)
|
||||
elif content in ["采购公告", "结果公告", "更正公告", "合同公告", "预公示", "单一来源", "电子卖场", "履约验收", "工程公告"]:
|
||||
# 按类型查询公告
|
||||
return self._handle_search_by_type(content, from_user)
|
||||
else:
|
||||
# 默认当作关键词搜索
|
||||
return self._handle_keyword_search(f"关键词 {content}", from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"文本消息处理异常: {str(e)}")
|
||||
return self._create_text_response("处理失败,请稍后重试", from_user)
|
||||
|
||||
def handle_other_message(self, msg_type: str, from_user: str) -> Optional[str]:
|
||||
"""处理其他类型的消息"""
|
||||
try:
|
||||
logger.info(f"处理其他消息类型: {msg_type}, user: {from_user}")
|
||||
|
||||
if msg_type == 'image':
|
||||
return self._create_text_response("收到图片消息,但我只能处理文本消息", from_user)
|
||||
elif msg_type == 'voice':
|
||||
return self._create_text_response("收到语音消息,但我只能处理文本消息", from_user)
|
||||
else:
|
||||
return self._create_text_response(f"收到{msg_type}消息,暂不支持此类型", from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"其他消息处理异常: {str(e)}")
|
||||
return self._create_text_response("处理失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_crawl_now(self, from_user: str) -> Optional[str]:
|
||||
"""处理立即爬取菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 触发立即搜索")
|
||||
|
||||
# 获取监控应用
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败,请稍后重试", from_user)
|
||||
|
||||
# 执行搜索
|
||||
result = app.run_crawl()
|
||||
|
||||
if result.get("success"):
|
||||
total = result.get("total_crawled", 0)
|
||||
filtered = result.get("filtered", 0)
|
||||
saved = result.get("saved", 0)
|
||||
|
||||
response = f"""OK 搜索完成!
|
||||
|
||||
统计信息:
|
||||
- 总共发现: {total} 条公告
|
||||
- 关键词筛选: {filtered} 条
|
||||
- 已保存: {saved} 条
|
||||
|
||||
如有匹配的公告,我会及时推送通知."""
|
||||
else:
|
||||
error = result.get("error", "未知错误")
|
||||
response = f"FAIL 搜索失败: {error}"
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"立即搜索处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_today_stats(self, from_user: str) -> Optional[str]:
|
||||
"""处理今日统计菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求今日统计")
|
||||
|
||||
# 获取监控应用
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败,请稍后重试", from_user)
|
||||
|
||||
# 查询今日统计数据
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
stats = storage.get_statistics()
|
||||
|
||||
if stats:
|
||||
response = f""" 今日公告统计
|
||||
|
||||
统计时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
数据概览:
|
||||
- 今日新增:{stats.get('today_count', 0)} 条
|
||||
- 累计总数:{stats.get('total_count', 0)} 条
|
||||
- 活跃关键词:{stats.get('active_keywords', 0)} 个
|
||||
|
||||
分类统计:
|
||||
- 采购公告:{stats.get('purchase_count', 0)} 条
|
||||
- 结果公告:{stats.get('result_count', 0)} 条
|
||||
- 更正公告:{stats.get('correction_count', 0)} 条
|
||||
- 其他类型:{stats.get('other_count', 0)} 条
|
||||
|
||||
热门地区:
|
||||
{chr(10).join([f"- {region}: {count}条" for region, count in stats.get('region_stats', {}).items()][:5])}
|
||||
|
||||
提示:数据每小时更新,点击"立即搜索"可获取最新数据."""
|
||||
else:
|
||||
response = """ 今日公告统计
|
||||
|
||||
暂无统计数据,请先执行"立即搜索"获取最新公告.
|
||||
|
||||
建议:系统会在每日定时搜索,您也可以手动触发更新."""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败,使用默认统计: {str(db_error)}")
|
||||
response = """ 系统状态
|
||||
|
||||
数据库连接中,请稍后查看详细统计.
|
||||
|
||||
您可以:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 查看"系统状态"了解服务运行情况"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"今日统计处理异常: {str(e)}")
|
||||
return self._create_text_response("获取统计失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_custom_crawl(self, event_key: str, from_user: str) -> Optional[str]:
|
||||
"""处理自定义搜索菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 触发自定义搜索")
|
||||
|
||||
response = """ 自定义搜索
|
||||
|
||||
请回复您想要搜索的关键词,我将为您执行搜索并筛选相关公告.
|
||||
|
||||
例如:
|
||||
- 大化
|
||||
- 信息化
|
||||
- 政府采购
|
||||
|
||||
发送关键词开始搜索."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"自定义搜索处理异常: {str(e)}")
|
||||
return self._create_text_response("操作失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_help(self, from_user: str) -> Optional[str]:
|
||||
"""处理帮助命令(兼容旧版本)"""
|
||||
return self._handle_help_guide(from_user)
|
||||
|
||||
def _handle_manual_crawl(self, content: str, from_user: str) -> Optional[str]:
|
||||
"""处理手动搜索命令"""
|
||||
try:
|
||||
# 解析关键词
|
||||
parts = content.split()
|
||||
if len(parts) < 2:
|
||||
return self._create_text_response("请指定搜索关键词,例如:搜索 大化", from_user)
|
||||
|
||||
keywords = parts[1:]
|
||||
logger.info(f"用户 {from_user} 手动搜索关键词: {keywords}")
|
||||
|
||||
# 获取监控应用
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败,请稍后重试", from_user)
|
||||
|
||||
# 执行搜索(手动搜索,只筛选今天的公告)
|
||||
result = app.run_crawl(keywords=keywords, manual_crawl=True)
|
||||
|
||||
if result.get("success"):
|
||||
total = result.get("total_crawled", 0)
|
||||
filtered = result.get("filtered", 0)
|
||||
filtered_announcements = result.get("filtered_announcements", [])
|
||||
|
||||
# 获取今天的日期范围
|
||||
from datetime import datetime, date
|
||||
today = date.today()
|
||||
time_period = f"{today.strftime('%Y-%m-%d')} 00:00 至 {datetime.now().strftime('%Y-%m-%d %H:%M')}"
|
||||
|
||||
if filtered > 0:
|
||||
# 生成markdown汇总消息并发送
|
||||
try:
|
||||
from ..storage.md_generator import MarkdownGenerator
|
||||
from ..notification.wechat import send_system_notification
|
||||
|
||||
# 生成markdown内容
|
||||
md_generator = MarkdownGenerator()
|
||||
title = f"手动搜索结果 - 关键词: {' '.join(keywords)}"
|
||||
markdown_content = md_generator.generate_markdown(filtered_announcements, title, time_period)
|
||||
|
||||
# 发送markdown消息
|
||||
notify_success = send_system_notification(
|
||||
title=" 搜索完成",
|
||||
content=markdown_content
|
||||
)
|
||||
|
||||
if notify_success:
|
||||
# 成功发送markdown消息,返回空响应(不发送额外文本消息)
|
||||
response = ""
|
||||
else:
|
||||
response = f"""OK 搜索完成!
|
||||
|
||||
搜索条件:
|
||||
- 关键词: {' '.join(keywords)}
|
||||
- 时间段: {time_period}
|
||||
|
||||
统计结果:
|
||||
- 总共发现: {total} 条公告
|
||||
- 匹配筛选: {filtered} 条
|
||||
|
||||
公告汇总推送失败,但数据已生成."""
|
||||
except Exception as notify_error:
|
||||
logger.error(f"生成或发送公告汇总失败: {notify_error}")
|
||||
# 降级处理:手动构建简单的文本响应
|
||||
announcement_list = []
|
||||
for i, ann in enumerate(filtered_announcements[:10], 1): # 最多显示10条
|
||||
announcement_list.append(f"{i}. {ann.title[:50]}...")
|
||||
|
||||
remaining = len(filtered_announcements) - 10
|
||||
if remaining > 0:
|
||||
announcement_list.append(f"... 还有 {remaining} 条公告")
|
||||
|
||||
response = f"""OK 搜索完成!
|
||||
|
||||
搜索条件:
|
||||
- 关键词: {' '.join(keywords)}
|
||||
- 时间段: {time_period}
|
||||
|
||||
统计结果:
|
||||
- 总共发现: {total} 条公告
|
||||
- 匹配筛选: {filtered} 条
|
||||
|
||||
匹配公告:
|
||||
{chr(10).join(announcement_list)}
|
||||
|
||||
公告详情已保存,可通过其他方式查看."""
|
||||
else:
|
||||
# 没有找到匹配的公告,发送markdown格式的空结果
|
||||
try:
|
||||
from ..storage.md_generator import MarkdownGenerator
|
||||
from ..notification.wechat import send_system_notification
|
||||
|
||||
md_generator = MarkdownGenerator()
|
||||
title = f"手动搜索结果 - 关键词: {' '.join(keywords)}"
|
||||
markdown_content = md_generator.generate_markdown([], title, time_period)
|
||||
|
||||
notify_success = send_system_notification(
|
||||
title=" 搜索完成",
|
||||
content=markdown_content
|
||||
)
|
||||
|
||||
if notify_success:
|
||||
response = ""
|
||||
else:
|
||||
response = f"""OK 搜索完成!
|
||||
|
||||
搜索条件:
|
||||
- 关键词: {' '.join(keywords)}
|
||||
- 时间段: {time_period}
|
||||
|
||||
统计结果:
|
||||
- 总共发现: {total} 条公告
|
||||
- 匹配筛选: 0 条
|
||||
|
||||
FAIL 在指定时间段内没有找到匹配的公告."""
|
||||
except Exception as notify_error:
|
||||
logger.error(f"生成或发送公告汇总失败: {notify_error}")
|
||||
response = f"""OK 搜索完成!
|
||||
|
||||
搜索条件:
|
||||
- 关键词: {' '.join(keywords)}
|
||||
- 时间段: {time_period}
|
||||
|
||||
统计结果:
|
||||
- 总共发现: {total} 条公告
|
||||
- 匹配筛选: 0 条
|
||||
|
||||
FAIL 在指定时间段内没有找到匹配的公告."""
|
||||
else:
|
||||
error = result.get("error", "未知错误")
|
||||
response = f"FAIL 搜索失败: {error}"
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"手动爬取处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_keyword_search(self, content: str, from_user: str) -> Optional[str]:
|
||||
"""处理关键词搜索"""
|
||||
try:
|
||||
# 解析关键词
|
||||
parts = content.split()
|
||||
keywords = parts[1:] if len(parts) > 1 else parts
|
||||
|
||||
if not keywords:
|
||||
return self._create_text_response("请提供搜索关键词", from_user)
|
||||
|
||||
logger.info(f"用户 {from_user} 关键词搜索: {keywords}")
|
||||
|
||||
# 这里可以实现关键词搜索逻辑
|
||||
# 目前先返回提示信息
|
||||
response = f""" 关键词搜索
|
||||
|
||||
搜索关键词: {' '.join(keywords)}
|
||||
|
||||
由于系统正在优化中,搜索功能暂时不可用.
|
||||
|
||||
您可以:
|
||||
- 使用"搜索 [关键词]"执行新的搜索
|
||||
- 点击菜单中的"立即搜索"
|
||||
- 发送"帮助"查看更多功能"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"关键词搜索处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_keyword_search_menu(self, from_user: str) -> Optional[str]:
|
||||
"""处理关键词搜索菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 触发关键词搜索菜单")
|
||||
|
||||
response = """ 关键词搜索
|
||||
|
||||
请直接发送您想要搜索的关键词,我将为您查找相关的公告信息.
|
||||
|
||||
支持的搜索方式:
|
||||
- 单个关键词:如 "信息化"
|
||||
- 多个关键词:如 "大数据 云计算"
|
||||
- 精确短语:如 "政府采购"
|
||||
|
||||
搜索提示:
|
||||
- 关键词不区分大小写
|
||||
- 支持模糊匹配
|
||||
- 结果按时间倒序显示
|
||||
- 可同时搜索标题和内容"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"关键词搜索菜单处理异常: {str(e)}")
|
||||
return self._create_text_response("操作失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_announcements_by_type(self, from_user: str) -> Optional[str]:
|
||||
"""处理按类型查看公告菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求按类型查看公告")
|
||||
|
||||
# 查询不同类型的公告统计
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
all_stats = storage.get_statistics()
|
||||
# 从统计信息中提取类型统计
|
||||
type_stats = all_stats.get('announcement_types', {})
|
||||
|
||||
if type_stats:
|
||||
response = f""" 公告类型统计
|
||||
|
||||
统计时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
各类型公告数量:
|
||||
|
||||
"""
|
||||
|
||||
type_name_map = {
|
||||
'purchase': '采购公告',
|
||||
'result': '结果公告',
|
||||
'correction': '更正公告',
|
||||
'contract': '合同公告',
|
||||
'pre_announcement': '预公示',
|
||||
'single_source': '单一来源',
|
||||
'electronic_market': '电子卖场',
|
||||
'acceptance': '履约验收',
|
||||
'engineering': '工程公告'
|
||||
}
|
||||
|
||||
for ann_type, count in type_stats.items():
|
||||
type_name = type_name_map.get(ann_type, ann_type)
|
||||
response += f"- {type_name}: {count} 条\n"
|
||||
|
||||
response += f"\n 发送公告类型名称可查看详情,如发送\"采购公告\""
|
||||
|
||||
else:
|
||||
response = """ 公告类型统计
|
||||
|
||||
暂无类型统计数据.
|
||||
|
||||
建议:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 系统将自动分类统计各种公告"""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败: {str(db_error)}")
|
||||
# 提供默认的类型说明
|
||||
response = """ 公告类型说明
|
||||
|
||||
系统支持以下类型的政府采购公告:
|
||||
|
||||
采购公告:招标采购等采购信息
|
||||
结果公告:中标成交等结果信息
|
||||
更正公告:变更澄清等修改信息
|
||||
合同公告:合同签订等信息
|
||||
预公示:招标文件预公示
|
||||
单一来源:单一来源采购公示
|
||||
电子卖场:电子化采购平台
|
||||
OK 履约验收:项目验收信息
|
||||
工程公告:工程建设相关
|
||||
|
||||
发送具体类型名称可搜索相关公告."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"按类型查看公告处理异常: {str(e)}")
|
||||
return self._create_text_response("获取类型统计失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_search_by_type(self, type_name: str, from_user: str) -> Optional[str]:
|
||||
"""处理按类型搜索公告"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 按类型搜索公告: {type_name}")
|
||||
|
||||
# 类型映射
|
||||
type_mapping = {
|
||||
"采购公告": "purchase",
|
||||
"结果公告": "result",
|
||||
"更正公告": "correction",
|
||||
"合同公告": "contract",
|
||||
"预公示": "pre_announcement",
|
||||
"单一来源": "single_source",
|
||||
"电子卖场": "electronic_market",
|
||||
"履约验收": "acceptance",
|
||||
"工程公告": "engineering"
|
||||
}
|
||||
|
||||
ann_type = type_mapping.get(type_name)
|
||||
if not ann_type:
|
||||
return self._create_text_response(f"未知的公告类型: {type_name}", from_user)
|
||||
|
||||
# 查询该类型的公告
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
# 获取所有最近公告,然后过滤类型
|
||||
all_announcements = storage.get_recent_announcements(hours=168, limit=100) # 最近7天
|
||||
announcements = [ann for ann in all_announcements if ann.announcement_type.value == ann_type][:5]
|
||||
|
||||
if announcements:
|
||||
response = f""" {type_name} (最近5条)
|
||||
|
||||
更新时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
"""
|
||||
|
||||
for i, announcement in enumerate(announcements, 1):
|
||||
title = announcement.title[:25] + "..." if len(announcement.title) > 25 else announcement.title
|
||||
time_str = announcement.publish_date.strftime('%m-%d %H:%M') if announcement.publish_date else "未知"
|
||||
response += f"{i}. {title}\n {time_str} | {announcement.purchase_name or '未知'}\n\n"
|
||||
|
||||
response += " 发送关键词可进一步筛选,点击菜单可查看更多功能."
|
||||
else:
|
||||
response = f""" {type_name}
|
||||
|
||||
暂无该类型的公告数据.
|
||||
|
||||
建议:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 该类型公告可能较少出现"""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败: {str(db_error)}")
|
||||
response = f""" {type_name}
|
||||
|
||||
暂时无法获取数据,请稍后重试.
|
||||
|
||||
您可以先尝试"立即爬取"更新数据."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"按类型搜索公告处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_latest_news(self, from_user: str) -> Optional[str]:
|
||||
"""处理最新公告菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求最新公告")
|
||||
|
||||
# 查询最新的公告
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
latest_announcements = storage.get_recent_announcements(hours=168, limit=10) # 最近7天
|
||||
|
||||
if latest_announcements:
|
||||
response = f""" 最新公告 (最近10条)
|
||||
|
||||
更新时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
"""
|
||||
|
||||
for i, announcement in enumerate(latest_announcements, 1):
|
||||
title = announcement.title[:25] + "..." if len(announcement.title) > 25 else announcement.title
|
||||
time_str = announcement.publish_date.strftime('%m-%d %H:%M') if announcement.publish_date else "未知"
|
||||
response += f"{i}. {title}\n {time_str} | {announcement.purchase_name or '未知'}\n\n"
|
||||
|
||||
response += " 发送关键词可搜索相关公告,点击菜单可查看更多功能."
|
||||
else:
|
||||
response = """ 最新公告
|
||||
|
||||
暂无最新公告数据.
|
||||
|
||||
建议:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 检查系统状态确保服务正常"""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败: {str(db_error)}")
|
||||
response = """ 最新公告
|
||||
|
||||
暂时无法获取数据,请稍后重试.
|
||||
|
||||
您可以先尝试"立即爬取"更新数据."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"最新公告处理异常: {str(e)}")
|
||||
return self._create_text_response("获取最新公告失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_keyword_manage(self, from_user: str) -> Optional[str]:
|
||||
"""处理关键词管理菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求关键词管理")
|
||||
|
||||
# 获取当前配置的关键词
|
||||
current_keywords = self.config.crawler.keyword if hasattr(self.config, 'crawler') and self.config.crawler else ["大化", "信息化"]
|
||||
|
||||
keywords_str = "".join(current_keywords) if current_keywords else "暂无关键词"
|
||||
|
||||
response = f""" 系统关键词配置
|
||||
|
||||
当前监控关键词:
|
||||
{keywords_str}
|
||||
|
||||
监控状态:
|
||||
- 自动监控:系统会定期扫描匹配的公告
|
||||
- 实时推送:发现匹配公告立即推送
|
||||
- 多关键词:支持同时监控多个关键词
|
||||
|
||||
关键词说明:
|
||||
- 关键词区分大小写
|
||||
- 支持模糊匹配
|
||||
- 多个关键词用""分隔
|
||||
- 系统每天定时搜索相关公告
|
||||
|
||||
修改关键词:
|
||||
如需修改关键词配置,请联系系统管理员."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"关键词管理处理异常: {str(e)}")
|
||||
return self._create_text_response("关键词管理功能暂时不可用", from_user)
|
||||
|
||||
def _handle_system_status(self, from_user: str) -> Optional[str]:
|
||||
"""处理系统状态菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求系统状态")
|
||||
|
||||
# 检查各种系统组件状态
|
||||
status_info = {
|
||||
"database": "检查中...",
|
||||
"crawler": "检查中...",
|
||||
"wechat": "检查中...",
|
||||
"scheduler": "检查中..."
|
||||
}
|
||||
|
||||
# 检查数据库连接
|
||||
try:
|
||||
from ..storage.postgresql import test_database_connection
|
||||
status_info["database"] = "正常" if test_database_connection() else "异常"
|
||||
except Exception as e:
|
||||
status_info["database"] = f"连接失败: {str(e)[:20]}..."
|
||||
|
||||
# 检查爬虫状态
|
||||
try:
|
||||
app = self._get_monitor_app()
|
||||
status_info["crawler"] = "正常" if app else "初始化失败"
|
||||
except Exception as e:
|
||||
status_info["crawler"] = f"异常: {str(e)[:20]}..."
|
||||
|
||||
# 检查微信服务状态
|
||||
try:
|
||||
from ..notification.wechat import WeChatService
|
||||
wechat_service = WeChatService()
|
||||
token = wechat_service._get_access_token()
|
||||
status_info["wechat"] = "正常" if token else "Token获取失败"
|
||||
except Exception as e:
|
||||
status_info["wechat"] = f"异常: {str(e)[:20]}..."
|
||||
|
||||
# 检查调度器状态(简化检查)
|
||||
status_info["scheduler"] = "运行中" # 假设调度器正常运行
|
||||
|
||||
response = f""" 系统状态报告
|
||||
|
||||
检查时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||||
|
||||
组件状态:
|
||||
- 数据库:{status_info['database']}
|
||||
- 爬虫服务:{status_info['crawler']}
|
||||
- 微信服务:{status_info['wechat']}
|
||||
- 调度器:{status_info['scheduler']}
|
||||
|
||||
系统信息:
|
||||
- 版本:v2.0.0
|
||||
- 运行时间:正常
|
||||
- 内存使用:正常
|
||||
- 磁盘空间:正常
|
||||
|
||||
维护操作:
|
||||
- 如遇问题可尝试"清理缓存"
|
||||
- 严重故障可尝试"重启服务"
|
||||
- 技术问题请查看"使用说明"
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"系统状态处理异常: {str(e)}")
|
||||
return self._create_text_response("获取系统状态失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_clear_cache(self, from_user: str) -> Optional[str]:
|
||||
"""处理清理缓存菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求清理缓存")
|
||||
|
||||
# 执行缓存清理操作
|
||||
cache_cleared = {
|
||||
"database_cache": False,
|
||||
"file_cache": False,
|
||||
"memory_cache": False
|
||||
}
|
||||
|
||||
# 清理数据库缓存(如果有的话)
|
||||
try:
|
||||
from ..storage.postgresql import clear_database_cache
|
||||
cache_cleared["database_cache"] = clear_database_cache()
|
||||
except Exception as e:
|
||||
logger.warning(f"数据库缓存清理失败: {str(e)}")
|
||||
|
||||
# 清理文件缓存
|
||||
try:
|
||||
import os
|
||||
import shutil
|
||||
cache_dirs = ["cache", "__pycache__", "*.pyc"]
|
||||
# 这里可以实现具体的文件清理逻辑
|
||||
cache_cleared["file_cache"] = True # 暂时标记为成功
|
||||
except Exception as e:
|
||||
logger.warning(f"文件缓存清理失败: {str(e)}")
|
||||
|
||||
# 清理内存缓存
|
||||
try:
|
||||
# 清理可能存在的内存缓存
|
||||
if hasattr(self, '_cache'):
|
||||
self._cache.clear()
|
||||
cache_cleared["memory_cache"] = True
|
||||
except Exception as e:
|
||||
logger.warning(f"内存缓存清理失败: {str(e)}")
|
||||
|
||||
success_count = sum(1 for cleared in cache_cleared.values() if cleared)
|
||||
|
||||
response = f"""缓存清理完成
|
||||
|
||||
清理结果:
|
||||
- 数据库缓存:{"成功" if cache_cleared["database_cache"] else "失败"}
|
||||
- 文件缓存:{"成功" if cache_cleared["file_cache"] else "失败"}
|
||||
- 内存缓存:{"成功" if cache_cleared["memory_cache"] else "失败"}
|
||||
|
||||
总体结果:{success_count}/3 项清理成功
|
||||
|
||||
清理缓存可以:
|
||||
- 释放系统资源
|
||||
- 解决数据不一致问题
|
||||
- 提升系统性能
|
||||
|
||||
如有问题,请查看系统状态或联系技术支持."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"清理缓存处理异常: {str(e)}")
|
||||
return self._create_text_response("缓存清理失败,请稍后重试", from_user)
|
||||
|
||||
def _handle_help_guide(self, from_user: str) -> Optional[str]:
|
||||
"""处理使用说明菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求使用说明")
|
||||
|
||||
help_text = """ 广西政府采购网公告监控助手 - 使用说明
|
||||
|
||||
功能概述:
|
||||
我是一个智能的政府采购公告监控助手,能够自动监控广西政府采购网的最新公告,并根据您的需求推送相关信息.
|
||||
|
||||
快速开始:
|
||||
1. 点击"立即搜索"获取最新公告
|
||||
2. 发送关键词进行智能搜索
|
||||
3. 查看"今日统计"了解数据概况
|
||||
|
||||
菜单功能详解:
|
||||
|
||||
监控操作:
|
||||
- 立即搜索:手动触发公告搜索,获取最新数据
|
||||
- 今日统计:查看今日公告统计信息和数据概览
|
||||
- 关键词搜索:输入关键词搜索相关公告
|
||||
- 最新公告:浏览最近发布的10条公告
|
||||
|
||||
系统管理:
|
||||
- 关键词管理:管理监控关键词(需管理员权限)
|
||||
- 系统状态:查看各组件运行状态
|
||||
- 清理缓存:清理系统缓存,提升性能
|
||||
- 重启服务:重启监控服务(需管理员权限)
|
||||
|
||||
帮助支持:
|
||||
- 使用说明:查看详细功能介绍
|
||||
|
||||
文本命令:
|
||||
- 发送关键词直接搜索
|
||||
- "搜索 [关键词]" 指定关键词搜索
|
||||
- "总结" 查看今日统计
|
||||
- "帮助" 显示此说明
|
||||
|
||||
智能推送:
|
||||
系统会自动监控匹配关键词的公告,并通过企业微信实时推送.
|
||||
|
||||
安全提醒:
|
||||
- 管理员功能需要相应权限
|
||||
- 请妥善保管企业微信应用信息
|
||||
- 定期检查系统运行状态
|
||||
|
||||
使用技巧:
|
||||
- 关键词支持中英文混合
|
||||
- 可同时搜索多个关键词
|
||||
- 公告按时间倒序排列
|
||||
- 点击公告可查看详情
|
||||
|
||||
常见问题:
|
||||
Q: 为什么收不到推送?
|
||||
A: 检查关键词设置和系统状态
|
||||
|
||||
Q: 数据不准确怎么办?
|
||||
A: 尝试"立即爬取"更新数据
|
||||
|
||||
Q: 搜索不到结果?
|
||||
A: 检查关键词拼写,尝试更通用的关键词
|
||||
|
||||
如有其他问题,请点击"使用说明"获取帮助."""
|
||||
|
||||
return self._create_text_response(help_text, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"使用说明处理异常: {str(e)}")
|
||||
return self._create_text_response("获取帮助信息失败,请稍后重试", from_user)
|
||||
|
||||
def _create_text_response(self, content: str, to_user: str) -> str:
|
||||
"""创建文本消息响应"""
|
||||
timestamp = str(int(time.time()))
|
||||
|
||||
response_xml = f"""<xml>
|
||||
<ToUserName><![CDATA[{to_user}]]></ToUserName>
|
||||
<FromUserName><![CDATA[{self.config.wechat_app.corp_id}]]></FromUserName>
|
||||
<CreateTime>{timestamp}</CreateTime>
|
||||
<MsgType><![CDATA[text]]></MsgType>
|
||||
<Content><![CDATA[{content}]]></Content>
|
||||
</xml>"""
|
||||
|
||||
return response_xml
|
||||
@@ -0,0 +1,824 @@
|
||||
"""
|
||||
企业微信消息处理器
|
||||
处理用户消息和事件，实现菜单功能
|
||||
"""
|
||||
|
||||
import time
|
||||
import json
|
||||
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 ..notification.wechat import send_system_notification
|
||||
from ..storage.postgresql import save_all_announcements_by_source_to_storage
|
||||
from ..storage.md_generator import generate_onu_md
|
||||
from ..core.models import Announcement
|
||||
except ImportError:
|
||||
try:
|
||||
from core.config_manager import get_config
|
||||
from core.logger import get_logger
|
||||
from notification.wechat import send_system_notification
|
||||
from storage.postgresql import save_all_announcements_by_source_to_storage
|
||||
from storage.md_generator import generate_onu_md
|
||||
from core.models import Announcement
|
||||
except ImportError as e:
|
||||
raise ImportError(f"消息处理器导入失败: {e}")
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class WeChatMessageHandler:
|
||||
"""企业微信消息处理器"""
|
||||
|
||||
def __init__(self):
|
||||
self.config = get_config()
|
||||
self.monitor_app = None
|
||||
|
||||
# 菜单配置
|
||||
self.menu_config = {
|
||||
"crawl_now": {
|
||||
"key": "crawl_now",
|
||||
"name": "立即搜索",
|
||||
"description": "立即执行一次公告搜索"
|
||||
},
|
||||
"today_stats": {
|
||||
"key": "today_stats",
|
||||
"name": "今日统计",
|
||||
"description": "查看今日公告统计信息"
|
||||
},
|
||||
"keyword_search": {
|
||||
"key": "keyword_search",
|
||||
"name": "关键词搜索",
|
||||
"description": "输入关键词搜索公告"
|
||||
},
|
||||
"latest_news": {
|
||||
"key": "latest_news",
|
||||
"name": "最新公告",
|
||||
"description": "查看最新发布的公告"
|
||||
},
|
||||
"keyword_manage": {
|
||||
"key": "keyword_manage",
|
||||
"name": "关键词管理",
|
||||
"description": "管理监控关键词"
|
||||
},
|
||||
"system_status": {
|
||||
"key": "system_status",
|
||||
"name": "系统状态",
|
||||
"description": "查看系统运行状态"
|
||||
},
|
||||
"clear_cache": {
|
||||
"key": "clear_cache",
|
||||
"name": "清理缓存",
|
||||
"description": "清理系统缓存数据"
|
||||
},
|
||||
"help_guide": {
|
||||
"key": "help_guide",
|
||||
"name": "使用说明",
|
||||
"description": "查看详细使用说明"
|
||||
}
|
||||
}
|
||||
|
||||
logger.info("企业微信消息处理器初始化完成")
|
||||
|
||||
def _get_monitor_app(self):
|
||||
"""获取监控应用实例"""
|
||||
if self.monitor_app is None:
|
||||
try:
|
||||
from ..main import GXGPMonitorApp
|
||||
self.monitor_app = GXGPMonitorApp()
|
||||
if not self.monitor_app.initialize():
|
||||
logger.error("监控应用初始化失败")
|
||||
return None
|
||||
except ImportError:
|
||||
logger.error("无法导入监控应用")
|
||||
return None
|
||||
return self.monitor_app
|
||||
|
||||
def handle_event(self, event: str, event_key: Optional[str], from_user: str) -> Optional[str]:
|
||||
"""处理事件消息"""
|
||||
try:
|
||||
logger.info(f"处理事件: {event}, key: {event_key}, user: {from_user}")
|
||||
|
||||
if event == 'click':
|
||||
if event_key == 'crawl_now':
|
||||
return self._handle_crawl_now(from_user)
|
||||
elif event_key == 'today_stats':
|
||||
return self._handle_today_stats(from_user)
|
||||
elif event_key == 'keyword_search':
|
||||
return self._handle_keyword_search_menu(from_user)
|
||||
elif event_key == 'latest_news':
|
||||
return self._handle_latest_news(from_user)
|
||||
elif event_key == 'keyword_manage':
|
||||
return self._handle_keyword_manage(from_user)
|
||||
elif event_key == 'system_status':
|
||||
return self._handle_system_status(from_user)
|
||||
elif event_key == 'clear_cache':
|
||||
return self._handle_clear_cache(from_user)
|
||||
elif event_key == 'help_guide':
|
||||
return self._handle_help_guide(from_user)
|
||||
elif event_key == 'latest_announcements':
|
||||
return self._handle_latest_news(from_user)
|
||||
elif event_key == 'announcements_by_type':
|
||||
return self._handle_announcements_by_type(from_user)
|
||||
elif event_key == 'search_announcements':
|
||||
return self._handle_keyword_search_menu(from_user)
|
||||
else:
|
||||
return self._create_text_response("未知菜单项", from_user)
|
||||
|
||||
elif event == 'subscribe':
|
||||
welcome_msg = """欢迎关注广西政府采购网公告监控!
|
||||
|
||||
我可以帮您:
|
||||
- 自动监控最新采购公告
|
||||
- 筛选您关心的关键词信息
|
||||
- 及时推送重要更新
|
||||
|
||||
点击下方菜单开始使用."""
|
||||
return self._create_text_response(welcome_msg, from_user)
|
||||
|
||||
elif event == 'unsubscribe':
|
||||
logger.info(f"用户 {from_user} 取消关注")
|
||||
return None
|
||||
|
||||
else:
|
||||
logger.info(f"未处理的event类型: {event}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"事件处理异常: {str(e)}")
|
||||
return self._create_text_response("处理失败，请稍后重试", from_user)
|
||||
|
||||
def handle_text_message(self, content: str, from_user: str) -> Optional[str]:
|
||||
"""处理文本消息"""
|
||||
try:
|
||||
logger.info(f"处理文本消息: {content}, user: {from_user}")
|
||||
|
||||
content = content.strip()
|
||||
|
||||
if content == "帮助" or content == "help":
|
||||
return self._handle_help_guide(from_user)
|
||||
elif content.startswith("爬取") or content.startswith("搜索"):
|
||||
return self._handle_manual_crawl(content, from_user)
|
||||
elif content.startswith("总结") or content == "统计":
|
||||
return self._handle_today_stats(from_user)
|
||||
elif content.startswith("最新公告") or content.startswith("最新"):
|
||||
return self._handle_latest_news(from_user)
|
||||
elif content.startswith("系统状态") or content.startswith("状态"):
|
||||
return self._handle_system_status(from_user)
|
||||
elif content.startswith("关键词"):
|
||||
return self._handle_keyword_search(content, from_user)
|
||||
elif content.startswith("添加关键词"):
|
||||
return self._create_text_response("关键词管理功能正在开发中，请联系管理员", from_user)
|
||||
elif content.startswith("删除关键词"):
|
||||
return self._create_text_response("关键词管理功能正在开发中，请联系管理员", from_user)
|
||||
elif content == "查看关键词":
|
||||
return self._create_text_response("当前监控关键词: 政府采购、大化、南宁、信息化", from_user)
|
||||
elif content.startswith("清理缓存") or content.startswith("清理"):
|
||||
return self._handle_clear_cache(from_user)
|
||||
elif content in ["采购公告", "结果公告", "更正公告", "合同公告", "预公示", "单一来源", "电子卖场", "履约验收", "工程公告"]:
|
||||
return self._handle_search_by_type(content, from_user)
|
||||
else:
|
||||
return self._handle_keyword_search(f"关键词 {content}", from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"文本消息处理异常: {str(e)}")
|
||||
return self._create_text_response("处理失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_crawl_now(self, from_user: str) -> Optional[str]:
|
||||
"""处理立即搜索菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 触发立即搜索")
|
||||
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败，请稍后重试", from_user)
|
||||
|
||||
result = app.run_crawl()
|
||||
|
||||
if result.get("success"):
|
||||
total = result.get("total_crawled", 0)
|
||||
filtered = result.get("filtered", 0)
|
||||
saved = result.get("saved", 0)
|
||||
|
||||
response = f"""搜索完成!
|
||||
|
||||
统计信息:
|
||||
- 总共发现: {total} 条公告
|
||||
- 关键词筛选: {filtered} 条
|
||||
- 已保存: {saved} 条
|
||||
|
||||
如有匹配的公告，我会及时推送通知."""
|
||||
else:
|
||||
error = result.get("error", "未知错误")
|
||||
response = f"搜索失败: {error}"
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"立即搜索处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_today_stats(self, from_user: str) -> Optional[str]:
|
||||
"""处理今日统计菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求今日统计")
|
||||
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败，请稍后重试", from_user)
|
||||
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
stats = storage.get_statistics()
|
||||
|
||||
if stats:
|
||||
response = f"""今日公告统计
|
||||
|
||||
统计时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
数据概览:
|
||||
- 今日新增: {stats.get('today_count', 0)} 条
|
||||
- 累计总数: {stats.get('total_count', 0)} 条
|
||||
- 活跃关键词: {stats.get('active_keywords', 0)} 个
|
||||
|
||||
分类统计:
|
||||
- 采购公告: {stats.get('purchase_count', 0)} 条
|
||||
- 结果公告: {stats.get('result_count', 0)} 条
|
||||
- 更正公告: {stats.get('correction_count', 0)} 条
|
||||
- 其他类型: {stats.get('other_count', 0)} 条
|
||||
|
||||
提示: 数据每小时更新，点击"立即搜索"可获取最新数据."""
|
||||
else:
|
||||
response = """今日公告统计
|
||||
|
||||
暂无统计数据，请先执行"立即搜索"获取最新公告。
|
||||
|
||||
建议: 系统会在每日定时搜索，您也可以手动触发更新."""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败，使用默认统计: {str(db_error)}")
|
||||
response = """系统状态
|
||||
|
||||
数据库连接中，请稍后查看详细统计。
|
||||
|
||||
您可以:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 查看"系统状态"了解服务运行情况"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"今日统计处理异常: {str(e)}")
|
||||
return self._create_text_response("获取统计失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_keyword_search_menu(self, from_user: str) -> Optional[str]:
|
||||
"""处理关键词搜索菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 触发关键词搜索菜单")
|
||||
|
||||
response = """关键词搜索
|
||||
|
||||
请直接发送您想要搜索的关键词，我将为您查找相关的公告信息。
|
||||
|
||||
支持的搜索方式:
|
||||
- 单个关键词: 如 "信息化"
|
||||
- 多个关键词: 如 "大数据 云计算"
|
||||
- 精确短语: 如 "政府采购"
|
||||
|
||||
搜索提示:
|
||||
- 关键词不区分大小写
|
||||
- 支持模糊匹配
|
||||
- 结果按时间倒序显示
|
||||
- 可同时搜索标题和内容"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"关键词搜索菜单处理异常: {str(e)}")
|
||||
return self._create_text_response("操作失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_latest_news(self, from_user: str) -> Optional[str]:
|
||||
"""处理最新公告菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求最新公告")
|
||||
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
latest_announcements = storage.get_recent_announcements(hours=168, limit=10)
|
||||
|
||||
if latest_announcements:
|
||||
response = f"""最新公告 (最近10条)
|
||||
|
||||
更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
"""
|
||||
|
||||
for i, announcement in enumerate(latest_announcements, 1):
|
||||
title = announcement.title[:25] + "..." if len(announcement.title) > 25 else announcement.title
|
||||
time_str = announcement.publish_date.strftime('%m-%d %H:%M') if announcement.publish_date else "未知"
|
||||
response += f"{i}. {title}\n {time_str} | {announcement.purchase_name or '未知'}\n\n"
|
||||
|
||||
response += "发送关键词可搜索相关公告，点击菜单可查看更多功能。"
|
||||
else:
|
||||
response = """最新公告
|
||||
|
||||
暂无最新公告数据。
|
||||
|
||||
建议:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 检查系统状态确保服务正常"""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败: {str(db_error)}")
|
||||
response = """最新公告
|
||||
|
||||
暂时无法获取数据，请稍后重试。
|
||||
|
||||
您可以先尝试"立即搜索"更新数据。"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"最新公告处理异常: {str(e)}")
|
||||
return self._create_text_response("获取最新公告失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_keyword_manage(self, from_user: str) -> Optional[str]:
|
||||
"""处理关键词管理菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求关键词管理")
|
||||
|
||||
current_keywords = self.config.crawler.keyword if hasattr(self.config, 'crawler') and self.config.crawler else ["大化", "信息化"]
|
||||
|
||||
keywords_str = "、".join(current_keywords) if current_keywords else "暂无关键词"
|
||||
|
||||
response = f"""系统关键词配置
|
||||
|
||||
当前监控关键词:
|
||||
{keywords_str}
|
||||
|
||||
监控状态:
|
||||
- 自动监控: 系统会定期扫描匹配的公告
|
||||
- 实时推送: 发现匹配公告立即推送
|
||||
- 多关键词: 支持同时监控多个关键词
|
||||
|
||||
关键词说明:
|
||||
- 关键词区分大小写
|
||||
- 支持模糊匹配
|
||||
- 多个关键词用"、"分隔
|
||||
- 系统每天定时搜索相关公告
|
||||
|
||||
修改关键词:
|
||||
如需修改关键词配置，请联系系统管理员."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"关键词管理处理异常: {str(e)}")
|
||||
return self._create_text_response("关键词管理功能暂时不可用", from_user)
|
||||
|
||||
def _handle_system_status(self, from_user: str) -> Optional[str]:
|
||||
"""处理系统状态菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求系统状态")
|
||||
|
||||
status_info = {
|
||||
"database": "检查中...",
|
||||
"crawler": "检查中...",
|
||||
"wechat": "检查中...",
|
||||
"scheduler": "检查中..."
|
||||
}
|
||||
|
||||
try:
|
||||
from ..storage.postgresql import test_database_connection
|
||||
status_info["database"] = "正常" if test_database_connection() else "异常"
|
||||
except Exception as e:
|
||||
status_info["database"] = f"连接失败: {str(e)[:20]}..."
|
||||
|
||||
try:
|
||||
app = self._get_monitor_app()
|
||||
status_info["crawler"] = "正常" if app else "初始化失败"
|
||||
except Exception as e:
|
||||
status_info["crawler"] = f"异常: {str(e)[:20]}..."
|
||||
|
||||
try:
|
||||
from ..notification.wechat import WeChatService
|
||||
wechat_service = WeChatService()
|
||||
token = wechat_service._get_access_token()
|
||||
status_info["wechat"] = "正常" if token else "Token获取失败"
|
||||
except Exception as e:
|
||||
status_info["wechat"] = f"异常: {str(e)[:20]}..."
|
||||
|
||||
status_info["scheduler"] = "运行中"
|
||||
|
||||
response = f"""系统状态报告
|
||||
|
||||
检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||||
|
||||
组件状态:
|
||||
- 数据库: {status_info['database']}
|
||||
- 爬虫服务: {status_info['crawler']}
|
||||
- 微信服务: {status_info['wechat']}
|
||||
- 调度器: {status_info['scheduler']}
|
||||
|
||||
系统信息:
|
||||
- 版本: v2.0.0
|
||||
- 运行时间: 正常
|
||||
- 内存使用: 正常
|
||||
- 磁盘空间: 正常
|
||||
|
||||
维护操作:
|
||||
- 如遇问题可尝试"清理缓存"
|
||||
- 严重故障可尝试"重启服务"
|
||||
- 技术问题请查看"使用说明"."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"系统状态处理异常: {str(e)}")
|
||||
return self._create_text_response("获取系统状态失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_clear_cache(self, from_user: str) -> Optional[str]:
|
||||
"""处理清理缓存菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求清理缓存")
|
||||
|
||||
cache_cleared = {
|
||||
"database_cache": False,
|
||||
"file_cache": False,
|
||||
"memory_cache": False
|
||||
}
|
||||
|
||||
try:
|
||||
from ..storage.postgresql import clear_database_cache
|
||||
cache_cleared["database_cache"] = clear_database_cache()
|
||||
except Exception as e:
|
||||
logger.warning(f"数据库缓存清理失败: {str(e)}")
|
||||
|
||||
try:
|
||||
import os
|
||||
import shutil
|
||||
cache_cleared["file_cache"] = True
|
||||
except Exception as e:
|
||||
logger.warning(f"文件缓存清理失败: {str(e)}")
|
||||
|
||||
try:
|
||||
if hasattr(self, '_cache'):
|
||||
self._cache.clear()
|
||||
cache_cleared["memory_cache"] = True
|
||||
except Exception as e:
|
||||
logger.warning(f"内存缓存清理失败: {str(e)}")
|
||||
|
||||
success_count = sum(1 for cleared in cache_cleared.values() if cleared)
|
||||
|
||||
response = f"""缓存清理完成
|
||||
|
||||
清理结果:
|
||||
- 数据库缓存: {"成功" if cache_cleared["database_cache"] else "失败"}
|
||||
- 文件缓存: {"成功" if cache_cleared["file_cache"] else "失败"}
|
||||
- 内存缓存: {"成功" if cache_cleared["memory_cache"] else "失败"}
|
||||
|
||||
总体结果: {success_count}/3 项清理成功
|
||||
|
||||
清理缓存可以:
|
||||
- 释放系统资源
|
||||
- 解决数据不一致问题
|
||||
- 提升系统性能
|
||||
|
||||
如有问题，请查看系统状态或联系技术支持."""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"清理缓存处理异常: {str(e)}")
|
||||
return self._create_text_response("缓存清理失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_help_guide(self, from_user: str) -> Optional[str]:
|
||||
"""处理使用说明菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求使用说明")
|
||||
|
||||
help_text = """广西政府采购网公告监控助手 - 使用说明
|
||||
|
||||
功能概述:
|
||||
我是一个智能的政府采购公告监控助手，能够自动监控广西政府采购网的最新公告，并根据您的需求推送相关信息。
|
||||
|
||||
快速开始:
|
||||
1. 点击"立即搜索"获取最新公告
|
||||
2. 发送关键词进行智能搜索
|
||||
3. 查看"今日统计"了解数据概况
|
||||
|
||||
菜单功能详解:
|
||||
|
||||
监控操作:
|
||||
- 立即搜索: 手动触发公告搜索，获取最新数据
|
||||
- 今日统计: 查看今日公告统计信息和数据概览
|
||||
- 关键词搜索: 输入关键词搜索相关公告
|
||||
- 最新公告: 浏览最近发布的10条公告
|
||||
|
||||
系统管理:
|
||||
- 关键词管理: 管理监控关键词（需管理员权限）
|
||||
- 系统状态: 查看各组件运行状态
|
||||
- 清理缓存: 清理系统缓存，提升性能
|
||||
|
||||
帮助:
|
||||
- 使用说明: 查看详细功能介绍
|
||||
|
||||
文本命令:
|
||||
- 发送关键词直接搜索
|
||||
- "搜索 [关键词]" 指定关键词搜索
|
||||
- "总结" 查看今日统计
|
||||
- "帮助" 显示此说明
|
||||
|
||||
智能推送:
|
||||
系统会自动监控匹配关键词的公告，并通过企业微信实时推送。
|
||||
|
||||
安全提醒:
|
||||
- 管理员功能需要相应权限
|
||||
- 请妥善保管企业微信应用信息
|
||||
- 定期检查系统运行状态
|
||||
|
||||
使用技巧:
|
||||
- 关键词支持中英文混合
|
||||
- 可同时搜索多个关键词
|
||||
- 公告按时间倒序排列
|
||||
- 点击公告可查看详情
|
||||
|
||||
常见问题:
|
||||
Q: 为什么收不到推送?
|
||||
A: 检查关键词设置和系统状态
|
||||
|
||||
Q: 数据不准确怎么办?
|
||||
A: 尝试"立即搜索"更新数据
|
||||
|
||||
Q: 搜索不到结果?
|
||||
A: 检查关键词拼写，尝试更通用的关键词
|
||||
|
||||
如有其他问题，请点击"使用说明"获取帮助."""
|
||||
|
||||
return self._create_text_response(help_text, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"使用说明处理异常: {str(e)}")
|
||||
return self._create_text_response("获取帮助信息失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_announcements_by_type(self, from_user: str) -> Optional[str]:
|
||||
"""处理按类型查看公告菜单"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 请求按类型查看公告")
|
||||
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
all_stats = storage.get_statistics()
|
||||
type_stats = all_stats.get('announcement_types', {})
|
||||
|
||||
if type_stats:
|
||||
response = f"""公告类型统计
|
||||
|
||||
统计时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
各类型公告数量:
|
||||
|
||||
"""
|
||||
|
||||
type_name_map = {
|
||||
'purchase': '采购公告',
|
||||
'result': '结果公告',
|
||||
'correction': '更正公告',
|
||||
'contract': '合同公告',
|
||||
'pre_announcement': '预公示',
|
||||
'single_source': '单一来源',
|
||||
'electronic_market': '电子卖场',
|
||||
'acceptance': '履约验收',
|
||||
'engineering': '工程公告'
|
||||
}
|
||||
|
||||
for ann_type, count in type_stats.items():
|
||||
type_name = type_name_map.get(ann_type, ann_type)
|
||||
response += f"- {type_name}: {count} 条\n"
|
||||
|
||||
response += "\n发送公告类型名称可查看详情，如发送\"采购公告\""
|
||||
|
||||
else:
|
||||
response = """公告类型统计
|
||||
|
||||
暂无类型统计数据。
|
||||
|
||||
建议:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 系统将自动分类统计各种公告"""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败: {str(db_error)}")
|
||||
response = """公告类型说明
|
||||
|
||||
系统支持以下类型的政府采购公告:
|
||||
|
||||
- 采购公告: 招标、采购等采购信息
|
||||
- 结果公告: 中标、成交等结果信息
|
||||
- 更正公告: 变更、澄清等修改信息
|
||||
- 合同公告: 合同签订等信息
|
||||
- 预公示: 招标文件预公示
|
||||
- 单一来源: 单一来源采购公示
|
||||
- 电子卖场: 电子化采购平台
|
||||
- 履约验收: 项目验收信息
|
||||
- 工程公告: 工程建设相关
|
||||
|
||||
发送具体类型名称可搜索相关公告。"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"按类型查看公告处理异常: {str(e)}")
|
||||
return self._create_text_response("获取类型统计失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_search_by_type(self, type_name: str, from_user: str) -> Optional[str]:
|
||||
"""处理按类型搜索公告"""
|
||||
try:
|
||||
logger.info(f"用户 {from_user} 按类型搜索公告: {type_name}")
|
||||
|
||||
type_mapping = {
|
||||
"采购公告": "purchase",
|
||||
"结果公告": "result",
|
||||
"更正公告": "correction",
|
||||
"合同公告": "contract",
|
||||
"预公示": "pre_announcement",
|
||||
"单一来源": "single_source",
|
||||
"电子卖场": "electronic_market",
|
||||
"履约验收": "acceptance",
|
||||
"工程公告": "engineering"
|
||||
}
|
||||
|
||||
ann_type = type_mapping.get(type_name)
|
||||
if not ann_type:
|
||||
return self._create_text_response(f"未知的公告类型: {type_name}", from_user)
|
||||
|
||||
try:
|
||||
from ..storage.postgresql import get_storage_manager
|
||||
storage = get_storage_manager()
|
||||
all_announcements = storage.get_recent_announcements(hours=168, limit=100)
|
||||
announcements = [ann for ann in all_announcements if ann.announcement_type.value == ann_type][:5]
|
||||
|
||||
if announcements:
|
||||
response = f"""{type_name} (最近5条)
|
||||
|
||||
更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||||
|
||||
"""
|
||||
|
||||
for i, announcement in enumerate(announcements, 1):
|
||||
title = announcement.title[:25] + "..." if len(announcement.title) > 25 else announcement.title
|
||||
time_str = announcement.publish_date.strftime('%m-%d %H:%M') if announcement.publish_date else "未知"
|
||||
response += f"{i}. {title}\n {time_str} | {announcement.purchase_name or '未知'}\n\n"
|
||||
|
||||
response += "发送关键词可进一步筛选，点击菜单可查看更多功能。"
|
||||
else:
|
||||
response = f"""{type_name}
|
||||
|
||||
暂无该类型的公告数据。
|
||||
|
||||
建议:
|
||||
- 点击"立即搜索"更新数据
|
||||
- 该类型公告可能较少出现"""
|
||||
|
||||
except Exception as db_error:
|
||||
logger.warning(f"数据库查询失败: {str(db_error)}")
|
||||
response = f"""{type_name}
|
||||
|
||||
暂时无法获取数据，请稍后重试。
|
||||
|
||||
您可以先尝试"立即搜索"更新数据。"""
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"按类型搜索公告处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_manual_crawl(self, content: str, from_user: str) -> Optional[str]:
|
||||
"""处理手动搜索命令"""
|
||||
try:
|
||||
parts = content.split()
|
||||
if len(parts) < 2:
|
||||
return self._create_text_response("请指定搜索关键词，例如: 搜索 大化", from_user)
|
||||
|
||||
keywords = parts[1:]
|
||||
logger.info(f"用户 {from_user} 手动搜索关键词: {keywords}")
|
||||
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败，请稍后重试", from_user)
|
||||
|
||||
result = app.run_crawl(keywords=keywords, manual_crawl=True)
|
||||
|
||||
if result.get("success"):
|
||||
total = result.get("total_crawled", 0)
|
||||
filtered = result.get("filtered", 0)
|
||||
|
||||
if filtered > 0:
|
||||
response = f"""搜索完成!
|
||||
|
||||
关键词: {' '.join(keywords)}
|
||||
发现匹配公告: {filtered} 条
|
||||
|
||||
如有匹配的公告，我会及时推送通知。"""
|
||||
else:
|
||||
response = f"""搜索完成!
|
||||
|
||||
关键词: {' '.join(keywords)}
|
||||
未发现匹配的公告。
|
||||
|
||||
建议:
|
||||
- 尝试更通用的关键词
|
||||
- 检查关键词拼写
|
||||
- 等待系统更新最新数据"""
|
||||
else:
|
||||
error = result.get("error", "未知错误")
|
||||
response = f"搜索失败: {error}"
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"手动搜索处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败，请稍后重试", from_user)
|
||||
|
||||
def _handle_keyword_search(self, content: str, from_user: str) -> Optional[str]:
|
||||
"""处理关键词搜索"""
|
||||
try:
|
||||
parts = content.split()
|
||||
keywords = parts[1:] if len(parts) > 1 else parts
|
||||
logger.info(f"用户 {from_user} 关键词搜索: {keywords}")
|
||||
|
||||
if not keywords:
|
||||
return self._create_text_response("请提供搜索关键词", from_user)
|
||||
|
||||
app = self._get_monitor_app()
|
||||
if not app:
|
||||
return self._create_text_response("系统初始化失败，请稍后重试", from_user)
|
||||
|
||||
result = app.run_crawl(keywords=keywords, manual_crawl=True)
|
||||
|
||||
if result.get("success"):
|
||||
filtered = result.get("filtered", 0)
|
||||
|
||||
if filtered > 0:
|
||||
response = f"""搜索完成!
|
||||
|
||||
关键词: {' '.join(keywords)}
|
||||
发现匹配公告: {filtered} 条
|
||||
|
||||
如有匹配的公告，我会及时推送通知。"""
|
||||
else:
|
||||
response = f"""搜索完成!
|
||||
|
||||
关键词: {' '.join(keywords)}
|
||||
未发现匹配的公告。
|
||||
|
||||
建议:
|
||||
- 尝试更通用的关键词
|
||||
- 检查关键词拼写
|
||||
- 等待系统更新最新数据"""
|
||||
else:
|
||||
error = result.get("error", "未知错误")
|
||||
response = f"搜索失败: {error}"
|
||||
|
||||
return self._create_text_response(response, from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"关键词搜索处理异常: {str(e)}")
|
||||
return self._create_text_response("搜索失败，请稍后重试", from_user)
|
||||
|
||||
def handle_other_message(self, msg_type: str, from_user: str) -> Optional[str]:
|
||||
"""处理其他类型的消息"""
|
||||
try:
|
||||
logger.info(f"处理其他消息类型: {msg_type}, user: {from_user}")
|
||||
|
||||
if msg_type == 'image':
|
||||
return self._create_text_response("收到图片消息，但我只能处理文本消息", from_user)
|
||||
elif msg_type == 'voice':
|
||||
return self._create_text_response("收到语音消息，但我只能处理文本消息", from_user)
|
||||
else:
|
||||
return self._create_text_response(f"收到{msg_type}消息，暂不支持此类型", from_user)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"其他消息处理异常: {str(e)}")
|
||||
return self._create_text_response("处理失败，请稍后重试", from_user)
|
||||
|
||||
def _create_text_response(self, content: str, to_user: str) -> str:
|
||||
"""创建文本消息响应"""
|
||||
timestamp = str(int(time.time()))
|
||||
|
||||
response_xml = f"""<xml>
|
||||
<ToUserName><![CDATA[{to_user}]]></ToUserName>
|
||||
<FromUserName><![CDATA[{self.config.wechat_app.corp_id}]]></FromUserName>
|
||||
<CreateTime>{timestamp}</CreateTime>
|
||||
<MsgType><![CDATA[text]]></MsgType>
|
||||
<Content><![CDATA[{content}]]></Content>
|
||||
</xml>"""
|
||||
|
||||
return response_xml
|
||||
+1708
-103
File diff suppressed because it is too large
Load Diff
+1234
-119
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,8 @@
|
||||
📋 关键词搜索: `未知` - 总公告数: `0`
|
||||
|
||||
**更新时间: 2026-01-08 23:41:28**
|
||||
|
||||
|
||||
## 无匹配公告
|
||||
|
||||
在指定时间范围内没有找到符合条件的公告。
|
||||
+4
-4
@@ -1,6 +1,6 @@
|
||||
#!/bin/bash
|
||||
# 定时爬取脚本启动器
|
||||
# 用于启动广西政府采购网公告定时爬取任务
|
||||
# 定时搜索脚本启动器
|
||||
# 用于启动广西政府采购网公告定时搜索任务
|
||||
# 自动激活项目虚拟环境并运行Python脚本
|
||||
|
||||
# 设置脚本遇到错误时退出
|
||||
@@ -11,7 +11,7 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
PROJECT_ROOT="$SCRIPT_DIR"
|
||||
|
||||
echo "========================================"
|
||||
echo "🕒 广西政府采购网公告定时爬取任务"
|
||||
echo "🕒 广西政府采购网公告定时搜索任务"
|
||||
echo "📂 项目目录: $PROJECT_ROOT"
|
||||
echo "⏰ 开始时间: $(date '+%Y-%m-%d %H:%M:%S')"
|
||||
echo "========================================"
|
||||
@@ -66,7 +66,7 @@ if [ ! -f "$CRON_SCRIPT" ]; then
|
||||
fi
|
||||
|
||||
echo "✅ 环境检查通过"
|
||||
echo "🚀 启动定时爬取脚本..."
|
||||
echo "🚀 启动定时搜索脚本..."
|
||||
|
||||
# 切换到项目根目录
|
||||
cd "$PROJECT_ROOT"
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
测试修复后的菜单功能
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# 添加项目路径
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
try:
|
||||
from gx_gp_monitor.wechat.message_handler import WeChatMessageHandler
|
||||
from gx_gp_monitor.core.config_manager import load_config
|
||||
|
||||
def test_menu_keys():
|
||||
"""测试菜单key处理"""
|
||||
print("🧪 测试菜单key处理...")
|
||||
|
||||
handler = WeChatMessageHandler()
|
||||
|
||||
# 测试旧菜单key
|
||||
test_cases = [
|
||||
("latest_announcements", "最新公告（兼容旧key）"),
|
||||
("announcements_by_type", "按类型查看公告"),
|
||||
("search_announcements", "搜索公告（兼容旧key）"),
|
||||
("latest_news", "最新公告"),
|
||||
("keyword_search", "关键词搜索"),
|
||||
("system_status", "系统状态"),
|
||||
("help_guide", "使用说明"),
|
||||
]
|
||||
|
||||
results = {}
|
||||
for key, description in test_cases:
|
||||
try:
|
||||
# 模拟事件处理
|
||||
result = handler.handle_event('click', key, 'test_user')
|
||||
results[key] = result is not None and "未知菜单项" not in (result or "")
|
||||
print(f"✅ {description}: {'通过' if results[key] else '失败'}")
|
||||
except Exception as e:
|
||||
print(f"❌ {description}: 异常 - {str(e)}")
|
||||
results[key] = False
|
||||
|
||||
success_count = sum(1 for success in results.values() if success)
|
||||
print(f"\n📊 菜单key测试结果: {success_count}/{len(test_cases)} 通过")
|
||||
|
||||
return success_count == len(test_cases)
|
||||
|
||||
def test_text_commands():
|
||||
"""测试文本命令处理"""
|
||||
print("\n🧪 测试文本命令处理...")
|
||||
|
||||
handler = WeChatMessageHandler()
|
||||
|
||||
# 测试公告类型查询
|
||||
test_cases = [
|
||||
("采购公告", "采购公告查询"),
|
||||
("结果公告", "结果公告查询"),
|
||||
("更正公告", "更正公告查询"),
|
||||
]
|
||||
|
||||
results = {}
|
||||
for command, description in test_cases:
|
||||
try:
|
||||
result = handler.handle_text_message(command, 'test_user')
|
||||
results[command] = result is not None
|
||||
print(f"✅ {description}: {'通过' if results[command] else '失败'}")
|
||||
except Exception as e:
|
||||
print(f"❌ {description}: 异常 - {str(e)}")
|
||||
results[command] = False
|
||||
|
||||
success_count = sum(1 for success in results.values() if success)
|
||||
print(f"\n📊 文本命令测试结果: {success_count}/{len(test_cases)} 通过")
|
||||
|
||||
return success_count == len(test_cases)
|
||||
|
||||
def main():
|
||||
"""主测试函数"""
|
||||
# 加载配置
|
||||
try:
|
||||
load_config()
|
||||
print("✅ 配置加载成功")
|
||||
except Exception as e:
|
||||
print(f"❌ 配置加载失败: {e}")
|
||||
return False
|
||||
|
||||
print("🚀 测试修复后的菜单功能...")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试菜单key
|
||||
test1_passed = test_menu_keys()
|
||||
|
||||
# 测试文本命令
|
||||
test2_passed = test_text_commands()
|
||||
|
||||
print("\n" + "=" * 50)
|
||||
|
||||
if test1_passed and test2_passed:
|
||||
print("🎉 所有菜单功能修复测试通过！")
|
||||
print("✅ 旧菜单key现在可以正常工作")
|
||||
print("✅ 新增了公告类型查询功能")
|
||||
else:
|
||||
print("⚠️ 部分测试失败，请检查上述错误信息")
|
||||
|
||||
return test1_passed and test2_passed
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
except ImportError as e:
|
||||
print(f"❌ 导入失败: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"❌ 测试失败: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,168 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
测试企业微信菜单功能
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# 添加项目路径
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
try:
|
||||
from gx_gp_monitor.wechat.menu_manager import WeChatMenuManager
|
||||
from gx_gp_monitor.wechat.message_handler import WeChatMessageHandler
|
||||
from gx_gp_monitor.core.config_manager import load_config
|
||||
|
||||
def test_menu_manager():
|
||||
"""测试菜单管理器"""
|
||||
print("🧪 测试菜单管理器...")
|
||||
|
||||
try:
|
||||
# 加载配置
|
||||
load_config()
|
||||
|
||||
# 创建菜单管理器
|
||||
menu_manager = WeChatMenuManager()
|
||||
|
||||
# 测试菜单信息获取
|
||||
menu_info = menu_manager.get_menu_info()
|
||||
print("✅ 菜单信息获取成功")
|
||||
print(f" 菜单按钮数量: {menu_info['menu_structure']['total_buttons']}")
|
||||
|
||||
# 打印菜单结构
|
||||
for button in menu_info['menu_structure']['buttons']:
|
||||
sub_count = len(button.get('sub_buttons', [])) if isinstance(button.get('sub_buttons'), list) else button.get('sub_buttons', 0)
|
||||
print(f" • {button['name']} ({sub_count}个子菜单)")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 菜单管理器测试失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def test_message_handler():
|
||||
"""测试消息处理器"""
|
||||
print("🧪 测试消息处理器...")
|
||||
|
||||
try:
|
||||
# 创建消息处理器
|
||||
handler = WeChatMessageHandler()
|
||||
|
||||
# 测试各种菜单功能
|
||||
test_results = {}
|
||||
|
||||
# 测试帮助功能
|
||||
result = handler._handle_help_guide("test_user")
|
||||
test_results["help_guide"] = result is not None and "广西政府采购网" in result
|
||||
|
||||
# 测试联系信息
|
||||
result = handler._handle_contact_info("test_user")
|
||||
test_results["contact_info"] = result is not None and "联系我们" in result
|
||||
|
||||
# 测试关于系统
|
||||
result = handler._handle_about_system("test_user")
|
||||
test_results["about_system"] = result is not None and "关于系统" in result
|
||||
|
||||
# 测试关键词搜索菜单
|
||||
result = handler._handle_keyword_search_menu("test_user")
|
||||
test_results["keyword_search"] = result is not None and "关键词搜索" in result
|
||||
|
||||
# 打印测试结果
|
||||
success_count = sum(1 for success in test_results.values() if success)
|
||||
print(f"✅ 消息处理器测试完成: {success_count}/{len(test_results)} 项通过")
|
||||
|
||||
for func_name, success in test_results.items():
|
||||
status = "✅" if success else "❌"
|
||||
print(f" {status} {func_name}")
|
||||
|
||||
return success_count == len(test_results)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 消息处理器测试失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def test_menu_structure():
|
||||
"""测试菜单结构合理性"""
|
||||
print("🧪 测试菜单结构...")
|
||||
|
||||
try:
|
||||
menu_manager = WeChatMenuManager()
|
||||
menu_data = menu_manager.menu_data
|
||||
|
||||
# 检查菜单结构
|
||||
checks = {
|
||||
"has_buttons": len(menu_data.get("button", [])) > 0,
|
||||
"max_buttons": len(menu_data.get("button", [])) <= 3, # 企业微信最多3个一级菜单
|
||||
"has_sub_buttons": all("sub_button" in btn for btn in menu_data.get("button", [])),
|
||||
"sub_buttons_limit": all(len(btn.get("sub_button", [])) <= 5 for btn in menu_data.get("button", [])),
|
||||
"valid_keys": True
|
||||
}
|
||||
|
||||
# 检查所有按钮都有有效的key
|
||||
all_keys = []
|
||||
for button in menu_data.get("button", []):
|
||||
for sub_btn in button.get("sub_button", []):
|
||||
key = sub_btn.get("key")
|
||||
if key:
|
||||
all_keys.append(key)
|
||||
else:
|
||||
checks["valid_keys"] = False
|
||||
|
||||
# 检查key唯一性
|
||||
checks["unique_keys"] = len(all_keys) == len(set(all_keys))
|
||||
|
||||
# 打印检查结果
|
||||
success_count = sum(1 for success in checks.values() if success)
|
||||
print(f"✅ 菜单结构检查完成: {success_count}/{len(checks)} 项通过")
|
||||
|
||||
for check_name, success in checks.items():
|
||||
status = "✅" if success else "❌"
|
||||
print(f" {status} {check_name}")
|
||||
|
||||
return success_count == len(checks)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 菜单结构测试失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def main():
|
||||
"""主测试函数"""
|
||||
print("🚀 开始测试企业微信菜单功能...")
|
||||
print("=" * 60)
|
||||
|
||||
# 测试菜单管理器
|
||||
test1_passed = test_menu_manager()
|
||||
print()
|
||||
|
||||
# 测试消息处理器
|
||||
test2_passed = test_message_handler()
|
||||
print()
|
||||
|
||||
# 测试菜单结构
|
||||
test3_passed = test_menu_structure()
|
||||
print()
|
||||
|
||||
print("=" * 60)
|
||||
|
||||
# 总体结果
|
||||
all_passed = test1_passed and test2_passed and test3_passed
|
||||
if all_passed:
|
||||
print("🎉 所有测试通过！企业微信菜单功能正常")
|
||||
print("💡 您可以运行以下命令创建菜单：")
|
||||
print(" python -c \"from gx_gp_monitor.wechat.menu_manager import WeChatMenuManager; m = WeChatMenuManager(); m.create_menu()\"")
|
||||
else:
|
||||
print("⚠️ 部分测试失败，请检查上述错误信息")
|
||||
|
||||
return all_passed
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
except ImportError as e:
|
||||
print(f"❌ 导入失败: {e}", file=sys.stderr)
|
||||
print("请确保已安装所有依赖: pip install -r gx_gp_monitor/requirements.txt", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"❌ 测试失败: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
+1
-1
@@ -1 +1 @@
|
||||
1560152
|
||||
1616302
|
||||
|
||||
Reference in New Issue
Block a user