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
69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
import requests
|
|
import json
|
|
import os
|
|
|
|
DINGTALK_APP_KEY = "ding4ursdp0l2giat4bj"
|
|
DINGTALK_APP_SECRET = "J0gBicjKiIHoKla7WfKKhRs1Tv8L6Xd5UhW3EVQByF16G7Vn7UUcRhP6u-PBCQNo"
|
|
ROBOT_CODE = "ding4ursdp0l2giat4bj"
|
|
OPEN_CONVERSATION_ID = "cidcjYshXVtKck5LfOO9AqOJg=="
|
|
|
|
FILE_PATH = r"C:\Users\ALC\.openclaw\workspace\前后端功能与开源可修改性分析报告.docx"
|
|
FILE_NAME = "前后端功能与开源可修改性分析报告.docx"
|
|
|
|
def get_access_token():
|
|
url = "https://api.dingtalk.com/v1.0/oauth2/accessToken"
|
|
headers = {"Content-Type": "application/json"}
|
|
data = {"appKey": DINGTALK_APP_KEY, "appSecret": DINGTALK_APP_SECRET}
|
|
r = requests.post(url, headers=headers, json=data, timeout=10)
|
|
return r.json()["accessToken"]
|
|
|
|
def upload_media_v1(access_token, file_path, file_type="file"):
|
|
url = "https://api.dingtalk.com/v1.0/media/upload"
|
|
files = {'media': (os.path.basename(file_path), open(file_path, 'rb'))}
|
|
params = {'type': file_type}
|
|
headers = {'x-acs-dingtalk-access-token': access_token}
|
|
r = requests.post(url, params=params, headers=headers, files=files, timeout=30)
|
|
result = r.json()
|
|
if "mediaId" in result:
|
|
return result["mediaId"]
|
|
raise Exception(f"Upload failed: {result}")
|
|
|
|
def send_message_v1(access_token, media_id):
|
|
url = "https://api.dingtalk.com/v1.0/robot/orgGroup/send"
|
|
headers = {'x-acs-dingtalk-access-token': access_token, 'Content-Type': 'application/json'}
|
|
payload = {
|
|
"openConversationId": OPEN_CONVERSATION_ID,
|
|
"robotCode": ROBOT_CODE,
|
|
"msgKey": "sampleFile",
|
|
"msgParam": json.dumps({"mediaId": media_id, "fileName": FILE_NAME}, ensure_ascii=False)
|
|
}
|
|
r = requests.post(url, headers=headers, json=payload, timeout=30)
|
|
return r.json()
|
|
|
|
def main():
|
|
try:
|
|
print("=== Direct File Upload Test ===")
|
|
token = get_access_token()
|
|
print("Token OK")
|
|
|
|
print("Uploading file...")
|
|
media_id = upload_media_v1(token, FILE_PATH, "file")
|
|
print(f"Media ID: {media_id}")
|
|
|
|
print("Sending to group...")
|
|
result = send_message_v1(token, media_id)
|
|
print(f"Result: {json.dumps(result, ensure_ascii=False)}")
|
|
|
|
if "processQueryKey" in result:
|
|
print(f"\n=== SUCCESS ===\nprocessQueryKey: {result['processQueryKey']}")
|
|
else:
|
|
print(f"\n=== FAILED ===\n{result}")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|