diff --git a/deploy/bot/xmpp_agent_core.py b/deploy/bot/xmpp_agent_core.py index 0f4ae0f5..489b839b 100644 --- a/deploy/bot/xmpp_agent_core.py +++ b/deploy/bot/xmpp_agent_core.py @@ -1,505 +1,523 @@ -#!/usr/bin/env python3 -""" -Core XMPP Agent — shared logic for zhiwei / mohe / xxm bots. -Imports by xmpp_zhiwei_bot.py / xmpp_mohe_bot.py with --agent flag. -""" -import os, sys, json, time, logging, threading, traceback -from http.server import HTTPServer, BaseHTTPRequestHandler -from urllib.parse import urlparse, parse_qs -from hashlib import md5 - -import slixmpp -from slixmpp import JID -import asyncio - -# ── Per-agent configuration ── -PER_AGENT = { - "mohe": { - "jid": "mohe@yoin.fun", - "password": "hermes123", - "nick": "mohe", - "http_port": 5808, - "gateway_url": "http://localhost:8642/v1/chat/completions", - "gateway_api_key": "hermes123", - "session_id": "xmpp-mohe-v2", - "name_cn": "莫荷", - "mention": "@mohe/@莫荷", - }, - "zhiwei": { - "jid": "zhiwei@yoin.fun", - "password": "2nw4psra", - "nick": "zhiwei", - "http_port": 5805, - "gateway_url": "http://localhost:8643/v1/chat/completions", - "gateway_api_key": "hermes123", - "session_id": "xmpp-zhiwei-v3", - "name_cn": "知微", - "mention": "@知微/zhiwei", - }, -} -_DEFAULT_AGENT = "mohe" - -# ── Module-level config (populated by _apply_config after agent detection) ── -AGENT_NAME = "" -XMPP_JID = "" -XMPP_PASSWORD = "" -MUC_ROOM = "coregroup@conference.yoin.fun" -MUC_NICK = "" -AGENT_MENTION = "" -HTTP_PORT = 5808 -AGENT_NICK = "" -ACK_DELAY = 120 # 真卡死才提示（普通 LLM 冷启动 20-100s 不应触发） -GATEWAY_URL = "" -GATEWAY_API_KEY = "" -GATEWAY_SESSION_ID = "" -GATEWAY_DEADLINE_SECONDS = 600 -CALL_HERMES_TIMEOUT = 600 # agent 带工具调用（读文件/查数据）需几分钟 -FALLBACK_REPLY = "请稍等，我在处理..." - -# ── 图片 OCR 配置（截图消息 → SenseNova vision）── -OCR_CONFIG_FILE = "/home/hmo/.config/mofin/ocr_config.json" -OCR_TIMEOUT = 90 -IMAGE_EXTS = ('.jpg', '.jpeg', '.png', '.webp', '.gif') - - -def _load_ocr_config(): - """从 /home/hmo/.config/mofin/ocr_config.json 读取 SenseNova 配置。 - 文件不存在或未配置 → 返回 (None, None, None)，图片功能降级为仅提示收到。""" - try: - with open(OCR_CONFIG_FILE, 'r', encoding='utf-8') as f: - cfg = json.load(f) - return cfg.get('key'), cfg.get('base_url'), cfg.get('model', 'sensenova-6.7-flash-lite') - except Exception: - return None, None, None - - -def _download_image(url, timeout=30): - """下载图片字节。失败返回 None。""" - import urllib.request - import ssl - try: - ctx = ssl.create_default_context() - ctx.check_hostname = False - ctx.verify_mode = ssl.CERT_NONE - req = urllib.request.Request(url, headers={'User-Agent': 'curl/8.5.0'}) - with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: - data = resp.read() - return data if len(data) > 100 else None - except Exception as e: - log.error(f"图片下载失败 {url[:80]}: {e}") - return None - - -def _ocr_image(img_data, prompt=None): - """调 SenseNova vision OCR。返回 (ok, text)。""" - key, base, model = _load_ocr_config() - if not (key and base): - return False, "OCR未配置" - import base64 - import urllib.request - b64 = base64.b64encode(img_data).decode() - text_prompt = prompt or "请识别这张图片中的所有文字内容，包括数字、股票名称、金额、日期。用中文回复。" - payload = json.dumps({ - "model": model, - "messages": [{ - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}, - {"type": "text", "text": text_prompt}, - ] - }], - "max_tokens": 1500, - }).encode() - req = urllib.request.Request( - f"{base.rstrip('/')}/chat/completions", data=payload, - headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"}) - try: - with urllib.request.urlopen(req, timeout=OCR_TIMEOUT) as resp: - data = json.loads(resp.read().decode()) - msg = data.get('choices', [{}])[0].get('message', {}) - text = msg.get('content', '') or msg.get('reasoning', '') - return True, text.strip() if text.strip() else "(OCR无内容)" - except Exception as e: - return False, f"OCR失败: {str(e)[:120]}" - - -def _is_image_url(url): - u = url.lower().split('?')[0] - return ('upload.yoin.fun' in u or '/upload/' in u or u.endswith(IMAGE_EXTS)) - - -def _process_image_message(url): - """下载+OCR 一张截图，返回注入 LLM 的上下文文本。""" - img = _download_image(url) - if not img: - return "[老爸发来一张截图，但下载失败，无法识别]" - ok, text = _ocr_image(img) - if not ok: - return f"[老爸发来一张截图，OCR识别失败: {text}]" - return (f"[老爸发来一张截图，OCR识别内容如下]\n{text}\n" - f"[截图内容结束] 请基于截图内容回应老爸。") - -# ── 全局队列 ── -_outbound_queue = [] -_outbound_lock = threading.Lock() -_inbound_queue = [] -_inbound_lock = threading.Lock() - -RECENT_SENT_MAX = 50 - -# ── XMPP 消息日志（dashboard 健康Tab「最近对话」数据源）── -try: - sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..')) - from xmpp_logger import log_xmpp as _log_xmpp -except Exception: - def _log_xmpp(*a, **kw): - pass - - -def _rs(p): - """Parse agent arg from sys.argv, returns agent name string.""" - agent = _DEFAULT_AGENT - skip_next = False - for i, a in enumerate(sys.argv[1:]): - if skip_next: - skip_next = False - continue - if a.startswith('--agent='): - agent = a.split('=', 1)[1] - elif a == '--agent' and i + 1 < len(sys.argv[1:]): - agent = sys.argv[i + 2] - skip_next = True - return agent - - -agent, is_mohe = _rs(None), None - -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s %(levelname)s %(message)s', - stream=sys.stdout, -) -log = logging.getLogger('xmpp_agent') - - -def _apply_config(agent_name): - """Set module-level config variables from PER_AGENT dict + env overrides.""" - global AGENT_NAME, XMPP_JID, XMPP_PASSWORD, MUC_NICK, AGENT_MENTION - global HTTP_PORT, AGENT_NICK, GATEWAY_URL, GATEWAY_API_KEY, GATEWAY_SESSION_ID - cfg = PER_AGENT.get(agent_name, PER_AGENT.get(_DEFAULT_AGENT, {})) - AGENT_NAME = agent_name - XMPP_JID = os.environ.get('XMPP_JID', cfg.get('jid', '')) - XMPP_PASSWORD = os.environ.get('XMPP_PASSWORD', cfg.get('password', '')) - MUC_NICK = os.environ.get('MUC_NICK', cfg.get('nick', agent_name)) - AGENT_MENTION = os.environ.get('AGENT_MENTION', cfg.get('mention', '')) - HTTP_PORT = int(os.environ.get('HTTP_PORT', cfg.get('http_port', 5808))) - AGENT_NICK = os.environ.get('AGENT_NICK', cfg.get('nick', agent_name)) - GATEWAY_URL = os.environ.get('GATEWAY_URL', cfg.get('gateway_url', '')) - GATEWAY_API_KEY = os.environ.get('GATEWAY_API_KEY', cfg.get('gateway_api_key', '')) - GATEWAY_SESSION_ID = os.environ.get('GATEWAY_SESSION_ID', cfg.get('session_id', '')) - - -# ── Periodic ACK task ── -class AckManager: - def __init__(self): - self._active = {} - self._lock = threading.Lock() - - def start(self, session_id, to_jid, msg_body): - """Record an active LLM analysis and schedule the ACK.""" - with self._lock: - self._active[session_id] = { - 'to_jid': to_jid, - 'body': msg_body[:80], - 'started': time.time(), - 'acked': False, - } - - def ack(self, session_id): - with self._lock: - self._active.pop(session_id, None) - - def tick(self, bot): - now = time.time() - to_send = [] - with self._lock: - for sid, info in list(self._active.items()): - if not info['acked'] and now - info['started'] >= ACK_DELAY: - info['acked'] = True - to_send.append((info['to_jid'], FALLBACK_REPLY)) - for jid, msg in to_send: - try: - bot.send_message(mto=jid, mbody=msg, mtype='chat') - except Exception: - pass - - -ack_mgr = AckManager() - - -# ── Slixmpp Bot ── -class XmppAgent(slixmpp.ClientXMPP): - def __init__(self, jid, password, room, nick): - super().__init__(jid, password) - self._room = room - self._nick = nick - self._muc_joined = False - self._recent_sent = [] - self._reconnecting = False # 防递归重连 - self.add_event_handler('session_start', self.on_start) - self.add_event_handler('message', self.on_msg) - self.add_event_handler('disconnected', self.on_disconnect) - self.register_plugin('xep_0030') - self.register_plugin('xep_0045') - self.register_plugin('xep_0199') - self.register_plugin('xep_0066') # OOB 附件（截图 URL） - - async def on_start(self, event): - self.send_presence() - await self.get_roster() - try: - await self.plugin['xep_0045'].join_muc(self._room, self._nick) - self._muc_joined = True - log.info(f"{AGENT_NAME} XMPP 就绪 (已加入 {self._room})") - except Exception as e: - log.error(f"{AGENT_NAME} MUC加入失败: {e}") - - def on_disconnect(self, event): - self._muc_joined = False - log.info(f"{AGENT_NAME} XMPP 断开") - if self._reconnecting: - log.warning(f"{AGENT_NAME} 已在重连中，跳过递归重连") - return - self._reconnecting = True - try: - self.reconnect(wait=5.0, reason="断线自动重连") - except Exception as e: - log.warning(f"{AGENT_NAME} 重连失败: {e}") - finally: - self._reconnecting = False - - def on_msg(self, msg): - if msg['type'] in ('chat', 'groupchat'): - body = str(msg['body']).strip() - # body 为空时检查 OOB 附件（XEP-0066：截图/文件以 OOB url 送达） - if not body: - try: - oob_url = str(msg['oob']['url'] or '').strip() - except Exception: - oob_url = '' - if oob_url and _is_image_url(oob_url): - body = f"[IMAGE] {oob_url}" - log.info(f"📩 收到图片(OOB): from={msg['from']} url={oob_url[:80]}") - else: - return - # body 本身是上传 URL（部分客户端把 URL 直接放 body） - elif body.startswith('http') and _is_image_url(body): - log.info(f"📩 收到图片(URL body): from={msg['from']} url={body[:80]}") - body = f"[IMAGE] {body}" - log.info(f"📩 收到: from={msg['from']} type={msg['type']} body={body[:60]}") - _log_xmpp('in', sender_str := str(msg['from']), f"{AGENT_NAME}@yoin.fun", body) - if ('[executor]' in body and 'gateway_zhiwei' in body): - log.info(f"过滤 executor 消息: {body[:80]}...") - return - sender = str(msg['from']) - msg_type = msg['type'] - for s in self._recent_sent: - if body[:50] in s or s in body[:50]: - return - if msg_type == 'groupchat': - nick = sender.split('/')[-1] if '/' in sender else '' - if nick == AGENT_NICK: - return - mention_list = AGENT_MENTION.replace('@', '').split('/') - is_for_me = any(m in body for m in ['@' + m for m in mention_list] + mention_list) - if not is_for_me: - return - with _inbound_lock: - _inbound_queue.append((sender, body, msg_type)) - - def mark_sent(self, body: str): - self._recent_sent.append(body[:80]) - if len(self._recent_sent) > RECENT_SENT_MAX * 2: - self._recent_sent = self._recent_sent[-RECENT_SENT_MAX:] - - -# ── Deliver loop ── -def _deliver_loop(bot): - global _outbound_queue - while True: - try: - items = [] - with _outbound_lock: - items, _outbound_queue = _outbound_queue[:], [] - for target, text, msg_type in items: - try: - async def _send(to, body, mtype): - bot.send_message(mto=to, mbody=body, mtype=mtype) - asyncio.run(_send(target, text, msg_type)) - bot.mark_sent(text) - _log_xmpp('out', f"{AGENT_NAME}@yoin.fun", target, text) - log.info(f" 已发送到 {target}: {text[:80]}") - except Exception as e: - _log_xmpp('out', f"{AGENT_NAME}@yoin.fun", target, text, status='error', error=str(e)[:150]) - log.error(f" 发送到 {target} 失败: {e}") - time.sleep(0.3) - except Exception as e: - log.error(f"_deliver_loop error: {e}") - time.sleep(1) - - -# ── Inbound processing loop ── -def _inbound_loop(bot): - global _inbound_queue - while True: - try: - time.sleep(0.2) - with _inbound_lock: - if not _inbound_queue: - continue - sender, body, msg_type = _inbound_queue.pop(0) - log.info(f"🔄 inbound处理: {body[:40]}") - ack_mgr.start(body[:40], sender, body) - # 截图消息：下载 + OCR → 注入上下文 - if body.startswith('[IMAGE]'): - img_url = body[len('[IMAGE]'):].strip() - log.info(f"🖼️ 图片消息处理: {img_url[:80]}") - body = _process_image_message(img_url) - log.info(f"🖼️ OCR完成: {body[:100]}") - reply = call_hermes(body) - ack_mgr.ack(body[:40]) - if reply: - with _outbound_lock: - _outbound_queue.append((sender, reply, 'chat')) - except Exception as e: - log.error(f"_inbound_loop error: {e}") - time.sleep(1) - - -# ── HTTP SendHandler ── -class SendHandler(BaseHTTPRequestHandler): - def do_POST(self): - content_len = int(self.headers.get('Content-Length', 0)) - post_body = self.rfile.read(content_len) - try: - data = json.loads(post_body) - target = data.get('to', '') - text = data.get('body', '') - if not target or not text: - self.send_response(400) - self.end_headers() - self.wfile.write(b'{"error":"missing to or body"}') - return - if '修复失败' in text and 'gateway_zhiwei' in text: - self.send_response(200) - self.end_headers() - self.wfile.write(b'{"ok":true,"filtered":true}') - return - msg_type = data.get('type', 'chat') - if text: - _outbound_queue.append((target, text, msg_type)) - self.send_response(200) - self.end_headers() - self.wfile.write(b'{"ok":true}') - log.info(f"SendHandler: enqueued -> {target}: {text[:80]}") - except Exception as e: - traceback.print_exc() - self.send_response(500) - self.end_headers() - self.wfile.write(str({'error': str(e)}).encode()) - - def log_message(self, format, *args): - pass - - -def _run_http_server(): - server = HTTPServer(('127.0.0.1', HTTP_PORT), SendHandler) - log.info(f"HTTP SendHandler listening on 127.0.0.1:{HTTP_PORT}") - server.serve_forever() - - -# ── call_hermes ── -def call_hermes(content: str, session_id=None) -> str: - now_str = time.strftime("[%Y-%m-%d %H:%M %A]", time.localtime()) - timed_content = f"{now_str}\n{content}" - payload = { - 'model': 'hermes-agent', - 'messages': [ - {'role': 'user', 'content': timed_content}, - ], - 'stream': False, - } - if session_id is None: - session_id = GATEWAY_SESSION_ID - headers = { - 'Content-Type': 'application/json', - 'X-Hermes-Session-Id': session_id, - } - if GATEWAY_API_KEY: - headers['Authorization'] = f'Bearer {GATEWAY_API_KEY}' - - import urllib.request - data_bytes = json.dumps(payload).encode('utf-8') - req = urllib.request.Request(GATEWAY_URL, data=data_bytes, headers=headers, method='POST') - try: - resp = urllib.request.urlopen(req, timeout=CALL_HERMES_TIMEOUT) - resp_data = json.loads(resp.read().decode('utf-8')) - reply = '' - if 'choices' in resp_data and len(resp_data['choices']) > 0: - choice = resp_data['choices'][0] - if 'message' in choice and 'content' in choice['message']: - reply = choice['message']['content'] - elif 'delta' in choice and 'content' in choice['delta']: - reply = choice['delta']['content'] - if not reply: - reply = resp_data.get('response', '') - if not reply: - reply = str(resp_data) - return reply.strip() - except urllib.request.HTTPError as e: - err_body = e.read().decode('utf-8', errors='replace') - log.error(f"call_hermes HTTP {e.code}: {err_body[:200]}") - return '' - except Exception as e: - log.error(f"call_hermes error: {type(e).__name__}: {e}") - return '' - - -# ── ACK manager tick ── -def _ack_tick(bot): - while True: - try: - ack_mgr.tick(bot) - time.sleep(5) - except Exception: - time.sleep(5) - - -# ── Main ── -def main(): - global is_mohe - agent_name = _rs(None) - _apply_config(agent_name) - is_mohe = (agent_name == 'mohe') - - log.info(f"Starting XMPP Agent: {agent_name} (mohe={is_mohe})") - log.info(f" JID={XMPP_JID} HTTP_PORT={HTTP_PORT}") - log.info(f" GATEWAY={GATEWAY_URL}") - log.info(f" SESSION_ID={GATEWAY_SESSION_ID}") - - bot = XmppAgent(XMPP_JID, XMPP_PASSWORD, MUC_ROOM, MUC_NICK) - bot.connect(host='127.0.0.1', port=5222) - t_deliver = threading.Thread(target=_deliver_loop, args=(bot,), daemon=True) - t_deliver.start() - t_inbound = threading.Thread(target=_inbound_loop, args=(bot,), daemon=True) - t_inbound.start() - t_ack = threading.Thread(target=_ack_tick, args=(bot,), daemon=True) - t_ack.start() - t_http = threading.Thread(target=_run_http_server, daemon=True) - t_http.start() - - bot.loop.run_forever() - - -if __name__ == '__main__': - main() +#!/usr/bin/env python3 +""" +Core XMPP Agent — shared logic for zhiwei / mohe / xxm bots. +Imports by xmpp_zhiwei_bot.py / xmpp_mohe_bot.py with --agent flag. +""" +import os, sys, json, time, logging, threading, traceback +from http.server import HTTPServer, BaseHTTPRequestHandler +from urllib.parse import urlparse, parse_qs +from hashlib import md5 + +import slixmpp +from slixmpp import JID +import asyncio + +# ── Per-agent configuration ── +PER_AGENT = { + "mohe": { + "jid": "mohe@yoin.fun", + "password": "hermes123", + "nick": "mohe", + "http_port": 5808, + "gateway_url": "http://localhost:8642/v1/chat/completions", + "gateway_api_key": "hermes123", + "session_id": "xmpp-mohe-v2", + "name_cn": "莫荷", + "mention": "@mohe/@莫荷", + }, + "zhiwei": { + "jid": "zhiwei@yoin.fun", + "password": "2nw4psra", + "nick": "zhiwei", + "http_port": 5805, + "gateway_url": "http://localhost:8643/v1/chat/completions", + "gateway_api_key": "hermes123", + "session_id": "xmpp-zhiwei-v3", + "name_cn": "知微", + "mention": "@知微/zhiwei", + }, +} +_DEFAULT_AGENT = "mohe" + +# ── Module-level config (populated by _apply_config after agent detection) ── +AGENT_NAME = "" +XMPP_JID = "" +XMPP_PASSWORD = "" +MUC_ROOM = "coregroup@conference.yoin.fun" +MUC_NICK = "" +AGENT_MENTION = "" +HTTP_PORT = 5808 +AGENT_NICK = "" +ACK_DELAY = 120 # 真卡死才提示（普通 LLM 冷启动 20-100s 不应触发） +GATEWAY_URL = "" +GATEWAY_API_KEY = "" +GATEWAY_SESSION_ID = "" +GATEWAY_DEADLINE_SECONDS = 600 +CALL_HERMES_TIMEOUT = 600 # agent 带工具调用（读文件/查数据）需几分钟 +FALLBACK_REPLY = "请稍等，我在处理..." + +# ── 图片 OCR 配置（截图消息 → SenseNova vision）── +OCR_CONFIG_FILE = "/home/hmo/.config/mofin/ocr_config.json" +OCR_TIMEOUT = 90 +IMAGE_EXTS = ('.jpg', '.jpeg', '.png', '.webp', '.gif') + + +def _load_ocr_config(): + """从 /home/hmo/.config/mofin/ocr_config.json 读取 SenseNova 配置。 + 文件不存在或未配置 → 返回 (None, None, None)，图片功能降级为仅提示收到。""" + try: + with open(OCR_CONFIG_FILE, 'r', encoding='utf-8') as f: + cfg = json.load(f) + return cfg.get('key'), cfg.get('base_url'), cfg.get('model', 'sensenova-6.7-flash-lite') + except Exception: + return None, None, None + + +def _download_image(url, timeout=30, retries=3): + """下载图片字节，带重试。失败返回 None。 + + 重试原因：客户端上传和发消息是并发的，bot 收到消息时上传可能还没写完， + 立即 GET 会 404（<100 字节错误页）。等 2 秒再试即可。 + """ + import urllib.request + import ssl + import time as _t + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + for attempt in range(1, retries + 1): + try: + req = urllib.request.Request(url, headers={'User-Agent': 'curl/8.5.0'}) + with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: + data = resp.read() + if len(data) > 100: + return data + log.warning(f"图片下载第{attempt}次: 响应过小({len(data)}B, 可能上传未完成)，重试...") + except Exception as e: + log.warning(f"图片下载第{attempt}次失败: {str(e)[:100]}") + if attempt < retries: + _t.sleep(2) + log.error(f"图片下载最终失败({retries}次尝试): {url}") + return None + + +def _ocr_image(img_data, prompt=None): + """调 SenseNova vision OCR。返回 (ok, text)。""" + key, base, model = _load_ocr_config() + if not (key and base): + return False, "OCR未配置" + import base64 + import urllib.request + b64 = base64.b64encode(img_data).decode() + text_prompt = prompt or "请识别这张图片中的所有文字内容，包括数字、股票名称、金额、日期。用中文回复。" + payload = json.dumps({ + "model": model, + "messages": [{ + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}, + {"type": "text", "text": text_prompt}, + ] + }], + "max_tokens": 1500, + }).encode() + req = urllib.request.Request( + f"{base.rstrip('/')}/chat/completions", data=payload, + headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"}) + try: + with urllib.request.urlopen(req, timeout=OCR_TIMEOUT) as resp: + data = json.loads(resp.read().decode()) + msg = data.get('choices', [{}])[0].get('message', {}) + text = msg.get('content', '') or msg.get('reasoning', '') + return True, text.strip() if text.strip() else "(OCR无内容)" + except Exception as e: + return False, f"OCR失败: {str(e)[:120]}" + + +def _is_image_url(url): + u = url.lower().split('?')[0] + return ('upload.yoin.fun' in u or '/upload/' in u or u.endswith(IMAGE_EXTS)) + + +def _process_image_message(url): + """下载+OCR 一张截图，返回注入 LLM 的上下文文本。""" + img = _download_image(url) + if not img: + return (f"[老爸发来一张截图，bot 下载失败（HTTP错误），无法识别。" + f"图片URL: {url}\n" + f"注：bot 的识图管道是 SenseNova（商汤云OCR），不是小果也不是 tesseract。" + f"下载失败可能是 URL 被截断或 ejabberd 上传过期，请老爸重发一次试试。") + ok, text = _ocr_image(img) + if not ok: + return (f"[老爸发来一张截图，SenseNova OCR 识别失败: {text}。" + f"注：识图管道是 SenseNova，如持续失败请检查 OCR key 配额。") + return (f"[老爸发来一张截图，SenseNova OCR 识别内容如下]\n{text}\n" + f"[截图内容结束] 请基于截图内容回应老爸。") + +# ── 全局队列 ── +_outbound_queue = [] +_outbound_lock = threading.Lock() +_inbound_queue = [] +_inbound_lock = threading.Lock() + +RECENT_SENT_MAX = 50 + +# ── XMPP 消息日志（dashboard 健康Tab「最近对话」数据源）── +try: + sys.path.insert(0, '/home/hmo/MoFin') # 绝对路径（红线#7），相对解析会指到 / + from xmpp_logger import log_xmpp as _log_xmpp + _LOG_XMPP_OK = True +except Exception as _e: + _LOG_XMPP_OK = False + def _log_xmpp(*a, **kw): + pass + print(f"[WARN] xmpp_logger import 失败，消息日志降级为 no-op: {_e}", file=sys.stderr) + + +def _rs(p): + """Parse agent arg from sys.argv, returns agent name string.""" + agent = _DEFAULT_AGENT + skip_next = False + for i, a in enumerate(sys.argv[1:]): + if skip_next: + skip_next = False + continue + if a.startswith('--agent='): + agent = a.split('=', 1)[1] + elif a == '--agent' and i + 1 < len(sys.argv[1:]): + agent = sys.argv[i + 2] + skip_next = True + return agent + + +agent, is_mohe = _rs(None), None + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s %(levelname)s %(message)s', + stream=sys.stdout, +) +log = logging.getLogger('xmpp_agent') + + +def _apply_config(agent_name): + """Set module-level config variables from PER_AGENT dict + env overrides.""" + global AGENT_NAME, XMPP_JID, XMPP_PASSWORD, MUC_NICK, AGENT_MENTION + global HTTP_PORT, AGENT_NICK, GATEWAY_URL, GATEWAY_API_KEY, GATEWAY_SESSION_ID + cfg = PER_AGENT.get(agent_name, PER_AGENT.get(_DEFAULT_AGENT, {})) + AGENT_NAME = agent_name + XMPP_JID = os.environ.get('XMPP_JID', cfg.get('jid', '')) + XMPP_PASSWORD = os.environ.get('XMPP_PASSWORD', cfg.get('password', '')) + MUC_NICK = os.environ.get('MUC_NICK', cfg.get('nick', agent_name)) + AGENT_MENTION = os.environ.get('AGENT_MENTION', cfg.get('mention', '')) + HTTP_PORT = int(os.environ.get('HTTP_PORT', cfg.get('http_port', 5808))) + AGENT_NICK = os.environ.get('AGENT_NICK', cfg.get('nick', agent_name)) + GATEWAY_URL = os.environ.get('GATEWAY_URL', cfg.get('gateway_url', '')) + GATEWAY_API_KEY = os.environ.get('GATEWAY_API_KEY', cfg.get('gateway_api_key', '')) + GATEWAY_SESSION_ID = os.environ.get('GATEWAY_SESSION_ID', cfg.get('session_id', '')) + + +# ── Periodic ACK task ── +class AckManager: + def __init__(self): + self._active = {} + self._lock = threading.Lock() + + def start(self, session_id, to_jid, msg_body): + """Record an active LLM analysis and schedule the ACK.""" + with self._lock: + self._active[session_id] = { + 'to_jid': to_jid, + 'body': msg_body[:80], + 'started': time.time(), + 'acked': False, + } + + def ack(self, session_id): + with self._lock: + self._active.pop(session_id, None) + + def tick(self, bot): + now = time.time() + to_send = [] + with self._lock: + for sid, info in list(self._active.items()): + if not info['acked'] and now - info['started'] >= ACK_DELAY: + info['acked'] = True + to_send.append((info['to_jid'], FALLBACK_REPLY)) + for jid, msg in to_send: + try: + bot.send_message(mto=jid, mbody=msg, mtype='chat') + except Exception: + pass + + +ack_mgr = AckManager() + + +# ── Slixmpp Bot ── +class XmppAgent(slixmpp.ClientXMPP): + def __init__(self, jid, password, room, nick): + super().__init__(jid, password) + self._room = room + self._nick = nick + self._muc_joined = False + self._recent_sent = [] + self._reconnecting = False # 防递归重连 + self.add_event_handler('session_start', self.on_start) + self.add_event_handler('message', self.on_msg) + self.add_event_handler('disconnected', self.on_disconnect) + self.register_plugin('xep_0030') + self.register_plugin('xep_0045') + self.register_plugin('xep_0199') + self.register_plugin('xep_0066') # OOB 附件（截图 URL） + + async def on_start(self, event): + self.send_presence() + await self.get_roster() + try: + await self.plugin['xep_0045'].join_muc(self._room, self._nick) + self._muc_joined = True + log.info(f"{AGENT_NAME} XMPP 就绪 (已加入 {self._room})") + except Exception as e: + log.error(f"{AGENT_NAME} MUC加入失败: {e}") + + def on_disconnect(self, event): + self._muc_joined = False + log.info(f"{AGENT_NAME} XMPP 断开") + if self._reconnecting: + log.warning(f"{AGENT_NAME} 已在重连中，跳过递归重连") + return + self._reconnecting = True + try: + self.reconnect(wait=5.0, reason="断线自动重连") + except Exception as e: + log.warning(f"{AGENT_NAME} 重连失败: {e}") + finally: + self._reconnecting = False + + def on_msg(self, msg): + if msg['type'] in ('chat', 'groupchat'): + body = str(msg['body']).strip() + # body 为空时检查 OOB 附件（XEP-0066：截图/文件以 OOB url 送达） + if not body: + try: + oob_url = str(msg['oob']['url'] or '').strip() + except Exception: + oob_url = '' + if oob_url and _is_image_url(oob_url): + body = f"[IMAGE] {oob_url}" + log.info(f"📩 收到图片(OOB): from={msg['from']} url={oob_url}") + else: + return + # body 本身是上传 URL（部分客户端把 URL 直接放 body） + elif body.startswith('http') and _is_image_url(body): + log.info(f"📩 收到图片(URL body): from={msg['from']} url={body}") + body = f"[IMAGE] {body}" + log.info(f"📩 收到: from={msg['from']} type={msg['type']} body={body[:60]}") + _log_xmpp('in', sender_str := str(msg['from']), f"{AGENT_NAME}@yoin.fun", body) + if ('[executor]' in body and 'gateway_zhiwei' in body): + log.info(f"过滤 executor 消息: {body[:80]}...") + return + sender = str(msg['from']) + msg_type = msg['type'] + for s in self._recent_sent: + if body[:50] in s or s in body[:50]: + return + if msg_type == 'groupchat': + nick = sender.split('/')[-1] if '/' in sender else '' + if nick == AGENT_NICK: + return + mention_list = AGENT_MENTION.replace('@', '').split('/') + is_for_me = any(m in body for m in ['@' + m for m in mention_list] + mention_list) + if not is_for_me: + return + with _inbound_lock: + _inbound_queue.append((sender, body, msg_type)) + + def mark_sent(self, body: str): + self._recent_sent.append(body[:80]) + if len(self._recent_sent) > RECENT_SENT_MAX * 2: + self._recent_sent = self._recent_sent[-RECENT_SENT_MAX:] + + +# ── Deliver loop ── +def _deliver_loop(bot): + global _outbound_queue + while True: + try: + items = [] + with _outbound_lock: + items, _outbound_queue = _outbound_queue[:], [] + for target, text, msg_type in items: + try: + async def _send(to, body, mtype): + bot.send_message(mto=to, mbody=body, mtype=mtype) + asyncio.run(_send(target, text, msg_type)) + bot.mark_sent(text) + _log_xmpp('out', f"{AGENT_NAME}@yoin.fun", target, text) + log.info(f" 已发送到 {target}: {text[:80]}") + except Exception as e: + _log_xmpp('out', f"{AGENT_NAME}@yoin.fun", target, text, status='error', error=str(e)[:150]) + log.error(f" 发送到 {target} 失败: {e}") + time.sleep(0.3) + except Exception as e: + log.error(f"_deliver_loop error: {e}") + time.sleep(1) + + +# ── Inbound processing loop ── +def _inbound_loop(bot): + global _inbound_queue + while True: + try: + time.sleep(0.2) + with _inbound_lock: + if not _inbound_queue: + continue + sender, body, msg_type = _inbound_queue.pop(0) + log.info(f"🔄 inbound处理: {body[:40]}") + ack_mgr.start(body[:40], sender, body) + # 截图消息：下载 + OCR → 注入上下文 + if body.startswith('[IMAGE]'): + img_url = body[len('[IMAGE]'):].strip() + log.info(f"🖼️ 图片消息处理: {img_url[:80]}") + body = _process_image_message(img_url) + log.info(f"🖼️ OCR完成: {body[:100]}") + reply = call_hermes(body) + ack_mgr.ack(body[:40]) + if reply: + with _outbound_lock: + _outbound_queue.append((sender, reply, 'chat')) + except Exception as e: + log.error(f"_inbound_loop error: {e}") + time.sleep(1) + + +# ── HTTP SendHandler ── +class SendHandler(BaseHTTPRequestHandler): + def do_POST(self): + content_len = int(self.headers.get('Content-Length', 0)) + post_body = self.rfile.read(content_len) + try: + data = json.loads(post_body) + target = data.get('to', '') + text = data.get('body', '') + if not target or not text: + self.send_response(400) + self.end_headers() + self.wfile.write(b'{"error":"missing to or body"}') + return + if '修复失败' in text and 'gateway_zhiwei' in text: + self.send_response(200) + self.end_headers() + self.wfile.write(b'{"ok":true,"filtered":true}') + return + msg_type = data.get('type', 'chat') + if text: + _outbound_queue.append((target, text, msg_type)) + self.send_response(200) + self.end_headers() + self.wfile.write(b'{"ok":true}') + log.info(f"SendHandler: enqueued -> {target}: {text[:80]}") + except Exception as e: + traceback.print_exc() + self.send_response(500) + self.end_headers() + self.wfile.write(str({'error': str(e)}).encode()) + + def log_message(self, format, *args): + pass + + +def _run_http_server(): + server = HTTPServer(('127.0.0.1', HTTP_PORT), SendHandler) + log.info(f"HTTP SendHandler listening on 127.0.0.1:{HTTP_PORT}") + server.serve_forever() + + +# ── call_hermes ── +def call_hermes(content: str, session_id=None) -> str: + now_str = time.strftime("[%Y-%m-%d %H:%M %A]", time.localtime()) + timed_content = f"{now_str}\n{content}" + payload = { + 'model': 'hermes-agent', + 'messages': [ + {'role': 'user', 'content': timed_content}, + ], + 'stream': False, + } + if session_id is None: + session_id = GATEWAY_SESSION_ID + headers = { + 'Content-Type': 'application/json', + 'X-Hermes-Session-Id': session_id, + } + if GATEWAY_API_KEY: + headers['Authorization'] = f'Bearer {GATEWAY_API_KEY}' + + import urllib.request + data_bytes = json.dumps(payload).encode('utf-8') + req = urllib.request.Request(GATEWAY_URL, data=data_bytes, headers=headers, method='POST') + try: + resp = urllib.request.urlopen(req, timeout=CALL_HERMES_TIMEOUT) + resp_data = json.loads(resp.read().decode('utf-8')) + reply = '' + if 'choices' in resp_data and len(resp_data['choices']) > 0: + choice = resp_data['choices'][0] + if 'message' in choice and 'content' in choice['message']: + reply = choice['message']['content'] + elif 'delta' in choice and 'content' in choice['delta']: + reply = choice['delta']['content'] + if not reply: + reply = resp_data.get('response', '') + if not reply: + reply = str(resp_data) + return reply.strip() + except urllib.request.HTTPError as e: + err_body = e.read().decode('utf-8', errors='replace') + log.error(f"call_hermes HTTP {e.code}: {err_body[:200]}") + return '' + except Exception as e: + log.error(f"call_hermes error: {type(e).__name__}: {e}") + return '' + + +# ── ACK manager tick ── +def _ack_tick(bot): + while True: + try: + ack_mgr.tick(bot) + time.sleep(5) + except Exception: + time.sleep(5) + + +# ── Main ── +def main(): + global is_mohe + agent_name = _rs(None) + _apply_config(agent_name) + is_mohe = (agent_name == 'mohe') + + log.info(f"Starting XMPP Agent: {agent_name} (mohe={is_mohe})") + log.info(f" JID={XMPP_JID} HTTP_PORT={HTTP_PORT}") + log.info(f" GATEWAY={GATEWAY_URL}") + log.info(f" SESSION_ID={GATEWAY_SESSION_ID}") + + bot = XmppAgent(XMPP_JID, XMPP_PASSWORD, MUC_ROOM, MUC_NICK) + bot.connect(host='127.0.0.1', port=5222) + t_deliver = threading.Thread(target=_deliver_loop, args=(bot,), daemon=True) + t_deliver.start() + t_inbound = threading.Thread(target=_inbound_loop, args=(bot,), daemon=True) + t_inbound.start() + t_ack = threading.Thread(target=_ack_tick, args=(bot,), daemon=True) + t_ack.start() + t_http = threading.Thread(target=_run_http_server, daemon=True) + t_http.start() + + bot.loop.run_forever() + + +if __name__ == '__main__': + main() diff --git a/deploy/profile-scripts/functional_health_check.py b/deploy/profile-scripts/functional_health_check.py index e47315ed..ecbb7509 100644 --- a/deploy/profile-scripts/functional_health_check.py +++ b/deploy/profile-scripts/functional_health_check.py @@ -50,6 +50,9 @@ REGISTRY = [ {"module": "gateway_llm", "function": "LLM调用链可用(gateway agent.log)", "check": {"type": "agent_log", "max_age_min": 30, "when": "always"}, "repair": {"action": "llm_diagnose"}}, + {"module": "sense_ocr", "function": "识图服务(SenseNova OCR配置+API可达)", + "check": {"type": "ocr_health", "when": "always"}, + "repair": {"action": "llm_diagnose"}}, {"module": "xmpp_bot", "function": "XMPP消息收发(bot journal)", "check": {"type": "bot_activity", "when": "always"}, "repair": {"action": "llm_diagnose"}}, @@ -127,6 +130,31 @@ def check_bot_activity(chk, now): return "fail", f"检查失败: {e}" +def check_ocr_health(chk, now): + """识图服务健康：OCR 配置存在 + SenseNova API 可达（TCP 443）。 + 不发真实 OCR 请求（省钱），真实端到端验证走 K 测试。""" + import json as _json, socket + # 1. OCR 配置存在且含 key + cfg_path = '/home/hmo/.config/mofin/ocr_config.json' + if not os.path.exists(cfg_path): + return "fail", "OCR 配置文件不存在: /home/hmo/.config/mofin/ocr_config.json" + try: + cfg = _json.load(open(cfg_path)) + if not cfg.get('key') or not cfg.get('base_url'): + return "fail", "OCR 配置缺 key 或 base_url" + except Exception as e: + return "fail", f"OCR 配置解析失败: {str(e)[:60]}" + # 2. SenseNova API TCP 可达 + try: + host = cfg['base_url'].split('//')[1].split('/')[0].split(':')[0] + port = int(cfg['base_url'].split('//')[1].split('/')[0].split(':')[1]) if ':' in cfg['base_url'].split('//')[1].split('/')[0] else 443 + s = socket.create_connection((host, port), timeout=5) + s.close() + except Exception as e: + return "fail", f"SenseNova API 不可达 ({host}): {str(e)[:60]}" + return "ok", f"配置 OK + {host}:{port} 可达" + + def check_cron_engine(chk, now): """cron 引擎：最近 max_age_min 内是否有任何 job 运行过""" try: @@ -180,6 +208,8 @@ def main(): status, reason = check_agent_log(chk, now) elif t == "bot_activity": status, reason = check_bot_activity(chk, now) + elif t == "ocr_health": + status, reason = check_ocr_health(chk, now) elif t == "cron_engine": status, reason = check_cron_engine(chk, now) else: diff --git a/static/mofin_health.json b/static/mofin_health.json index 5c13d547..37959b04 100644 --- a/static/mofin_health.json +++ b/static/mofin_health.json @@ -1,5 +1,5 @@ { - "generated_at": "2026-07-20 21:15:58", + "generated_at": "2026-07-20 21:30:05", "feature_tree": { "label": "MoFin 系统", "status": "ok", @@ -161,7 +161,7 @@ "script": "stale_detector.py", "schedule": "0 21 * * 1-5", "status": "ok", - "last_run": "2026-07-17T21:18", + "last_run": "2026-07-20T21:17", "type": "LLM", "profile": "position-analyst" }, @@ -537,7 +537,7 @@ "script": "mofin_health.py", "schedule": "*/15 9-16,20-22 * * 1-5", "status": "ok", - "last_run": "2026-07-20T21:01", + "last_run": "2026-07-20T21:16", "type": "no_agent", "profile": "position-analyst" } @@ -624,7 +624,7 @@ "script": "self_todo_executor.py", "schedule": "*/10 8-22 * * 1-5", "status": "ok", - "last_run": "2026-07-20T21:10", + "last_run": "2026-07-20T21:20", "type": "no_agent", "profile": "position-analyst" } @@ -885,7 +885,7 @@ "script": "fix_gateway_port.py", "schedule": "every 10m", "status": "ok", - "last_run": "2026-07-20T21:13", + "last_run": "2026-07-20T21:24", "type": "no_agent", "profile": "position-analyst" } @@ -917,7 +917,7 @@ "script": "functional_health_check.py", "schedule": "*/15 9-16,20-22 * * 1-5", "status": "ok", - "last_run": "2026-07-20T21:00", + "last_run": "2026-07-20T21:15", "type": "no_agent", "profile": "position-analyst" } @@ -1081,14 +1081,14 @@ "rows": 185, "readers": [ "promote_candidates", - "mofin_db", "candidate_filter", + "mofin_db", "accumulation_scanner" ], "writers": [ - "market_scanner", "promote_candidates", "candidate_filter", + "market_scanner", "accumulation_scanner" ], "has_input": true, @@ -1113,9 +1113,9 @@ "rows": 1, "readers": [ "per_stock_reassess", + "batch_reassess", "mofin_db", - "merge_third_db", - "batch_reassess" + "merge_third_db" ], "writers": [ "mofin_db", @@ -1145,8 +1145,8 @@ "prepare_report_data" ], "writers": [ - "mofin_db", - "mo_data" + "mo_data", + "mofin_db" ], "has_input": true, "has_output": true, @@ -1195,25 +1195,25 @@ "desc": "每只股票的完整策略参数", "rows": 227, "readers": [ - "run_all_tests", - "system_audit", - "capital_flow_collector", - "verify_300308", - "strategy_lifecycle", - "watchlist_auto_exit", - "mofin_db", - "verify_reassess_pipeline", + "mofin_collect", + "promote_candidates", + "check_db_state", "data_governance", - "promote_candidates" + "watchlist_auto_exit", + "mo_data", + "price_monitor", + "generate_report", + "system_health_check", + "sync_decisions_to_db" ], "writers": [ - "mofin_db", - "per_stock_reassess", - "data_governance", "promote_candidates", + "data_governance", + "sync_decisions_to_db", "batch_reassess", "watchlist_auto_exit", - "sync_decisions_to_db" + "per_stock_reassess", + "mofin_db" ], "has_input": true, "has_output": true, @@ -1241,21 +1241,21 @@ "desc": "当前持仓（权威源）", "rows": 14, "readers": [ + "mofin_collect", + "prepare_recommendation", + "mo_data", + "price_monitor", + "system_health_check", "run_all_tests", - "xiaoguo_signal_consumer", - "system_audit", - "capital_flow_collector", - "strategy_lifecycle", - "mofin_db", - "server", - "trend_detector", + "refresh_mtf_cache", "prepare_report_data", - "mofin_collect" + "capital_flow_collector", + "per_stock_reassess" ], "writers": [ "mofin_db", - "import_holding_xls", - "price_monitor" + "price_monitor", + "import_holding_xls" ], "has_input": true, "has_output": true, @@ -1283,18 +1283,18 @@ "desc": "所有持仓+自选最新实时价", "rows": 232, "readers": [ - "mofin_db", "stale_push_wlin", + "generate_report", "verify_reassess_pipeline", "candidate_filter", - "system_audit", "cron_health_monitor", "mo_data", - "generate_report" + "mofin_db", + "system_audit" ], "writers": [ - "mofin_db", "mo_data", + "mofin_db", "price_monitor" ], "has_input": true, @@ -1321,15 +1321,15 @@ "desc": "宏观上下文（大盘偏向/指数）", "rows": 81, "readers": [ - "divergence_detector", - "xiaoguo_signal_consumer", - "system_audit", - "stock_profile", "strategy_tree", - "per_stock_reassess", + "stale_push_wlin", + "divergence_detector", "batch_reassess", + "stock_profile", "strategy_lifecycle", - "stale_push_wlin" + "per_stock_reassess", + "system_audit", + "xiaoguo_signal_consumer" ], "writers": [ "refresh_macro_context" @@ -1384,14 +1384,14 @@ "desc": "大盘指数快照（每10分）", "rows": 989, "readers": [ - "mofin_db", - "merge_third_db", - "system_audit", - "trend_detector", - "prepare_report_data", - "mofin_query", "market_screener", - "market_scanner" + "trend_detector", + "mofin_query", + "market_scanner", + "prepare_report_data", + "merge_third_db", + "mofin_db", + "system_audit" ], "writers": [ "mofin_db", @@ -1420,13 +1420,13 @@ "desc": "多周期均线缓存", "rows": 64, "readers": [ - "multi_timeframe", "mofin_db", - "technical_analysis" + "technical_analysis", + "multi_timeframe" ], "writers": [ - "multi_timeframe", - "mofin_db" + "mofin_db", + "multi_timeframe" ], "has_input": true, "has_output": true, @@ -1451,18 +1451,18 @@ "desc": "总资产/现金/仓位汇总", "rows": 1, "readers": [ - "mofin_db", - "mo_data", - "prepare_report_data", - "preflight_verify", - "price_monitor", "check_db_state", - "import_holding_xls" + "prepare_report_data", + "price_monitor", + "import_holding_xls", + "mo_data", + "preflight_verify", + "mofin_db" ], "writers": [ "mofin_db", - "import_holding_xls", - "price_monitor" + "price_monitor", + "import_holding_xls" ], "has_input": true, "has_output": true, @@ -1489,21 +1489,21 @@ "desc": "价格区间突破事件日志", "rows": 6353, "readers": [ - "backfill_price_events", - "mofin_db", - "test_db_only", - "test_dual_write", "test_raw_insert", + "check_price_events", + "test_dual_write", + "mofin_db", "test_db_write", - "check_price_events" + "test_db_only", + "backfill_price_events" ], "writers": [ - "backfill_price_events", - "mofin_db", - "test_dual_write", - "test_db_only", "test_raw_insert", - "test_db_write" + "test_dual_write", + "mofin_db", + "test_db_write", + "test_db_only", + "backfill_price_events" ], "has_input": true, "has_output": true, @@ -1525,15 +1525,15 @@ "desc": "行业信号（趋势检测产出）", "rows": 653, "readers": [ - "trend_detector", "xiaoguo_news_processor", + "server", "mofin_news", - "server" + "trend_detector" ], "writers": [ - "trend_detector", "xiaoguo_news_processor", - "mofin_news" + "mofin_news", + "trend_detector" ], "has_input": true, "has_output": true, @@ -1560,13 +1560,13 @@ "desc": "行业板块数据", "rows": 68408, "readers": [ - "mofin_db", - "inspect_third_db", - "merge_third_db", - "trend_detector", - "strategy_lifecycle", "market_screener", - "market_scanner" + "trend_detector", + "market_scanner", + "merge_third_db", + "inspect_third_db", + "strategy_lifecycle", + "mofin_db" ], "writers": [ "mofin_db", @@ -1595,22 +1595,22 @@ "desc": "信号相关新闻", "rows": 1287, "readers": [ + "macro_signal_consumer", "intraday_health_check", - "xiaoguo_signal_consumer", "server", - "system_audit", - "per_stock_reassess", "batch_reassess", - "macro_signal_consumer" + "per_stock_reassess", + "system_audit", + "xiaoguo_signal_consumer" ], "writers": [ - "divergence_detector", - "xiaoguo_signal_consumer", + "xiaoguo_scanner", + "macro_signal_consumer", "mofin_news", + "divergence_detector", "xiaoguo_news_processor", "macro_context_collector", - "macro_signal_consumer", - "xiaoguo_scanner" + "xiaoguo_signal_consumer" ], "has_input": true, "has_output": true, @@ -1726,12 +1726,12 @@ "desc": "股票行业映射", "rows": 64, "readers": [ - "mofin_db", "trend_detector", - "per_stock_reassess", "mofin_news", "xiaoguo_news_processor", - "strategy_lifecycle" + "strategy_lifecycle", + "per_stock_reassess", + "mofin_db" ], "writers": [ "mofin_db" @@ -1776,19 +1776,19 @@ "desc": "全量股票代码", "rows": 5575, "readers": [ - "mofin_db", "trend_detector", - "import_full_stocks", - "mofin_news", "accumulation_scanner", + "mofin_news", "xiaoguo_news_processor", + "import_full_stocks", + "mofin_db", "check_stocks_table" ], "writers": [ - "backfill_price_events", - "mofin_db", "import_full_stocks", - "price_monitor" + "mofin_db", + "price_monitor", + "backfill_price_events" ], "has_input": true, "has_output": true, @@ -1813,8 +1813,8 @@ "desc": "策略重评历史记录", "rows": 217, "readers": [ - "mofin_db", - "verify_reassess_pipeline" + "verify_reassess_pipeline", + "mofin_db" ], "writers": [ "mofin_collect" @@ -1870,19 +1870,19 @@ "readers": [ "intraday_health_check", "merge_third_db", - "strategy-staleness-check", - "morning_health_check", - "self_todo_executor" - ], - "writers": [ - "intraday_health_check", - "merge_third_db", - "strategy-staleness-check", - "preflight_verify", "morning_health_check", "self_todo_executor", + "strategy-staleness-check" + ], + "writers": [ "mofin_collect", - "cron_health_monitor" + "intraday_health_check", + "merge_third_db", + "morning_health_check", + "self_todo_executor", + "strategy-staleness-check", + "cron_health_monitor", + "preflight_verify" ], "has_input": true, "has_output": true, @@ -1929,20 +1929,20 @@ "desc": "自选股列表", "rows": 69, "readers": [ - "mofin_db", - "run_all_tests", - "xiaoguo_scanner", - "xiaoguo_signal_consumer", - "trend_detector", - "per_stock_reassess", - "mo_alphasift_bridge", - "price_monitor", "mofin_collect", - "refresh_mtf_cache" + "trend_detector", + "stale_push_wlin", + "xiaoguo_scanner", + "refresh_mtf_cache", + "mofin_db", + "stock_quote", + "per_stock_reassess", + "run_all_tests", + "mo_alphasift_bridge" ], "writers": [ - "mofin_db", - "per_stock_reassess" + "per_stock_reassess", + "mofin_db" ], "has_input": true, "has_output": true, @@ -1968,8 +1968,8 @@ "desc": "小果扫描跟踪", "rows": 527, "readers": [ - "xiaoguo_scanner", - "server" + "server", + "xiaoguo_scanner" ], "writers": [ "xiaoguo_scanner" @@ -2109,11 +2109,11 @@ "readers": [ "analyze_health", "verify_self_check_section", - "verify_health_json", - "verify_deployment" + "verify_deployment", + "verify_health_json" ], "writers": [], - "last_modified": "07-20 21:01", + "last_modified": "07-20 21:16", "warn": false, "migrated_to_db": null }, @@ -2245,7 +2245,7 @@ "script": "fix_gateway_port.py", "schedule": "every 10m", "status": "ok", - "last_run": "2026-07-20T21:13:56", + "last_run": "2026-07-20T21:24:02", "profile": "position-analyst" }, { @@ -2362,7 +2362,7 @@ "script": "mofin_health.py", "schedule": "*/15 9-16,20-22 * * 1-5", "status": "ok", - "last_run": "2026-07-20T21:01:01", + "last_run": "2026-07-20T21:16:11", "profile": "position-analyst" }, { @@ -2416,7 +2416,7 @@ "script": "functional_health_check.py", "schedule": "*/15 9-16,20-22 * * 1-5", "status": "ok", - "last_run": "2026-07-20T21:00:56", + "last_run": "2026-07-20T21:15:59", "profile": "position-analyst" }, { @@ -2434,7 +2434,7 @@ "script": "", "schedule": "{'kind': 'cron', 'expr': '*/10 * * * *'}", "status": "ok", - "last_run": "2026-07-20T21:11:48", + "last_run": "2026-07-20T21:21:29", "profile": "default" }, { @@ -2704,7 +2704,7 @@ "script": "stale_detector.py", "schedule": "0 21 * * 1-5", "status": "ok", - "last_run": "2026-07-17T21:18:03", + "last_run": "2026-07-20T21:17:10", "profile": "position-analyst" }, { @@ -2749,7 +2749,7 @@ "script": "self_todo_executor.py", "schedule": "*/10 8-22 * * 1-5", "status": "ok", - "last_run": "2026-07-20T21:10:55", + "last_run": "2026-07-20T21:20:59", "profile": "position-analyst" }, { @@ -2866,41 +2866,41 @@ "table": "mtf_cache", "label": "多周期均线缓存", "last_record": "07-20 17:08", - "age_hours": 4.1, + "age_hours": 4.4, "warn": false }, { "table": "macro_context_log", "label": "宏观上下文", "last_record": "07-20 15:31", - "age_hours": 5.7, + "age_hours": 6.0, "warn": false }, { "table": "market_snapshots", "label": "市场快照", "last_record": "07-20 15:50", - "age_hours": 5.4, + "age_hours": 5.7, "warn": false }, { "table": "live_prices", "label": "实时价格", "last_record": "07-20 18:12", - "age_hours": 3.1, + "age_hours": 3.3, "warn": false }, { "table": "price_events", "label": "价格事件", "last_record": "07-20 17:06", - "age_hours": 4.2, + "age_hours": 4.4, "warn": false } ], "self_check": { "functional": { - "generated_at": "2026-07-20 21:15:58", + "generated_at": "2026-07-20 21:30:05", "status": "ok", "summary": { "total": 9, @@ -2944,7 +2944,7 @@ "module": "premarket", "function": "盘前全量重评(premarket summary)", "status": "ok", - "reason": "785min前", + "reason": "799min前", "repair": { "action": "rerun_script", "script": "premarket_full_review.py" @@ -2954,7 +2954,7 @@ "module": "gateway_llm", "function": "LLM调用链可用(gateway agent.log)", "status": "ok", - "reason": "latency=8.5s", + "reason": "latency=9.5s", "repair": { "action": "llm_diagnose" }