from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException from tencentcloud.faceid.v20180301 import faceid_client, models from tencentcloud.sms.v20210111 import sms_client, models as sms_models from tencentcloud.faceid.v20180301 import models as faceid_models from qcloud_cos import CosConfig, CosS3Client from app.core.config import settings import json import random import string import uuid from datetime import datetime from fastapi import UploadFile class QCloudManager: """腾讯云服务管理类""" def __init__(self): """初始化认证信息""" self.cred = credential.Credential( settings.TENCENT_SECRET_ID, settings.TENCENT_SECRET_KEY ) # 配置 HTTP self.http_profile = HttpProfile() # 配置 Client self.client_profile = ClientProfile() self.client_profile.httpProfile = self.http_profile # 初始化客户端 self.sms_client = None self.faceid_client = None self.cos_client = None def _init_faceid_client(self): """初始化人脸识别客户端""" if not self.faceid_client: self.http_profile.endpoint = "faceid.tencentcloudapi.com" self.faceid_client = faceid_client.FaceidClient( self.cred, settings.TENCENT_REGION, self.client_profile ) def _init_sms_client(self): """初始化短信客户端""" if not self.sms_client: self.http_profile.endpoint = "sms.tencentcloudapi.com" self.sms_client = sms_client.SmsClient( self.cred, settings.TENCENT_REGION ) def _init_cos_client(self): """初始化 COS 客户端""" if not self.cos_client: config = CosConfig( Region=settings.COS_REGION, SecretId=settings.TENCENT_SECRET_ID, SecretKey=settings.TENCENT_SECRET_KEY ) self.cos_client = CosS3Client(config) def generate_phone_code(self, length: int = 4) -> str: """生成验证码""" return ''.join(random.choices(string.digits, k=length)) async def send_sms_code(self, phone: str) -> tuple[str, str]: """ 发送短信验证码 Args: phone: 手机号 Returns: tuple: (验证码, 请求ID) Raises: Exception: 发送失败时抛出异常 """ try: self._init_sms_client() # 生成验证码 code = self.generate_phone_code() # 构建请求 req = sms_models.SendSmsRequest() req.SmsSdkAppId = settings.SMS_SDK_APP_ID req.SignName = settings.SMS_SIGN_NAME req.TemplateId = settings.SMS_TEMPLATE_ID req.TemplateParamSet = [code] req.PhoneNumberSet = [f"+86{phone}"] # 发送短信 response = self.sms_client.SendSms(req) # 检查发送结果 if response.SendStatusSet[0].Code != "Ok": raise Exception(response.SendStatusSet[0].Message) return code, response.RequestId except TencentCloudSDKException as e: raise Exception(f"发送短信失败: {str(e)}") async def send_sms_order_receive(self, phone: str) -> tuple[str, str]: """ 发送订单接单短信 """ try: self._init_sms_client() # 构建请求 req = sms_models.SendSmsRequest() req.SmsSdkAppId = settings.SMS_SDK_APP_ID req.SignName = settings.SMS_SIGN_NAME req.TemplateId = settings.SMS_TEMPLATE_ID_ORDER_RECEIVE req.PhoneNumberSet = [f"+86{phone}"] # 发送短信 response = self.sms_client.SendSms(req) # 检查发送结果 if response.SendStatusSet[0].Code != "Ok": raise Exception(response.SendStatusSet[0].Message) return response.RequestId except TencentCloudSDKException as e: raise Exception(f"发送订单接单短信失败: {str(e)}") async def send_sms_order_complete(self, phone: str) -> tuple[str, str]: """ 发送订单完成短信 """ try: self._init_sms_client() # 构建请求 req = sms_models.SendSmsRequest() req.SmsSdkAppId = settings.SMS_SDK_APP_ID req.SignName = settings.SMS_SIGN_NAME req.TemplateId = settings.SMS_TEMPLATE_ID_ORDER_COMPLETE req.PhoneNumberSet = [f"+86{phone}"] # 发送短信 response = self.sms_client.SendSms(req) # 检查发送结果 if response.SendStatusSet[0].Code != "Ok": raise Exception(response.SendStatusSet[0].Message) return response.RequestId except TencentCloudSDKException as e: raise Exception(f"发送订单完成短信失败: {str(e)}") async def send_sms_code_additional_fee(self, phone: str) -> tuple[str, str]: """ 发送加价短信验证码 """ try: self._init_sms_client() # 构建请求 req = sms_models.SendSmsRequest() req.SmsSdkAppId = settings.SMS_SDK_APP_ID req.SignName = settings.SMS_SIGN_NAME req.TemplateId = settings.SMS_TEMPLATE_ID_ADDITIONAL_FEE req.PhoneNumberSet = [f"+86{phone}"] # 发送短信 response = self.sms_client.SendSms(req) # 检查发送结果 if response.SendStatusSet[0].Code != "Ok": raise Exception(response.SendStatusSet[0].Message) return response.RequestId except TencentCloudSDKException as e: raise Exception(f"发送加价短信验证码失败: {str(e)}") async def verify_id_card(self, id_card: str, name: str) -> dict: """ 身份证实名认证 Args: id_card: 身份证号 name: 姓名 Returns: dict: { "Result": "0"/"1"/"2", # 0:一致 1/2:不一致 "Description": str, # 结果描述 "RequestId": str # 请求ID } Raises: TencentCloudSDKException: 调用API失败 """ try: self._init_faceid_client() # 构建请求 req = faceid_models.IdCardVerificationRequest() params = { "IdCard": id_card, "Name": name } req.from_json_string(json.dumps(params)) # 发送请求 response = self.faceid_client.IdCardVerification(req) # 解析结果 result = json.loads(response.to_json_string()) return { "Result": result.get("Result", "0"), "Description": result.get("Description", "验证失败"), "RequestId": result.get("RequestId", "") } except TencentCloudSDKException as e: raise Exception(f"身份证实名认证失败: {str(e)}") async def upload_file_bytes(self, file_bytes: bytes, key: str) -> str: """ 上传文件字节流到 COS """ try: self._init_cos_client() # 上传文件 self.cos_client.put_object( Bucket=settings.COS_BUCKET, Body=file_bytes, Key=key, ) return f"https://{settings.COS_BASE_URL}/{key}" except Exception as e: raise Exception(f"文件上传失败: {str(e)}") async def upload_file(self, file: UploadFile, folder: str = None) -> str: """ 上传文件到 COS Args: file: 上传的文件 folder: 存储文件夹名称，默认使用日期 Returns: str: 文件访问URL Raises: Exception: 上传失败时抛出异常 """ try: self._init_cos_client() # 生成存储路径 folder = folder or datetime.now().strftime('%Y%m%d') # 获取文件名 if file.filename: ext = file.filename.split('.')[-1] if '.' in file.filename else '' name = file.filename.split('.')[0] if '.' in file.filename else file.filename key = f"{folder}/{name}_{uuid.uuid4()}.{ext}" else: key = f"{folder}/{uuid.uuid4()}.{ext}" # 上传文件 self.cos_client.put_object( Bucket=settings.COS_BUCKET, Body=await file.read(), Key=key, ContentType=file.content_type, ACL="public-read" ) # 返回文件URL return f"https://{settings.COS_BASE_URL}/{key}" except Exception as e: raise Exception(f"文件上传失败: {str(e)}") def get_upload_url(self, key: str, expire: int = 300) -> str: """ 获取预签名上传URL Args: key: 文件路径 expire: 链接有效期(秒) Returns: str: 预签名URL """ try: self._init_cos_client() return self.cos_client.get_presigned_url( Method='PUT', Bucket=settings.COS_BUCKET, Key=key, Expired=expire ) except Exception as e: raise Exception(f"获取上传URL失败: {str(e)}") async def verify_bank_card(self, name: str, id_card: str, card_number: str) -> dict: """ 银行卡三要素核验 Args: name: 姓名 id_card: 身份证号 card_number: 银行卡号 Returns: dict: { "Result": str, # "0":认证通过 "1":认证未通过 "2":认证不确定 "Description": str, # 结果描述 "RequestId": str # 请求ID } Raises: Exception: 验证失败时抛出异常 """ try: self._init_faceid_client() # 构建请求 req = faceid_models.BankCardVerificationRequest() params = { "Name": name, "IdCard": id_card, "BankCard": card_number } req.from_json_string(json.dumps(params)) # 发送请求 response = self.faceid_client.BankCardVerification(req) # 解析结果 result = json.loads(response.to_json_string()) return { "Result": result.get("Result", "2"), "Description": result.get("Description", "验证失败"), "RequestId": result.get("RequestId", "") } except TencentCloudSDKException as e: raise Exception(f"银行卡三要素核验失败: {str(e)}") # 创建全局实例 qcloud_manager = QCloudManager()