# @Author : Rocky # @File : UserController.py # @Time : 2024/11/27 16:35 import hashlib import hmac import os import time import requests from Ansjer.config import LOGGER from django.views.generic.base import View from Model.models import WeChatMiniProgram, Device_User, Device_Info, DeviceNameLanguage from Object.Enums.WeChatEnum import WeChatMiniProgramAPIEnum, WeChatMiniProgramConfigEnum from Object.RedisObject import RedisObject from Object.ResponseObject import ResponseObject class UserView(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.GET, request, operation) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.POST, request, operation) def validation(self, request_dict, request, operation): lang = request_dict.get('lang', 'cn') response = ResponseObject(lang) if operation == 'getPhoneNumber': return self.get_phone_number(request_dict, response) elif operation == 'userLogin': return self.user_login(request_dict, response) else: user_id = self.check_session_key_and_get_user_id(request) if not user_id: return response.json(309) if operation == 'deviceList': return self.device_list(user_id, response) @classmethod def check_session_key_and_get_user_id(cls, request): """ 校验 session_key 是否有效 https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/user-login/checkSessionKey.html @return: bool """ try: login_status = request.META.get('HTTP_AUTHORIZATION') if not login_status: return False wechat_mini_program_qs = WeChatMiniProgram.objects.filter(login_status=login_status). \ values('openid', 'session_key', 'phone_number') if not wechat_mini_program_qs.exists(): return False openid = wechat_mini_program_qs[0]['openid'] session_key = wechat_mini_program_qs[0]['session_key'] signature = cls.generate_signature(session_key) access_token = cls.get_access_token() params = { 'openid': openid, 'access_token': access_token, 'signature': signature, 'sig_method': 'hmac_sha256' } r = requests.get(url=WeChatMiniProgramAPIEnum.checkSessionKeyAPI.value, params=params, timeout=5) result = eval(r.content) errcode = result.get('errcode') assert errcode == 0 # 根据手机号查询用户id phone_number = wechat_mini_program_qs[0]['phone_number'] device_user_qs = Device_User.objects.filter(phone=phone_number).values('userID') if not device_user_qs.exists(): return False return device_user_qs[0]['userID'] except Exception as e: return False @staticmethod def get_access_token(): """ 获取小程序全局唯一后台接口调用凭据，token有效期为7200s https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/mp-access-token/getAccessToken.html @return: access_token """ try: redis_obj = RedisObject() access_token_key = WeChatMiniProgramConfigEnum.AccessTokenKey.value expires_time = redis_obj.get_ttl(access_token_key) # 如果有效时间大于十分钟,返回token,否则刷新token if expires_time > 10 * 60: access_token = redis_obj.get_data(access_token_key) return access_token else: params = { 'grant_type': 'client_credential', 'appid': WeChatMiniProgramConfigEnum.AppID.value, 'secret': WeChatMiniProgramConfigEnum.AppSecret.value } r = requests.get(url=WeChatMiniProgramAPIEnum.getAccessTokenAPI.value, params=params, timeout=5) result = eval(r.content) access_token = result.get('access_token') assert access_token expires_in = result.get('expires_in') # 保存到Redis redis_obj.set_ex_data(key=access_token_key, val=access_token, expire=expires_in) return access_token except Exception as e: LOGGER.info('微信小程序获取token异常:error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return None @classmethod def get_phone_number(cls, request_dict, response): """ 获取手机号码 @param request_dict: @param response: @return: res """ code = request_dict.get('code', None) if not code: return response.json(444) try: access_token = cls.get_access_token() url = WeChatMiniProgramAPIEnum.getPhoneNumberAPI.value.format(access_token) data = { 'code': code } r = requests.post(url=url, json=data, timeout=5) result = eval(r.content) errcode = result.get('errcode') assert errcode == 0 phone_number = result['phone_info']['purePhoneNumber'] res = { 'phone_number': phone_number } return response.json(0, res) except Exception as e: LOGGER.info('微信小程序获取手机号码异常:error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @classmethod def user_login(cls, request_dict, response): """ 小程序登录 https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/user-login/code2Session.html https://developers.weixin.qq.com/miniprogram/dev/framework/open-ability/login.html @param request_dict: @param response: @return: """ js_code = request_dict.get('js_code', None) phone_number = request_dict.get('phone_number', None) if not all([js_code, phone_number]): return response.json(444) try: params = { 'js_code': js_code, 'grant_type': 'authorization_code', 'appid': WeChatMiniProgramConfigEnum.AppID.value, 'secret': WeChatMiniProgramConfigEnum.AppSecret.value } r = requests.get(url=WeChatMiniProgramAPIEnum.code2SessionAPI.value, params=params, timeout=5) result = eval(r.content) # {'session_key': 'xxx' 会话密钥, 'openid': 'xxx' 用户唯一标识} openid = result.get('openid') assert openid openid = result['openid'] session_key = result['session_key'] # 生成登录态 login_status = cls.generate_login_status(openid, session_key) now_time = int(time.time()) wechat_mini_program_qs = WeChatMiniProgram.objects.filter(openid=openid) if wechat_mini_program_qs.exists(): wechat_mini_program_qs.update( login_status=login_status, session_key=session_key, phone_number=phone_number, updated_time=now_time ) else: WeChatMiniProgram.objects.create( login_status=login_status, openid=openid, session_key=session_key, phone_number=phone_number, created_time=now_time, updated_time=now_time ) res = { 'login_status': login_status } return response.json(0, res) except Exception as e: LOGGER.info('微信小程序登录异常:error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def generate_login_status(openid, session_key): """ 生成登录态 @param openid: @param session_key: @return: login_status """ salt = os.urandom(16).hex() # 将openid, session_key和salt拼接成一个字符串 to_hash = openid + session_key + salt # 使用hashlib生成SHA256哈希值 hash_object = hashlib.sha256(to_hash.encode()) # 获取十六进制格式的哈希值 login_status = hash_object.hexdigest() return login_status @staticmethod def generate_signature(session_key): """ 用户登录态签名，用session_key对空字符串签名得到的结果。即 signature = hmac_sha256(session_key, "") @param session_key: 会话密钥 @return: signature """ # 将session_key和要签名的数据（这里是空字符串）转换为字节类型 session_key_bytes = session_key.encode('utf-8') data = b"" # 空字符串转换为字节类型 # 创建一个新的hmac对象，使用sha256作为hash函数 hmac_object = hmac.new(session_key_bytes, data, hashlib.sha256) # 获取十六进制格式的签名结果 signature = hmac_object.hexdigest() return signature @staticmethod def device_list(user_id, response): """ 查询设备列表 @param user_id: 用户id @param response: @return: """ try: img_url_1 = DeviceNameLanguage.objects.filter(name='智能摄像机 C520M').values('app_device_type__iconV2')[0]['app_device_type__iconV2'] img_url_2 = DeviceNameLanguage.objects.filter(name='智能摄像机 C518').values('app_device_type__iconV2')[0]['app_device_type__iconV2'] res = [ { 'device_name': '智能摄像机520', 'service_status': '服务使用中 2025-11-25', 'img_url': img_url_1}, { 'device_name': '智能摄像机518', 'service_status': '服务到期', 'img_url': img_url_2} ] return response.json(0, res) except Exception as e: return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))