90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""迁移码：按目标俱乐部 AppID 生成无限小程序码并上传 COS。"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import logging
|
|
import time
|
|
from typing import Optional, Tuple
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
from jituan.services.club_migrate import SCENE_PREFIX, scene_for_token
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RECEIVE_PAGE = 'pages/migrate/receive'
|
|
|
|
|
|
def generate_migrate_qrcode(*, to_club_id: str, token: str) -> Tuple[str, str]:
|
|
"""返回 (relative_url, err)。成功时 err 为空。"""
|
|
from utils.weixin_token import get_club_mini_access_token, is_weixin_token_invalid
|
|
|
|
club_id = (to_club_id or '').strip()
|
|
if not club_id:
|
|
return '', '缺少目标俱乐部'
|
|
scene = scene_for_token(token)
|
|
if len(scene) > 32:
|
|
return '', 'scene 超长'
|
|
if not scene.startswith(SCENE_PREFIX):
|
|
return '', 'scene 非法'
|
|
|
|
access_token = get_club_mini_access_token(club_id)
|
|
if not access_token:
|
|
return '', f'俱乐部 {club_id} 获取微信 token 失败'
|
|
|
|
post_data = {
|
|
'scene': scene,
|
|
'page': RECEIVE_PAGE,
|
|
'width': 430,
|
|
'auto_color': False,
|
|
'line_color': {'r': 0, 'g': 0, 'b': 0},
|
|
'check_path': False,
|
|
}
|
|
|
|
def _req(tok: str):
|
|
url = f'https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={tok}'
|
|
return requests.post(url, json=post_data, timeout=15)
|
|
|
|
wx_resp = _req(access_token)
|
|
if wx_resp.headers.get('content-type', '').startswith('application/json'):
|
|
try:
|
|
err = wx_resp.json()
|
|
if is_weixin_token_invalid(err.get('errcode'), err.get('errmsg', '')):
|
|
access_token = get_club_mini_access_token(club_id, force_refresh=True)
|
|
if access_token:
|
|
wx_resp = _req(access_token)
|
|
except Exception:
|
|
pass
|
|
|
|
content_type = wx_resp.headers.get('content-type', '')
|
|
if wx_resp.status_code != 200 or 'image' not in content_type:
|
|
errmsg = '未知错误'
|
|
try:
|
|
error_info = wx_resp.json()
|
|
errmsg = error_info.get('errmsg', errmsg)
|
|
logger.error('migrate qr fail club=%s err=%s', club_id, error_info)
|
|
except Exception:
|
|
logger.error('migrate qr fail club=%s status=%s', club_id, wx_resp.status_code)
|
|
return '', f'生成小程序码失败: {errmsg}'
|
|
|
|
from utils.oss_utils import upload_to_oss
|
|
|
|
filename = f"migrate_{club_id}_{token}_{int(time.time())}.png"
|
|
oss_path = f"club_migrate/{club_id}/{filename}"
|
|
file_obj = io.BytesIO(wx_resp.content)
|
|
try:
|
|
full_url = upload_to_oss(file_obj, oss_path)
|
|
except Exception as e:
|
|
logger.exception('migrate qr upload fail: %s', e)
|
|
return '', f'上传小程序码失败: {e}'
|
|
if not full_url:
|
|
return '', '上传小程序码失败'
|
|
|
|
cos_domain = getattr(settings, 'COS_DOMAIN', '').rstrip('/')
|
|
if cos_domain and full_url.startswith(cos_domain):
|
|
relative_path = full_url.replace(f'{cos_domain}/', '', 1)
|
|
else:
|
|
relative_path = oss_path
|
|
return relative_path, ''
|