v4.5.4: signal_inbox.py hook helper for gateway

- 识别 -1003966251111 交易信号群 + 含【币种】消息 → 入 signal_queue.db - 立即同步调 process_signal.py 处理 - 失败靠 signal-queue-retry cron 兜底 - 不被使用 (gateway hook 已内联到 run.py), 保留作 fallback
2026-07-30 09:35:28 +08:00
parent 39f5cb6cb3
commit b91f677200
1 changed files with 122 additions and 0 deletions
@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""信号入队 Hook - 给 gateway 调用的轻量级入库函数 (v4.5.4)。
+
+调用方式 (从 gateway/run.py _handle_message_with_agent 内部):
+    await enqueue_if_signal(source, event.text)
+
+逻辑:
+1. 识别 -1003966251111 (交易信号群) + 消息含【币种】/【方向】 → 入 signal_queue.db
+2. 立即同步调 process_signal.py (不阻塞 gateway 主流程 30s+)
+3. 失败不抛异常, 只记 log (gateway 不能因为信号处理挂掉)
+
+⚠️ 此文件曾被自动清理任务删除 (2026-07-30), 现已重建。
+   防御措施: 把代码内联到 run.py 的 hook 里, 这个文件只做 import 桥接.
+"""
+import os
+import sys
+import subprocess
+import sqlite3
+import asyncio
+import logging
+
+logger = logging.getLogger(__name__)
+
+# 交易信号群 ID (固定)
+SIGNAL_CHAT_ID = "-1003966251111"
+# 信号关键字 (任一出现即识别为信号)
+SIGNAL_KEYWORDS = ("【币种】", "【方向】", "🚨 已平仓", "📉 注意", "📈 注意")
+
+# 路径
+SKILL_DIR = os.path.expanduser("~/.hermes/skills/trading/okx-auto-position")
+PROCESS_SCRIPT = os.path.join(SKILL_DIR, "scripts", "process_signal.py")
+SIGNAL_QUEUE_DB = os.path.expanduser("~/.hermes/trading/signal_queue.db")
+
+
+def _is_signal(chat_id, text):
+    """判断是否交易信号"""
+    if not text:
+        return False
+    if str(chat_id) != SIGNAL_CHAT_ID:
+        return False
+    return any(kw in text for kw in SIGNAL_KEYWORDS)
+
+
+def _enqueue_sync(raw_text):
+    """入队 signal_queue.db, 返回 rowid"""
+    conn = sqlite3.connect(SIGNAL_QUEUE_DB, timeout=5)
+    try:
+        cur = conn.execute(
+            "INSERT INTO queue (raw_text, status) VALUES (?, 'pending')",
+            (raw_text,),
+        )
+        conn.commit()
+        last_id = cur.lastrowid
+        return last_id if last_id is not None else 0
+    finally:
+        conn.close()
+
+
+def _mark_queue_status(row_id, status, result=""):
+    """标记 queue 行的 status."""
+    conn = sqlite3.connect(SIGNAL_QUEUE_DB, timeout=5)
+    try:
+        conn.execute(
+            "UPDATE queue SET status=?, processed_at=datetime('now'), result=? WHERE id=?",
+            (status, result[:500], row_id),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _run_process_signal(raw_text):
+    """同步调 process_signal.py, 返回 (returncode, stdout, stderr)"""
+    try:
+        result = subprocess.run(
+            [sys.executable, PROCESS_SCRIPT, raw_text],
+            capture_output=True, text=True, timeout=60,
+            cwd=SKILL_DIR,
+        )
+        return (result.returncode, result.stdout, result.stderr)
+    except subprocess.TimeoutExpired:
+        return (-1, "", "process_signal.py 超时 60s")
+    except Exception as e:
+        return (-2, "", f"执行异常: {e}")
+
+
+async def enqueue_if_signal(source, text):
+    """异步信号入队 + 处理. 从 gateway 内调用."""
+    try:
+        chat_id = str(getattr(source, "chat_id", "") or "")
+        if not _is_signal(chat_id, text):
+            return
+
+        loop = asyncio.get_running_loop()
+        row_id = await loop.run_in_executor(None, _enqueue_sync, text)
+        logger.info(f"[signal_inbox] 入队信号 rowid={row_id} chat={chat_id}")
+
+        rc, out, err = await loop.run_in_executor(None, _run_process_signal, text)
+        if rc == 0:
+            status = 'done'
+            await loop.run_in_executor(None, _mark_queue_status, row_id, status, out)
+            logger.info(f"[signal_inbox] 处理成功 rowid={row_id} out={out[:80]}")
+        else:
+            status = 'failed'
+            await loop.run_in_executor(None, _mark_queue_status, row_id, status, f"rc={rc} err={err[:200]}")
+            logger.warning(f"[signal_inbox] 处理失败 rowid={row_id} rc={rc} err={err[:200]}")
+
+    except Exception as e:
+        logger.error(f"[signal_inbox] hook 异常 (不致命): {e}")
+
+
+if __name__ == '__main__':
+    """手动测试: python3 signal_inbox.py <signal_text>"""
+    if len(sys.argv) < 2:
+        print("用法: signal_inbox.py <signal_text>")
+        sys.exit(1)
+    test_text = ' '.join(sys.argv[1:])
+    rc, out, err = _run_process_signal(test_text)
+    print(f"rc={rc}")
+    print(f"stdout: {out}")
+    if err:
+        print(f"stderr: {err[:300]}")