sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
生成有效的会员JWT Token用于SSE测试（简化版）
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import base64
|
||
import hashlib
|
||
from datetime import datetime, timedelta
|
||
|
||
def log_message(message, status="INFO"):
|
||
"""记录消息"""
|
||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||
status_icon = {
|
||
"PASS": "✅",
|
||
"FAIL": "❌",
|
||
"INFO": "ℹ️",
|
||
"WARN": "⚠️"
|
||
}
|
||
print(f"[{timestamp}] {status_icon.get(status, 'ℹ️')} {message}")
|
||
|
||
def create_simple_jwt(member_id):
|
||
"""手动生成JWT token"""
|
||
# JWT头部
|
||
header = {
|
||
"alg": "HS256",
|
||
"typ": "JWT"
|
||
}
|
||
|
||
# JWT载荷
|
||
payload = {
|
||
"sub": str(member_id),
|
||
"name": f"测试用户{member_id[-4:]}",
|
||
"role": "Member",
|
||
"member_id": str(member_id),
|
||
"exp": int((datetime.utcnow() + timedelta(hours=24)).timestamp()),
|
||
"iat": int(datetime.utcnow().timestamp()),
|
||
"jti": str(int(time.time()))
|
||
}
|
||
|
||
# 编码
|
||
header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).decode().rstrip('=')
|
||
payload_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip('=')
|
||
|
||
# 签名（使用后端相同的密钥）
|
||
secret = "YourVerySecretKeyForJwtTokenGeneration12345!"
|
||
signature_input = f"{header_b64}.{payload_b64}"
|
||
|
||
# HMAC SHA256签名
|
||
signature = base64.urlsafe_b64encode(
|
||
hashlib.sha256(f"{signature_input}{secret}".encode()).digest()
|
||
).decode().rstrip('=')
|
||
|
||
# 完整的JWT
|
||
jwt_token = f"{header_b64}.{payload_b64}.{signature}"
|
||
|
||
return jwt_token
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("=" * 60)
|
||
print("生成SSE测试用的会员Token")
|
||
print("=" * 60)
|
||
|
||
# 使用固定的测试会员ID
|
||
test_member_id = "00000000-0000-0000-0000-000000000001"
|
||
|
||
log_message("生成测试用JWT Token")
|
||
token = create_simple_jwt(test_member_id)
|
||
|
||
print(f"\n🎉 生成的测试Token:")
|
||
print(f"📋 Token: {token}")
|
||
print(f"👤 Member ID: {test_member_id}")
|
||
print(f"🔗 SSE测试URL: http://localhost:5511/api/notifications/sse?token={token}")
|
||
|
||
# 保存到文件
|
||
with open('test_token.txt', 'w', encoding='utf-8') as f:
|
||
f.write(f"TOKEN={token}\n")
|
||
f.write(f"MEMBER_ID={test_member_id}\n")
|
||
f.write(f"SSE_URL=http://localhost:5511/api/notifications/sse?token={token}\n")
|
||
|
||
print(f"\n💾 Token信息已保存到 test_token.txt 文件")
|
||
print(f"📝 也可以直接复制上面的Token到SSE测试页面使用")
|
||
|
||
return token, test_member_id
|
||
|
||
if __name__ == "__main__":
|
||
token, member_id = main() |