ASJServer/Roomumy/Controller/RTCController.py

# @Author    : Rocky
# @File      : RTCController.py
# @Time      : 2025/3/14 8:49
import random
import time
from hashlib import sha1
import hmac
import base64
import binascii
from optparse import OptionParser
from django.views.generic.base import View
from Model.models import Device_Info
from Object.Enums.RTCEnum import RTCConfigEnum
from Service.CommonService import CommonService


class RTCView(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):
        token_code, user_id, response = CommonService.verify_token_get_user_id(request_dict, request)
        if token_code != 0:
            return response.json(token_code)
        if operation == 'getToken':  # 获取token
            return self.get_token(request_dict, response)

    @classmethod
    def get_token(cls, request_dict, response):
        """
        获取token
        https://cloud.baidu.com/doc/RTC/s/Qjxbh7jpu#%E5%9C%A8appserver%E4%B8%8A%E9%83%A8%E7%BD%B2token%E7%94%9F%E6%88%90%E6%9C%8D%E5%8A%A1
        @param request_dict:
        @param response:
        @return: res
        """
        uid = request_dict.get('uid', None)
        if not uid:
            return response.json(444)
        try:
            # 查询主用户user id
            device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID')
            if not device_info_qs.exists():
                return response.json(173)
            pri_user_id = device_info_qs[0]['vodPrimaryUserID']
            # 签名
            now_time = int(time.time())
            expect_time = 999999999
            random_str = cls.generate_random_str()
            data = "ACS{}{}{}{}{}{}".format(RTCConfigEnum.AppID.value, now_time, random_str, uid, pri_user_id, expect_time)
            signature = hmac.new(RTCConfigEnum.AppKey.value, data, sha1).digest()  # .encode('base64').rstrip()
            signature = binascii.b2a_hex(signature)
            token = '{}{}{}{}{}'.format('004', signature, now_time, random_str, expect_time)
            res = {
                'token': token
            }
            return response.json(0, res)
        except Exception as e:
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    @staticmethod
    def generate_random_str():
        # 生成一个随机整数
        random_int = random.randint(0, 2 ** 32 - 1)  # 生成一个 0 到 2^32 - 1 的随机整数
        # 将整数转换为 16 进制字符串
        hex_string = hex(random_int)[2:]  # 去掉前缀 '0x'
        # 补齐至 8 位
        random_str = hex_string.zfill(8)
        return random_str