feat: update ali oss client
This commit is contained in:
yanghao05
@@ -69,12 +69,12 @@ func Provide(opts ...opt.Option) error {
|
||||
return err
|
||||
}
|
||||
if err := container.Container.Provide(func(
|
||||
ali *ali.Config,
|
||||
app *app.Config,
|
||||
oss *ali.OSSClient,
|
||||
) (*uploads, error) {
|
||||
obj := &uploads{
|
||||
ali: ali,
|
||||
app: app,
|
||||
oss: oss,
|
||||
}
|
||||
|
||||
return obj, nil
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
_ "go.ipao.vip/atom"
|
||||
_ "go.ipao.vip/atom/contracts"
|
||||
. "go.ipao.vip/atom/fen"
|
||||
"mime/multipart"
|
||||
"quyun/app/requests"
|
||||
)
|
||||
|
||||
@@ -82,26 +81,10 @@ func (r *Routes) Register(router fiber.Router) {
|
||||
))
|
||||
|
||||
// 注册路由组: uploads
|
||||
router.Post("/v1/admin/uploads/:md5/chunks/:idx", Func3(
|
||||
r.uploads.Chunks,
|
||||
PathParam[string]("md5"),
|
||||
PathParam[string]("idx"),
|
||||
File[multipart.FileHeader]("file"),
|
||||
))
|
||||
|
||||
router.Post("/v1/admin/uploads/:md5/complete", Func2(
|
||||
r.uploads.Complete,
|
||||
PathParam[string]("md5"),
|
||||
Body[UploadFileInfo]("body"),
|
||||
))
|
||||
|
||||
router.Get("/v1/admin/uploads/token", DataFunc0(
|
||||
r.uploads.Token,
|
||||
))
|
||||
|
||||
router.Get("/v1/admin/uploads/pre-uploaded-check/:md5", Func1(
|
||||
router.Get("/v1/admin/uploads/pre-uploaded-check/:md5.:ext", DataFunc2(
|
||||
r.uploads.PreUploadCheck,
|
||||
PathParam[string]("md5"),
|
||||
PathParam[string]("ext"),
|
||||
))
|
||||
|
||||
router.Post("/v1/admin/uploads/post-uploaded-action", Func1(
|
||||
|
||||
@@ -3,20 +3,16 @@ package admin
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"mime/multipart"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"quyun/app/models"
|
||||
"quyun/database/schemas/public/model"
|
||||
"quyun/pkg/utils"
|
||||
"quyun/providers/ali"
|
||||
"quyun/providers/app"
|
||||
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
|
||||
"github.com/go-jet/jet/v2/qrm"
|
||||
"github.com/gofiber/fiber/v3"
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const UPLOAD_PATH = "quyun"
|
||||
@@ -24,140 +20,30 @@ const UPLOAD_PATH = "quyun"
|
||||
// @provider
|
||||
type uploads struct {
|
||||
app *app.Config
|
||||
ali *ali.Config
|
||||
oss *ali.OSSClient
|
||||
}
|
||||
|
||||
func (up *uploads) storagePath() string {
|
||||
return filepath.Join(up.app.StoragePath, "uploads")
|
||||
}
|
||||
|
||||
type UploadChunk struct {
|
||||
Chunk int `query:"chunk"`
|
||||
Md5 string `query:"md5"`
|
||||
}
|
||||
|
||||
type UploadFileInfo struct {
|
||||
Md5 string `json:"md5"`
|
||||
Filename string `json:"filename"`
|
||||
Mime string `json:"mime"`
|
||||
Chunks int `json:"chunks"`
|
||||
}
|
||||
|
||||
// Upload chunks
|
||||
// @Router /v1/admin/uploads/:md5/chunks/:idx [post]
|
||||
// @Bind md5 path
|
||||
// @Bind idx path
|
||||
// @Bind file file
|
||||
func (up *uploads) Chunks(ctx fiber.Ctx, md5, idx string, file *multipart.FileHeader) error {
|
||||
tmpPath := filepath.Join(up.storagePath(), md5, idx)
|
||||
|
||||
// if tmpPath not exists, create it
|
||||
if _, err := os.Stat(tmpPath); os.IsNotExist(err) {
|
||||
if err := os.MkdirAll(filepath.Dir(tmpPath), os.ModePerm); err != nil {
|
||||
log.WithError(err).Errorf("create tmpPath failed %s", tmpPath)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// save file to tmpPath
|
||||
if err := ctx.SaveFile(file, tmpPath); err != nil {
|
||||
log.WithError(err).Errorf("save file to tmpPath failed %s", tmpPath)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Complete uploads
|
||||
// @Router /v1/admin/uploads/:md5/complete [post]
|
||||
// @Bind md5 path
|
||||
// @Bind body body
|
||||
func (up *uploads) Complete(ctx fiber.Ctx, md5 string, body *UploadFileInfo) error {
|
||||
// merge chunks
|
||||
path := filepath.Join(up.storagePath(), md5)
|
||||
defer os.RemoveAll(path)
|
||||
|
||||
targetFile := filepath.Join(up.storagePath(), md5, body.Filename)
|
||||
|
||||
// if targetFile not exists, create it
|
||||
tf, err := os.Create(targetFile)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for i := 0; i < body.Chunks; i++ {
|
||||
tmpPath := filepath.Join(up.storagePath(), md5, fmt.Sprintf("%d", i))
|
||||
|
||||
// open chunk file
|
||||
chunkFile, err := os.Open(tmpPath)
|
||||
if err != nil {
|
||||
tf.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
// copy chunk file to target file
|
||||
if _, err := tf.ReadFrom(chunkFile); err != nil {
|
||||
chunkFile.Close()
|
||||
tf.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
chunkFile.Close()
|
||||
}
|
||||
tf.Close()
|
||||
|
||||
// validate md5
|
||||
ok, err := utils.CompareFileMd5(targetFile, md5)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok {
|
||||
return errors.New("md5 not match")
|
||||
}
|
||||
|
||||
// save file to target path
|
||||
targetPath := filepath.Join(up.storagePath(), md5+filepath.Ext(body.Filename))
|
||||
if err := os.Rename(targetFile, targetPath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fState, err := os.Stat(targetPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
model := &model.Medias{
|
||||
CreatedAt: time.Now(),
|
||||
Name: body.Filename,
|
||||
MimeType: body.Mime,
|
||||
Size: fState.Size(), // Updated to use fState.Size()
|
||||
Path: targetPath,
|
||||
}
|
||||
|
||||
// save to db
|
||||
if err := models.Medias.Create(ctx.Context(), model); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Infof("File %s uploaded successfully", body.Filename)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Token
|
||||
// @Router /v1/admin/uploads/token [get]
|
||||
func (up *uploads) Token(ctx fiber.Ctx) (*ali.PolicyToken, error) {
|
||||
return up.ali.GetToken(UPLOAD_PATH)
|
||||
type PreCheckResp struct {
|
||||
Exists bool `json:"exists"`
|
||||
PreSign *oss.PresignResult `json:"pre_sign"`
|
||||
}
|
||||
|
||||
// PreUploadCheck
|
||||
// @Router /v1/admin/uploads/pre-uploaded-check/:md5 [get]
|
||||
// @Router /v1/admin/uploads/pre-uploaded-check/:md5.:ext [get]
|
||||
// @Bind md5 path
|
||||
func (up *uploads) PreUploadCheck(ctx fiber.Ctx, md5 string) error {
|
||||
// @Bind ext path
|
||||
func (up *uploads) PreUploadCheck(ctx fiber.Ctx, md5, ext string) (*PreCheckResp, error) {
|
||||
_, err := models.Medias.GetByHash(ctx.Context(), md5)
|
||||
if err != nil && errors.Is(err, qrm.ErrNoRows) {
|
||||
return ctx.SendString("ok")
|
||||
preSign, err := up.oss.PreSignUpload(ctx.Context(), fmt.Sprintf("%s%s", md5, ext))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &PreCheckResp{Exists: false, PreSign: preSign}, nil
|
||||
}
|
||||
return ctx.SendString("exists")
|
||||
|
||||
return &PreCheckResp{Exists: true}, nil
|
||||
}
|
||||
|
||||
type PostUploadedForm struct {
|
||||
|
||||
49
backend/app/jobs/download_from_alioss.go
Normal file
49
backend/app/jobs/download_from_alioss.go
Normal file
@@ -0,0 +1,49 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
. "github.com/riverqueue/river"
|
||||
log "github.com/sirupsen/logrus"
|
||||
_ "go.ipao.vip/atom"
|
||||
"go.ipao.vip/atom/contracts"
|
||||
_ "go.ipao.vip/atom/contracts"
|
||||
)
|
||||
|
||||
var _ contracts.JobArgs = (*WechatCallback)(nil)
|
||||
|
||||
type DownloadFromAliOSS struct {
|
||||
Bucket string
|
||||
Path string
|
||||
}
|
||||
|
||||
func (s DownloadFromAliOSS) InsertOpts() InsertOpts {
|
||||
return InsertOpts{
|
||||
Queue: QueueDefault,
|
||||
Priority: PriorityDefault,
|
||||
}
|
||||
}
|
||||
|
||||
func (s DownloadFromAliOSS) Kind() string { return "download_from_ali_oss" }
|
||||
func (a DownloadFromAliOSS) UniqueID() string { return a.Kind() }
|
||||
|
||||
var _ Worker[DownloadFromAliOSS] = (*DownloadFromAliOSSWorker)(nil)
|
||||
|
||||
// @provider(job)
|
||||
type DownloadFromAliOSSWorker struct {
|
||||
WorkerDefaults[DownloadFromAliOSS]
|
||||
}
|
||||
|
||||
func (w *DownloadFromAliOSSWorker) Work(ctx context.Context, job *Job[DownloadFromAliOSS]) error {
|
||||
log := log.WithField("job", job.Args.Kind())
|
||||
|
||||
log.Infof("[Start] Working on job with strings: %+v", job.Args)
|
||||
defer log.Infof("[End] Finished %s", job.Args.Kind())
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *DownloadFromAliOSSWorker) NextRetry(job *Job[DownloadFromAliOSS]) time.Time {
|
||||
return time.Now().Add(30 * time.Second)
|
||||
}
|
||||
@@ -37,6 +37,18 @@ func Provide(opts ...opt.Option) error {
|
||||
}, atom.GroupInitial); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := container.Container.Provide(func(
|
||||
__job *job.Job,
|
||||
) (contracts.Initial, error) {
|
||||
obj := &DownloadFromAliOSSWorker{}
|
||||
if err := river.AddWorkerSafely(__job.Workers, obj); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return obj, nil
|
||||
}, atom.GroupInitial); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := container.Container.Provide(func(
|
||||
__job *job.Job,
|
||||
) (contracts.Initial, error) {
|
||||
|
||||
Reference in New Issue
Block a user