15c4480db1
- Add agent configuration files (AGENTS.md, USER.md, IDENTITY.md, SOUL.md) - Add git configuration and skills management scripts - Add frontend/backend analysis tools and reports - Add DingTalk media sender utilities and documentation - Fix OpenClaw runtime environment (Node.js and Python) - Configure git remotes and push scripts
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
import requests
|
|
import json
|
|
import os
|
|
import base64
|
|
from requests_toolbelt.multipart.encoder import MultipartEncoder
|
|
|
|
DINGTALK_APP_KEY = "ding4ursdp0l2giat4bj"
|
|
DINGTALK_APP_SECRET = "J0gBicjKiIHoKla7WfKKhRs1Tv8L6Xd5UhW3EVQByF16G7Vn7UUcRhP6u-PBCQNo"
|
|
ROBOT_CODE = "ding4ursdp0l2giat4bj"
|
|
OPEN_CONVERSATION_ID = "cidcjYshXVtKck5LfOO9AqOJg=="
|
|
|
|
IMAGE_PATH = r"C:\Users\ALC\.openclaw\media\browser\c0ae1018-cfaf-41ed-967a-eebcb93881e8.jpg"
|
|
|
|
def get_token():
|
|
url = "https://oapi.dingtalk.com/gettoken"
|
|
params = {'appkey': DINGTALK_APP_KEY, 'appsecret': DINGTALK_APP_SECRET}
|
|
r = requests.get(url, params=params, timeout=10)
|
|
result = r.json()
|
|
if result.get('errcode') == 0:
|
|
return result['access_token']
|
|
raise Exception(f"Get token failed: {result}")
|
|
|
|
def upload_image(access_token, image_path):
|
|
url = "https://oapi.dingtalk.com/media/upload"
|
|
|
|
with open(image_path, 'rb') as f:
|
|
fields = {
|
|
'access_token': access_token,
|
|
'type': 'image',
|
|
'media': (os.path.basename(image_path), f, 'image/jpeg')
|
|
}
|
|
|
|
m = MultipartEncoder(fields=fields)
|
|
headers = {'Content-Type': m.content_type}
|
|
|
|
r = requests.post(url, headers=headers, data=m, timeout=30)
|
|
return r.json()
|
|
|
|
def send_image(access_token, media_id):
|
|
# 使用 sampleMarkdown 格式发送图片（之前测试成功）
|
|
url = "https://oapi.dingtalk.com/robot/send"
|
|
|
|
# 移除 media_id 前面的 @ 符号
|
|
clean_media_id = media_id.replace('@', '')
|
|
|
|
payload = {
|
|
"msgtype": "markdown",
|
|
"markdown": {
|
|
"title": "CNBlogs首页截图",
|
|
"text": f""
|
|
},
|
|
"openConversationId": OPEN_CONVERSATION_ID
|
|
}
|
|
|
|
params = {'access_token': access_token}
|
|
|
|
r = requests.post(url, params=params, json=payload, timeout=30)
|
|
return r.json()
|
|
|
|
def main():
|
|
try:
|
|
result_path = r"C:\Users\ALC\.openclaw\workspace\send_image_result.txt"
|
|
|
|
with open(result_path, "w", encoding="utf-8") as f:
|
|
f.write("=== DingTalk Image Send ===\n\n")
|
|
|
|
token = get_token()
|
|
f.write(f"Token obtained\n\n")
|
|
|
|
f.write("Upload Image:\n")
|
|
upload_result = upload_image(token, IMAGE_PATH)
|
|
f.write(f"{json.dumps(upload_result, ensure_ascii=False)}\n\n")
|
|
|
|
if upload_result.get('errcode') == 0:
|
|
media_id = upload_result['media_id']
|
|
f.write(f"Media ID: {media_id}\n\n")
|
|
|
|
f.write("Send Image\n")
|
|
send_result = send_image(token, media_id)
|
|
f.write(f"{json.dumps(send_result, ensure_ascii=False)}\n\n")
|
|
|
|
if send_result.get('errcode') == 0 and send_result.get('processQueryKey'):
|
|
f.write("=== SUCCESS ===\n")
|
|
f.write(f"ProcessQueryKey: {send_result['processQueryKey']}\n")
|
|
else:
|
|
f.write("=== FAILED ===\n")
|
|
else:
|
|
f.write("=== UPLOAD FAILED ===\n")
|
|
|
|
print("Check send_image_result.txt")
|
|
|
|
except Exception as e:
|
|
with open(r"C:\Users\ALC\.openclaw\workspace\send_image_error.txt", "w", encoding="utf-8") as f:
|
|
f.write(f"Error: {e}\n")
|
|
import traceback
|
|
f.write(traceback.format_exc())
|
|
print(f"Error: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|