jx-callback/business/jxutils/datares/datares.go

package datares

import (
	"fmt"
	"strings"
	"time"

	"git.rosy.net.cn/baseapi/utils"
	"git.rosy.net.cn/jx-callback/business/jxutils"
	"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
	"git.rosy.net.cn/jx-callback/business/jxutils/tasksch"
	"git.rosy.net.cn/jx-callback/business/model"
	"git.rosy.net.cn/jx-callback/business/model/dao"
	"git.rosy.net.cn/jx-callback/business/partner"
	"git.rosy.net.cn/jx-callback/globals"
	"git.rosy.net.cn/jx-callback/globals/api"
	"github.com/qiniu/api.v7/storage"
)

const (
	qiniuTokenExpires   = 300 // 七牛TOKEN有效时间，5分钟
	dataResCacheTimeout = 24 * time.Hour
)

var (
	dataResCache jxutils.SyncMapWithTimeout
)

type UploadResTokenInfo struct {
	Token    string `json:"token,omitempty"`
	Expires  uint64 `json:"expires,omitempty"`
	FileName string `json:"fileName,omitempty"`
	Hit      bool   `json:"hit,omitempty"`
	Img      string `json:"img,omitempty"`
}

func GetQiniuUploadToken(ctx *jxcontext.Context, suffix, hashCode string) (upTokenInfo *UploadResTokenInfo, err error) {
	imgURL := ""
	if hashCode != "" {
		imgURL, _ = GetDataResource(ctx, hashCode)
	}

	putPolicy := storage.PutPolicy{
		Scope:   globals.QiniuBucket,
		Expires: qiniuTokenExpires,
	}
	upTokenInfo = &UploadResTokenInfo{
		Token:    putPolicy.UploadToken(api.QiniuAPI),
		Expires:  putPolicy.Expires,
		FileName: jxutils.GenPicFileName(suffix),
		Hit:      imgURL != "",
		Img:      imgURL,
	}
	dataResCache.StoreWithTimeout(imgURL, hashCode, dataResCacheTimeout)
	return upTokenInfo, err
}

func suffix2MimeType(suffix string) (mimeType string) {
	suffix = strings.Trim(suffix, ". ")
	for k, v := range model.ValidMimeTypes {
		for _, v2 := range v {
			if v2 == suffix {
				mimeType = k
				break
			}
		}
	}
	return mimeType
}

func getMimeTypeFromURL(resourceURL string) (mimeType string) {
	index := strings.LastIndex(resourceURL, ".")
	if index >= 0 {
		mimeType = suffix2MimeType(resourceURL[index:])
	}
	return mimeType
}

func RegisterDataResource(ctx *jxcontext.Context, name, resourceURL, mimeType, hashCode string, resBinary []byte, isUpload2Vendor, isAsync bool) (dataRes *model.DataResource, err error) {
	if model.ValidMimeTypes[mimeType] == nil {
		return nil, fmt.Errorf("MIME type:%s非法", mimeType)
	}
	dataRes = &model.DataResource{
		Name:         name,
		HashCode:     hashCode,
		ResourceType: mimeType,
		MainURL:      resourceURL,
		IsVendor:     int8(utils.Bool2Int(isUpload2Vendor)),
	}

	vendorID := jxutils.GuessDataResourceVendor(resourceURL)
	switch vendorID {
	case model.VendorIDQiNiuCloud:
		dataRes.QiniuURL = resourceURL
	case model.VendorIDEBAI:
		dataRes.EbaiURL = resourceURL
	}

	dao.WrapAddIDCULEntity(dataRes, ctx.GetUserName())
	if err = dao.CreateEntity(dao.GetDB(), dataRes); err != nil {
		if dao.IsDuplicateError(err) {
			dataRes, err = dao.GetDataResource(dao.GetDB(), dataRes.HashCode, dataRes.MainURL)
		} else {
			dataRes = nil
		}
		return dataRes, err
	}
	if isUpload2Vendor {
		// 忽略上传错误
		UploadImage2Vendors(ctx, nil, dataRes, resBinary, isAsync)
	}
	return dataRes, err
}

func TryRegisterDataResource(ctx *jxcontext.Context, name, resourceURL string, isAllowDownLoad, isUpload2Vendor, isAsync bool) (dataRes *model.DataResource, err error) {
	dataRes = &model.DataResource{
		MainURL: resourceURL,
	}
	db := dao.GetDB()
	err = dao.GetEntity(db, dataRes, "MainURL")
	if err != nil {
		if !dao.IsNoRowsError(err) {
			return nil, err
		}
	} else {
		return dataRes, nil
	}
	value, ok := dataResCache.Load(resourceURL)
	var hashCode string
	var resBinary []byte
	if !ok {
		if !isAllowDownLoad {
			return nil, nil
		}
		if resBinary, hashCode, err = jxutils.DownloadFileByURL(resourceURL); err != nil {
			return nil, err
		}
	} else {
		hashCode = value.(string)
	}
	return RegisterDataResource(ctx, name, resourceURL, getMimeTypeFromURL(resourceURL), hashCode, resBinary, isUpload2Vendor, isAsync)
}

