yzk_wechat_event/apps/system/utils.py

import requests
from django.conf import settings

from apps.system.choices import ActivityType, AuthorizationPlatformType, QuickCodeCreateMethodType, QuickCodeType, \
    DbEnum
from django.core.files.storage import default_storage
from datetime import date, datetime
from apps.system.models import Authorization, Activity, QuickCode
from libs.kz import KZ
from libs.topsdk.top import TopUtils
from utils.exceptions import CustomProjectException
from utils.tools import InviteCode
from yzk.settings.constant import MiddlePageConstant


def get_alipay_link_by_bid(bid):
    return settings.ALIPAY_ACTIVITY_LINK.format(bid)


def get_response_json_by_activity(activity):
    activity_type = activity.activity_type
    request_json = activity.request_json
    data = None
    if activity_type == ActivityType.ELM:
        authorization_id = request_json.get('authorization_id')
        authorization = Authorization.objects.get(
            pk=authorization_id, platform_type=AuthorizationPlatformType.ELM)
        app_key_secret_info = authorization.app_key_secret_info
        appkey = app_key_secret_info.get('appkey')
        secret = app_key_secret_info.get('secret')
        pid = request_json.get('pid')
        sid = request_json.get('sid')
        activity_id = request_json.get(
            'activity_id', MiddlePageConstant.ELM_ACTIVITY_ID)
        top_utils = TopUtils(appkey, secret)
        response_json = top_utils.get_official_activity(
            pid, activity_id, sid=sid)
        activity.response_json = response_json
        activity.save()
        data = {
            'kl': response_json.get('link').get('eleme_word'),
            'l': response_json.get('link').get('h5_url'),
            'r': activity.title,
        }
    elif activity_type == ActivityType.ALIPAY:
        bid = request_json.get('bid')
        data = {
            'link': get_alipay_link_by_bid(bid)
        }
        activity.response_json = data
        activity.save()
    return data


def get_activity_by_short_chain_key(short_chain_key, select_user=False):
    activity_id = InviteCode.decode(short_chain_key)
    queryset = Activity.objects.filter(id=activity_id, is_delete=False)
    if select_user:
        queryset = queryset.select_related('user')
    activity = queryset.first()
    return activity


def create_quick_code(instance, link, domain=None, create_method=QuickCodeCreateMethodType.PROGRAM):
    domain = domain or instance.domain
    if domain is None:
        return None, None
    app_key = domain.app_key
    secret = domain.secret
    success, quick_code_link = KZ(
        appkey=app_key, secret=secret).gen_short_link(link, domain=domain)
    if not success:
        raise CustomProjectException(detail=quick_code_link)
    quick_code = None
    if quick_code_link:
        quick_code = QuickCode.objects.create(
            user=instance.user,
            activity=instance,
            quick_code_url=quick_code_link,
            converted_url=link,
            create_method=create_method,
            quick_code_type=QuickCodeType.PRIVATE if domain.user_id > 0 else QuickCodeType.OFFICIAL
        )
    return quick_code_link, quick_code


def save_files(files, business=None, return_path=False, user_id=0):
    data = []
    folder = business or DbEnum.HDY
    for file in files:
        fs = default_storage
        timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
        file_name = f'{folder}/{user_id}/{timestamp}_{file.name}'
        fs.save(file_name, file)
        file_url = fs.url(file_name)
        if return_path:
            data.append({
                'path': file_name,
                'url': file_url,
            })
        else:
            data.append(f'{file_url}')
    return data


def test_authorization(instance: Authorization, field):
    field_info = getattr(instance, field)
    if not field_info:
        return False
    platform_type = instance.platform_type
    success = False
    if platform_type == AuthorizationPlatformType.ELM:
        if field == 'cookie_info':
            cookies = []
            xsrf_token = ''
            for info in field_info:
                cookie = f'{info.get("name")}={info.get("value")}'
                cookies.append(cookie)
                if info.get('name') == 'XSRF-TOKEN':
                    xsrf_token = info.get('value')
            cookie = ';'.join(cookies)
            url = f'https://union.ele.me/user/userinfo?_csrf={xsrf_token}'
            headers = {
                'Cookie': cookie,
                'X-Csrf-Token': xsrf_token,
                'X-Xsrf-Token': xsrf_token,
            }
            res = requests.get(url, headers=headers).json()
            success = res.get('success') and str(res.get('result', {}).get('userId')) == instance.uid
        if field == 'app_key_secret_info':
            appkey = field_info.get('appkey')
            secret = field_info.get('secret')
            top_util = TopUtils(appkey=appkey, secret=secret)
            res = top_util.get_zone_list()
            success = res is not None
    elif platform_type == AuthorizationPlatformType.MT:
        if field == 'cookie_info':
            cookies = []
            for info in field_info:
                cookie = f'{info.get("name")}={info.get("value")}'
                cookies.append(cookie)
            cookie = ';'.join(cookies)
            headers = {
                'Cookie': cookie,
            }
            url = f'https://union.meituan.com/api/account/info'
            res = requests.get(url, headers=headers)
            if res.status_code < 400:
                res = res.json()
                success = res.get('status') == 0 and str(res.get('data', {}).get('uid')) == instance.uid
    if field == 'cookie_info':
        instance.cookie_check_result = int(success)
        instance.cookie_check_time = datetime.now()
        instance.save()
    return success