GX-gp-notify/app/wechat/handler.py

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()

# 防重入：最新公告爬取，60秒内不重复执行
_last_latest_time: float = 0.0
_latest_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)
        elif event_key == "toggle_ai":
            return await self._handle_toggle_ai(from_user)
        return None

    async def handle_text(self, content: str, from_user: str) -> str | None:
        """文本消息：返回帮助提示"""
        help_text = (
            "请使用菜单操作：\n"
            "---\n"
            "📋 最新公告 - 获取最新公告\n"
            "📊 查询 → 监控配置/系统状态/今日工作日\n"
            "⚙️ 系统管理 → 立即爬取/同步节假日/AI 分析"
        )
        ok = await self.client.send_text(help_text, from_user)
        if not ok:
            logger.warning(f"发送帮助消息失败, touser={from_user}")
        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:
                remaining = int(60 - (now - _last_crawl_time))
                logger.info(f"爬取请求被忽略（防重入），距上次 {now - _last_crawl_time:.1f}s")
                await self.client.send_text(
                    f"爬取任务进行中，请 {remaining} 秒后再试", from_user
                )
                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
        from app.services.ai_state import get_ai_status_text
        try:
            keywords = settings.crawler_keywords
            sources = json.loads(settings.announcement_sources)
            source_names = "、".join(v["name"] for v in sources.values())
            text = (
                f"📋 监控配置\n"
                f"---\n"
                f"监控关键词: {', '.join(keywords)}\n"
                f"爬取页数: {settings.crawler_max_pages} 页\n"
                f"定时规则: {settings.scheduler_cron}\n"
                f"公告来源: {source_names}\n"
                f"---\n"
                f"🤖 AI 分析: {get_ai_status_text()}\n"
                f"AI 模型: {settings.ai_model}\n"
                f"重点标记: {settings.ai_analysis_title}"
            )
            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
        from app.services.ai_state import get_ai_status_text

        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

                # AI 标记统计
                ai_relevant_result = await db.execute(
                    select(func.count()).where(
                        Announcement.ai_relevant == True,  # noqa: E712
                    ).select_from(Announcement)
                )
                ai_relevant = ai_relevant_result.scalar() or 0

                scheduler_status = "已启用" if settings.scheduler_enabled else "已禁用"
                text = (
                    f"📊 系统状态\n"
                    f"---\n"
                    f"累计公告: {total} 条\n"
                    f"今日新增: {today} 条\n"
                    f"待推送: {unsent} 条\n"
                    f"---\n"
                    f"定时任务: {scheduler_status}\n"
                    f"定时规则: {settings.scheduler_cron}\n"
                    f"---\n"
                    f"🤖 AI 分析: {get_ai_status_text()}\n"
                    f"AI 标记项目: {ai_relevant} 条"
                )
                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:
        global _last_latest_time
        from app.api.deps import get_db, get_crawl_service
        from sqlalchemy import select, asc, or_
        from app.models.announcement import Announcement

        # 防重入：60秒内整个函数只执行一次（含推送）
        async with _latest_lock:
            now = time.monotonic()
            if (now - _last_latest_time) < 60:
                logger.info("最新公告请求被忽略（防重入）")
                return None
            _last_latest_time = now

        try:
            await self.client.send_text("正在获取最新公告，请稍候...", from_user)
            service = get_crawl_service()
            await service.run_all()

            # 查：广西政采网关键词匹配 + 大化县政府网全部，取最新6条后按时间升序推送
            async for db in get_db():
                from sqlalchemy import select, desc, or_
                result = await db.execute(
                    select(Announcement)
                    .where(or_(
                        Announcement.keyword_matched == True,  # noqa: E712
                        Announcement.source_code == "dahuagov",
                    ))
                    .order_by(desc(Announcement.publish_date))
                    .limit(6)
                )
                items = result.scalars().all()

                if not items:
                    await self.client.send_text("暂无公告", from_user)
                    return None

                # 按时间升序推送，企微里向上滑即为时间正序
                for ann in sorted(items, key=lambda a: a.publish_date):
                    title = ann.title if len(ann.title) <= 128 else ann.title[:125] + "..."
                    purchase_name = ann.purchase_name or ""
                    if len(purchase_name) > 25:
                        purchase_name = purchase_name[:22] + "..."
                    time_str = ann.publish_date.strftime("%Y-%m-%d %H:%M") if ann.publish_date else "时间未知"
                    source_name = ann.source_name or ann.source_code or ""
                    description = f"{source_name} | {purchase_name} | {time_str}"
                    await self.client.send_textcard(title, description, ann.content_url or "", from_user)
                    await asyncio.sleep(1)
        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_toggle_ai(self, from_user: str) -> str | None:
        from app.services.ai_state import (
            get_ai_status_text,
            get_whitelist,
            is_ai_enabled,
            is_whitelisted,
            set_ai_enabled,
        )

        # 白名单校验
        whitelist = get_whitelist()
        if whitelist and not is_whitelisted(from_user):
            await self.client.send_text(
                f"⚠️ 你没有权限操作 AI 分析开关\n"
                f"当前仅以下用户可操作：\n{', '.join(whitelist)}",
                from_user,
            )
            return None

        current = is_ai_enabled()
        changed = set_ai_enabled(not current)
        if not changed:
            await self.client.send_text(
                f"AI 分析当前已是「{get_ai_status_text()}」状态，无需切换",
                from_user,
            )
            return None

        await self.client.send_text(
            f"✅ AI 分析已切换为「{get_ai_status_text()}」\n"
            f"下次爬取触发时生效",
            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)