func GetDataResource(ctx *jxcontext.Context, hashCode string) (resourceURL string, err error) {
	db := dao.GetDB()
	dataRes, err := dao.GetDataResource(db, hashCode, "")
	if err != nil {
		if dao.IsNoRowsError(err) {
			skuName, err2 := dao.GetSkuNameByHashCode(db, hashCode)
			if err = err2; err == nil {
				resourceURL = skuName.Img
			} else if dao.IsNoRowsError(err) {
				err = nil
			}
		}
	} else {
		resourceURL = dataRes.MainURL
	}
	return resourceURL, err
}

// 这个函数，可能部分平台成功，部分失败
func UploadImage2Vendors(ctx *jxcontext.Context, parentTask tasksch.ITask, dataRes *model.DataResource, imgData []byte, isAsync bool) (hint string, err error) {
	var vendorIDs []int
	if dataRes.EbaiURL == "" {
		vendorIDs = append(vendorIDs, model.VendorIDEBAI)
	}
	if dataRes.MtwmURL == "" {
		vendorIDs = append(vendorIDs, model.VendorIDMTWM)
	}
	if len(vendorIDs) > 0 {
		imgName := jxutils.GetShortNameFromURL(dataRes.MainURL)
		task := tasksch.NewSeqTask(fmt.Sprintf("上传图片至平台1:%s,%s", dataRes.Name, dataRes.MainURL), ctx,
			func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
				switch step {
				case 0:
					if imgData == nil {
						if imgData, _, err = jxutils.DownloadFileByURL(dataRes.MainURL); err != nil {
							return nil, err
						}
					}
				case 1:
					uploadTask := tasksch.NewParallelTask(fmt.Sprintf("上传图片至平台2:%s,%s", dataRes.Name, imgName),
						tasksch.NewParallelConfig().SetIsContinueWhenError(true), ctx,
						func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
							vendorID := batchItemList[0].(int)
							if handler := partner.GetPurchasePlatformFromVendorID(vendorID); handler != nil {
								imgHint, err2 := handler.UploadImg(ctx, dataRes.MainURL, imgData, imgName)
								if err = err2; err == nil {
									retVal = [][]interface{}{
										[]interface{}{
											vendorID,
											imgHint,
										},
									}
								}
							}
							return retVal, err
						}, vendorIDs)
					tasksch.HandleTask(uploadTask, task, false).Run()
					resultList, err2 := uploadTask.GetResult(0)
					err = err2
					if len(resultList) > 0 {
						db := dao.GetDB()
						for _, v := range resultList {
							result := v.([]interface{})
							vendorID := result[0].(int)
							imgHint := result[1].(string)
							updateField := ""
							if vendorID == model.VendorIDEBAI {
								dataRes.EbaiURL = imgHint
								updateField = "EbaiURL"
							} else if vendorID == model.VendorIDMTWM {
								dataRes.MtwmURL = imgHint
								updateField = "MtwmURL"
							}
							dao.UpdateEntity(db, dataRes, updateField)
						}
					}
				}
				return result, err
			}, 2)
		tasksch.HandleTask(task, parentTask, false).Run()
		if !isAsync {
			if _, err = task.GetResult(0); err == nil {
				hint = "1"
			}
		} else {
			hint = task.GetID()
		}
	}
	return hint, err
}

// func UploadImg2Platforms(ctx *jxcontext.Context, parentTask tasksch.ITask, imgURL string, imgData []byte, imgName string) (imgHintMap map[int]string, err error) {
// 	task := tasksch.NewParallelTask("UploadImg2Platforms", nil, ctx,
// 		func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
// 			vendorID := batchItemList[0].(int)
// 			if handler := partner.GetPurchasePlatformFromVendorID(vendorID); handler != nil {
// 				imgHint, err2 := handler.UploadImg(ctx, imgURL, imgData, imgName)
// 				if err = err2; err == nil {
// 					return [][]interface{}{
// 						[]interface{}{
// 							vendorID,
// 							imgHint,
// 						},
// 					}, nil
// 				}
// 			}
// 			return nil, err
// 		}, []int{model.VendorIDEBAI})
// 	tasksch.HandleTask(task, parentTask, false).Run()
// 	resultList, err := task.GetResult(0)
// 	if err == nil {
// 		imgHintMap = make(map[int]string)
// 		for _, v := range resultList {
// 			vList := v.([]interface{})
// 			imgHintMap[vList[0].(int)] = vList[1].(string)
// 		}
// 	}
// 	return imgHintMap, err
// }