Compare commits

...
16 Commits
9 changed files with 705 additions and 62 deletions
+4
View File
@@ -173,3 +173,7 @@ cython_debug/
# PyPI configuration file
.pypirc
.*
state.json
botpy.log*
joined_group.cfg
+3 -3
View File
@@ -70,18 +70,18 @@ class bobbySession(requests.Session):
for i in codes :
res.append(i.group())
return res
def getInviteCode(self,step=0)->str:
def getInviteCode(self,ctx:str,step=0)->str:
if step == 5:
raise SystemError("验证码查找失败")
try :
codes = self.__getInvite()
if codes.__len__() == 0:
self.__GenerateInvite()
return self.getInviteCode(step+1)
return self.getInviteCode(ctx,step+1)
rand =random.randrange(0,codes.__len__())
return codes[rand]
except:
self.Refresh()
return self.getInviteCode(step+1)
return self.getInviteCode(ctx,step+1)
+198 -17
View File
@@ -1,28 +1,166 @@
import botpy
from botpy.message import Message
from typing import Any, List
import group
from botpy.connection import ConnectionSession
from botpy.message import (Message,GroupMessage)
from botpy.flags import (Flag,fill_with_flags)
from botpy import( logger,Intents,Client)
from botpy.gateway import (BotWebSocket)
from botpy.types.gateway import (MessagePayload)
import queue
from botpy.types.session import Session
from botpy.types.message import MarkdownPayload
from dotenv import load_dotenv
import os
import types
from bobby import bobbySession
from mcwatcher import watcher
from threading import Thread
import asyncio
from serverGET import ServerGet
#环境配置
class djtuWebsocket (botpy.Client) :
async def on_group_at_message_create(self, message: Message):
class Intents_ex(Intents):
def __init__(self, **kwargs: bool) -> None:
arg_tmp = dict[str,bool]()
arg_ex = dict[str,bool]()
for key, value in kwargs.items():
if key in ['member_event']:
arg_ex[key] = value
else:
arg_tmp[key] = value
super().__init__(**arg_tmp)
for key , value in arg_ex.items() :
setattr(self, key, value)
@Flag
def member_event(self):
return 1 << 24
class BotWebsocket_ex(BotWebSocket):
def __init__(self, session: Session, _connection: ConnectionSession):
super().__init__(session, _connection)
self._parser['group_member_add'] = lambda aaw:self._connection.dispatch("group_member_add",aaw)
self._parser['group_message_create'] = lambda aaw:self._connection.dispatch("group_message_create",aaw)
class djtuWebsocket (Client) :
def __init__(
self, intents: 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 bot_connect(self, session):
"""
newConnect 启动一个新的连接,如果连接在监听过程中报错了,或者被远端关闭了链接,需要识别关闭的原因,能否继续 resume
如果能够 resume,则往 sessionChan 中放入带有 sessionID 的 session
如果不能,则清理掉 sessionID,将 session 放入 sessionChan 中
session 的启动,交给 start 中的 for 循环执行,session 不自己递归进行重连,避免递归深度过深
param session: session对象
"""
logger.info("[botpy] 会话启动中...")
client = BotWebsocket_ex(session, self._connection)
try:
await client.ws_connect()
except (Exception, KeyboardInterrupt, SystemExit) as e:
await client.on_error(e)
async def on_group_at_message_create(self, message: Message):
await self.parseCommand(message)
async def on_group_message_create(self,message:Message):
async def on_group_message_create(self,message):
# 检验是否有自身@消息
try:
msg = message['d']
mention = msg['mentions']
for m in mention :
if m['is_you'] == True:
return await self.parseCommand(GroupMessage(self.api,msg['id'],MessagePayload(message['d'])))
except:
'没有@消息欸'
await self.parseGlobalMessage(message)
return await self.parseGlobalMessage(message)
async def parseCommand(self,msg :Message):
async def on_group_member_add(self,payload):
# await message.reply(content="awdwa")
openid = 'null'
memberid = 'null'
eid ='null'
try :
eid = payload['id']
data = payload['d']
openid = data['group_openid']
memberid = data['member_openid']
except:
logger.warning("data解析失败")
return
logger.info("入群信息:"+' 群: '+openid+" openid: "+memberid)
await self.api.post_group_message(openid,event_id=eid,content='欢迎加入大海啸!\n入服教程请看链接:\nhttps://hiso.wgfire.top/archives/ifnew')
# 触发概率不大所以可以直接提取关键字
async def parseCommand(self,msg :GroupMessage):
# 加入群聊
if msg.group_openid:
self.group_add(msg.group_openid)
commands:dict[str,types.MethodType|types.FunctionType] = {
r'/邀请码' : bobby.getInviteCode,
r'吃什么' : lambda :"是啊吃什么"
r'吃什么' : lambda ctx:"是啊吃什么" ,
r'/服务器' : ServerGet.GetNames,
}
required = msg.content
reply = ""
for cmd,func in commands.items():
if required.find(cmd) != -1:
try:
ctx = required.split(cmd)[1]
reply = func(ctx)
except:
reply = "服务异常"
break
if reply.__len__() == 0:
reply = "你在发什么东西我怎么看不懂?"
md = MarkdownPayload(content=reply)
await msg.reply(msg_type=2, markdown=md)
def group_add(self,gid :str):
# 加入群聊
new_group = True
if not groups is None :
for i in groups.groups :
if i.find(gid) != -1 :
new_group = False
if new_group :
groups.groups.append(gid)
groups.save()
# 触发概率大,以后要做好解析防止刷屏
async def parseGlobalMessage(self,payload):
try :
msg = payload['d']
gid = msg['group_id']
content = msg['content']
except:
return
# 群聊加入
self.group_add(gid)
# 解析指令
commands:dict[str,types.MethodType|types.FunctionType] = {
r'吃什么' : lambda :"是啊吃什么"
}
required = content
reply = ""
for cmd,func in commands.items():
if required.find(cmd) != -1:
try:
@@ -31,20 +169,38 @@ class djtuWebsocket (botpy.Client) :
reply = "服务异常"
break
if reply.__len__() == 0:
reply = "你在发什么东西我怎么看不懂?"
await msg.reply(content=reply)
return
await self.api.post_group_message(gid,content=reply)
# 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()
for i in groups.groups:
asyncio.create_task(self.api.post_group_message(i,content=msg))
except Exception as e:
logger.error("MCWatcher发送群聊任务失败:\n"+str(e))
async def parseGlobalMessage(self,msg:Message):
await msg.reply(content="awda")
# 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")
groups = group.GroupCtl()
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 :
@@ -53,6 +209,31 @@ if appid is None or secret_bot is None :
raise SystemError("未找到bot配置")
bobby = bobbySession(bbyAccount,bbyPasswd)
#bot 启动
ws = djtuWebsocket(botpy.Intents(public_messages=True))
ws.run(appid=appid,secret=secret_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(Intents_ex(public_messages=True,member_event=True))
# 启动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))
+4
View File
@@ -0,0 +1,4 @@
SECRET=""
APP_ID=""
BOBBY_PASSWORD=""
BOBBY_ACCOUNT=""
+22
View File
@@ -0,0 +1,22 @@
import os
# 记录加入的群组
class GroupCtl :
path = os.path.join(os.getcwd(),"joined_group.cfg")
groups:list[str] = []
def __init__(self) -> None:
with open(self.openfile()) as awa:
self.groups = awa.readlines()
def save(self):
# 重新加载
fd = self.openfile()
os.truncate(self.path,0)
os.write(fd,bytes('\n'.join(self.groups),'utf-8'))
os.close(fd)
def openfile(self):
return os.open(self.path,os.O_RDWR|os.O_CREAT)
+6
View File
@@ -0,0 +1,6 @@
{
"last_snapshot_id": "26.3",
"last_release_id": "26.3",
"targets": [],
"etag": null
}
+359
View File
@@ -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)
+1
View File
@@ -1,3 +1,4 @@
requests
qq-botpy
dotenv
httpx
+66
View File
@@ -0,0 +1,66 @@
import requests
from typing import List
class ServerGet :
url = r'https://mcserve-panel.hiso.de5.net/api/overview'
@staticmethod
def __getNames():
try:
res = requests.get(ServerGet.url,timeout=5)
data = res.json()
servers:list = data['servers']
return servers
except Exception as e :
return None
@staticmethod
def __getIndex(ind:int):
if ind <1 :
return None
try:
res = requests.get(ServerGet.url,timeout=5)
data = res.json()
servers:list = data['servers']
if ind>servers.__len__():
return None
server = servers[ind-1]
return server
except Exception as e :
return None
@staticmethod
def GetNames(ctx:str):
faultmsg = "(っ °Д °;)っ出错了呜....."
servers = ServerGet.__getNames()
if servers is None :
return faultmsg
if servers.__len__() == 0:
return "(っ °Д °;)っ好像...没有服务器呢"
# 试探是否有ctx对应的信息
try :
for s in servers:
if ctx.find(s['name']) != -1:
name = s['name']
s_id = s['id']
status = s['status']
online = s['online']
url = s['url']
addr = s['address']
msg = "{0} 信息如下:\n \nid: {1}\n状态: {2}\n在线人数: {3}\n信息页面: {4}\n地址: {5}".format(name,s_id,status,online,url,addr)
return msg
except:
'交给轮询试试'
# 服务器轮训
msg = 'mc服务器如下:\n \n'
try :
for i in servers:
msg += '' + """- <qqbot-cmd-input text="/服务器 {0}" show="{0}" reference="false" />""".format(str(i['name'])) +' ' + "状态:"+str(i['status'])
msg +='\n'
msg += ' \n点击蓝字获取详情~'
return msg
except:
return faultmsg