42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
from backend.app.core.error_codes import ErrorCode
|
|
from backend.app.core.exceptions import AppException
|
|
|
|
|
|
class FileService:
|
|
def __init__(self) -> None:
|
|
self.attachments: list[dict] = []
|
|
self.next_attachment_id = 8001
|
|
|
|
def create_upload_token(self, payload: dict) -> dict:
|
|
if not payload["file_name"].strip():
|
|
raise AppException(code=ErrorCode.PARAM_ERROR, message="文件名不能为空", status_code=400)
|
|
|
|
expire_at = datetime.now() + timedelta(hours=1)
|
|
return {
|
|
"upload_url": "https://oss-upload-url",
|
|
"file_url": f"https://oss-public-url/{payload['biz_type']}/{payload['biz_id']}/{payload['file_name']}",
|
|
"headers": {},
|
|
"expire_at": expire_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
def save_attachment(self, payload: dict) -> dict:
|
|
attachment = {
|
|
"attachment_id": self.next_attachment_id,
|
|
"biz_type": payload["biz_type"],
|
|
"biz_id": payload["biz_id"],
|
|
"file_name": payload["file_name"],
|
|
"file_url": payload["file_url"],
|
|
"file_type": payload["file_type"],
|
|
"file_size": payload["file_size"],
|
|
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
self.next_attachment_id += 1
|
|
self.attachments.append(attachment)
|
|
return attachment
|
|
|
|
|
|
file_service = FileService()
|
|
|