#!/usr/bin/env python3 """XMPP Bot xiaoguo@yoin.fun - 跑在 Linux 上""" import asyncio, logging, ssl, json, urllib.request, subprocess, re from xml.sax.saxutils import escape logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s') GATEWAY = "http://localhost:8645/v1/chat/completions" API_KEY = "hermes123" _opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) def send(from_jid, to_jid, body): safe = escape(body) subprocess.run(["docker","exec","ejabberd","ejabberdctl","send_stanza", from_jid, to_jid, f"{safe}" ], capture_output=True, timeout=10) class XiaoGuoBot: def __init__(self): import slixmpp self.xmpp = slixmpp.ClientXMPP('xiaoguo@yoin.fun', 'hermes123') self.xmpp.add_event_handler('session_bind', self.on_bind) self.xmpp.add_event_handler('message', self.on_msg) self.xmpp.add_event_handler('disconnected', self.on_disconnect) ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE self.xmpp.ssl_context = ctx self.ready = asyncio.Event() self._call_seq = 0 async def on_bind(self, event): self.xmpp.send_presence() self.xmpp.get_roster() # 加入内核组 self.xmpp.plugin['xep_0045'].join_muc('coregroup@conference.yoin.fun', 'xiaoguo') self.ready.set() logging.info("✅ 小果上线") async def on_disconnect(self, event): self.ready.clear() logging.warning("⚠️ 小果断线") async def on_msg(self, msg): body = msg['body'] sender = str(msg['from']) msg_type = msg['type'] if not body: return # 群聊 if msg_type == 'groupchat': if 'xiaoguo@yoin.fun' in sender: return nickname = sender.split('/')[-1] if '/' in sender else '' if nickname in ('hmo', 'xxm'): logging.info(f"📩 群消息 [{sender}]: {body[:80]}") room = sender.split('/')[0] ctx_body = f"[核心群 {room}] {nickname} 说: {body}" await self.call_hermes(ctx_body, room, is_group=True) return # 私聊 if msg_type == 'chat' and 'hmo@yoin.fun' in sender: self._call_seq += 1 logging.info(f"📩 老爸(#{self._call_seq}): {body}") await self.call_hermes(body, sender) async def call_hermes(self, content, sender, is_group=False): msg_type = 'groupchat' if is_group else 'chat' try: payload = json.dumps({ "model": "hermes-agent", "messages": [{"role": "user", "content": f"[xiaoguo] {content}"}] }).encode() req = urllib.request.Request(GATEWAY, data=payload, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {API_KEY}") req.add_header("X-Hermes-Session-Id", "xmpp-xiaoguo") loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, lambda: _opener.open(req, timeout=600)) data = json.loads(result.read()) reply = data.get("choices", [{}])[0].get("message", {}).get("content", "") finish = data.get("choices", [{}])[0].get("finish_reason", "") # 处理 __SILENT__ 和 __REPLY__ 标记（和莫荷保持一致） stripped = reply.strip() if stripped.startswith('__SILENT__'): logging.info("⏭️ 小果决定沉默，不发送") return # 安全网：过滤沉默宣告类文本（防止 LLM 不按规则走） if re.match(r'^(保持安静|不插嘴|我沉默了|收到|明白|好的|在的?|在呢|来了|沉默|安静)([,，!！。.？\s]|$)', stripped, re.IGNORECASE): logging.info(f"⏭️ 小果沉默宣告被拦截: {stripped[:60]}") return reply = re.sub(r'^__REPLY__\s*', '', reply) if reply.strip() and finish != "silent": if is_group: self.xmpp.send_message(mto=sender, mbody=reply, mtype='groupchat') else: send("xiaoguo@yoin.fun", sender, reply) logging.info(f"✅ 小果回复: {reply[:80]}") except Exception as e: logging.error(f"❌ 小果错误: {e}") async def main(): while True: try: z = XiaoGuoBot() z.xmpp.register_plugin('xep_0030') z.xmpp.register_plugin('xep_0045') z.xmpp.register_plugin('xep_0199') z.xmpp.connect(host='127.0.0.1', port=5222) await asyncio.wait_for(z.ready.wait(), timeout=30) logging.info("小果就绪") await asyncio.Event().wait() except Exception as e: logging.error(f"小果main错误: {e}") await asyncio.sleep(3) if __name__ == '__main__': asyncio.run(main())