feat: 添加 H5 跳转小程序功能

- 后端: 新增 JS-SDK 签名接口和 URL Scheme 生成接口
- 前端: 创建 MiniProgramLauncher 组件,支持环境自适应
  - 微信内 H5: 使用 wx-open-launch-weapp 开放标签
  - 外部浏览器: 使用 URL Scheme 拉起微信
  - PC 端: 显示小程序码引导扫码
- 引入微信 JS-SDK (jweixin-1.6.0.js)
- 新增 miniprogramService 服务层封装 API 调用

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
zdl
2025-12-12 16:56:04 +08:00
parent 7be5e3b9e1
commit 154bb76212
8 changed files with 1200 additions and 0 deletions

329
app.py
View File

@@ -333,6 +333,15 @@ WECHAT_OPEN_APPSECRET = 'eedef95b11787fd7ca7f1acc6c9061bc'
WECHAT_MP_APPID = 'wx8afd36f7c7b21ba0'
WECHAT_MP_APPSECRET = 'c3ec5a227ddb26ad8a1d4c55efa1cf86'
# 微信小程序配置H5 跳转小程序用)
WECHAT_MINIPROGRAM_APPID = 'wx0edeaab76d4fa414'
WECHAT_MINIPROGRAM_APPSECRET = os.environ.get('WECHAT_MINIPROGRAM_APPSECRET', '0d0c70084f05a8c1411f6b89da7e815d')
WECHAT_MINIPROGRAM_ORIGINAL_ID = 'gh_fd2fd8dd2fb5'
# Redis 缓存键前缀(微信 token
WECHAT_ACCESS_TOKEN_PREFIX = "wechat:access_token:"
WECHAT_JSAPI_TICKET_PREFIX = "wechat:jsapi_ticket:"
# 微信回调地址
WECHAT_REDIRECT_URI = 'https://valuefrontier.cn/api/auth/wechat/callback'
@@ -4751,6 +4760,326 @@ def unbind_wechat_account():
return jsonify({'error': '解绑失败,请重试'}), 500
# ============ H5 跳转小程序相关 API ============
def get_wechat_access_token_cached(appid, appsecret):
"""
获取微信 access_tokenRedis 缓存,支持多 Worker
Args:
appid: 微信 AppID公众号或小程序
appsecret: 对应的 AppSecret
Returns:
access_token 字符串,失败返回 None
"""
cache_key = f"{WECHAT_ACCESS_TOKEN_PREFIX}{appid}"
# 1. 尝试从 Redis 获取缓存
try:
cached = redis_client.get(cache_key)
if cached:
data = json.loads(cached)
# 提前 5 分钟刷新,避免临界问题
if data.get('expires_at', 0) > time.time() + 300:
print(f"[access_token] 使用缓存: appid={appid[:8]}...")
return data['token']
except Exception as e:
print(f"[access_token] Redis 读取失败: {e}")
# 2. 请求新 token
url = "https://api.weixin.qq.com/cgi-bin/token"
params = {
'grant_type': 'client_credential',
'appid': appid,
'secret': appsecret
}
try:
response = requests.get(url, params=params, timeout=10)
result = response.json()
if 'access_token' in result:
token = result['access_token']
expires_in = result.get('expires_in', 7200)
# 3. 存入 RedisTTL 比 token 有效期短 60 秒)
cache_data = {
'token': token,
'expires_at': time.time() + expires_in
}
redis_client.setex(
cache_key,
expires_in - 60,
json.dumps(cache_data)
)
print(f"[access_token] 获取成功: appid={appid[:8]}..., expires_in={expires_in}s")
return token
else:
print(f"[access_token] 获取失败: errcode={result.get('errcode')}, errmsg={result.get('errmsg')}")
return None
except Exception as e:
print(f"[access_token] 请求异常: {e}")
return None
def get_jsapi_ticket_cached(appid, appsecret):
"""
获取 jsapi_ticketRedis 缓存)
用于 JS-SDK 签名
"""
cache_key = f"{WECHAT_JSAPI_TICKET_PREFIX}{appid}"
# 1. 尝试从缓存获取
try:
cached = redis_client.get(cache_key)
if cached:
data = json.loads(cached)
if data.get('expires_at', 0) > time.time() + 300:
print(f"[jsapi_ticket] 使用缓存")
return data['ticket']
except Exception as e:
print(f"[jsapi_ticket] Redis 读取失败: {e}")
# 2. 获取 access_token
access_token = get_wechat_access_token_cached(appid, appsecret)
if not access_token:
return None
# 3. 请求 jsapi_ticket
url = "https://api.weixin.qq.com/cgi-bin/ticket/getticket"
params = {
'access_token': access_token,
'type': 'jsapi'
}
try:
response = requests.get(url, params=params, timeout=10)
result = response.json()
if result.get('errcode') == 0:
ticket = result['ticket']
expires_in = result.get('expires_in', 7200)
# 存入 Redis
cache_data = {
'ticket': ticket,
'expires_at': time.time() + expires_in
}
redis_client.setex(
cache_key,
expires_in - 60,
json.dumps(cache_data)
)
print(f"[jsapi_ticket] 获取成功, expires_in={expires_in}s")
return ticket
else:
print(f"[jsapi_ticket] 获取失败: errcode={result.get('errcode')}, errmsg={result.get('errmsg')}")
return None
except Exception as e:
print(f"[jsapi_ticket] 请求异常: {e}")
return None
def generate_jssdk_signature(url, appid, appsecret):
"""
生成 JS-SDK 签名配置
Args:
url: 当前页面 URL不含 # 及其后的部分)
appid: 公众号 AppID
appsecret: 公众号 AppSecret
Returns:
签名配置字典,失败返回 None
"""
import hashlib
# 获取 jsapi_ticket
ticket = get_jsapi_ticket_cached(appid, appsecret)
if not ticket:
return None
# 生成签名参数
timestamp = int(time.time())
nonce_str = uuid.uuid4().hex
# 签名字符串(必须按字典序排序!)
sign_str = f"jsapi_ticket={ticket}&noncestr={nonce_str}&timestamp={timestamp}&url={url}"
# SHA1 签名
signature = hashlib.sha1(sign_str.encode('utf-8')).hexdigest()
return {
'appId': appid,
'timestamp': timestamp,
'nonceStr': nonce_str,
'signature': signature,
'jsApiList': ['updateAppMessageShareData', 'updateTimelineShareData'],
'openTagList': ['wx-open-launch-weapp']
}
@app.route('/api/wechat/jssdk-config', methods=['POST'])
def api_wechat_jssdk_config():
"""获取微信 JS-SDK 签名配置(用于开放标签)"""
try:
data = request.get_json() or {}
url = data.get('url')
if not url:
return jsonify({
'code': 400,
'message': '缺少必要参数 url',
'data': None
}), 400
# URL 校验:必须是允许的域名
from urllib.parse import urlparse
parsed = urlparse(url)
allowed_domains = ['valuefrontier.cn', 'www.valuefrontier.cn', 'localhost', '127.0.0.1']
if parsed.netloc.split(':')[0] not in allowed_domains:
return jsonify({
'code': 400,
'message': 'URL 域名不在允许范围内',
'data': None
}), 400
# URL 处理:移除 hash 部分
if '#' in url:
url = url.split('#')[0]
# 生成签名(使用公众号配置)
config = generate_jssdk_signature(
url=url,
appid=WECHAT_MP_APPID,
appsecret=WECHAT_MP_APPSECRET
)
if not config:
return jsonify({
'code': 500,
'message': '获取签名配置失败,请稍后重试',
'data': None
}), 500
return jsonify({
'code': 200,
'message': 'success',
'data': config
})
except Exception as e:
print(f"[JS-SDK Config] 异常: {e}")
import traceback
traceback.print_exc()
return jsonify({
'code': 500,
'message': '服务器内部错误',
'data': None
}), 500
@app.route('/api/miniprogram/url-scheme', methods=['POST'])
def api_miniprogram_url_scheme():
"""生成小程序 URL Scheme外部浏览器跳转小程序用"""
try:
# 频率限制
client_ip = request.headers.get('X-Forwarded-For', request.remote_addr)
if client_ip:
client_ip = client_ip.split(',')[0].strip()
rate_key = f"rate_limit:urlscheme:{client_ip}"
current = redis_client.incr(rate_key)
if current == 1:
redis_client.expire(rate_key, 60)
if current > 30: # 每分钟最多 30 次
return jsonify({
'code': 429,
'message': '请求过于频繁,请稍后再试',
'data': None
}), 429
data = request.get_json() or {}
# 参数校验
path = data.get('path')
if path and not path.startswith('/'):
path = '/' + path # 自动补全 /
# 获取小程序 access_token
access_token = get_wechat_access_token_cached(
WECHAT_MINIPROGRAM_APPID,
WECHAT_MINIPROGRAM_APPSECRET
)
if not access_token:
return jsonify({
'code': 500,
'message': '获取访问令牌失败',
'data': None
}), 500
# 构建请求参数
wx_url = f"https://api.weixin.qq.com/wxa/generatescheme?access_token={access_token}"
expire_type = data.get('expire_type', 1)
expire_interval = min(data.get('expire_interval', 30), 30) # 最长30天
payload = {
"is_expire": expire_type == 1
}
# 跳转信息
if path or data.get('query'):
payload["jump_wxa"] = {}
if path:
payload["jump_wxa"]["path"] = path
if data.get('query'):
payload["jump_wxa"]["query"] = data.get('query')
# 有效期设置
if expire_type == 1:
if data.get('expire_time'):
payload["expire_time"] = data.get('expire_time')
else:
payload["expire_interval"] = expire_interval
response = requests.post(wx_url, json=payload, timeout=10)
result = response.json()
if result.get('errcode') == 0:
return jsonify({
'code': 200,
'message': 'success',
'data': {
'openlink': result['openlink'],
'expire_time': data.get('expire_time') or (int(time.time()) + expire_interval * 86400),
'created_at': datetime.utcnow().isoformat() + 'Z'
}
})
else:
print(f"[URL Scheme] 生成失败: errcode={result.get('errcode')}, errmsg={result.get('errmsg')}")
return jsonify({
'code': 500,
'message': f"生成 URL Scheme 失败: {result.get('errmsg', '未知错误')}",
'data': None
}), 500
except Exception as e:
print(f"[URL Scheme] 异常: {e}")
import traceback
traceback.print_exc()
return jsonify({
'code': 500,
'message': '服务器内部错误',
'data': None
}), 500
# 评论模型
class EventComment(db.Model):
"""事件评论"""