ASJServer/Object/UVerifyObject.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import base64
import hashlib
import hmac
import json
import logging
import random
import string
import time
import requests
from collections import OrderedDict
from datetime import datetime, timedelta


class UVerifyObject:

    '''
    【友盟+】智能认证 U-Verify（一键登录）
    开发文档:
    https://developer.umeng.com/docs/143070/detail/144785
    https://developer.umeng.com/docs/143070/detail/144783
    '''

    def __init__(self, token):
        # aliyun config
        self.ali_appkey = '204024884'
        self.ali_app_secret = 'lD7AXxc0ji0HzrNXZmfJaz3AecAboItD'

        # umeng config
        self.host_url = 'https://verify5.market.alicloudapi.com'
        self.path_url = '/api/v1/mobile/info'
        self.umeng_app_key = '5d8da0670cafb2a6b5000bc3'

        self.token = token
        self.phone = ''

        self.body_data_str = json.dumps({'token': self.token})
        content_md5_str = base64.b64encode(hashlib.md5(self.body_data_str.encode("UTF-8")).digest()).decode("UTF-8")

        now_date_time = datetime.utcnow() + timedelta(hours=8)
        timestamp = str(int(time.mktime(now_date_time.timetuple()) * 1000))

        sign_headers_dict = {
            'X-Ca-Key': self.ali_appkey,
            'X-Ca-Stage': 'RELEASE',
            'X-Ca-Timestamp': timestamp,
            'X-Ca-Version': '1'
        }

        sign_headers_str = ''
        for key in sorted(sign_headers_dict.keys()):
            value = sign_headers_dict[key]
            sign_headers_str = sign_headers_str + key + ':' + value + '\n'

        # 需要使用有序字典,否则StringToSign错误
        sign_dict = OrderedDict(
            [('HTTPMethod', 'POST'),
            ('Accept', 'application/json'),
            ('Content-MD5', content_md5_str),
            ('Content-Type', 'application/json; charset=UTF-8'),
            ('Date', None,),
            ('Headers', sign_headers_str),
            ('Url', self.path_url + '?' + 'appkey=' + self.umeng_app_key)]
        )

        string_to_sign = ""
        for key in sign_dict.keys():
            value = sign_dict[key]
            if key not in ['Headers', 'Url']:
                if value:
                    string_to_sign = string_to_sign + value + "\n"
                else:
                    string_to_sign = string_to_sign + "\n"
            else:
                string_to_sign = string_to_sign + value

        logger = logging.getLogger('info')
        logger.info('string_to_sign: {}'.format(string_to_sign))

        key_bytes = self.ali_app_secret.encode('utf-8')
        text_bytes = string_to_sign.encode('utf-8')
        hash_bytes = hmac.new(key_bytes, text_bytes, hashlib.sha256).digest()
        signature_str = base64.b64encode(hash_bytes).decode('utf-8')

        nonce_str = self.generate_code(8) + '-' + self.generate_code(4) + '-' + self.generate_code(4) + '-' + \
                    self.generate_code(4) + '-' + self.generate_code(12)

        self.headers_dict = {
            'Content-Type': 'application/json; charset=UTF-8',
            'Accept': 'application/json',
            'X-Ca-Version': '1',
            'X-Ca-Signature-Headers': 'X-Ca-Version,X-Ca-Stage,X-Ca-Key,X-Ca-Timestamp',
            'X-Ca-Stage': 'RELEASE',
            'X-Ca-Key': self.ali_appkey,
            'X-Ca-Timestamp': timestamp,
            'X-Ca-Nonce': nonce_str,
            'Content-MD5': content_md5_str,
            'X-Ca-Signature': signature_str
        }

        self.request_url = self.host_url + self.path_url + '?' + 'appkey=' + self.umeng_app_key


    def generate_code(self, length):
        key = string.digits + 'abcdef'
        str_list = random.sample(key + key, length)  # length >= len(key+key)
        result_str = ''.join(str_list)
        return result_str

    def verify_request(self):
        logger = logging.getLogger('info')
        logger.info('request_url: {}, data: {}, headers: {}'.format(self.request_url, self.body_data_str, self.headers_dict))
        response = requests.post(url=self.request_url, data=self.body_data_str.encode("UTF-8"), headers=self.headers_dict)
        logger.info('response: {}, headers: {}, text: {}'.format(response, response.headers, response.text))
        if response.status_code != 200:
            return False
        # 替换字符串,防止将字符串转为字典时报错
        res_text = response.text.replace('true', '"true"').replace('false', '"false"').replace('null', '"null"')
        logger.info('res_text: {}'.format(res_text))
        res = eval(res_text)
        print(res['success'])
        if res['success'] == 'false':
            return False
        self.phone = res['data']['mobile']  # 获取手机号
        return True