fix : 暂时关闭全量消息推送

bot.py

@@ -1,14 +1,32 @@
+from typing import Any, List
+
 import botpy
-from botpy.message import Message
+from botpy.message import (Message)
+from botpy import logger
+import queue
 from dotenv import load_dotenv
 import os 
 import types
 from bobby import bobbySession
+from mcwatcher import watcher  
+from threading import Thread
+import asyncio
 #环境配置  
 
 class djtuWebsocket (botpy.Client) :
+    def __init__(
+            self, intents: botpy.Intents,
+            timeout: int = 5,
+            is_sandbox=False, 
+            log_config: str | dict | None = None, 
+            log_format: str| None = None, 
+            log_level: int |None= None, 
+            bot_log: bool | None = True, 
+            ext_handlers: dict | List[dict] | bool = True):
+        self.q = queue.Queue()
+        super().__init__(intents, timeout, is_sandbox, log_config, log_format, log_level, bot_log, ext_handlers)
+    
     async def on_group_at_message_create(self, message: Message):
-
         await self.parseCommand(message)
 
     async def on_group_message_create(self,message:Message):
@@ -16,6 +34,7 @@ class djtuWebsocket (botpy.Client) :
         await self.parseGlobalMessage(message)
 
 
+# 触发概率不大所以可以直接提取关键字
     async def parseCommand(self,msg :Message):
         commands:dict[str,types.MethodType|types.FunctionType] = {
             r'/邀请码' : bobby.getInviteCode,
@@ -34,17 +53,38 @@ class djtuWebsocket (botpy.Client) :
             reply = "你在发什么东西我怎么看不懂?"
         await msg.reply(content=reply)
 
-    async def parseGlobalMessage(self,msg:Message):
+# 触发概率大，以后要做好解析防止刷屏
-        await msg.reply(content="awda")
+    # async def parseGlobalMessage(self,msg:Message):
+        # await msg.reply(content="awda")
+
+# 监听队列，接受其他线程的消息
+    async def mcloop(self):
+        while True:
+            if self.q.empty():
+                # time.sleep会阻塞整个线程我去
+                await asyncio.sleep(20)
+            else:
+                try :
+                    msg = self.q.get()
+                    if group_open_id is None:
+                        logger.error("群聊不存在")
+                        continue
+                    asyncio.create_task(self.api.post_group_message(group_open_id,content=msg))
+                except Exception as e:
+                    logger.error("MCWatcher发送群聊任务失败:\n"+str(e))
+
 # source
 load_dotenv(".env")
-print(os.getenv("APP_ID"))
-
 
 bbyAccount = os.getenv("BOBBY_ACCOUNT")
 bbyPasswd = os.getenv("BOBBY_PASSWORD")
 appid = os.getenv("APP_ID")
 secret_bot = os.getenv("SECRET")
+group_open_id = os.getenv("GROUP_OPEN_ID")
+
+if appid is None or secret_bot is None :
+    raise SystemExit("未找到bot配置!")
+logger.info("登录账号:"+appid)
 
 # bobby Session 
 if bbyAccount is None or bbyPasswd is None :
@@ -52,7 +92,32 @@ if bbyAccount is None or bbyPasswd is None :
 if appid is None or secret_bot is None :
     raise SystemError("未找到bot配置")
 bobby = bobbySession(bbyAccount,bbyPasswd)
-
+     
-#bot 启动 
+async def main(ws:djtuWebsocket, *args: Any, **kwargs: Any) -> None:
+    async with ws as ws:
+        logger.info("启动并发异步ing...")
+        mcloop = ws.mcloop()
+        # 你*的为什么retcoro true的话会报错啊
+        # 协程对象不会不是这个异步回话而是httpx吧？
+        bot = ws.start(*args, **kwargs,ret_coro=False)
+        if bot is None:
+            logger.error("bot不给协程啊")
+            return
+        try:
+            await asyncio.gather(bot,mcloop)
+        except Exception as e :
+            logger.error(e)
+            
+# bot初始化
 ws = djtuWebsocket(botpy.Intents(public_messages=True))
-ws.run(appid=appid,secret=secret_bot)
+
+# 启动mcwatcher线程
+def pushQuene(cont:str):
+    ws.q.put(cont)
+Thread(target=watcher.MCWatcher,name="watcher",kwargs={
+    "hookfunction":pushQuene,
+    "logging":logger
+}).start()
+
+# bot线程运行
+asyncio.run(main(ws,appid=appid,secret=secret_bot))

mcwatcher/state.json

@@ -0,0 +1,6 @@
+{
+  "last_snapshot_id": "26.3",
+  "last_release_id": "26.3",
+  "targets": [],
+  "etag": null
+}

mcwatcher/watcher.py

@@ -0,0 +1,359 @@
+
+import asyncio
+import json
+from types import (FunctionType,MethodType)
+import re
+from pathlib import Path
+from datetime import datetime
+from typing import Dict, Any, Optional, Set
+import time
+import httpx
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+import logging
+
+# from astrbot.api.all import (
+#     register, command, Context, Star, AstrMessageEvent,
+#     MessageChain, Plain, logger
+# )
+
+# from astrbot.api.star import StarTools
+
+
+MANIFEST_URL = "https://piston-meta.mojang.com/mc/game/version_manifest_v2.json"
+
+
+def _ts():
+    return datetime.now().strftime("[%Y-%m-%d %H:%M:%S]")
+
+
+# ========== 已改进版本的文章链接生成函数 ==========
+def format_article_url(lang: str, version_id: str) -> str:
+    vid = (version_id or "").lower()
+
+    def _join_parts(match: re.Match) -> str:
+        parts = [match.group("major"), match.group("minor")]
+        patch = match.group("patch")
+        if patch is not None:
+            parts.append(patch)
+        return "-".join(parts)
+
+    # pre-release: minecraft-1-a-b-pre-release-c
+    pre_match = re.match(
+        r"^(?P<major>\d+)\.(?P<minor>\d+)(?:\.(?P<patch>\d+))?-pre(?P<build>\d+)$",
+        vid,
+    )
+    if pre_match:
+        slug = _join_parts(pre_match)
+        build = pre_match.group("build")
+        return f"https://www.minecraft.net/{lang}/article/minecraft-{slug}-pre-release-{build}"
+
+    # RC: minecraft-1-a-b-release-candidate-c
+    rc_match = re.match(
+        r"^(?P<major>\d+)\.(?P<minor>\d+)(?:\.(?P<patch>\d+))?-rc(?P<build>\d+)$",
+        vid,
+    )
+    if rc_match:
+        slug = _join_parts(rc_match)
+        build = rc_match.group("build")
+        return f"https://www.minecraft.net/{lang}/article/minecraft-{slug}-release-candidate-{build}"
+
+    # Snapshot，例如 25w46a
+    if "w" in vid:
+        slug = vid.replace(" ", "").replace("_", "-")
+        return f"https://www.minecraft.net/{lang}/article/minecraft-snapshot-{slug}"
+
+    return ""
+
+
+# ===================================================
+
+
+class State:
+    def __init__(self, path: Path):
+        self.path = path
+        self.last_snapshot_id: Optional[str] = None
+        self.last_release_id: Optional[str] = None
+        self.targets: Set[str] = set()
+        self.etag: Optional[str] = None
+
+    def load(self):
+        if not self.path.exists():
+            return
+        try:
+            data = json.loads(self.path.read_text("utf-8"))
+            self.last_snapshot_id = data.get("last_snapshot_id")
+            self.last_release_id = data.get("last_release_id")
+            self.targets = set(data.get("targets", []))
+            self.etag = data.get("etag")
+        except (json.JSONDecodeError, KeyError, OSError) as e:
+            logger.warning(f"{_ts()} state 文件异常：{e}")
+
+    def save(self):
+        data = {
+            "last_snapshot_id": self.last_snapshot_id,
+            "last_release_id": self.last_release_id,
+            "targets": list(self.targets),
+            "etag": self.etag,
+        }
+        try:
+            self.path.write_text(json.dumps(data, ensure_ascii=False, indent=2), "utf-8")
+        except OSError as e:
+            logger.warning(f"{_ts()} state 写入失败：{e}")
+
+
+# @register(
+#     "astrbot_plugin_mcwatcher",
+#     "noname2309-bot",
+#     "MCWatcher 是一个基于 AstrBot 的插件，用于 自动监测 Mojang 的最新 Minecraft 版本",
+#     "0.2.4",
+#     "https://github.com/wadadawsd/astrbot_plugin_mcwatcher"
+# )
+class MCWatcher:
+
+    # def __init__(self, context: Context, config: Optional[Dict[str, Any]] = None, **kwargs):
+    def __init__(   self,
+                    config: Optional[Dict[str, Any]] = None,
+                    hookfunction:FunctionType | MethodType = print,
+                    logging =logging,
+                    **kwargs):
+        # super().__init__(context)
+
+        self.config: Dict[str, Any] = config or {}
+        # self.ctx = context
+
+        def cfg(key: str, default):
+            return self.config.get(key, default)
+        global logger 
+        logger = logging
+        self.poll_seconds = int(cfg("interval_seconds", cfg("poll_seconds", 120)))
+        self.tz_name = cfg("timezone", "Asia/Shanghai")
+        # self.watch_channels = set(cfg("watch_channels", ["snapshot"]))
+        self.watch_channels = set(cfg("watch_channels", ["snapshot",'release']))
+        self.article_lang = cfg("mc_article_lang", "en-us")
+
+        #设置自建回调
+        self.hookfunction = hookfunction 
+        # data_dir = Path(StarTools.get_data_dir("astrbot_plugin_mcwatcher"))
+        data_dir = Path("MCWatcher")
+        data_dir.mkdir(parents=True, exist_ok=True)
+
+        self.state = State(data_dir / "state.json")
+        self.state.load()
+
+        self._stop = False
+        self._task =  asyncio.run(self._poll_loop())
+        # self._task = asyncio.create_task(self._poll_loop())
+
+        logger.info(
+            f"{_ts()} MCWatcher started. interval={self.poll_seconds}s tz={self.tz_name} watch={self.watch_channels}"
+        )
+
+    async def terminate(self):
+        self._stop = True
+        if self._task:
+            self._task.cancel()
+            # try:
+            #     await self._task
+            # except asyncio.CancelledError:
+            #     pass
+        self.state.save()
+
+    # ========== 工具方法 ==========
+
+    # def _get_plain_text(self, event: AstrMessageEvent) -> str:
+    #     try:
+    #         mc = getattr(event, "message_chain", None)
+    #         if mc:
+    #             parts = []
+    #             for seg in mc:
+    #                 if isinstance(seg, Plain):
+    #                     parts.append(getattr(seg, "text", "") or getattr(seg, "content", ""))
+    #             return "".join(parts)
+    #     except (AttributeError, TypeError) as e:
+    #         logger.warning(f"Failed to parse plain text: {e}", exc_info=True)
+    #     return ""
+
+    def _parse_fake_version(self, raw: str, keyword: str, default_vid: str) -> str:
+        toks = raw.replace("\u3000", " ").split()
+        tokens_lower = [t.lower() for t in toks]
+
+        if keyword in tokens_lower:
+            idx = tokens_lower.index(keyword)
+            if idx + 1 < len(toks):
+                return toks[idx + 1]
+            return default_vid
+
+        if len(toks) <= 2:
+            return default_vid
+
+        return toks[2]
+
+    # ========== 指令 ==========
+
+    # # @command("mcwatch bind", alias={"mcwatch on", "mc订阅"})
+    # async def bind_here(self, event: AstrMessageEvent):
+    #     sid = event.unified_msg_origin
+    #     self.state.targets.add(sid)
+    #     self.state.save()
+    #     yield event.plain_result("已绑定本会话。")
+
+    # # @command("mcwatch unbind", alias={"mcwatch off"})
+    # async def unbind_here(self, event: AstrMessageEvent):
+    #     sid = event.unified_msg_origin
+    #     if sid in self.state.targets:
+    #         self.state.targets.remove(sid)
+    #         self.state.save()
+    #         yield event.plain_result("已取消绑定。")
+    #     else:
+    #         yield event.plain_result("本会话未绑定。")
+
+    # # @command("mcwatch list")
+    # async def list_targets(self, event: AstrMessageEvent):
+    #     if not self.state.targets:
+    #         yield event.plain_result("暂无绑定会话。")
+    #     else:
+    #         yield event.plain_result("已绑定：\n" + "\n".join(sorted(self.state.targets)))
+
+    # # @command("mcwatch now")
+    # async def check_now(self, event: AstrMessageEvent):
+    #     await self._check_once(force_push=True)
+    #     yield event.plain_result("OK，已主动检查一次。")
+
+    # ========== fake 推送（仅当前会话，不广播） ==========
+
+    # @command("mcwatch fake")
+    # async def fake_snapshot(self, event: AstrMessageEvent):
+    #     raw = self._get_plain_text(event).strip()
+    #     vid = self._parse_fake_version(raw, "fake", "25w45a")
+    #     msg = self._build_message("snapshot", vid, datetime.now().isoformat())
+
+    #     await self.ctx.send_message(event.unified_msg_origin, MessageChain([Plain(msg)]))
+    #     yield event.plain_result(f"已模拟 snapshot 推送：{vid}")
+
+    # # @command("mcwatch fake_release")
+    # async def fake_release(self, event: AstrMessageEvent):
+    #     raw = self._get_plain_text(event).strip()
+    #     vid = self._parse_fake_version(raw, "fake_release", "1.21.3")
+    #     msg = self._build_message("release", vid, datetime.now().isoformat())
+
+    #     await self.ctx.send_message(event.unified_msg_origin, MessageChain([Plain(msg)]))
+    #     yield event.plain_result(f"已模拟 release 推送：{vid}")
+
+    # ========== 轮询 ==========
+
+    async def _poll_loop(self):
+        logger.info("等待bot完毕~")
+        # 防止bot未初始化
+        time.sleep(10)
+        try:
+            await self._check_once(force_push=False)
+        except Exception as e:
+            logger.warning(f"MCWatcher 首次检查异常：{e}", exc_info=True)
+
+        while not self._stop:
+            try:
+                await asyncio.sleep(self.poll_seconds)
+                await self._check_once(force_push=False)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.warning(f"MCWatcher loop error: {e}", exc_info=True)
+
+    async def _fetch_manifest(self, ignore_cache: bool = False):
+        headers = {}
+        if not ignore_cache and self.state.etag:
+            headers["If-None-Match"] = self.state.etag
+
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            resp = await client.get(MANIFEST_URL, headers=headers)
+
+            if resp.status_code == 304:
+                return None
+
+            resp.raise_for_status()
+
+            etag = resp.headers.get("ETag")
+            if etag:
+                self.state.etag = etag
+                self.state.save()
+
+            return resp.json()
+
+    async def _check_once(self, force_push: bool):
+        data = await self._fetch_manifest(ignore_cache=force_push)
+        if data is None and not force_push:
+            return None
+
+        latest = (data or {}).get("latest", {})
+        msgs = []
+
+        if "snapshot" in self.watch_channels:
+            sid = latest.get("snapshot")
+            if sid and sid != self.state.last_snapshot_id:
+                t = self._lookup_release_time(data, sid)
+                msgs.append(self._build_message("snapshot", sid, t))
+                self.state.last_snapshot_id = sid
+
+        if "release" in self.watch_channels:
+            rid = latest.get("release")
+            if rid and rid != self.state.last_release_id:
+                t = self._lookup_release_time(data, rid)
+                msgs.append(self._build_message("release", rid, t))
+                self.state.last_release_id = rid
+
+        if msgs:
+            self.state.save()
+            await self._broadcast("\n\n".join(msgs), self.state.targets)
+
+    def _lookup_release_time(self, data: Optional[Dict[str, Any]], vid: str):
+        if not data:
+            return None
+        for v in data.get("versions", []):
+            if v.get("id") == vid:
+                return v.get("releaseTime")
+        return None
+
+    def _build_message(self, vtype: str, vid: str, release_iso: Optional[str]):
+        dt_str = "未知时间"
+        if release_iso:
+            try:
+                dt = datetime.fromisoformat(
+                    release_iso.replace("Z", "+00:00")
+                ).astimezone(ZoneInfo(self.tz_name))
+                dt_str = dt.strftime("%Y-%m-%d %H:%M:%S %z")
+            except (ValueError, ZoneInfoNotFoundError):
+                dt_str = release_iso
+
+        article = format_article_url(self.article_lang, vid) if vtype == "snapshot" else ""
+
+        lines = [
+            f"MC 更新：{vid} ({vtype})",
+            f"时间：{dt_str}"
+        ]
+        if article:
+            lines.append(f"日志：{article}")
+
+        return "\n".join(lines)
+
+    # ========== 高性能并发广播 ==========
+    # async def _broadcast(self, text: str, targets: Set[str]):
+    #     if not targets:
+    #         return
+
+    #     mc = MessageChain([Plain(text)])
+    #     sid_list = list(targets)
+
+    #     tasks = [self.ctx.send_message(sid, mc) for sid in sid_list]
+    #     results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    #     ok = 0
+    #     for sid, result in zip(sid_list, results):
+    #         if isinstance(result, Exception):
+    #             logger.warning(f"{_ts()} 推送失败 {sid}: {result}", exc_info=True)
+    #         elif result:
+    #             ok += 1
+
+    #     logger.info(f"{_ts()} MCWatcher 已推送 {ok}/{len(targets)} 个会话。")
+    async def _broadcast(self, text: str, targets: Set[str]):
+         self.hookfunction(text)

requirements.txt

@@ -1,3 +1,4 @@
 requests
 qq-botpy
-dotenv
+dotenv
+httpx