推送信息到QQ机器人

通过 https://q.qq.com/qqbot/openclaw/ 界面 快速申请一个 机器人 。

然后 使用 以下脚本进行测试获取自己的 OPENID

import asyncio
import json
import os
import time
import requests

# ================= 配置区 =================
# 机器人凭据
APP_ID = ""
CLIENT_SECRET = ""

# 目标用户的 OpenID
# 如果留空 "" 或保持原样，运行脚本时会自动进入“获取 OpenID”模式
OPENID = "" 

# 缓存文件路径
TOKEN_CACHE = "qqbot_token_cache.json"
# =========================================

def get_access_token():
    """获取鉴权 Token (带本地文件缓存)"""
    if os.path.exists(TOKEN_CACHE):
        try:
            with open(TOKEN_CACHE, "r") as f:
                cache = json.load(f)
                if cache.get("expires_at", 0) > time.time() + 300:
                    return cache["access_token"]
        except Exception:
            pass

    print("[网络] 正在获取新 AccessToken...")
    url = "https://bots.qq.com/app/getAppAccessToken"
    payload = {"appId": APP_ID, "clientSecret": CLIENT_SECRET}
    try:
        response = requests.post(url, json=payload, timeout=10)
        if response.status_code == 200:
            data = response.json()
            token = data.get("access_token")
            try:
                expires_in = int(data.get("expires_in", 7200))
            except:
                expires_in = 7200
            
            with open(TOKEN_CACHE, "w") as f:
                json.dump({"access_token": token, "expires_at": time.time() + expires_in}, f)
            return token
    except Exception as e:
        print(f"[错误] 获取 Token 失败: {e}")
    return None

async def listen_for_openid(token):
    """通过 WebSocket 实时捕获 OpenID"""
    try:
        import websockets
    except ImportError:
        print("\n[错误] 缺少 'websockets' 库。请运行: pip install websockets")
        return None

    # 获取 Gateway 地址
    gw_url = "https://api.sgroup.qq.com/gateway"
    gw_resp = requests.get(gw_url, headers={"Authorization": f"QQBot {token}"})
    ws_url = gw_resp.json().get("url")

    print("\n" + "="*40)
    print(" 进入 [获取 OpenID] 模式")
    print(" 👉 请现在用手机 QQ 给机器人发送一条 [私聊] 消息...")
    print("="*40 + "\n")

    try:
        async with websockets.connect(ws_url) as ws:
            # 鉴权
            intents = (1 << 0) | (1 << 9) | (1 << 12) | (1 << 25) | (1 << 30)
            auth_payload = {
                "op": 2,
                "d": {
                    "token": f"QQBot {token}",
                    "intents": intents,
                    "shard": [0, 1],
                    "properties": {}
                }
            }
            await ws.send(json.dumps(auth_payload))
            
            while True:
                msg = await ws.recv()
                data = json.loads(msg)
                
                # 捕获私聊消息
                if data.get("t") == "C2C_MESSAGE_CREATE":
                    author = data["d"].get("author", {})
                    openid = author.get("user_openid")
                    username = author.get("username", "未知用户")
                    print(f"\n✅ 成功捕获!")
                    print(f"昵称: {username}")
                    print(f"您的 OpenID 是: {openid}")
                    print("-" * 40)
                    print(f"建议将此 ID 填入脚本第 13 行的 OPENID 变量中。")
                    return openid
    except Exception as e:
        print(f"[错误] 监听中断: {e}")
        return None

def send_push_message(token, openid, text):
    """推送文本消息"""
    url = f"https://api.sgroup.qq.com/v2/users/{openid}/messages"
    payload = {"content": text, "msg_type": 0}
    headers = {"Authorization": f"QQBot {token}", "Content-Type": "application/json"}
    
    try:
        response = requests.post(url, json=payload, headers=headers, timeout=10)
        if response.status_code == 200:
            print(f"[成功] 消息已发送至 {openid}")
            return True
        else:
            print(f"[失败] 状态码: {response.status_code}, 详情: {response.text}")
    except Exception as e:
        print(f"[异常] 发送失败: {e}")
    return False

async def main():
    token = get_access_token()
    if not token:
        return

    target_id = OPENID.strip()
    
    # 判断是否需要进入获取流程
    if not target_id or target_id == "在此填入你的_OPENID":
        captured_id = await listen_for_openid(token)
        if captured_id:
            confirm = input(f"\n是否立即向该 ID 发送一条测试消息? (y/n): ")
            if confirm.lower() == 'y':
                send_push_message(token, captured_id, "你好！这是自动捕获 OpenID 后发送的测试消息。")
    else:
        # 已有 ID，直接发送
        send_push_message(token, target_id, "这是一条主动推送的测试消息。")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n已退出。")

测试通过以后，让AI改成自己需要的语言版本即可。