mike
657dc41c46
- OKX交易自动化 (okx-auto-position, okx-crypto, okx-exchange) - 交易信号处理 (signal-confirmation-templates, trading-signal-aggregator) - 量化因子挖掘 (quant-factor-mining) - 长桥集成 (longbridge-cli, longbridge-python-sdk) - 六合彩分析 (lottery-hk) - 股息投资 (dividend-investing, dividend-scanner) - 日内交易 (intraday-trading) - 同花顺 (tonghuashun)
82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
QQ Bot API 直推脚本（备用推送方式）
|
|
|
|
当 hermes send 因 delivery context 跳过时使用。
|
|
直接从 ~/.hermes/.env 读取凭证，通过 QQ Bot API 发送 C2C 消息。
|
|
|
|
用法：
|
|
python3 qq_push.py "消息内容"
|
|
echo "消息" | python3 qq_push.py
|
|
|
|
凭证：从 ~/.hermes/.env 读取 QQ_APP_ID, QQ_CLIENT_SECRET, QQ_ALLOWED_USERS
|
|
"""
|
|
|
|
import os, sys, json, urllib.request
|
|
|
|
def read_env(path):
|
|
"""从 .env 文件读取变量"""
|
|
creds = {}
|
|
for line in open(path).read().splitlines():
|
|
line = line.strip()
|
|
if '=' in line and not line.startswith('#'):
|
|
k, v = line.split('=', 1)
|
|
creds[k.strip()] = v.strip().strip("'\"").strip('"')
|
|
return creds
|
|
|
|
def send_qq_msg(msg, app_id, secret, openid):
|
|
"""通过 QQ Bot API 发送 C2C 消息"""
|
|
# 1. 获取 access token
|
|
token_data = json.dumps({
|
|
'appId': app_id,
|
|
'clientSecret': secret
|
|
}).encode()
|
|
req = urllib.request.Request(
|
|
'https://bots.qq.com/app/getAppAccessToken',
|
|
data=token_data,
|
|
headers={'Content-Type': 'application/json'},
|
|
method='POST'
|
|
)
|
|
resp = urllib.request.urlopen(req, timeout=15)
|
|
token = json.loads(resp.read())['access_token']
|
|
|
|
# 2. 发送消息
|
|
body = json.dumps({'content': msg, 'msg_type': 0}).encode()
|
|
req2 = urllib.request.Request(
|
|
f'https://api.sgroup.qq.com/v2/users/{openid}/messages',
|
|
data=body,
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'QQBot {token}'
|
|
},
|
|
method='POST'
|
|
)
|
|
resp2 = urllib.request.urlopen(req2, timeout=15)
|
|
result = json.loads(resp2.read())
|
|
return result.get('id', 'unknown')
|
|
|
|
if __name__ == '__main__':
|
|
msg = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read().strip()
|
|
if not msg:
|
|
print('Usage: qq_push.py "message"', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
env = read_env(os.path.expanduser('~/.hermes/.env'))
|
|
app_id = env.get('QQ_APP_ID', '')
|
|
secret = env.get('QQ_CLIENT_SECRET', '')
|
|
openid = env.get('QQ_ALLOWED_USERS', 'B1EF50442496D57C1B4F3890501C34C2')
|
|
|
|
if not app_id or not secret:
|
|
print('❌ QQ credentials not found in ~/.hermes/.env', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
msg_id = send_qq_msg(msg, app_id, secret, openid)
|
|
print(f'✅ Sent! msg_id: {msg_id}')
|
|
except urllib.error.HTTPError as e:
|
|
print(f'❌ HTTP {e.code}: {e.read().decode()[:200]}', file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f'❌ {e}', file=sys.stderr)
|
|
sys.exit(1)
|