send_message.py
4.14 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
#!/usr/bin/env python3
"""
send_message.py - 发送飞书消息（Python 版本，自动处理 JSON 转义）
用法:
python3 send_message.py <chat_id> <message> [msg_type]
示例:
python3 send_message.py oc_xxxxxx "Hello World"
python3 send_message.py oc_xxxxxx "{\"text\":\"Hello\"}" text
python3 send_message.py oc_xxxxxx "{\"title\":\"Title\",\"content\":\"Body\"}" post
"""
import json
import sys
import os
import urllib.request
def get_token():
"""获取 tenant_access_token"""
app_id = os.environ.get('FEISHU_APP_ID', 'cli_a91b01debab81cd1')
app_secret = os.environ.get('FEISHU_APP_SECRET')
if not app_secret:
# 尝试从 .env 文件读取
env_file = os.path.expanduser('~/.hermes/.env')
if os.path.exists(env_file):
with open(env_file, 'r') as f:
for line in f:
if line.startswith('FEISHU_APP_SECRET='):
app_secret = line.split('=', 1)[1].strip()
break
if not app_secret:
print("Error: FEISHU_APP_SECRET not found")
sys.exit(1)
token_data = json.dumps({
"app_id": app_id,
"app_secret": app_secret
}).encode('utf-8')
token_req = urllib.request.Request(
'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal',
data=token_data,
headers={'Content-Type': 'application/json'},
method='POST'
)
with urllib.request.urlopen(token_req, timeout=30) as response:
token_result = json.loads(response.read().decode('utf-8'))
if token_result.get('code') != 0:
print(f"Error getting token: {token_result}")
sys.exit(1)
return token_result['tenant_access_token']
def send_message(chat_id, message, msg_type='text'):
"""发送消息到飞书"""
token = get_token()
# 处理 content 格式
if msg_type == 'text':
# 如果 message 已经是 JSON 字符串，直接使用
try:
json.loads(message)
content = message
except json.JSONDecodeError:
# 否则包装为 JSON 字符串
content = json.dumps({"text": message})
elif msg_type == 'post':
try:
json.loads(message)
content = message
except json.JSONDecodeError:
content = json.dumps({
"zh_cn": {
"title": message,
"content": [[{"tag": "text", "text": message}]]
}
})
elif msg_type == 'interactive':
# 卡片消息，必须是 JSON 字符串
try:
json.loads(message)
content = message
except json.JSONDecodeError:
print("Error: interactive message must be valid JSON")
sys.exit(1)
else:
content = message
message_data = {
"receive_id": chat_id,
"msg_type": msg_type,
"content": content
}
msg_req = urllib.request.Request(
'https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id',
data=json.dumps(message_data).encode('utf-8'),
headers={
'Authorization': f'Bearer {token}',
'Content-Type': 'application/json'
},
method='POST'
)
with urllib.request.urlopen(msg_req, timeout=30) as response:
result = json.loads(response.read().decode('utf-8'))
if result.get('code') == 0:
print(f"✅ Message sent successfully!")
print(f"Message ID: {result['data']['message_id']}")
else:
print(f"❌ Error: {result.get('msg', 'Unknown error')}")
print(f"Full response: {json.dumps(result, indent=2, ensure_ascii=False)}")
sys.exit(1)
def main():
if len(sys.argv) < 3:
print("Usage: python3 send_message.py <chat_id> <message> [msg_type]")
print(" msg_type: text (default), post, interactive")
sys.exit(1)
chat_id = sys.argv[1]
message = sys.argv[2]
msg_type = sys.argv[3] if len(sys.argv) > 3 else 'text'
send_message(chat_id, message, msg_type)
if __name__ == '__main__':
main()