338 lines
11 KiB
Python
338 lines
11 KiB
Python
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_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() |