v6ole
0121017272
- latest_announcements 不再过滤 keyword_matched，返回全部最新5条 - 每条公告显示来源网站（广西政府采购网/大化县政府网采购公告） - 菜单「今日公告」改为顶级按钮「最新公告」，移除子菜单重复项
290 lines
12 KiB
Python
290 lines
12 KiB
Python
import asyncio
|
|
import logging
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from app.config import settings
|
|
from app.wechat.crypto import WXBizMsgCrypt
|
|
from app.wechat.client import WeChatClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 防重入：记录最近一次触发爬取的时间戳，60秒内不重复执行
|
|
_last_crawl_time: float = 0.0
|
|
_crawl_lock = asyncio.Lock()
|
|
|
|
|
|
class WeChatMessageHandler:
|
|
def __init__(self):
|
|
self.wxcpt = WXBizMsgCrypt(
|
|
sToken=settings.wechat_token,
|
|
sEncodingAESKey=settings.wechat_encoding_aes_key,
|
|
sReceiveId=settings.wechat_corp_id,
|
|
)
|
|
self.client = WeChatClient()
|
|
|
|
def verify_url(
|
|
self, msg_signature: str, timestamp: str, nonce: str, echostr: str
|
|
) -> str | None:
|
|
ret, sEchoStr = self.wxcpt.VerifyURL(
|
|
msg_signature, timestamp, nonce, echostr
|
|
)
|
|
if ret == 0:
|
|
return (
|
|
sEchoStr.decode("utf-8")
|
|
if isinstance(sEchoStr, bytes)
|
|
else sEchoStr
|
|
)
|
|
return None
|
|
|
|
def decrypt_message(
|
|
self,
|
|
post_data: str,
|
|
msg_signature: str,
|
|
timestamp: str,
|
|
nonce: str,
|
|
) -> ET.Element | None:
|
|
ret, xml_content = self.wxcpt.DecryptMsg(
|
|
post_data, msg_signature, timestamp, nonce
|
|
)
|
|
if ret != 0:
|
|
return None
|
|
return ET.fromstring(xml_content)
|
|
|
|
def encrypt_response(
|
|
self, response_xml: str, nonce: str, timestamp: str
|
|
) -> str | None:
|
|
ret, encrypted = self.wxcpt.EncryptMsg(response_xml, nonce, timestamp)
|
|
if ret == 0:
|
|
return encrypted
|
|
return None
|
|
|
|
async def handle_event(
|
|
self, event: str, event_key: str | None, from_user: str
|
|
) -> str | None:
|
|
if event != "click" or not event_key:
|
|
return None
|
|
|
|
if event_key == "today_stats":
|
|
return await self._handle_today_stats(from_user)
|
|
elif event_key == "trigger_crawl":
|
|
return await self._handle_trigger_crawl(from_user)
|
|
elif event_key == "sync_holidays":
|
|
return await self._handle_sync_holidays(from_user)
|
|
elif event_key == "monitor_config":
|
|
return await self._handle_monitor_config(from_user)
|
|
elif event_key == "system_status":
|
|
return await self._handle_system_status(from_user)
|
|
elif event_key == "workday_status":
|
|
return await self._handle_workday_status(from_user)
|
|
elif event_key == "latest_announcements":
|
|
return await self._handle_latest_announcements(from_user)
|
|
elif event_key == "pause_scheduler":
|
|
return await self._handle_pause_scheduler(from_user)
|
|
elif event_key == "resume_scheduler":
|
|
return await self._handle_resume_scheduler(from_user)
|
|
return None
|
|
|
|
async def handle_text(self, content: str, from_user: str) -> str | None:
|
|
return None
|
|
|
|
async def _handle_today_stats(self, from_user: str) -> str | None:
|
|
from app.api.deps import get_db
|
|
|
|
try:
|
|
async for db in get_db():
|
|
from sqlalchemy import func, select
|
|
from app.models.announcement import Announcement
|
|
|
|
total_result = await db.execute(
|
|
select(func.count()).select_from(Announcement)
|
|
)
|
|
total = total_result.scalar() or 0
|
|
|
|
today_result = await db.execute(
|
|
select(func.count()).where(
|
|
func.date(Announcement.publish_date) == func.current_date()
|
|
).select_from(Announcement)
|
|
)
|
|
today = today_result.scalar() or 0
|
|
|
|
text = f"今日新增: {today} 条\n累计公告: {total} 条"
|
|
await self.client.send_text(text, from_user)
|
|
except Exception as e:
|
|
logger.error(f"查询统计失败: {e}")
|
|
await self.client.send_text("查询失败，请稍后再试", from_user)
|
|
|
|
async def _handle_trigger_crawl(self, from_user: str) -> str | None:
|
|
global _last_crawl_time
|
|
from app.api.deps import get_crawl_service
|
|
|
|
# 防重入：企业微信会对同一事件重试多次，60秒内只执行一次
|
|
async with _crawl_lock:
|
|
now = time.monotonic()
|
|
if now - _last_crawl_time < 60:
|
|
logger.info(f"爬取请求被忽略（防重入），距上次 {now - _last_crawl_time:.1f}s")
|
|
return None
|
|
_last_crawl_time = now
|
|
|
|
await self.client.send_text("开始爬取，请稍候...", from_user)
|
|
|
|
try:
|
|
service = get_crawl_service()
|
|
results = await service.run_all()
|
|
total = sum(r.total_count for r in results)
|
|
stored = sum(
|
|
r.pipeline_result.stored for r in results
|
|
if r.pipeline_result
|
|
)
|
|
notified = sum(
|
|
r.pipeline_result.notified for r in results
|
|
if r.pipeline_result
|
|
)
|
|
errors = [r.error_message for r in results if not r.success]
|
|
msg = f"爬取完成\n抓取: {total} 条\n新增: {stored} 条\n推送: {notified} 条"
|
|
if errors:
|
|
msg += f"\n异常: {errors[0][:50]}"
|
|
await self.client.send_text(msg, from_user)
|
|
except Exception as e:
|
|
logger.error(f"手动爬取失败: {e}")
|
|
await self.client.send_text(f"爬取失败: {e}", from_user)
|
|
|
|
async def _handle_sync_holidays(self, from_user: str) -> str | None:
|
|
from app.services.holiday_service import now_in_china, sync_holidays
|
|
from app.api.deps import get_db
|
|
|
|
try:
|
|
async for db in get_db():
|
|
year = now_in_china().year
|
|
count = await sync_holidays(db, year)
|
|
await self.client.send_text(
|
|
f"已同步 {year} 年节假日\n共 {count} 条记录", from_user,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"同步节假日失败: {e}")
|
|
await self.client.send_text(f"同步失败: {e}", from_user)
|
|
|
|
async def _handle_monitor_config(self, from_user: str) -> str | None:
|
|
import json
|
|
try:
|
|
keywords = settings.crawler_keywords
|
|
sources = json.loads(settings.announcement_sources)
|
|
source_names = "、".join(v["name"] for v in sources.values())
|
|
text = (
|
|
f"监控关键词: {', '.join(keywords)}\n"
|
|
f"爬取页数: {settings.crawler_max_pages} 页\n"
|
|
f"定时规则: {settings.scheduler_cron}\n"
|
|
f"公告来源: {source_names}"
|
|
)
|
|
await self.client.send_text(text, from_user)
|
|
except Exception as e:
|
|
logger.error(f"查询监控配置失败: {e}")
|
|
await self.client.send_text("查询失败，请稍后再试", from_user)
|
|
|
|
async def _handle_system_status(self, from_user: str) -> str | None:
|
|
from app.api.deps import get_db
|
|
from sqlalchemy import func, select
|
|
from app.models.announcement import Announcement
|
|
|
|
try:
|
|
async for db in get_db():
|
|
total_result = await db.execute(
|
|
select(func.count()).select_from(Announcement)
|
|
)
|
|
total = total_result.scalar() or 0
|
|
|
|
today_result = await db.execute(
|
|
select(func.count()).where(
|
|
func.date(Announcement.publish_date) == func.current_date()
|
|
).select_from(Announcement)
|
|
)
|
|
today = today_result.scalar() or 0
|
|
|
|
unsent_result = await db.execute(
|
|
select(func.count()).where(
|
|
Announcement.is_sent == False, # noqa: E712
|
|
Announcement.keyword_matched == True, # noqa: E712
|
|
).select_from(Announcement)
|
|
)
|
|
unsent = unsent_result.scalar() or 0
|
|
|
|
scheduler_status = "已启用" if settings.scheduler_enabled else "已禁用"
|
|
text = (
|
|
f"累计公告: {total} 条\n"
|
|
f"今日新增: {today} 条\n"
|
|
f"待推送: {unsent} 条\n"
|
|
f"定时任务: {scheduler_status}\n"
|
|
f"定时规则: {settings.scheduler_cron}"
|
|
)
|
|
await self.client.send_text(text, from_user)
|
|
except Exception as e:
|
|
logger.error(f"查询系统状态失败: {e}")
|
|
await self.client.send_text("查询失败，请稍后再试", from_user)
|
|
|
|
async def _handle_workday_status(self, from_user: str) -> str | None:
|
|
from app.services.holiday_service import now_in_china, is_workday
|
|
from app.api.deps import get_db
|
|
|
|
try:
|
|
async for db in get_db():
|
|
today = now_in_china()
|
|
workday = await is_workday(db, today)
|
|
status = "工作日，正常爬取" if workday else "非工作日，跳过爬取"
|
|
text = f"今天 {today.strftime('%Y-%m-%d %A')}\n{status}"
|
|
await self.client.send_text(text, from_user)
|
|
except Exception as e:
|
|
logger.error(f"查询工作日状态失败: {e}")
|
|
await self.client.send_text("查询失败，请稍后再试", from_user)
|
|
|
|
async def _handle_latest_announcements(self, from_user: str) -> str | None:
|
|
from app.api.deps import get_db
|
|
from sqlalchemy import select, desc
|
|
from app.models.announcement import Announcement
|
|
|
|
try:
|
|
async for db in get_db():
|
|
stmt = (
|
|
select(Announcement)
|
|
.order_by(desc(Announcement.publish_date))
|
|
.limit(5)
|
|
)
|
|
result = await db.execute(stmt)
|
|
items = result.scalars().all()
|
|
|
|
if not items:
|
|
await self.client.send_text("暂无公告", from_user)
|
|
return None
|
|
|
|
lines = ["最新公告（最近5条）:"]
|
|
for i, ann in enumerate(items, 1):
|
|
date_str = ann.publish_date.strftime("%m-%d %H:%M") if ann.publish_date else "?"
|
|
title = ann.title[:28] + "..." if len(ann.title) > 28 else ann.title
|
|
source = ann.source_name or ann.source_code
|
|
lines.append(f"{i}. [{date_str}] {source}")
|
|
lines.append(f" {title}")
|
|
await self.client.send_text("\n".join(lines), from_user)
|
|
except Exception as e:
|
|
logger.error(f"查询最新公告失败: {e}")
|
|
await self.client.send_text("查询失败，请稍后再试", from_user)
|
|
|
|
async def _handle_pause_scheduler(self, from_user: str) -> str | None:
|
|
try:
|
|
from app.scheduler.jobs import scheduler
|
|
if scheduler.running:
|
|
scheduler.pause()
|
|
await self.client.send_text("定时任务已暂停", from_user)
|
|
else:
|
|
await self.client.send_text("定时任务未在运行", from_user)
|
|
except Exception as e:
|
|
logger.error(f"暂停定时任务失败: {e}")
|
|
await self.client.send_text(f"操作失败: {e}", from_user)
|
|
|
|
async def _handle_resume_scheduler(self, from_user: str) -> str | None:
|
|
try:
|
|
from app.scheduler.jobs import scheduler
|
|
scheduler.resume()
|
|
await self.client.send_text("定时任务已恢复", from_user)
|
|
except Exception as e:
|
|
logger.error(f"恢复定时任务失败: {e}")
|
|
await self.client.send_text(f"操作失败: {e}", from_user)
|