476 lines
17 KiB
Python
476 lines
17 KiB
Python
import asyncio
|
||
import os
|
||
from pathlib import Path
|
||
from weixin_bot import WeixinBot
|
||
|
||
# ---------- 发送图片的核心函数（不依赖 OpenClaw）----------
|
||
import httpx
|
||
import json
|
||
import base64
|
||
import random
|
||
import time
|
||
import hashlib
|
||
import uuid
|
||
from Crypto.Cipher import AES
|
||
from Crypto.Util.Padding import pad
|
||
from urllib.parse import quote
|
||
|
||
def calculate_encrypted_size(raw_size: int) -> int:
|
||
return ((raw_size + 1 + 15) // 16) * 16
|
||
|
||
def prepare_image_upload(image_path: str):
|
||
if not Path(image_path).exists():
|
||
return {"success": False, "error": f"图片文件不存在: {image_path}"}
|
||
with open(image_path, 'rb') as f:
|
||
file_data = f.read()
|
||
rawsize = len(file_data)
|
||
rawfilemd5 = hashlib.md5(file_data).hexdigest()
|
||
filekey = ''.join(random.choices('0123456789abcdef', k=32))
|
||
aeskey_hex = ''.join(random.choices('0123456789abcdef', k=32))
|
||
filesize = calculate_encrypted_size(rawsize)
|
||
return {
|
||
"success": True,
|
||
"filekey": filekey,
|
||
"aeskey_hex": aeskey_hex,
|
||
"rawsize": rawsize,
|
||
"rawfilemd5": rawfilemd5,
|
||
"filesize": filesize,
|
||
"file_data": file_data
|
||
}
|
||
|
||
def aes_encrypt_file(file_path: str, aes_key_hex: str) -> bytes:
|
||
aes_key = bytes.fromhex(aes_key_hex)
|
||
with open(file_path, 'rb') as f:
|
||
file_data = f.read()
|
||
cipher = AES.new(aes_key, AES.MODE_ECB)
|
||
padded_data = pad(file_data, AES.block_size, style='pkcs7')
|
||
encrypted_data = cipher.encrypt(padded_data)
|
||
return encrypted_data
|
||
|
||
def encode_aes_key(aes_key_hex: str) -> str:
|
||
hex_bytes = aes_key_hex.encode('utf-8')
|
||
return base64.b64encode(hex_bytes).decode('utf-8')
|
||
|
||
def get_upload_params(bot_token: str, filekey: str, media_type: int, to_user_id: str,
|
||
rawsize: int, rawfilemd5: str, filesize: int, aeskey_hex: str):
|
||
random_uint32 = os.urandom(4)
|
||
x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8')
|
||
body = {
|
||
"filekey": filekey,
|
||
"media_type": media_type,
|
||
"to_user_id": to_user_id,
|
||
"rawsize": rawsize,
|
||
"rawfilemd5": rawfilemd5,
|
||
"filesize": filesize,
|
||
"no_need_thumb": True,
|
||
"aeskey": aeskey_hex,
|
||
"base_info": {"channel_version": "1.0.0"}
|
||
}
|
||
raw = json.dumps(body, ensure_ascii=False)
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"AuthorizationType": "ilink_bot_token",
|
||
"Authorization": f"Bearer {bot_token}",
|
||
"X-WECHAT-UIN": x_wechat_uin,
|
||
"Content-Length": str(len(raw.encode("utf-8"))),
|
||
}
|
||
import requests
|
||
resp = requests.post(
|
||
"https://ilinkai.weixin.qq.com/ilink/bot/getuploadurl",
|
||
headers=headers,
|
||
data=raw.encode('utf-8'),
|
||
timeout=15
|
||
)
|
||
if resp.status_code != 200:
|
||
return {"success": False, "error": f"申请上传参数失败，HTTP状态码: {resp.status_code}"}
|
||
result = resp.json()
|
||
upload_param = result.get("upload_param", "")
|
||
if not upload_param:
|
||
return {"success": False, "error": "响应中未包含upload_param"}
|
||
return {"success": True, "upload_param": upload_param}
|
||
|
||
def upload_to_cdn(upload_param: str, filekey: str, encrypted_data: bytes):
|
||
encoded_param = quote(upload_param, safe='')
|
||
cdn_url = f"https://novac2c.cdn.weixin.qq.com/c2c/upload?encrypted_query_param={encoded_param}&filekey={filekey}"
|
||
headers = {"Content-Type": "application/octet-stream"}
|
||
resp = httpx.post(cdn_url, headers=headers, content=encrypted_data, timeout=30)
|
||
if resp.status_code == 200:
|
||
encrypt_query_param = resp.headers.get('x-encrypted-param', '')
|
||
if not encrypt_query_param:
|
||
return {"success": False, "error": "未在响应头中找到x-encrypted-param"}
|
||
return {"success": True, "encrypt_query_param": encrypt_query_param}
|
||
else:
|
||
return {"success": False, "error": f"上传失败，HTTP状态码: {resp.status_code}"}
|
||
|
||
def send_weixin_image(bot_token: str, to_user_id: str, context_token: str, image_path: str) -> dict:
|
||
"""发送图片，需要提供正确的 context_token。返回 {"success": bool, "new_context_token": str}"""
|
||
prepare_result = prepare_image_upload(image_path)
|
||
if not prepare_result["success"]:
|
||
print(f"准备图片失败: {prepare_result.get('error')}")
|
||
return {"success": False, "new_context_token": ""}
|
||
|
||
filekey = prepare_result["filekey"]
|
||
aeskey_hex = prepare_result["aeskey_hex"]
|
||
rawsize = prepare_result["rawsize"]
|
||
rawfilemd5 = prepare_result["rawfilemd5"]
|
||
filesize = prepare_result["filesize"]
|
||
|
||
upload_params = get_upload_params(bot_token, filekey, 1, to_user_id,
|
||
rawsize, rawfilemd5, filesize, aeskey_hex)
|
||
if not upload_params["success"]:
|
||
print(f"申请上传失败: {upload_params.get('error')}")
|
||
return {"success": False, "new_context_token": ""}
|
||
|
||
encrypted_data = aes_encrypt_file(image_path, aeskey_hex)
|
||
upload_result = upload_to_cdn(upload_params["upload_param"], filekey, encrypted_data)
|
||
if not upload_result["success"]:
|
||
print(f"上传CDN失败: {upload_result.get('error')}")
|
||
return {"success": False, "new_context_token": ""}
|
||
encrypt_query_param = upload_result["encrypt_query_param"]
|
||
encoded_aes_key = encode_aes_key(aeskey_hex)
|
||
|
||
# 构造最终消息
|
||
timestamp_ms = int(time.time() * 1000)
|
||
random_suffix = uuid.uuid4().hex[:8]
|
||
client_id = f"weixin-bot:{timestamp_ms}-{random_suffix}"
|
||
random_uint32 = os.urandom(4)
|
||
x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8')
|
||
|
||
payload = {
|
||
"msg": {
|
||
"from_user_id": "",
|
||
"to_user_id": to_user_id,
|
||
"client_id": client_id,
|
||
"message_type": 2,
|
||
"message_state": 2,
|
||
"context_token": context_token,
|
||
"item_list": [
|
||
{
|
||
"type": 2,
|
||
"image_item": {
|
||
"media": {
|
||
"encrypt_query_param": encrypt_query_param,
|
||
"aes_key": encoded_aes_key,
|
||
"encrypt_type": 1
|
||
},
|
||
"mid_size": rawsize
|
||
}
|
||
}
|
||
]
|
||
},
|
||
"base_info": {"channel_version": "1.0.0"}
|
||
}
|
||
|
||
raw = json.dumps(payload, ensure_ascii=False)
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"AuthorizationType": "ilink_bot_token",
|
||
"Authorization": f"Bearer {bot_token}",
|
||
"X-WECHAT-UIN": x_wechat_uin,
|
||
"Content-Length": str(len(raw.encode("utf-8"))),
|
||
}
|
||
import requests
|
||
resp = requests.post(
|
||
"https://ilinkai.weixin.qq.com/ilink/bot/sendmessage",
|
||
headers=headers,
|
||
data=raw.encode('utf-8'),
|
||
timeout=15
|
||
)
|
||
if resp.status_code == 200:
|
||
resp_json = resp.json()
|
||
# 调试：打印完整响应，确认 context_token 位置
|
||
print(f"[DEBUG] sendmessage 响应: {json.dumps(resp_json, ensure_ascii=False)[:500]}")
|
||
# 尝试从 msg.context_token 或顶层 context_token 获取
|
||
new_ctx = (
|
||
resp_json.get("msg", {}).get("context_token", "")
|
||
or resp_json.get("context_token", "")
|
||
)
|
||
ret = resp_json.get("ret")
|
||
if ret == 0 or not resp_json:
|
||
print(f"✅ 图片发送成功, new_ctx={'有' if new_ctx else '无'}")
|
||
return {"success": True, "new_context_token": new_ctx}
|
||
else:
|
||
print(f"❌ 发送失败，ret={ret}")
|
||
return {"success": False, "new_context_token": new_ctx}
|
||
else:
|
||
print(f"❌ HTTP错误 {resp.status_code}")
|
||
return {"success": False, "new_context_token": ""}
|
||
|
||
|
||
def send_weixin_text(bot_token: str, to_user_id: str, context_token: str, text: str) -> dict:
|
||
"""发送文字消息，返回 {"success": bool, "new_context_token": str}"""
|
||
timestamp_ms = int(time.time() * 1000)
|
||
random_suffix = uuid.uuid4().hex[:8]
|
||
client_id = f"weixin-bot:{timestamp_ms}-{random_suffix}"
|
||
random_uint32 = os.urandom(4)
|
||
x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8')
|
||
|
||
payload = {
|
||
"msg": {
|
||
"from_user_id": "",
|
||
"to_user_id": to_user_id,
|
||
"client_id": client_id,
|
||
"message_type": 2,
|
||
"message_state": 2,
|
||
"context_token": context_token,
|
||
"item_list": [
|
||
{
|
||
"type": 1,
|
||
"text_item": {"text": text}
|
||
}
|
||
]
|
||
},
|
||
"base_info": {"channel_version": "1.0.0"}
|
||
}
|
||
|
||
raw = json.dumps(payload, ensure_ascii=False)
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"AuthorizationType": "ilink_bot_token",
|
||
"Authorization": f"Bearer {bot_token}",
|
||
"X-WECHAT-UIN": x_wechat_uin,
|
||
"Content-Length": str(len(raw.encode("utf-8"))),
|
||
}
|
||
import requests
|
||
resp = requests.post(
|
||
"https://ilinkai.weixin.qq.com/ilink/bot/sendmessage",
|
||
headers=headers,
|
||
data=raw.encode('utf-8'),
|
||
timeout=15
|
||
)
|
||
if resp.status_code == 200:
|
||
resp_json = resp.json()
|
||
new_ctx = (
|
||
resp_json.get("msg", {}).get("context_token", "")
|
||
or resp_json.get("context_token", "")
|
||
)
|
||
ret = resp_json.get("ret")
|
||
if ret == 0 or not resp_json:
|
||
print(f"✅ 文字发送成功")
|
||
return {"success": True, "new_context_token": new_ctx}
|
||
else:
|
||
print(f"❌ 文字发送失败，ret={ret}, errmsg={resp_json.get('errmsg', '')}")
|
||
return {"success": False, "new_context_token": new_ctx}
|
||
else:
|
||
print(f"❌ 文字HTTP错误 {resp.status_code}")
|
||
return {"success": False, "new_context_token": ""}
|
||
|
||
|
||
class weix:
|
||
def __init__(self):
|
||
self.bot = WeixinBot()
|
||
self.bot.login()
|
||
self.BOT_TOKEN = self.bot._credentials.token
|
||
self.MY_USER_ID = self.bot._credentials.user_id
|
||
self.bot.on_message(self.handle_message)
|
||
self.msg = ""
|
||
|
||
async def send_text(self, input_text):
|
||
# 不通过 bot.reply()，改用独立函数发送，避免 context_token 冲突
|
||
context_token = self.bot._context_tokens.get(self.msg.user_id)
|
||
result = await asyncio.to_thread(
|
||
send_weixin_text,
|
||
self.BOT_TOKEN,
|
||
self.msg.user_id,
|
||
context_token,
|
||
input_text
|
||
)
|
||
if result.get("new_context_token"):
|
||
self.bot._context_tokens[self.msg.user_id] = result["new_context_token"]
|
||
self.msg._context_token = result["new_context_token"]
|
||
return result["success"]
|
||
|
||
async def send_images(self, image_path):
|
||
# 发送图片，返回是否成功
|
||
context_token = self.bot._context_tokens.get(self.msg.user_id)
|
||
result = await asyncio.to_thread(
|
||
send_weixin_image,
|
||
self.BOT_TOKEN,
|
||
self.msg.user_id,
|
||
context_token,
|
||
image_path
|
||
)
|
||
# 用服务端返回的新 context_token 更新缓存，避免后续发送失败
|
||
if result.get("new_context_token"):
|
||
self.bot._context_tokens[self.msg.user_id] = result["new_context_token"]
|
||
self.msg._context_token = result["new_context_token"]
|
||
return result["success"]
|
||
|
||
async def send_video(self, video_path):
|
||
"""发送视频文件"""
|
||
context_token = self.bot._context_tokens.get(self.msg.user_id)
|
||
result = await asyncio.to_thread(
|
||
send_weixin_file,
|
||
self.BOT_TOKEN,
|
||
self.msg.user_id,
|
||
context_token,
|
||
video_path
|
||
)
|
||
return result
|
||
|
||
async def handle_message(self, msg):
|
||
print(f"收到消息: {msg.text} from {msg.user_id}")
|
||
self.msg = msg
|
||
|
||
|
||
def send_weixin_file(bot_token: str, to_user_id: str, context_token: str, file_path: str) -> bool:
|
||
"""
|
||
发送任意文件（图片、视频、音频、文档等）
|
||
返回 True/False
|
||
"""
|
||
# 判断文件类型
|
||
file_ext = Path(file_path).suffix.lower().lstrip('.')
|
||
image_exts = ['jpg','jpeg','png','gif','bmp','webp','tiff','svg']
|
||
video_exts = ['mp4','mov','avi','wmv','flv','mkv','webm','mpeg','mpg']
|
||
audio_exts = ['mp3','wav','aac','flac','m4a','ogg','wma']
|
||
doc_exts = ['pdf','doc','docx','xls','xlsx','ppt','pptx','txt']
|
||
archive_exts = ['zip','rar','7z','tar','gz']
|
||
|
||
if file_ext in image_exts:
|
||
media_type = 1
|
||
elif file_ext in video_exts:
|
||
media_type = 2
|
||
elif file_ext in audio_exts:
|
||
media_type = 4
|
||
elif file_ext in doc_exts or file_ext in archive_exts:
|
||
media_type = 3
|
||
else:
|
||
media_type = 3 # 默认当作普通文件
|
||
|
||
# 1. 准备上传参数
|
||
prepare_result = prepare_image_upload(file_path) # 复用之前的准备函数
|
||
if not prepare_result["success"]:
|
||
print(f"准备文件失败: {prepare_result.get('error')}")
|
||
return False
|
||
|
||
filekey = prepare_result["filekey"]
|
||
aeskey_hex = prepare_result["aeskey_hex"]
|
||
rawsize = prepare_result["rawsize"]
|
||
rawfilemd5 = prepare_result["rawfilemd5"]
|
||
filesize = prepare_result["filesize"]
|
||
|
||
# 2. 申请上传参数（媒体类型已确定）
|
||
upload_params = get_upload_params(bot_token, filekey, media_type, to_user_id,
|
||
rawsize, rawfilemd5, filesize, aeskey_hex)
|
||
if not upload_params["success"]:
|
||
print(f"申请上传失败: {upload_params.get('error')}")
|
||
return False
|
||
|
||
# 3. 加密文件
|
||
encrypted_data = aes_encrypt_file(file_path, aeskey_hex)
|
||
|
||
# 4. 上传到 CDN
|
||
upload_result = upload_to_cdn(upload_params["upload_param"], filekey, encrypted_data)
|
||
if not upload_result["success"]:
|
||
print(f"上传CDN失败: {upload_result.get('error')}")
|
||
return False
|
||
encrypt_query_param = upload_result["encrypt_query_param"]
|
||
encoded_aes_key = encode_aes_key(aeskey_hex)
|
||
|
||
# 5. 构造消息体（根据 media_type 不同）
|
||
# 公共字段
|
||
timestamp_ms = int(time.time() * 1000)
|
||
random_suffix = uuid.uuid4().hex[:8]
|
||
client_id = f"weixin-bot:{timestamp_ms}-{random_suffix}"
|
||
random_uint32 = os.urandom(4)
|
||
x_wechat_uin = base64.b64encode(random_uint32).decode('utf-8')
|
||
|
||
# 基础消息结构
|
||
msg_base = {
|
||
"msg": {
|
||
"from_user_id": "",
|
||
"to_user_id": to_user_id,
|
||
"client_id": client_id,
|
||
"message_type": 2,
|
||
"message_state": 2,
|
||
"context_token": context_token,
|
||
"item_list": []
|
||
},
|
||
"base_info": {"channel_version": "1.0.0"}
|
||
}
|
||
|
||
# 根据类型填充 item
|
||
if media_type == 1: # 图片
|
||
item = {
|
||
"type": 2,
|
||
"image_item": {
|
||
"media": {
|
||
"encrypt_query_param": encrypt_query_param,
|
||
"aes_key": encoded_aes_key,
|
||
"encrypt_type": 1
|
||
},
|
||
"mid_size": rawsize
|
||
}
|
||
}
|
||
elif media_type == 2: # 视频
|
||
item = {
|
||
"type": 5,
|
||
"video_item": {
|
||
"media": {
|
||
"encrypt_query_param": encrypt_query_param,
|
||
"aes_key": encoded_aes_key,
|
||
"encrypt_type": 1
|
||
},
|
||
"video_size": rawsize # 明文大小
|
||
}
|
||
}
|
||
elif media_type == 3: # 普通文件
|
||
item = {
|
||
"type": 4,
|
||
"file_item": {
|
||
"media": {
|
||
"encrypt_query_param": encrypt_query_param,
|
||
"aes_key": encoded_aes_key,
|
||
"encrypt_type": 1
|
||
},
|
||
"file_name": Path(file_path).name
|
||
# 注意：main_send_file.py 中注释说 len 和 md5 写了可能发不出去，所以不加
|
||
}
|
||
}
|
||
elif media_type == 4: # 音频（通常也是文件类型，但微信可能支持语音，此处按文件处理）
|
||
# 微信音频消息类型可能是 3？文档不详，暂按文件处理
|
||
# 但为了兼容，我们可以也作为普通文件发送
|
||
item = {
|
||
"type": 4,
|
||
"file_item": {
|
||
"media": {
|
||
"encrypt_query_param": encrypt_query_param,
|
||
"aes_key": encoded_aes_key,
|
||
"encrypt_type": 1
|
||
},
|
||
"file_name": Path(file_path).name
|
||
}
|
||
}
|
||
else:
|
||
print(f"不支持的媒体类型: {media_type}")
|
||
return False
|
||
|
||
msg_base["msg"]["item_list"].append(item)
|
||
|
||
# 6. 发送请求
|
||
raw = json.dumps(msg_base, ensure_ascii=False)
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"AuthorizationType": "ilink_bot_token",
|
||
"Authorization": f"Bearer {bot_token}",
|
||
"X-WECHAT-UIN": x_wechat_uin,
|
||
"Content-Length": str(len(raw.encode("utf-8"))),
|
||
}
|
||
import requests
|
||
resp = requests.post(
|
||
"https://ilinkai.weixin.qq.com/ilink/bot/sendmessage",
|
||
headers=headers,
|
||
data=raw.encode('utf-8'),
|
||
timeout=15
|
||
)
|
||
if resp.status_code == 200:
|
||
resp_json = resp.json()
|
||
if resp_json.get('ret') == 0:
|
||
print(f"✅ {Path(file_path).name} 发送成功")
|
||
return True
|
||
else:
|
||
print(f"❌ 发送失败，ret={resp_json.get('ret')}")
|
||
return False
|
||
else:
|
||
print(f"❌ HTTP错误 {resp.status_code}")
|
||
return False |