73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from fastapi import UploadFile
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.common.errors.app_error import NotFoundAppError
|
|
from app.common.utils.id_gen import new_order_no
|
|
from app.core.storage import storage_service
|
|
from app.models.entities import MediaAsset
|
|
from app.modules.assets.repository import AssetsRepository
|
|
|
|
|
|
class AssetsService:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.repository = AssetsRepository(db)
|
|
|
|
def create_upload_token(self, media_type: str) -> dict:
|
|
return {
|
|
"uploadToken": new_order_no("upload"),
|
|
"mediaType": media_type,
|
|
"uploadMode": "multipart",
|
|
}
|
|
|
|
def save_asset(self, user_id: int, file: UploadFile, media_type: str) -> dict:
|
|
stored = storage_service.save_upload(file, folder=f"uploads/{media_type}")
|
|
asset = MediaAsset(
|
|
asset_no=new_order_no("asset"),
|
|
user_id=user_id,
|
|
media_type=media_type,
|
|
source_type="upload",
|
|
original_filename=file.filename or "upload.bin",
|
|
mime_type=file.content_type or "application/octet-stream",
|
|
file_ext=Path(file.filename or "").suffix,
|
|
file_size=stored["file_size"],
|
|
storage_provider="local",
|
|
storage_bucket="local",
|
|
storage_key=stored["storage_key"],
|
|
public_url=stored["public_url"],
|
|
sha256=stored["sha256"],
|
|
status="active",
|
|
)
|
|
self.db.add(asset)
|
|
self.db.commit()
|
|
self.db.refresh(asset)
|
|
return self.serialize(asset)
|
|
|
|
def list_assets(self, user_id: int) -> list[dict]:
|
|
return [self.serialize(item) for item in self.repository.list_assets(user_id).all()]
|
|
|
|
def delete_asset(self, user_id: int, asset_id: int) -> dict:
|
|
asset = self.repository.get_asset(user_id, asset_id)
|
|
if not asset:
|
|
raise NotFoundAppError("asset not found", code=40003)
|
|
asset.status = "deleted"
|
|
asset.deleted_at = datetime.utcnow()
|
|
self.db.commit()
|
|
return {"assetId": asset_id, "deleted": True}
|
|
|
|
@staticmethod
|
|
def serialize(asset: MediaAsset) -> dict:
|
|
return {
|
|
"id": asset.id,
|
|
"assetNo": asset.asset_no,
|
|
"mediaType": asset.media_type,
|
|
"originalFilename": asset.original_filename,
|
|
"fileSize": asset.file_size,
|
|
"publicUrl": asset.public_url,
|
|
"createdAt": asset.created_at.isoformat(),
|
|
}
|
|
|