103 lines
4.8 KiB
Python
103 lines
4.8 KiB
Python
"""微信主动推送工具(DESIGN §8.7 渠道抽象)。host-side,按 user_id 注入。
|
|
|
|
agent 在(交互 / 定时)run 里调:把一段文本 + 可选产物文件**主动推**到用户已绑的微信
|
|
(当前 ClawBot 个人微信;企业微信渠道后续在 `service.send_to_user` 内追加)。仅当渠道开关
|
|
在才挂(沿用「有开关才注册」§3.4)。
|
|
|
|
ClawBot 推送有 **24h 窗口**约束:用户超期未在微信里跟机器人说过话 / 从未开口 → 推不出,
|
|
工具返回明确 reason,agent 可改走 `send_email`。密钥/凭据只在 host 进程,绝不进沙箱。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from uuid import UUID
|
|
|
|
from core.wechat import service
|
|
from .base import FileOutOfBounds, Tool
|
|
|
|
|
|
def wechat_push_available() -> bool:
|
|
"""任一微信渠道开着就挂工具(ClawBot 个人微信 / 企业微信)。渠道清单的唯一真相源
|
|
是 `service.active_channels()`,与 `send_to_user` 的投递口径同源,不再各列各的。"""
|
|
return bool(service.active_channels())
|
|
|
|
|
|
_REASON_HINT = {
|
|
"no_binding": "用户还没绑定微信(让其在 zcbot 网页扫码绑定)",
|
|
"never_opened": "用户绑了但从未在微信里跟机器人说过话(主动推需先开口一次)",
|
|
"token_stale": "距用户上次在微信互动已超 24h,主动推送窗口已过(让其发条消息即可恢复)",
|
|
}
|
|
|
|
|
|
class WechatPushTool(Tool):
|
|
name = "wechat_push"
|
|
description = (
|
|
"Proactively push a short text message (and optionally one result file, e.g. a .docx/.pdf "
|
|
"report) to the user's bound WeChat (personal WeChat via ClawBot, or WeCom/企业微信). Use "
|
|
"when the user asks to send something to their WeChat, or when a scheduled task should "
|
|
"deliver its output there. IMPORTANT: if the user names a specific WeChat — '企业微信'/'企微'/"
|
|
"WeCom → channel='wecom'; '个人微信'/'微信' (personal, via ClawBot) → channel='clawbot' — pass "
|
|
"that channel so it ONLY goes there. Omit channel to broadcast to all bound WeChat channels. "
|
|
"NOTE: the ~24h-window constraint applies ONLY to the personal WeChat (ClawBot) channel — WeCom "
|
|
"(企业微信) has no window limit. If it returns a window/binding error, fall back to send_email. "
|
|
"The file path is relative to the working directory."
|
|
)
|
|
parameters = {
|
|
"type": "object",
|
|
"properties": {
|
|
"text": {"type": "string", "description": "Message text to push (plain text)."},
|
|
"file": {
|
|
"type": "string",
|
|
"description": "Optional path (relative to working dir) of one file to attach, e.g. 'report.docx'.",
|
|
},
|
|
"channel": {
|
|
"type": "string",
|
|
"enum": ["wecom", "clawbot"],
|
|
"description": (
|
|
"Optional. Target a single WeChat channel: 'wecom' = WeCom/企业微信, "
|
|
"'clawbot' = personal WeChat. Omit to broadcast to all bound channels."
|
|
),
|
|
},
|
|
},
|
|
"required": ["text"],
|
|
}
|
|
|
|
def __init__(self, user_id: UUID, base_dir=None, user_root=None) -> None:
|
|
super().__init__(base_dir=base_dir, user_root=user_root)
|
|
self.user_id = user_id
|
|
|
|
def execute(
|
|
self, text: str = "", file: Optional[str] = None, channel: Optional[str] = None
|
|
) -> str:
|
|
text = (text or "").strip()
|
|
if not text and not file:
|
|
return "[Error] text 不能为空"
|
|
|
|
channel = (channel or "").strip().lower() or None
|
|
if channel is not None and channel not in ("wecom", "clawbot"):
|
|
return f"[Error] 未知渠道: {channel}(只支持 wecom / clawbot)"
|
|
|
|
fpath: Optional[str] = None
|
|
if file and file.strip():
|
|
# host-side 解析:容器 /workspace 路径翻回宿主 + 强制落 user_root 内(防越界)
|
|
try:
|
|
p = self._resolve_user_file(file.strip())
|
|
except FileOutOfBounds:
|
|
return f"[Error] 文件路径越界(必须在工作目录内): {file}"
|
|
if not p.is_file():
|
|
return f"[Error] 文件不存在: {file}"
|
|
fpath = str(p)
|
|
|
|
report = service.send_to_user(self.user_id, text, fpath, channel=channel)
|
|
if report.delivered:
|
|
n = "(含 1 个文件)" if fpath else ""
|
|
return f"[ok] 已推送到微信 {n}".strip()
|
|
|
|
if not report.results:
|
|
return "[Error] 没有可用的微信渠道(未开启 / 未配置)"
|
|
# 取首个失败原因给 agent 决定是否改走 send_email
|
|
r = report.results[0]
|
|
hint = _REASON_HINT.get(r.reason, r.reason)
|
|
return f"[Error] 微信推送未送达({r.channel}: {r.reason})。{hint}。可改用 send_email。"
|