230 lines
8.8 KiB
Python
230 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
"""messenger.py — 统一消息处理者（生产者与消费者解耦）
|
||
|
||
核心原则：消费者(本模块)根据【生产者是谁】(哪个 cron job) 决定消息去向，
|
||
绝不根据消息内容判断。
|
||
|
||
通道配置（jobs.json 的 delivery 字段）：
|
||
broadcast : 只写 broadcast_messages 表（不推 xmpp）
|
||
xmpp : 写 broadcast 归档 + 推 xmpp
|
||
both : 同上（保留兼容）
|
||
|
||
用法：
|
||
from messenger import send
|
||
send(title="MoFin持仓异动", content="...") # 自动按当前脚本/job路由
|
||
"""
|
||
import os, sys, time, json, sqlite3
|
||
from datetime import datetime
|
||
|
||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||
JOBS_JSON = "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json"
|
||
DEFAULT_CHANNEL = "broadcast"
|
||
|
||
# 脚本名 → job delivery 配置缓存 {script_basename: channel}
|
||
_script_channel_cache = None
|
||
|
||
|
||
def _load_script_channels():
|
||
"""读取 jobs.json：建立 {脚本basename: delivery通道} 映射"""
|
||
global _script_channel_cache
|
||
if _script_channel_cache is not None:
|
||
return _script_channel_cache
|
||
mapping = {}
|
||
try:
|
||
if os.path.exists(JOBS_JSON):
|
||
with open(JOBS_JSON, encoding="utf-8") as f:
|
||
jobs = json.load(f).get("jobs", [])
|
||
for j in jobs:
|
||
if not isinstance(j, dict):
|
||
continue
|
||
script = str(j.get("script", "")).split("/")[-1]
|
||
if not script or not script.endswith(".py"):
|
||
continue
|
||
channel = j.get("delivery")
|
||
if channel:
|
||
if script not in mapping or channel == "xmpp":
|
||
mapping[script] = channel
|
||
except Exception as e:
|
||
print(f" [messenger] 读取 job 通道配置失败: {e}", file=sys.stderr)
|
||
_script_channel_cache = mapping
|
||
return mapping
|
||
|
||
|
||
def get_channel(job_script=None):
|
||
"""获取指定生产者脚本的通道。None=自动识别当前脚本"""
|
||
mapping = _load_script_channels()
|
||
if job_script:
|
||
return mapping.get(job_script, DEFAULT_CHANNEL)
|
||
script = os.path.basename(sys.argv[0]) if sys.argv else "mofin"
|
||
return mapping.get(script, DEFAULT_CHANNEL)
|
||
|
||
|
||
def _classify(title, content):
|
||
"""内容分类(仅为 broadcast 表展示用，不用于通道决策)"""
|
||
text = (title + " " + content).lower()
|
||
words = {
|
||
"trading": ["买入", "卖出", "止损", "止盈", "加仓", "减仓", "持仓", "区间", "操作", "推荐"],
|
||
"system_error": ["llm端点", "api错误", "连接失败", "超时", "异常", "失败", "error", "exception", "告警", "故障"],
|
||
"health": ["健康", "体检", "完整性", "守卫", "cron"],
|
||
"market": ["大盘", "市场", "板块", "行业", "指数", "行情"],
|
||
"news": ["新闻", "消息", "资讯", "公告", "政策"],
|
||
"strategy": ["策略", "重评", "评估", "温区", "regime", "12维"],
|
||
}
|
||
for cat, kws in words.items():
|
||
if any(kw in text for kw in kws):
|
||
return cat
|
||
return "general"
|
||
|
||
|
||
def send(title="", content="", source=None, channel=None, job_name=None):
|
||
"""统一消息发送。生产者调用即可，不关心去向。
|
||
|
||
通道决策优先级：
|
||
1. 显式 channel 参数
|
||
2. 当前脚本名对应的 job 配置（自动识别生产者）
|
||
3. 默认 broadcast
|
||
"""
|
||
now = datetime.now().isoformat()
|
||
producer = source or job_name or (os.path.basename(sys.argv[0]) if sys.argv else "mofin")
|
||
category = _classify(title, content)
|
||
|
||
# 决定通道（生产者驱动：按脚本名查 job 配置）
|
||
if channel is None:
|
||
mapping = _load_script_channels()
|
||
script_key = None
|
||
if source and source.endswith(".py"):
|
||
script_key = source
|
||
elif producer and producer.endswith(".py"):
|
||
script_key = producer
|
||
else:
|
||
script_key = os.path.basename(sys.argv[0]) if sys.argv else "mofin"
|
||
channel = mapping.get(script_key, DEFAULT_CHANNEL)
|
||
|
||
# 写 broadcast 表（所有通道归档，保证日志完整）
|
||
try:
|
||
conn = sqlite3.connect(DB, timeout=30)
|
||
conn.execute(
|
||
"INSERT INTO broadcast_messages (ts, category, title, content, source) VALUES (?,?,?,?,?)",
|
||
(now, category, title, content, producer))
|
||
conn.commit()
|
||
conn.close()
|
||
except Exception as e:
|
||
print(f" [messenger] broadcast 写入失败: {e}", file=sys.stderr)
|
||
|
||
# xmpp 通道：输出 stdout（hermes 捕获推 xmpp）
|
||
if channel in ("xmpp", "both"):
|
||
if title:
|
||
print(f"【{title}】{content}", flush=True)
|
||
else:
|
||
print(content, flush=True)
|
||
return "xmpp"
|
||
|
||
return "broadcast"
|
||
|
||
|
||
def send_action_recommendation(title, content, source=None):
|
||
"""操作推荐：强制 xmpp 通道（必须推送给用户）"""
|
||
return send(title=title, content=content, source=source, channel="xmpp")
|
||
|
||
|
||
|
||
class _MessengerStdout:
|
||
# 2026-08-24 事件块聚合：逐行碎片→一块一条消息（老莫：broadcast每条消息详情=摘要,无信息量）
|
||
# 块边界：静默>2秒 或 满25行；atexit强制flush。xmpp实时输出不缓冲(hermes靠stdout推送)。
|
||
MAX_BLOCK_LINES = 25
|
||
MAX_IDLE_SEC = 2.0
|
||
|
||
def __init__(self, orig, script):
|
||
self.orig = orig
|
||
self.script = script
|
||
self.channel = get_channel(script)
|
||
self._buf = []
|
||
self._last = 0.0
|
||
import atexit
|
||
atexit.register(self._flush_buf)
|
||
|
||
def write(self, s):
|
||
if self.channel in ("xmpp", "both"):
|
||
# xmpp 通道: 实时输出给 hermes(推xmpp)——不缓冲
|
||
self.orig.write(s); self.orig.flush()
|
||
now = time.time()
|
||
if self._buf and now - self._last > self.MAX_IDLE_SEC:
|
||
self._flush_buf() # 静默超2秒: 上一个事件块结束
|
||
self._buf.append(s)
|
||
self._last = now
|
||
if len(self._buf) >= self.MAX_BLOCK_LINES:
|
||
self._flush_buf()
|
||
return len(s)
|
||
|
||
def _flush_buf(self):
|
||
if self._buf:
|
||
text = "".join(self._buf)
|
||
self._buf = []
|
||
_flush_broadcast_block(text.split("\n"), self.script)
|
||
|
||
def flush(self):
|
||
if self.channel in ("xmpp", "both"):
|
||
self.orig.flush()
|
||
|
||
|
||
def _flush_broadcast_block(lines, script):
|
||
"""把一批 stdout 行聚合成一条 broadcast 消息（2026-08-24 事件块化）。
|
||
过滤噪音前缀 + 相邻重复行去重（根治逐行碎片+双写重复）+ 单连接一次写入。"""
|
||
skip_prefix = ("[DB]", "[SYNC", "[guard", "[regime] ", "⏱", "📊 ", "📥 ", "💾 ", "🔄 ", "[SILENT]")
|
||
out = []
|
||
for line in lines:
|
||
line = line.strip()
|
||
if not line or len(line) < 3:
|
||
continue
|
||
if line.startswith(skip_prefix):
|
||
continue
|
||
if line.startswith("【") and "】" in line and "MoFin·" in line:
|
||
continue # 跳过已格式化的 messenger 输出, 防递归
|
||
if out and out[-1] == line:
|
||
continue # 相邻重复去重
|
||
out.append(line)
|
||
if not out:
|
||
return
|
||
content = "\n".join(out)
|
||
try:
|
||
category = _classify(_job_title(script), content)
|
||
conn = sqlite3.connect(DB, timeout=30)
|
||
conn.execute(
|
||
"INSERT INTO broadcast_messages (ts, category, title, content, source) VALUES (?,?,?,?,?)",
|
||
(datetime.now().isoformat(), category, _job_title(script), content, script))
|
||
conn.commit()
|
||
conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _write_broadcast_from_stdout(text, script):
|
||
"""把 stdout 内容直接写入 broadcast_messages(不经 send, 避免 xmpp 递归)"""
|
||
if not text or not text.strip():
|
||
return
|
||
_flush_broadcast_block(text.strip().split("\n"), script)
|
||
|
||
|
||
def _job_title(script):
|
||
return script.replace(".py", "")
|
||
|
||
|
||
|
||
def install_stdio_hook(script_name=None):
|
||
"""在脚本入口调用：重定向 stdout 到 messenger 通道控制。
|
||
2026-08-24 幂等化：已 hook 则直接复用不重复包装——库文件（如 market_regime）被 import
|
||
时其模块级 _msh() 会变成 no-op，否则 stdout 被包 N 层、N 个 buffer 各 flush 一次，
|
||
broadcast 整块重复（双写根因，per_stock_reassess→strategy_lifecycle→market_regime 链实锤）。"""
|
||
if isinstance(sys.stdout, _MessengerStdout):
|
||
return sys.stdout.channel # 已 hook：幂等返回，不再包装
|
||
if script_name is None:
|
||
script_name = os.path.basename(sys.argv[0]) if sys.argv else "mofin"
|
||
sys.stdout = _MessengerStdout(sys.stdout, script_name)
|
||
return get_channel(script_name)
|
||
|
||
if __name__ == "__main__":
|
||
print("=== messenger 自测 ===")
|
||
mapping = _load_script_channels()
|
||
for s, c in sorted(mapping.items()):
|
||
if "price" in s or "market" in s or "per_stock" in s or "anomaly" in s:
|
||
print(f" {s}: {c}") |