Hermes-Skills/okx-auto-position/scripts/signal_inbox.py

#!/usr/bin/env python3
"""信号入队 Hook - 给 gateway 调用的轻量级入库函数 (v4.5.4 → v4.5.5)。

调用方式 (从 gateway/run.py _handle_message_with_agent 内部):
    import importlib, sys
    sys.path.insert(0, '~/.hermes/skills/trading/okx-auto-position/scripts')
    sig = importlib.import_module('signal_inbox')
    await sig.enqueue_if_signal(source, event.text or '')

逻辑 (v4.5.5):
1. 识别 -1003966251111 (交易信号群) + 消息含【币种】/【方向】 → 加时间戳后入 signal_queue.db
2. 时间戳用 hook 接收时刻 (datetime.now Asia/Shanghai), 而非原消息发送时间
3. 立即同步调 process_signal.py
4. 失败不抛异常, 只记 log

时间戳后缀格式: 全文末尾追加 "\n\n⏱信号时间: YYYY-MM-DD HH:MM:SS"

⚠️ v4.5.4 改动: forwarder 实际用 forward_messages 原生转发, 不带时间戳。
   改在 gateway hook 处加, 保证推 QQ 的消息带时间戳。
"""
import os
import sys
import subprocess
import sqlite3
import asyncio
import logging
from datetime import datetime
from zoneinfo import ZoneInfo

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")

# 时区
_TZ = ZoneInfo("Asia/Shanghai")


def get_now_str() -> str:
    """获取当前时刻 (Asia/Shanghai), 格式 YYYY-MM-DD HH:MM:SS."""
    return datetime.now(_TZ).strftime("%Y-%m-%d %H:%M:%S")


def append_timestamp(text: str) -> str:
    """在文本末尾追加 ⏱信号时间: {...} (如果没有就加)."""
    if not text:
        text = ""
    stamp = f"\n\n⏱信号时间: {get_now_str()}"
    # 如果已经有时间戳就不重复加
    if "⏱信号时间:" in text:
        return text
    return text + stamp


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: str) -> int:
    """入队 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: int, status: str, result: str = "") -> None:
    """标记 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: str) -> tuple:
    """同步调 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 内调用.

    v4.5.5: 在 text 末尾追加 hook 接收时刻的时间戳, 保证推 QQ 时带时间戳.
    """
    try:
        chat_id = str(getattr(source, "chat_id", "") or "")
        if not _is_signal(chat_id, text):
            return

        # v4.5.5: 追加时间戳 (hook 接收时刻)
        text_with_ts = append_timestamp(text)

        loop = asyncio.get_running_loop()
        row_id = await loop.run_in_executor(None, _enqueue_sync, text_with_ts)
        logger.info(f"[signal_inbox] 入队 rowid={row_id} chat={chat_id}")

        rc, out, err = await loop.run_in_executor(None, _run_process_signal, text_with_ts)
        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:])
    test_text_ts = append_timestamp(test_text)
    print(f"添加时间戳后: {test_text_ts[:200]}...")
    rc, out, err = _run_process_signal(test_text_ts)
    print(f"rc={rc}")
    print(f"stdout: {out}")
    if err:
        print(f"stderr: {err[:300]}")