155 lines
5.2 KiB
Python
155 lines
5.2 KiB
Python
from fastapi import APIRouter, HTTPException, Depends, UploadFile, File, Form, Query
|
||
from fastapi.responses import JSONResponse
|
||
from pydantic import BaseModel
|
||
import logging
|
||
from typing import List, Optional, Dict, Any
|
||
|
||
from app.services.qcloud_service import QCloudCOSService
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter()
|
||
|
||
class STSTokenResponse(BaseModel):
|
||
"""STS临时凭证响应"""
|
||
credentials: Dict[str, str]
|
||
expiration: str
|
||
request_id: str
|
||
bucket: str
|
||
region: str
|
||
|
||
class FileInfo(BaseModel):
|
||
"""文件信息"""
|
||
key: str
|
||
url: str
|
||
size: int
|
||
last_modified: str
|
||
etag: str
|
||
|
||
class ListFilesResponse(BaseModel):
|
||
"""列出文件响应"""
|
||
files: List[FileInfo]
|
||
is_truncated: bool
|
||
next_marker: Optional[str] = None
|
||
common_prefixes: Optional[List[Dict[str, Any]]] = None
|
||
|
||
@router.post("/upload", tags=["腾讯云COS"])
|
||
async def upload_file(
|
||
file: UploadFile = File(...),
|
||
directory: str = Form("uploads"),
|
||
qcloud_service: QCloudCOSService = Depends(lambda: QCloudCOSService())
|
||
):
|
||
"""
|
||
上传文件到腾讯云COS
|
||
|
||
- **file**: 要上传的文件
|
||
- **directory**: 存储目录（默认为"uploads"）
|
||
"""
|
||
try:
|
||
# 读取文件内容
|
||
file_content = await file.read()
|
||
|
||
# 上传文件
|
||
result = await qcloud_service.upload_file(
|
||
file_content=file_content,
|
||
file_name=file.filename,
|
||
directory=directory,
|
||
content_type=file.content_type
|
||
)
|
||
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"文件上传失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"文件上传失败: {str(e)}")
|
||
finally:
|
||
# 关闭文件
|
||
await file.close()
|
||
|
||
@router.delete("/files/{key:path}", tags=["腾讯云COS"])
|
||
async def delete_file(
|
||
key: str,
|
||
qcloud_service: QCloudCOSService = Depends(lambda: QCloudCOSService())
|
||
):
|
||
"""
|
||
从腾讯云COS删除文件
|
||
|
||
- **key**: 文件的对象键（COS路径）
|
||
"""
|
||
try:
|
||
success = await qcloud_service.delete_file(key=key)
|
||
if success:
|
||
return {"message": "文件删除成功", "key": key}
|
||
else:
|
||
raise HTTPException(status_code=404, detail=f"文件删除失败，文件可能不存在")
|
||
except Exception as e:
|
||
logger.error(f"文件删除失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"文件删除失败: {str(e)}")
|
||
|
||
@router.get("/files/url/{key:path}", tags=["腾讯云COS"])
|
||
async def get_file_url(
|
||
key: str,
|
||
expires: int = Query(3600, description="URL有效期（秒）"),
|
||
qcloud_service: QCloudCOSService = Depends(lambda: QCloudCOSService())
|
||
):
|
||
"""
|
||
获取COS文件的临时访问URL
|
||
|
||
- **key**: 文件的对象键（COS路径）
|
||
- **expires**: URL的有效期（秒），默认3600秒
|
||
"""
|
||
try:
|
||
url = await qcloud_service.get_file_url(key=key, expires=expires)
|
||
return {"url": url, "key": key, "expires": expires}
|
||
except Exception as e:
|
||
logger.error(f"获取文件URL失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"获取文件URL失败: {str(e)}")
|
||
|
||
@router.post("/sts-token", response_model=STSTokenResponse, tags=["腾讯云COS"])
|
||
async def generate_sts_token(
|
||
allow_prefix: str = Form("*", description="允许操作的对象前缀"),
|
||
duration_seconds: int = Form(1800, description="凭证有效期（秒）"),
|
||
qcloud_service: QCloudCOSService = Depends(lambda: QCloudCOSService())
|
||
):
|
||
"""
|
||
生成腾讯云COS临时访问凭证（STS），用于前端直传
|
||
|
||
- **allow_prefix**: 允许操作的对象前缀，默认为'*'表示所有对象
|
||
- **duration_seconds**: 凭证有效期（秒），默认1800秒
|
||
"""
|
||
try:
|
||
sts_result = await qcloud_service.generate_cos_sts_token(
|
||
allow_prefix=allow_prefix,
|
||
duration_seconds=duration_seconds
|
||
)
|
||
# 添加桶和区域信息，方便前端使用
|
||
sts_result['bucket'] = qcloud_service.bucket
|
||
sts_result['region'] = qcloud_service.region
|
||
|
||
return sts_result
|
||
except Exception as e:
|
||
logger.error(f"生成STS临时凭证失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"生成STS临时凭证失败: {str(e)}")
|
||
|
||
@router.get("/files", response_model=ListFilesResponse, tags=["腾讯云COS"])
|
||
async def list_files(
|
||
directory: str = Query("", description="目录前缀"),
|
||
limit: int = Query(100, description="返回的最大文件数"),
|
||
marker: str = Query("", description="分页标记"),
|
||
qcloud_service: QCloudCOSService = Depends(lambda: QCloudCOSService())
|
||
):
|
||
"""
|
||
列出COS中的文件
|
||
|
||
- **directory**: 目录前缀
|
||
- **limit**: 返回的最大文件数
|
||
- **marker**: 分页标记
|
||
"""
|
||
try:
|
||
result = await qcloud_service.list_files(
|
||
directory=directory,
|
||
limit=limit,
|
||
marker=marker
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"列出文件失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"列出文件失败: {str(e)}") |