deliveryman/deliveryman-api
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

合并腾讯 qcloud 相关代码

Browse Source
This commit is contained in:
aaron 2025-01-24 22:06:56 +08:00
parent fd28f612b4
commit f2891a17c0
4 changed files with 151 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
app/api/endpoints/upload.py
View File

@ -1,8 +1,6 @@
from fastapi import APIRouter, UploadFile, File, Depends
from typing import List
import uuid
from datetime import datetime
from app.core.cos import cos_manager
from app.core.qcloud import qcloud_manager
from app.core.config import settings
from app.models.upload import UploadResponse, MultiUploadResponse
from app.core.response import success_response, error_response, ResponseModel
@ -11,24 +9,6 @@ from app.models.user import UserDB
router = APIRouter()
async def upload_to_cos(file: UploadFile) -> str:
"""上传文件到腾讯云COS"""
# 生成唯一文件名
ext = file.filename.split('.')[-1] if '.' in file.filename else ''
filename = f"{datetime.now().strftime('%Y%m%d')}/{uuid.uuid4()}.{ext}"
# 上传文件
cos_manager.put_object(
Bucket=settings.COS_BUCKET,
Body=await file.read(),
Key=filename,
ContentType=file.content_type,
ACL="public-read"
)
# 返回文件URL
return f"https://{settings.COS_BASE_URL}/{filename}"
@router.post("/image", response_model=ResponseModel)
async def upload_image(
file: UploadFile = File(...),
@ -39,7 +19,7 @@ async def upload_image(
return error_response(code=400, message="只能上传图片文件")
try:
url = await upload_to_cos(file)
url = await qcloud_manager.upload_file(file)
return success_response(data=UploadResponse(url=url))
except Exception as e:
return error_response(code=500, message=f"上传失败: {str(e)}")
@ -58,7 +38,7 @@ async def upload_images(
for file in files:
if not file.content_type.startswith('image/'):
return error_response(code=400, message="只能上传图片文件")
url = await upload_to_cos(file)
url = await qcloud_manager.upload_file(file)
urls.append(url)
return success_response(data=MultiUploadResponse(urls=urls))
except Exception as e:
@ -72,7 +52,7 @@ async def get_presigned_url(
"""获取预签名上传URL"""
key = f"uploads/{current_user.userid}/{filename}"
try:
url = cos_manager.get_upload_url(key)
url = qcloud_manager.get_upload_url(key)
return success_response(data={
"url": url,
"key": key

27
app/api/endpoints/user.py
View File

@ -38,32 +38,11 @@ redis_client = redis.Redis(
async def send_verify_code(request: VerifyCodeRequest):
"""发送验证码"""
phone = request.phone
code = ''.join(random.choices(string.digits, k=6))
try:
# 实例化认证对象
cred = credential.Credential(
settings.TENCENT_SECRET_ID,
settings.TENCENT_SECRET_KEY
)
# 发送验证码
code, request_id = await qcloud_manager.send_sms_code(phone)
# 实例化短信客户端
client = sms_client.SmsClient(cred, "ap-guangzhou")
# 组装请求参数
req = 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}"]
# 发送短信
resp = client.SendSms(req)
if resp.SendStatusSet[0].Code != "Ok":
return error_response(message=f"短信发送失败: {resp.SendStatusSet[0].Message}")
# 存储验证码到 Redis
redis_client.setex(
f"verify_code:{phone}",
@ -73,7 +52,7 @@ async def send_verify_code(request: VerifyCodeRequest):
return success_response(message="验证码已发送")
except TencentCloudSDKException as e:
except Exception as e:
return error_response(message=f"发送验证码失败: {str(e)}")
@router.post("/login")

69
app/core/cos.py
View File

@ -1,69 +0,0 @@
from qcloud_cos import CosConfig, CosS3Client
from app.core.config import settings
import sys
import logging
# 正常情况日志级别使用INFO，需要定位时可以修改为DEBUG，此时SDK会打印和服务端的通信信息
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
class COSManager:
def __init__(self):
config = CosConfig(
Region=settings.COS_REGION,
SecretId=settings.TENCENT_SECRET_ID,
SecretKey=settings.TENCENT_SECRET_KEY
)
self.client = CosS3Client(config)
def get_upload_url(self, key: str, expires: int = 3600) -> str:
"""
获取预签名上传URL
Args:
key: 对象键
expires: 签名有效期（秒）
Returns:
预签名URL
"""
try:
url = self.client.get_presigned_url(
Method='PUT',
Bucket=settings.COS_BUCKET,
Key=key,
Expired=expires
)
return url
except Exception as e:
logging.error(f"获取上传URL失败: {str(e)}")
raise
def get_download_url(self, key: str, expires: int = 3600) -> str:
"""
获取预签名下载URL
Args:
key: 对象键
expires: 签名有效期（秒）
Returns:
预签名URL
"""
try:
url = self.client.get_presigned_url(
Method='GET',
Bucket=settings.COS_BUCKET,
Key=key,
Expired=expires
)
return url
except Exception as e:
logging.error(f"获取下载URL失败: {str(e)}")
raise
def put_object(self, **kwargs):
"""上传对象的封装方法"""
return self.client.put_object(**kwargs)
# 创建全局 COS 客户端实例
cos_manager = COSManager()

151
app/core/qcloud.py
View File

@ -3,8 +3,15 @@ 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 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:
"""腾讯云服务管理类"""
@ -18,19 +25,88 @@ class QCloudManager:
# 配置 HTTP
self.http_profile = HttpProfile()
self.http_profile.endpoint = "faceid.tencentcloudapi.com"
# 配置 Client
self.client_profile = ClientProfile()
self.client_profile.httpProfile = self.http_profile
# 初始化人脸识别客户端
self.faceid_client = faceid_client.FaceidClient(
self.cred,
settings.TENCENT_REGION,
self.client_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_verify_code(self, length: int = 6) -> 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_verify_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 verify_id_card(self, id_card: str, name: str) -> dict:
"""
身份证实名认证
@ -50,6 +126,8 @@ class QCloudManager:
TencentCloudSDKException: 调用API失败
"""
try:
self._init_faceid_client()
# 构建请求
req = models.IdCardVerificationRequest()
params = {
@ -73,5 +151,64 @@ class QCloudManager:
except TencentCloudSDKException 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')
ext = file.filename.split('.')[-1] if '.' in file.filename 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)}")
# 创建全局实例
qcloud_manager = QCloudManager()