85 lines
3.6 KiB
Python
85 lines
3.6 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 Tool
|
|
|
|
|
|
def wechat_push_available() -> bool:
|
|
"""任一微信渠道可用(当前 = ClawBot 开关在;后续 or 企业微信配齐)。"""
|
|
return service.clawbot_enabled()
|
|
|
|
|
|
_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. Use when the user asks to send something to their "
|
|
"WeChat, or when a scheduled task should deliver its output there. NOTE: WeChat push only "
|
|
"works if the user has messaged the bot within the last ~24h; 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'.",
|
|
},
|
|
},
|
|
"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) -> str:
|
|
text = (text or "").strip()
|
|
if not text and not file:
|
|
return "[Error] text 不能为空"
|
|
|
|
fpath: Optional[str] = None
|
|
if file and file.strip():
|
|
p = self._resolve(file.strip()).resolve()
|
|
if self.user_root is not None:
|
|
try:
|
|
p.relative_to(self.user_root.resolve())
|
|
except ValueError:
|
|
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)
|
|
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。"
|