NightStar
8917d0a102
- SKILL.md: remind-me 技能定义 - scripts/feishu_urgent.py: 飞书加急消息发送脚本 - user_id 通过环境变量或传参获取，不硬编码 - 支持自然语言设置定时提醒，到点飞书弹窗
132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
"""飞书加急消息发送脚本
|
||
用法:
|
||
python3 scripts/feishu_urgent.py "提醒内容" <user_id>
|
||
|
||
参数:
|
||
content 提醒消息内容（必填）
|
||
user_id 目标用户 open_id（必填，支持环境变量 FEISHU_USER_ID）
|
||
|
||
环境变量:
|
||
FEISHU_USER_ID 默认目标用户 open_id
|
||
FEISHU_CONFIG_PATH 飞书配置文件路径（默认 ~/.openclaw/openclaw.json）
|
||
|
||
示例:
|
||
python3 scripts/feishu_urgent.py "🧴 主人！该买洗发膏了！" ou_xxxxxx
|
||
FEISHU_USER_ID=ou_xxxxxx python3 scripts/feishu_urgent.py "开会了"
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import urllib.request
|
||
import os
|
||
|
||
# 配置
|
||
CONFIG_PATH = os.environ.get("FEISHU_CONFIG_PATH",
|
||
os.path.expanduser("~/.openclaw/openclaw.json"))
|
||
DEFAULT_USER_ID = os.environ.get("FEISHU_USER_ID", "")
|
||
BASE_URL = "https://open.feishu.cn/open-apis"
|
||
|
||
|
||
def load_config():
|
||
with open(CONFIG_PATH) as f:
|
||
cfg = json.load(f)
|
||
fs = cfg["channels"]["feishu"]
|
||
return fs["appId"], fs["appSecret"]
|
||
|
||
|
||
def get_token(app_id, app_secret):
|
||
req = urllib.request.Request(
|
||
f"{BASE_URL}/auth/v3/tenant_access_token/internal",
|
||
data=json.dumps({"app_id": app_id, "app_secret": app_secret}).encode(),
|
||
headers={"Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req) as resp:
|
||
data = json.loads(resp.read())
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"获取 token 失败: {data}")
|
||
return data["tenant_access_token"]
|
||
|
||
|
||
def send_message(token, user_id, content):
|
||
body = {
|
||
"receive_id": user_id,
|
||
"msg_type": "text",
|
||
"content": json.dumps({"text": content}),
|
||
}
|
||
req = urllib.request.Request(
|
||
f"{BASE_URL}/im/v1/messages?receive_id_type=open_id",
|
||
data=json.dumps(body).encode(),
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req) as resp:
|
||
data = json.loads(resp.read())
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"发送消息失败: {data}")
|
||
return data["data"]["message_id"]
|
||
|
||
|
||
def make_urgent(token, message_id, user_id):
|
||
body = {"urgent_type": "app", "user_id_list": [user_id]}
|
||
req = urllib.request.Request(
|
||
f"{BASE_URL}/im/v1/messages/{message_id}/urgent_app?user_id_type=open_id",
|
||
data=json.dumps(body).encode(),
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
method="PATCH",
|
||
)
|
||
with urllib.request.urlopen(req) as resp:
|
||
data = json.loads(resp.read())
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"加急失败: {data}")
|
||
return data
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
print("用法: python3 feishu_urgent.py <消息内容> <user_id>", file=sys.stderr)
|
||
print("或设置环境变量 FEISHU_USER_ID 作为默认 user_id", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
content = sys.argv[1]
|
||
user_id = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_USER_ID
|
||
|
||
if not user_id:
|
||
print("❌ 必须提供 user_id：传参或设置 FEISHU_USER_ID 环境变量", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
try:
|
||
app_id, app_secret = load_config()
|
||
print(f"[1/4] 已读取飞书配置", file=sys.stderr)
|
||
|
||
token = get_token(app_id, app_secret)
|
||
print(f"[2/4] 已获取 token", file=sys.stderr)
|
||
|
||
msg_id = send_message(token, user_id, content)
|
||
print(f"[3/4] 消息已发送", file=sys.stderr)
|
||
|
||
result = make_urgent(token, msg_id, user_id)
|
||
invalid = result.get("data", {}).get("invalid_user_id_list", [])
|
||
if invalid:
|
||
print(f"[4/4] ⚠️ 加急部分失败，无效用户", file=sys.stderr)
|
||
else:
|
||
print(f"[4/4] ✅ 加急成功！用户应已收到弹窗", file=sys.stderr)
|
||
|
||
print(json.dumps({"ok": True, "message_id": msg_id, "invalid_users": invalid}))
|
||
|
||
except RuntimeError as e:
|
||
print(f"\n❌ {e}", file=sys.stderr)
|
||
print(json.dumps({"ok": False, "error": str(e)}))
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|