19904408334
313bb692c7
更新 TenantSurplusResp 结构以匹配实际的嵌套 JSON 响应格式， 并修正 worker 中 billing 逻辑错误引用的租户ID变量。
334 lines
10 KiB
Go
334 lines
10 KiB
Go
package gateway
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"mime/multipart"
|
|
"model-gateway/model/entity"
|
|
"time"
|
|
|
|
commonHttp "gitea.redpowerfuture.com/red-future/common/http"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/util/guid"
|
|
)
|
|
|
|
type UploadFileResponse struct {
|
|
FileURL string `json:"fileURL"` // 文件 URL
|
|
FileSize int `json:"fileSize"` // 文件大小（字节）
|
|
FileName string `json:"fileName"` // 文件名
|
|
FileFormat string `json:"fileFormat"` // 文件格式
|
|
FileAddressPrefix string `json:"fileAddressPrefix"` // 文件地址前缀
|
|
}
|
|
|
|
// UploadByTask 通过任务上传文件
|
|
func UploadByTask(ctx context.Context, data []byte, fileExt string) (oss *UploadFileResponse, err error) {
|
|
// multipart
|
|
body := &bytes.Buffer{}
|
|
writer := multipart.NewWriter(body)
|
|
|
|
ext := fileExt
|
|
if ext == "" {
|
|
ext = ".bin"
|
|
}
|
|
if ext[0] != '.' {
|
|
ext = "." + ext
|
|
}
|
|
|
|
filename := fmt.Sprintf("asynch_%d_%s%s", time.Now().Unix(), guid.S(), ext)
|
|
part, err := writer.CreateFormFile("file", filename)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err = part.Write(data); err != nil {
|
|
return nil, err
|
|
}
|
|
//contentType := writer.FormDataContentType()
|
|
if err = writer.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
//headers := util.ForwardHeaders(ctx)
|
|
//headers["Content-Type"] = contentType
|
|
|
|
headers := make(map[string]string)
|
|
headers["Content-Type"] = writer.FormDataContentType()
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
if auth := r.Header.Get("Authorization"); auth != "" {
|
|
headers["Authorization"] = auth
|
|
}
|
|
}
|
|
|
|
fullURL := "oss/file/uploadFile"
|
|
g.Log().Infof(ctx, "[OSS] upload start url=%s filename=%s size=%d", fullURL, filename, len(data))
|
|
|
|
var resp UploadFileResponse
|
|
if err = commonHttp.Post(ctx, fullURL, headers, &resp, body.Bytes()); err != nil {
|
|
return nil, err
|
|
}
|
|
if &resp == nil {
|
|
return nil, errors.New("[OSS] 上传文件失败")
|
|
}
|
|
g.Log().Infof(ctx, "[OSS] 上传成功 url=%s size=%d format=%s", resp.FileURL, resp.FileSize, resp.FileFormat)
|
|
return &resp, nil
|
|
}
|
|
|
|
// CallbackPayload 回调请求体
|
|
type CallbackPayload struct {
|
|
TaskId string `json:"task_id"`
|
|
State int `json:"state"`
|
|
OssFile string `json:"oss_file"`
|
|
FileType string `json:"file_type"`
|
|
ErrorMsg string `json:"error_msg"`
|
|
BillingDate []map[string]any `json:"billing_data"`
|
|
}
|
|
|
|
// TriggerCallback 任务的回调
|
|
func TriggerCallback(ctx context.Context, t *entity.ModelGatewayTask) {
|
|
//headers := util.ForwardHeaders(ctx)
|
|
headers := make(map[string]string)
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Request.Header {
|
|
if len(v) > 0 {
|
|
headers[k] = v[0]
|
|
}
|
|
}
|
|
}
|
|
var resp struct{}
|
|
payload := CallbackPayload{
|
|
TaskId: t.TaskID,
|
|
State: t.State,
|
|
ErrorMsg: t.ErrorMsg,
|
|
BillingDate: t.BillingData,
|
|
}
|
|
if !g.IsEmpty(t.ResultFile) {
|
|
payload.OssFile = t.ResultFile.OssFile
|
|
payload.FileType = t.ResultFile.FileType
|
|
}
|
|
jsonData, err := json.Marshal(payload)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[回调] JSON序列化失败 taskId=%s 错误=%v", t.TaskID, err)
|
|
return
|
|
}
|
|
g.Log().Infof(ctx, "[回调] 开始发送 taskId=%s 回调地址=%s 请求头数量=%d 消息体大小=%d字节",
|
|
t.TaskID, t.CallbackURL, len(headers), len(jsonData))
|
|
|
|
err = commonHttp.Post(ctx, t.CallbackURL, headers, &resp, jsonData)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[回调] 发送失败 taskId=%s 回调地址=%s 错误=%v", t.TaskID, t.CallbackURL, err)
|
|
return
|
|
}
|
|
g.Log().Infof(ctx, "[回调] 发送成功 taskId=%s 回调地址=%s 消息体大小=%d字节", t.TaskID, t.CallbackURL, len(jsonData))
|
|
}
|
|
|
|
// PromptsCallbackPayload 提示词回调请求体
|
|
type PromptsCallbackPayload struct {
|
|
EpicycleId int64 `json:"epicycleId"`
|
|
Messages map[string]any `json:"messages"`
|
|
}
|
|
|
|
// TriggerPromptsCallback 任务成功后的提示词回调
|
|
func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epicycleId int64) {
|
|
callbackURL := "prompts-core/session/callback"
|
|
//headers := util.ForwardHeaders(ctx)
|
|
headers := make(map[string]string)
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Request.Header {
|
|
if len(v) > 0 {
|
|
headers[k] = v[0]
|
|
}
|
|
}
|
|
}
|
|
var resp struct{}
|
|
payload := PromptsCallbackPayload{
|
|
EpicycleId: epicycleId,
|
|
Messages: t.TextResult,
|
|
}
|
|
jsonData, err := json.Marshal(payload)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[提示词回调] JSON序列化失败 epicycleId=%d 错误=%v", epicycleId, err)
|
|
return
|
|
}
|
|
g.Log().Infof(ctx, "[提示词回调] 开始发送 epicycleId=%d 回调地址=%s 请求头数量=%d 消息体大小=%d字节",
|
|
t.EpicycleId, callbackURL, len(headers), len(jsonData))
|
|
|
|
err = commonHttp.Post(ctx, callbackURL, headers, &resp, jsonData)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[提示词回调] 发送失败 epicycleId=%d 回调地址=%s 错误=%v", t.EpicycleId, callbackURL, err)
|
|
return
|
|
}
|
|
g.Log().Infof(ctx, "[提示词回调] 发送成功 epicycleId=%d 回调地址=%s 消息体大小=%d字节", t.EpicycleId, callbackURL, len(jsonData))
|
|
}
|
|
|
|
// IsSuperAdmin 调用admin-go服务检查是否是超级管理员
|
|
func IsSuperAdmin(ctx context.Context) (res bool, err error) {
|
|
//headers := util.ForwardHeaders(ctx)
|
|
headers := make(map[string]string)
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Request.Header {
|
|
if len(v) > 0 {
|
|
headers[k] = v[0]
|
|
}
|
|
}
|
|
}
|
|
var r = make(map[string]bool)
|
|
if err = commonHttp.Get(ctx, "admin-go/api/v1/system/user/checkIsSuperAdmin", headers, &r); err != nil {
|
|
return false, err
|
|
}
|
|
return r["isSuperAdmin"], err
|
|
}
|
|
|
|
// VideoDurationResp 视频时长接口返回
|
|
type VideoDurationResp struct {
|
|
Videos []VideoInfo `json:"videos"`
|
|
Count int `json:"count"`
|
|
TotalDuration float64 `json:"totalDuration"`
|
|
TotalDurationStr string `json:"totalDurationStr"`
|
|
}
|
|
|
|
type VideoInfo struct {
|
|
Index int `json:"index"`
|
|
VideoUrl string `json:"videoUrl"`
|
|
Duration float64 `json:"duration"`
|
|
DurationStr string `json:"durationStr"`
|
|
}
|
|
|
|
// GetVideoDuration 获取视频时长
|
|
func GetVideoDuration(ctx context.Context, urls []string) (VideoDurationResp, error) {
|
|
apiURL := "media/video/duration"
|
|
headers := make(map[string]string)
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Request.Header {
|
|
if len(v) > 0 {
|
|
headers[k] = v[0]
|
|
}
|
|
}
|
|
}
|
|
|
|
body := map[string]any{"video_urls": urls}
|
|
jsonData, _ := json.Marshal(body)
|
|
|
|
var resp VideoDurationResp
|
|
err := commonHttp.Post(ctx, apiURL, headers, &resp, jsonData)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[视频时长] 获取失败 err=%v", err)
|
|
return resp, err
|
|
}
|
|
|
|
g.Log().Infof(ctx, "[视频时长] 获取成功 count=%d totalDuration=%.2f", resp.Count, resp.TotalDuration)
|
|
return resp, nil
|
|
}
|
|
|
|
// DeductBalanceReq 扣减余额请求
|
|
type DeductBalanceReq struct {
|
|
Id uint64 `json:"id"`
|
|
Surplus float64 `json:"surplus"`
|
|
}
|
|
|
|
// DeductBalance 扣减租户余额
|
|
func DeductBalance(ctx context.Context, tenantId uint64, amount float64) error {
|
|
apiURL := "admin-go/api/v1/system/tenant/edit"
|
|
headers := make(map[string]string)
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Request.Header {
|
|
if len(v) > 0 {
|
|
headers[k] = v[0]
|
|
}
|
|
}
|
|
}
|
|
|
|
body := DeductBalanceReq{
|
|
Id: tenantId,
|
|
Surplus: amount,
|
|
}
|
|
jsonData, _ := json.Marshal(body)
|
|
|
|
var resp struct{}
|
|
err := commonHttp.Put(ctx, apiURL, headers, &resp, jsonData)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[扣减余额] 失败 tenantId=%d amount=%.6f err=%v", tenantId, amount, err)
|
|
return err
|
|
}
|
|
g.Log().Infof(ctx, "[扣减余额] 成功 tenantId=%d amount=%.6f", tenantId, amount)
|
|
return nil
|
|
}
|
|
|
|
// TenantSurplusResp 租户余额返回
|
|
type TenantSurplusResp struct {
|
|
Tenant struct {
|
|
Surplus float64 `json:"surplus"`
|
|
} `json:"tenant"`
|
|
}
|
|
|
|
// GetTenantSurplus 获取租户余额
|
|
func GetTenantSurplus(ctx context.Context, tenantId uint64) (float64, error) {
|
|
apiURL := fmt.Sprintf("admin-go/api/v1/system/tenant/getTenantDetails?tenantId=%d", tenantId)
|
|
headers := make(map[string]string)
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Request.Header {
|
|
if len(v) > 0 {
|
|
headers[k] = v[0]
|
|
}
|
|
}
|
|
}
|
|
|
|
var resp TenantSurplusResp
|
|
err := commonHttp.Get(ctx, apiURL, headers, &resp, nil)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "[获取余额] 失败 tenantId=%d err=%v", tenantId, err)
|
|
return 0, err
|
|
}
|
|
return resp.Tenant.Surplus, nil
|
|
}
|
|
|
|
//// callback 向回调地址 POST 任务结果（与查询接口 GetTaskRes 出参一致）
|
|
//func (s *audioTaskService) callback(ctx context.Context, taskID, status, errMsg, callbackURL string) {
|
|
// if callbackURL == "" {
|
|
// return
|
|
// }
|
|
//
|
|
// task, _ := dao.TranscribeTask.GetByTaskID(ctx, taskID)
|
|
// if task == nil {
|
|
// g.Log().Errorf(ctx, "[回调 %s] 任务不存在", taskID)
|
|
// return
|
|
// }
|
|
//
|
|
// detailList, _ := dao.TranscribeTaskDetail.ListByTaskID(ctx, taskID)
|
|
// detailItems := make([]dto.TranscribeTaskDetailItem, 0, len(detailList))
|
|
// for i := range detailList {
|
|
// detailItems = append(detailItems, dao.DetailEntityToItem(&detailList[i]))
|
|
// }
|
|
//
|
|
// // 构建与查询接口一致的 taskInfo
|
|
// taskInfo := dao.EntityToItem(task)
|
|
//
|
|
// // 兼容历史数据: 从 result 中补全 scenes 等字段
|
|
// detailItems = enrichDetailsFromResult(task.Result, detailItems)
|
|
//
|
|
// payload := dto.CallbackPayload{
|
|
// TaskInfo: taskInfo,
|
|
// DetailList: detailItems,
|
|
// }
|
|
//
|
|
// body, _ := json.Marshal(payload)
|
|
//
|
|
// // 透传调用方的用户信息
|
|
// userJSON, _ := json.Marshal(beans.User{UserName: "admin", TenantId: 1})
|
|
//
|
|
// req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
|
|
// req.Header.Set("Content-Type", "application/json")
|
|
// req.Header.Set("X-User-Info", string(userJSON))
|
|
//
|
|
// resp, reqErr := http.DefaultClient.Do(req)
|
|
// if reqErr != nil {
|
|
// g.Log().Errorf(ctx, "[回调 %s] 请求失败: %v", taskID, reqErr)
|
|
// return
|
|
// }
|
|
// defer resp.Body.Close()
|
|
//
|
|
// respBody, _ := io.ReadAll(resp.Body)
|
|
// g.Log().Infof(ctx, "[回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
|
|
//}
|