mirror of
https://github.com/yangjian102621/geekai.git
synced 2026-09-17 10:47:15 +00:00
feat: 迁移 geekai-plus v4.3.1 到开源版
- 同步 Plus v4.3.1 功能源并移除商业 License 闭环 - 更新开源镜像命名、Docker 部署版本和 geekai 数据库配置 - 补充前端 ESLint 检查配置并修复存量解析与模板问题 - 保留 JWT、管理员权限、API Key 和 OAuth 等正常鉴权机制
This commit is contained in:
@@ -58,7 +58,7 @@ func (s *Service) PushTask(task types.ImageTask) {
|
||||
func (s *Service) Run() {
|
||||
// 将数据库中未提交的任务加载到队列
|
||||
var jobs []model.ImageJob
|
||||
s.db.Where("progress", 0).Find(&jobs)
|
||||
s.db.Where("task_id", "").Where("status IN ?", []string{model.ImageStatusPending, model.ImageStatusInProgress}).Find(&jobs)
|
||||
for _, v := range jobs {
|
||||
var task types.ImageTask
|
||||
err := utils.JsonDecode(v.Params, &task)
|
||||
@@ -81,11 +81,17 @@ func (s *Service) Run() {
|
||||
}
|
||||
logger.Infof("handle a new Image generation task: %+v", task)
|
||||
go func() {
|
||||
s.db.Model(&model.ImageJob{Id: task.Id}).UpdateColumns(map[string]any{
|
||||
"status": model.ImageStatusInProgress,
|
||||
"progress": 1,
|
||||
"err_msg": "",
|
||||
})
|
||||
_, err = s.Image(task, false)
|
||||
if err != nil {
|
||||
logger.Errorf("error with image task: %v", err)
|
||||
s.db.Model(&model.ImageJob{Id: task.Id}).UpdateColumns(map[string]interface{}{
|
||||
"progress": service.FailTaskProgress,
|
||||
"status": model.ImageStatusFailed,
|
||||
"progress": 0,
|
||||
"err_msg": err.Error(),
|
||||
})
|
||||
}
|
||||
@@ -189,8 +195,10 @@ func (s *Service) Image(task types.ImageTask, sync bool) (string, error) {
|
||||
s.db.Model(&apiKey).UpdateColumn("last_used_at", time.Now().Unix())
|
||||
var imgURL string
|
||||
var data = map[string]any{
|
||||
"status": model.ImageStatusDownloading,
|
||||
"progress": 100,
|
||||
"prompt": task.Prompt,
|
||||
"task_id": utils.RandString(12),
|
||||
}
|
||||
// 如果返回的是base64,则需要上传到oss
|
||||
if res.Data[0].B64Json != "" {
|
||||
@@ -224,18 +232,19 @@ func (s *Service) CheckTaskStatus() {
|
||||
for {
|
||||
// 检查未完成任务进度
|
||||
var jobs []model.ImageJob
|
||||
s.db.Where("progress < ?", 100).Find(&jobs)
|
||||
s.db.Where("status IN ?", []string{model.ImageStatusPending, model.ImageStatusInProgress, model.ImageStatusDownloading}).Find(&jobs)
|
||||
for _, job := range jobs {
|
||||
// 超时的任务标记为失败
|
||||
if time.Since(job.CreatedAt) > time.Minute*10 {
|
||||
job.Progress = service.FailTaskProgress
|
||||
job.Status = model.ImageStatusFailed
|
||||
job.Progress = 0
|
||||
job.ErrMsg = "任务超时"
|
||||
s.db.Updates(&job)
|
||||
}
|
||||
}
|
||||
|
||||
// 找出失败的任务,并恢复其扣减算力
|
||||
s.db.Where("progress", service.FailTaskProgress).Where("power > ?", 0).Find(&jobs)
|
||||
// 找出失败的任务,并恢复其扣减积分
|
||||
s.db.Where("status", model.ImageStatusFailed).Where("power > ?", 0).Find(&jobs)
|
||||
for _, job := range jobs {
|
||||
var task types.ImageTask
|
||||
err := utils.JsonDecode(job.Params, &task)
|
||||
@@ -245,7 +254,7 @@ func (s *Service) CheckTaskStatus() {
|
||||
err = s.userService.IncreasePower(job.UserId, job.Power, model.PowerLog{
|
||||
Type: types.PowerRefund,
|
||||
Model: task.ModelName,
|
||||
Remark: fmt.Sprintf("任务失败,退回算力。任务ID:%d,Err: %s", job.Id, job.ErrMsg),
|
||||
Remark: fmt.Sprintf("任务失败,退回积分。任务ID:%d,Err: %s", job.Id, job.ErrMsg),
|
||||
})
|
||||
if err != nil {
|
||||
continue
|
||||
@@ -262,7 +271,7 @@ func (s *Service) DownloadImages() {
|
||||
go func() {
|
||||
var items []model.ImageJob
|
||||
for {
|
||||
res := s.db.Where("img_url = ? AND progress = ?", "", 100).Find(&items)
|
||||
res := s.db.Where("img_url = ? AND status = ?", "", model.ImageStatusDownloading).Find(&items)
|
||||
if res.Error != nil {
|
||||
continue
|
||||
}
|
||||
@@ -297,7 +306,10 @@ func (s *Service) downloadImage(jobId uint, orgURL string) (string, error) {
|
||||
}
|
||||
|
||||
// update img_url
|
||||
res := s.db.Model(&model.ImageJob{Id: jobId}).UpdateColumn("img_url", imgURL)
|
||||
res := s.db.Model(&model.ImageJob{Id: jobId}).UpdateColumns(map[string]any{
|
||||
"img_url": imgURL,
|
||||
"status": model.ImageStatusSuccess,
|
||||
})
|
||||
if res.Error != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
package jimeng
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -22,6 +26,13 @@ import (
|
||||
|
||||
var logger = log.GetLogger()
|
||||
|
||||
const seedanceOfficialBaseURL = "https://ark.cn-beijing.volces.com/api/v3"
|
||||
|
||||
const (
|
||||
jimengMediaRepairInterval = 60 * time.Second
|
||||
jimengMediaRepairBatch = 50
|
||||
)
|
||||
|
||||
// Service 即梦服务(合并了消费者功能)
|
||||
type Service struct {
|
||||
db *gorm.DB
|
||||
@@ -61,6 +72,7 @@ func (s *Service) Start() {
|
||||
s.running = true
|
||||
go s.consumeTasks()
|
||||
go s.pollTaskStatus()
|
||||
go s.runJimengSuccessMediaRepairLoop()
|
||||
}
|
||||
|
||||
// Stop 停止服务
|
||||
@@ -145,11 +157,6 @@ func (s *Service) ProcessTask(jobId uint) error {
|
||||
return fmt.Errorf("get jimeng job failed: %w", err)
|
||||
}
|
||||
|
||||
// 更新任务状态为处理中
|
||||
if err := s.UpdateJobStatus(job.Id, types.JMTaskStatusGenerating, ""); err != nil {
|
||||
return fmt.Errorf("update job status failed: %w", err)
|
||||
}
|
||||
|
||||
// 解析任务参数
|
||||
var req types.JimengTaskRequest
|
||||
err := utils.JsonDecode(job.Params, &req)
|
||||
@@ -157,12 +164,6 @@ func (s *Service) ProcessTask(jobId uint) error {
|
||||
return fmt.Errorf("parse task params failed: %w", err)
|
||||
}
|
||||
|
||||
// 构建请求并提交任务
|
||||
params, err := s.buildTaskRequest(&req)
|
||||
if err != nil {
|
||||
return s.handleTaskError(job.Id, fmt.Sprintf("build task request failed: %v", err))
|
||||
}
|
||||
|
||||
// 数字人任务,先识别主体
|
||||
if req.TaskType == types.JMTaskTypeVirtualHuman {
|
||||
if err := s.client.AvatarRecognition(req.ImageUrls[0], req.RecognizeKey); err != nil {
|
||||
@@ -170,8 +171,8 @@ func (s *Service) ProcessTask(jobId uint) error {
|
||||
}
|
||||
}
|
||||
|
||||
// 同步任务 ,后台执行
|
||||
if req.ReqKey == DoubaoSeedream40ReqKey {
|
||||
// Seedream 同步生图(Ark)
|
||||
if IsSeedreamReqKey(req.ReqKey) {
|
||||
go func() {
|
||||
resp, err := s.client.SubmitSyncImageTask(req)
|
||||
if err != nil {
|
||||
@@ -191,18 +192,44 @@ func (s *Service) ProcessTask(jobId uint) error {
|
||||
return
|
||||
}
|
||||
|
||||
if len(resp.Data) == 0 || resp.Data[0] == nil || resp.Data[0].Url == nil || strings.TrimSpace(*resp.Data[0].Url) == "" {
|
||||
_ = s.db.Model(&model.JimengJob{}).Where("id = ?", job.Id).Update("raw_data", string(rawData)).Error
|
||||
_ = s.handleTaskError(job.Id, "seedream response has no image url")
|
||||
return
|
||||
}
|
||||
|
||||
remoteURL := strings.TrimSpace(*resp.Data[0].Url)
|
||||
ext := filepath.Ext(strings.Split(remoteURL, "?")[0])
|
||||
if ext == "" {
|
||||
ext = ".png"
|
||||
}
|
||||
|
||||
// 更新任务状态
|
||||
updates["status"] = types.JMTaskStatusSuccess
|
||||
// 下载图片
|
||||
imgUrl, err := s.uploader.GetUploadHandler().PutUrlFile(*resp.Data[0].Url, ".png", false)
|
||||
if err == nil {
|
||||
updates["img_url"] = imgUrl
|
||||
// 转存到本地/OSS(失败时回退为官方临时 URL,与即梦异步任务一致)
|
||||
imgURL, err := s.uploader.GetUploadHandler().PutUrlFile(remoteURL, ext, false)
|
||||
if err != nil {
|
||||
logger.Errorf("jimeng seedream upload image failed, job_id=%d: %v", job.Id, err)
|
||||
imgURL = remoteURL
|
||||
}
|
||||
updates["img_url"] = imgURL
|
||||
updates["progress"] = 100
|
||||
s.db.Model(&model.JimengJob{}).Where("id = ?", job.Id).Updates(updates)
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Seedance 视频任务(DoubaoAdapter)
|
||||
if IsSeedanceReqKey(req.ReqKey) {
|
||||
return s.submitSeedanceTask(job.Id, &req)
|
||||
}
|
||||
|
||||
// 其他请求走即梦 Visual 异步任务
|
||||
params, err := s.buildTaskRequest(&req)
|
||||
if err != nil {
|
||||
return s.handleTaskError(job.Id, fmt.Sprintf("build task request failed: %v", err))
|
||||
}
|
||||
|
||||
logger.Debugf("提交即梦任务: %+v", params)
|
||||
// 异步任务 ,前台执行
|
||||
resp, err := s.client.SubmitTask(params)
|
||||
@@ -228,6 +255,123 @@ func (s *Service) ProcessTask(jobId uint) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) submitSeedanceTask(jobId uint, req *types.JimengTaskRequest) error {
|
||||
jimengConfig, err := s.getJimengConfig()
|
||||
if err != nil {
|
||||
return s.handleTaskError(jobId, fmt.Sprintf("load jimeng config failed: %v", err))
|
||||
}
|
||||
|
||||
content := s.buildSeedanceContent(req)
|
||||
if len(content) == 0 {
|
||||
return s.handleTaskError(jobId, "seedance content 不能为空")
|
||||
}
|
||||
|
||||
payload := map[string]any{
|
||||
"model": req.ReqKey,
|
||||
"content": content,
|
||||
"duration": req.Duration,
|
||||
"ratio": req.AspectRatio,
|
||||
"resolution": req.Resolution,
|
||||
}
|
||||
// 兼容旧参数:0 让官方走默认值
|
||||
if req.Duration == 0 {
|
||||
delete(payload, "duration")
|
||||
}
|
||||
if req.AspectRatio == "" {
|
||||
delete(payload, "ratio")
|
||||
}
|
||||
if req.Resolution == "" {
|
||||
delete(payload, "resolution")
|
||||
}
|
||||
if req.ReturnLastFrame {
|
||||
payload["return_last_frame"] = req.ReturnLastFrame
|
||||
}
|
||||
if req.Watermark != nil {
|
||||
payload["watermark"] = *req.Watermark
|
||||
}
|
||||
if req.GenerateAudio != nil {
|
||||
payload["generate_audio"] = *req.GenerateAudio
|
||||
}
|
||||
|
||||
resp, rawData, err := s.callSeedanceCreate(payload, jimengConfig)
|
||||
if err != nil {
|
||||
return s.handleTaskError(jobId, fmt.Sprintf("submit seedance task failed: %v", err))
|
||||
}
|
||||
|
||||
logger.Debugf("seedance create response: %+v", resp)
|
||||
|
||||
if err := s.db.Model(&model.JimengJob{}).Where("id = ?", jobId).Updates(map[string]any{
|
||||
"task_id": resp.TaskID,
|
||||
"raw_data": rawData,
|
||||
"status": types.JMTaskStatusInQueue,
|
||||
"updated_at": time.Now(),
|
||||
}).Error; err != nil {
|
||||
logger.Errorf("update seedance task_id failed: %v", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) buildSeedanceContent(req *types.JimengTaskRequest) []types.JMContentItem {
|
||||
if len(req.Content) > 0 {
|
||||
return req.Content
|
||||
}
|
||||
|
||||
content := make([]types.JMContentItem, 0, 4)
|
||||
if req.Prompt != "" {
|
||||
content = append(content, types.JMContentItem{
|
||||
Type: "text",
|
||||
Text: req.Prompt,
|
||||
})
|
||||
}
|
||||
|
||||
if len(req.ImageUrls) > 0 {
|
||||
for index, imageURL := range req.ImageUrls {
|
||||
if imageURL == "" {
|
||||
continue
|
||||
}
|
||||
role := "reference_image"
|
||||
if len(req.ImageUrls) == 1 {
|
||||
role = "first_frame"
|
||||
} else if len(req.ImageUrls) == 2 {
|
||||
if index == 0 {
|
||||
role = "first_frame"
|
||||
} else {
|
||||
role = "last_frame"
|
||||
}
|
||||
}
|
||||
content = append(content, types.JMContentItem{
|
||||
Type: "image_url",
|
||||
ImageURL: &types.JMAssetRef{
|
||||
URL: imageURL,
|
||||
},
|
||||
Role: role,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
if req.VideoURL != "" {
|
||||
content = append(content, types.JMContentItem{
|
||||
Type: "video_url",
|
||||
VideoURL: &types.JMAssetRef{
|
||||
URL: req.VideoURL,
|
||||
},
|
||||
Role: "reference_video",
|
||||
})
|
||||
}
|
||||
|
||||
if req.AudioURL != "" {
|
||||
content = append(content, types.JMContentItem{
|
||||
Type: "audio_url",
|
||||
AudioURL: &types.JMAssetRef{
|
||||
URL: req.AudioURL,
|
||||
},
|
||||
Role: "reference_audio",
|
||||
})
|
||||
}
|
||||
|
||||
return content
|
||||
}
|
||||
|
||||
// buildTaskRequest 构建任务请求(统一的参数解析)
|
||||
func (s *Service) buildTaskRequest(req *types.JimengTaskRequest) (map[string]any, error) {
|
||||
var params map[string]any
|
||||
@@ -295,8 +439,15 @@ func (s *Service) pollTaskStatus() {
|
||||
continue
|
||||
}
|
||||
|
||||
// 豆包生图 4.0 是同步任务,不需要轮询
|
||||
if job.ReqKey == DoubaoSeedream40ReqKey {
|
||||
// Seedream 为同步任务,不需要轮询
|
||||
if IsSeedreamReqKey(job.ReqKey) {
|
||||
continue
|
||||
}
|
||||
|
||||
if IsSeedanceReqKey(job.ReqKey) {
|
||||
if err := s.pollSeedanceTask(&job); err != nil {
|
||||
s.handleTaskError(job.Id, err.Error())
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -333,6 +484,7 @@ func (s *Service) pollTaskStatus() {
|
||||
updates := map[string]any{
|
||||
"status": types.JMTaskStatusSuccess,
|
||||
"updated_at": time.Now(),
|
||||
"progress": 100,
|
||||
}
|
||||
|
||||
// 设置结果URL
|
||||
@@ -370,13 +522,13 @@ func (s *Service) pollTaskStatus() {
|
||||
|
||||
}
|
||||
|
||||
// 找出失败的任务,并恢复其扣减算力
|
||||
// 找出失败的任务,并恢复其扣减积分
|
||||
s.db.Where("status = ?", types.JMTaskStatusFailed).Where("power > ?", 0).Find(&jobs)
|
||||
for _, job := range jobs {
|
||||
err := s.userService.IncreasePower(job.UserId, job.Power, model.PowerLog{
|
||||
Type: types.PowerRefund,
|
||||
Model: job.ReqKey,
|
||||
Remark: fmt.Sprintf("任务失败,退回算力。任务ID:%d", job.Id),
|
||||
Remark: fmt.Sprintf("任务失败,退回积分。任务ID:%d", job.Id),
|
||||
})
|
||||
if err != nil {
|
||||
continue
|
||||
@@ -391,6 +543,159 @@ func (s *Service) pollTaskStatus() {
|
||||
|
||||
}
|
||||
|
||||
func (s *Service) pollSeedanceTask(job *model.JimengJob) error {
|
||||
jimengConfig, err := s.getJimengConfig()
|
||||
if err != nil {
|
||||
return fmt.Errorf("load jimeng config failed: %w", err)
|
||||
}
|
||||
|
||||
resp, rawData, err := s.callSeedanceQuery(job.TaskId, jimengConfig)
|
||||
if err != nil {
|
||||
return fmt.Errorf("query seedance task failed: %w", err)
|
||||
}
|
||||
|
||||
logger.Debugf("seedance query response: %+v", resp)
|
||||
|
||||
s.db.Model(&model.JimengJob{}).Where("id = ?", job.Id).Update("raw_data", rawData)
|
||||
|
||||
switch resp.Status {
|
||||
case "succeeded":
|
||||
updates := map[string]any{
|
||||
"status": types.JMTaskStatusSuccess,
|
||||
"updated_at": time.Now(),
|
||||
"progress": 100,
|
||||
}
|
||||
if resp.Content.VideoURL != "" {
|
||||
videoURL, upErr := s.uploader.GetUploadHandler().PutUrlFile(resp.Content.VideoURL, ".mp4", false)
|
||||
if upErr != nil {
|
||||
logger.Errorf("upload seedance video failed: %v", upErr)
|
||||
videoURL = resp.Content.VideoURL
|
||||
}
|
||||
updates["video_url"] = videoURL
|
||||
}
|
||||
return s.db.Model(&model.JimengJob{}).Where("id = ?", job.Id).Updates(updates).Error
|
||||
case "queued", "running":
|
||||
return s.UpdateJobStatus(job.Id, types.JMTaskStatusGenerating, "")
|
||||
case "failed", "cancelled":
|
||||
errMsg := resp.Error
|
||||
if errMsg == "" {
|
||||
errMsg = "seedance task failed"
|
||||
}
|
||||
return fmt.Errorf("%s", errMsg)
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
type seedanceCreateResponse struct {
|
||||
ID string `json:"id"`
|
||||
PlatformID string `json:"platform_id"`
|
||||
}
|
||||
|
||||
type seedanceCreateResult struct {
|
||||
TaskID string
|
||||
}
|
||||
|
||||
type seedanceQueryResponse struct {
|
||||
ID string `json:"id"`
|
||||
PlatformID string `json:"platform_id"`
|
||||
Status string `json:"status"`
|
||||
Error string `json:"error"`
|
||||
Content struct {
|
||||
VideoURL string `json:"video_url"`
|
||||
} `json:"content"`
|
||||
}
|
||||
|
||||
func (s *Service) callSeedanceCreate(payload map[string]any, jimengConfig *types.JimengConfig) (*seedanceCreateResult, string, error) {
|
||||
if jimengConfig == nil || strings.TrimSpace(jimengConfig.ApiKey) == "" {
|
||||
return nil, "", fmt.Errorf("jimeng api key 未配置")
|
||||
}
|
||||
bodyBytes, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
url := fmt.Sprintf("%s/contents/generations/tasks", seedanceOfficialBaseURL)
|
||||
req, err := http.NewRequest("POST", url, bytes.NewReader(bodyBytes))
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(jimengConfig.ApiKey))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
client := &http.Client{Timeout: 60 * time.Second}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
raw, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return nil, string(raw), fmt.Errorf("status=%d body=%s", resp.StatusCode, string(raw))
|
||||
}
|
||||
var parsed seedanceCreateResponse
|
||||
if err := json.Unmarshal(raw, &parsed); err != nil {
|
||||
return nil, string(raw), err
|
||||
}
|
||||
taskID := parsed.PlatformID
|
||||
if taskID == "" {
|
||||
taskID = parsed.ID
|
||||
}
|
||||
if taskID == "" {
|
||||
return nil, string(raw), fmt.Errorf("seedance create 响应缺少 task id: %s", string(raw))
|
||||
}
|
||||
return &seedanceCreateResult{TaskID: taskID}, string(raw), nil
|
||||
}
|
||||
|
||||
func (s *Service) callSeedanceQuery(taskID string, jimengConfig *types.JimengConfig) (*seedanceQueryResponse, string, error) {
|
||||
if jimengConfig == nil || strings.TrimSpace(jimengConfig.ApiKey) == "" {
|
||||
return nil, "", fmt.Errorf("jimeng api key 未配置")
|
||||
}
|
||||
url := fmt.Sprintf("%s/contents/generations/tasks/%s", seedanceOfficialBaseURL, taskID)
|
||||
req, err := http.NewRequest("GET", url, nil)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(jimengConfig.ApiKey))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
client := &http.Client{Timeout: 60 * time.Second}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
raw, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return nil, string(raw), fmt.Errorf("status=%d body=%s", resp.StatusCode, string(raw))
|
||||
}
|
||||
var parsed seedanceQueryResponse
|
||||
if err := json.Unmarshal(raw, &parsed); err != nil {
|
||||
return nil, string(raw), err
|
||||
}
|
||||
return &parsed, string(raw), nil
|
||||
}
|
||||
|
||||
func (s *Service) getJimengConfig() (*types.JimengConfig, error) {
|
||||
var configRow model.Config
|
||||
if err := s.db.Where("name = ?", types.ConfigKeyJimeng).First(&configRow).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var jimengConfig types.JimengConfig
|
||||
if err := utils.JsonDecode(configRow.Value, &jimengConfig); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if strings.TrimSpace(jimengConfig.ApiKey) == "" {
|
||||
return nil, fmt.Errorf("jimeng api key 未配置")
|
||||
}
|
||||
return &jimengConfig, nil
|
||||
}
|
||||
|
||||
// UpdateJobStatus 更新任务状态
|
||||
func (s *Service) UpdateJobStatus(jobId uint, status types.JMTaskStatus, errMsg string) error {
|
||||
updates := map[string]any{
|
||||
@@ -446,6 +751,142 @@ func (s *Service) GetTaskStats() (map[string]any, error) {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// runJimengSuccessMediaRepairLoop 定时修复:状态已是 success 但进度未满且媒体地址均为空的任务,从 raw_data 重新解析并转存。
|
||||
func (s *Service) runJimengSuccessMediaRepairLoop() {
|
||||
ticker := time.NewTicker(jimengMediaRepairInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-s.ctx.Done():
|
||||
logger.Info("Jimeng success-media repair loop stopped")
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.repairJimengSuccessJobsMediaOnce()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) repairJimengSuccessJobsMediaOnce() {
|
||||
var jobs []model.JimengJob
|
||||
err := s.db.Where("status = ?", types.JMTaskStatusSuccess).
|
||||
Where("progress <> ?", 100).
|
||||
Where("(COALESCE(img_url, '') = ? AND COALESCE(video_url, '') = ?)", "", "").
|
||||
Where("raw_data IS NOT NULL AND raw_data <> ?", "").
|
||||
Order("id ASC").
|
||||
Limit(jimengMediaRepairBatch).
|
||||
Find(&jobs).Error
|
||||
if err != nil {
|
||||
logger.Errorf("jimeng media repair query failed: %v", err)
|
||||
return
|
||||
}
|
||||
if len(jobs) == 0 {
|
||||
return
|
||||
}
|
||||
for i := range jobs {
|
||||
job := jobs[i]
|
||||
remoteImg, remoteVid := parseJimengRawMediaURLs(job.RawData)
|
||||
if strings.TrimSpace(remoteImg) == "" && strings.TrimSpace(remoteVid) == "" {
|
||||
logger.Warnf("jimeng media repair: job_id=%d no media url in raw_data", job.Id)
|
||||
continue
|
||||
}
|
||||
updates := map[string]any{
|
||||
"updated_at": time.Now(),
|
||||
"progress": 100,
|
||||
}
|
||||
if strings.TrimSpace(remoteImg) != "" {
|
||||
updates["img_url"] = s.putJimengRemoteMedia(strings.TrimSpace(remoteImg), ".png", false)
|
||||
}
|
||||
if strings.TrimSpace(remoteVid) != "" {
|
||||
updates["video_url"] = s.putJimengRemoteMedia(strings.TrimSpace(remoteVid), ".mp4", true)
|
||||
}
|
||||
if err := s.db.Model(&model.JimengJob{}).Where("id = ?", job.Id).Updates(updates).Error; err != nil {
|
||||
logger.Errorf("jimeng media repair update failed job_id=%d: %v", job.Id, err)
|
||||
} else {
|
||||
logger.Infof("jimeng media repair ok job_id=%d", job.Id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// parseJimengRawMediaURLs 从 raw_data 解析远程图片/视频地址(即梦异步、Ark Seedream、Seedance)。
|
||||
func parseJimengRawMediaURLs(raw string) (remoteImg, remoteVideo string) {
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" {
|
||||
return "", ""
|
||||
}
|
||||
var root map[string]json.RawMessage
|
||||
if err := json.Unmarshal([]byte(raw), &root); err != nil {
|
||||
return "", ""
|
||||
}
|
||||
|
||||
if dataRaw, ok := root["data"]; ok {
|
||||
dataBytes := []byte(dataRaw)
|
||||
trimmed := bytes.TrimSpace(dataBytes)
|
||||
if len(trimmed) > 0 && trimmed[0] == '[' {
|
||||
var items []struct {
|
||||
Url *string `json:"url"`
|
||||
}
|
||||
if json.Unmarshal(dataBytes, &items) == nil {
|
||||
for _, it := range items {
|
||||
if it.Url != nil && strings.TrimSpace(*it.Url) != "" {
|
||||
return strings.TrimSpace(*it.Url), ""
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
var qd struct {
|
||||
ImageUrls []string `json:"image_urls"`
|
||||
VideoUrl string `json:"video_url"`
|
||||
}
|
||||
if json.Unmarshal(dataBytes, &qd) == nil {
|
||||
img := ""
|
||||
if len(qd.ImageUrls) > 0 {
|
||||
img = strings.TrimSpace(qd.ImageUrls[0])
|
||||
}
|
||||
vid := strings.TrimSpace(qd.VideoUrl)
|
||||
if img != "" || vid != "" {
|
||||
return img, vid
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if contentRaw, ok := root["content"]; ok {
|
||||
var c struct {
|
||||
VideoURL string `json:"video_url"`
|
||||
}
|
||||
if json.Unmarshal(contentRaw, &c) == nil && strings.TrimSpace(c.VideoURL) != "" {
|
||||
return "", strings.TrimSpace(c.VideoURL)
|
||||
}
|
||||
}
|
||||
|
||||
return "", ""
|
||||
}
|
||||
|
||||
func (s *Service) putJimengRemoteMedia(remote, fallbackExt string, isVideo bool) string {
|
||||
remote = strings.TrimSpace(remote)
|
||||
if remote == "" {
|
||||
return ""
|
||||
}
|
||||
ext := fallbackExt
|
||||
if !isVideo {
|
||||
u := remote
|
||||
if i := strings.Index(u, "?"); i >= 0 {
|
||||
u = u[:i]
|
||||
}
|
||||
if e := filepath.Ext(u); e != "" {
|
||||
ext = e
|
||||
}
|
||||
} else {
|
||||
ext = ".mp4"
|
||||
}
|
||||
out, err := s.uploader.GetUploadHandler().PutUrlFile(remote, ext, false)
|
||||
if err != nil {
|
||||
logger.Errorf("jimeng putJimengRemoteMedia failed: %v", err)
|
||||
return remote
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// GetJob 获取任务
|
||||
func (s *Service) GetJob(jobId uint) (*model.JimengJob, error) {
|
||||
var job model.JimengJob
|
||||
|
||||
@@ -124,10 +124,17 @@ type CreateTaskRequest struct {
|
||||
}
|
||||
|
||||
const (
|
||||
ImageEffectReqKey = "i2i_multi_style_zx2x"
|
||||
DoubaoSeedream40ReqKey = "doubao-seedream-4-0-250828"
|
||||
ImageEffectReqKey = "i2i_multi_style_zx2x"
|
||||
)
|
||||
|
||||
func IsSeedreamReqKey(reqKey string) bool {
|
||||
return len(reqKey) >= len("doubao-seedream-") && reqKey[:len("doubao-seedream-")] == "doubao-seedream-"
|
||||
}
|
||||
|
||||
func IsSeedanceReqKey(reqKey string) bool {
|
||||
return len(reqKey) >= len("doubao-seedance-") && reqKey[:len("doubao-seedance-")] == "doubao-seedance-"
|
||||
}
|
||||
|
||||
const (
|
||||
ASyncActionSubmit = "CVSync2AsyncSubmitTask" // 异步提交任务
|
||||
SyncActionSubmit = "CVSubmitTask" // 同步提交任务
|
||||
|
||||
@@ -388,6 +388,14 @@ func (s *MigrationService) incrementalTableMigration() {
|
||||
if s.db.Migrator().HasColumn(&model.User{}, "chat_config") {
|
||||
s.db.Migrator().DropColumn(&model.User{}, "chat_config")
|
||||
}
|
||||
var hasChatRolesJSON int
|
||||
if s.db.Raw("SELECT COUNT(1) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'geekai_users' AND COLUMN_NAME = 'chat_roles_json'").Scan(&hasChatRolesJSON).Error == nil && hasChatRolesJSON > 0 {
|
||||
if err := s.db.Exec("ALTER TABLE geekai_users DROP COLUMN chat_roles_json").Error; err != nil {
|
||||
logger.Errorf("删除 geekai_users.chat_roles_json 失败: %v", err)
|
||||
} else {
|
||||
logger.Info("geekai_users: 已删除 chat_roles_json 列")
|
||||
}
|
||||
}
|
||||
if s.db.Migrator().HasColumn(&model.ChatModel{}, "category") {
|
||||
s.db.Migrator().DropColumn(&model.ChatModel{}, "category")
|
||||
}
|
||||
@@ -406,6 +414,7 @@ func (s *MigrationService) incrementalTableMigration() {
|
||||
if s.db.Migrator().HasColumn(&model.Product{}, "url") {
|
||||
s.db.Migrator().DropColumn(&model.Product{}, "url")
|
||||
}
|
||||
s.migrateProductStripePrice()
|
||||
if s.db.Migrator().HasColumn(&model.VideoJob{}, "water_url") {
|
||||
s.db.Migrator().DropColumn(&model.VideoJob{}, "water_url")
|
||||
}
|
||||
@@ -429,17 +438,7 @@ func (s *MigrationService) incrementalTableMigration() {
|
||||
}
|
||||
|
||||
// ========== 数据迁移:根据业务逻辑更新现有数据 ==========
|
||||
|
||||
// video_job: 根据 progress 填充 status
|
||||
if s.db.Migrator().HasColumn(&model.VideoJob{}, "status") {
|
||||
s.db.Exec(`UPDATE geekai_video_jobs SET status = CASE
|
||||
WHEN progress < 100 THEN 'in_progress'
|
||||
WHEN progress = 100 THEN 'success'
|
||||
WHEN progress = 101 THEN 'failed'
|
||||
WHEN progress = 102 THEN 'downloading'
|
||||
ELSE 'pending'
|
||||
END WHERE status = '' OR status IS NULL`)
|
||||
}
|
||||
s.migrateTaskStatuses()
|
||||
|
||||
// suno_job: 从 output 提取 tags/model_name 填入 params
|
||||
s.migrateSunoJobData()
|
||||
@@ -455,6 +454,96 @@ func (s *MigrationService) TableMigration() {
|
||||
s.migrateChatAppSystemPromptFromJSON()
|
||||
}
|
||||
|
||||
// migrateTaskStatuses 将历史 progress 魔法数与 status 对齐(幂等)。
|
||||
// 顺序:先处理 101/102/100 等离散值,再 1–99 进行中,最后 progress=0 为排队,避免旧逻辑里 progress<100 把 0 标成 in_progress。
|
||||
func (s *MigrationService) migrateTaskStatuses() {
|
||||
migrator := s.db.Migrator()
|
||||
exec := func(label, sql string, args ...interface{}) {
|
||||
if err := s.db.Exec(sql, args...).Error; err != nil {
|
||||
logger.Warnf("回填 %s 失败(可忽略若表/列不存在): %v", label, err)
|
||||
}
|
||||
}
|
||||
|
||||
// image_jobs:与 image_service 约定一致(失败 progress=0;下载中 progress=100)
|
||||
if migrator.HasTable("geekai_image_jobs") && migrator.HasColumn(&model.ImageJob{}, "status") {
|
||||
exec("geekai_image_jobs failed(101)",
|
||||
"UPDATE `geekai_image_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 101",
|
||||
model.ImageStatusFailed)
|
||||
exec("geekai_image_jobs downloading(102)",
|
||||
"UPDATE `geekai_image_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 102",
|
||||
model.ImageStatusDownloading)
|
||||
exec("geekai_image_jobs success(100)",
|
||||
"UPDATE `geekai_image_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 100",
|
||||
model.ImageStatusSuccess)
|
||||
exec("geekai_image_jobs in_progress(1-99)",
|
||||
"UPDATE `geekai_image_jobs` SET `status` = ? WHERE `progress` >= 1 AND `progress` <= 99",
|
||||
model.ImageStatusInProgress)
|
||||
exec("geekai_image_jobs pending(0)",
|
||||
"UPDATE `geekai_image_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 0",
|
||||
model.ImageStatusPending)
|
||||
}
|
||||
|
||||
// mj_jobs:含历史 progress=-1 失败
|
||||
if migrator.HasTable("geekai_mj_jobs") && migrator.HasColumn(&model.MidJourneyJob{}, "status") {
|
||||
exec("geekai_mj_jobs failed(101)",
|
||||
"UPDATE `geekai_mj_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 101",
|
||||
model.ImageStatusFailed)
|
||||
exec("geekai_mj_jobs failed(-1)",
|
||||
"UPDATE `geekai_mj_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = -1",
|
||||
model.ImageStatusFailed)
|
||||
exec("geekai_mj_jobs downloading(102)",
|
||||
"UPDATE `geekai_mj_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 102",
|
||||
model.ImageStatusDownloading)
|
||||
exec("geekai_mj_jobs success(100)",
|
||||
"UPDATE `geekai_mj_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 100",
|
||||
model.ImageStatusSuccess)
|
||||
exec("geekai_mj_jobs in_progress(1-99)",
|
||||
"UPDATE `geekai_mj_jobs` SET `status` = ? WHERE `progress` >= 1 AND `progress` <= 99",
|
||||
model.ImageStatusInProgress)
|
||||
exec("geekai_mj_jobs pending(0)",
|
||||
"UPDATE `geekai_mj_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 0",
|
||||
model.ImageStatusPending)
|
||||
}
|
||||
|
||||
// suno_jobs:与五态常量一致,去掉重复的 101 更新
|
||||
if migrator.HasTable("geekai_suno_jobs") && migrator.HasColumn(&model.SunoJob{}, "status") {
|
||||
exec("geekai_suno_jobs failed(101)",
|
||||
"UPDATE `geekai_suno_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 101",
|
||||
model.ImageStatusFailed)
|
||||
exec("geekai_suno_jobs downloading(102)",
|
||||
"UPDATE `geekai_suno_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 102",
|
||||
model.ImageStatusDownloading)
|
||||
exec("geekai_suno_jobs success(100)",
|
||||
"UPDATE `geekai_suno_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 100",
|
||||
model.ImageStatusSuccess)
|
||||
exec("geekai_suno_jobs in_progress(1-99)",
|
||||
"UPDATE `geekai_suno_jobs` SET `status` = ? WHERE `progress` >= 1 AND `progress` <= 99",
|
||||
model.ImageStatusInProgress)
|
||||
exec("geekai_suno_jobs pending(0)",
|
||||
"UPDATE `geekai_suno_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 0",
|
||||
model.ImageStatusPending)
|
||||
}
|
||||
|
||||
// video_jobs:状态字符串与 Image 五态相同
|
||||
if migrator.HasTable("geekai_video_jobs") && migrator.HasColumn(&model.VideoJob{}, "status") {
|
||||
exec("geekai_video_jobs failed(101)",
|
||||
"UPDATE `geekai_video_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 101",
|
||||
types.VideoStatusFailed)
|
||||
exec("geekai_video_jobs downloading(102)",
|
||||
"UPDATE `geekai_video_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 102",
|
||||
types.VideoStatusDownloading)
|
||||
exec("geekai_video_jobs success(100)",
|
||||
"UPDATE `geekai_video_jobs` SET `status` = ?, `progress` = 100 WHERE `progress` = 100",
|
||||
types.VideoStatusSuccess)
|
||||
exec("geekai_video_jobs in_progress(1-99)",
|
||||
"UPDATE `geekai_video_jobs` SET `status` = ? WHERE `progress` >= 1 AND `progress` <= 99",
|
||||
types.VideoStatusInProgress)
|
||||
exec("geekai_video_jobs pending(0)",
|
||||
"UPDATE `geekai_video_jobs` SET `status` = ?, `progress` = 0 WHERE `progress` = 0",
|
||||
types.VideoStatusPending)
|
||||
}
|
||||
}
|
||||
|
||||
// migrateChatAppSystemPromptFromJSON 将智能体 system_prompt 字段中历史 JSON 数组
|
||||
// 解析后取出 role 为 system 的 content,覆盖回 system_prompt(纯文本)
|
||||
func (s *MigrationService) migrateChatAppSystemPromptFromJSON() {
|
||||
@@ -542,6 +631,11 @@ func (s *MigrationService) migratePaymentConfig(config *types.AppConfig) error {
|
||||
Alipay: config.AlipayConfig,
|
||||
Epay: config.GeekPayConfig,
|
||||
WxPay: config.WechatPayConfig,
|
||||
Stripe: types.StripeConfig{
|
||||
Enabled: false,
|
||||
Currency: "usd",
|
||||
Domain: config.ProxyURL,
|
||||
},
|
||||
}
|
||||
if err := s.saveConfig(types.ConfigKeyPayment, paymentConfig); err != nil {
|
||||
return err
|
||||
@@ -703,3 +797,36 @@ func (s *MigrationService) migrateSunoJobData() {
|
||||
logger.Infof("SunoJob 数据合并完成,共更新 %d 条记录", updatedCount)
|
||||
s.redisClient.Set(context.Background(), key, "1", 0)
|
||||
}
|
||||
|
||||
// migrateProductStripePrice 回填 Stripe 专用价格。
|
||||
// 仅对 stripe_price 为 0 的记录执行回填,避免覆盖已手工配置的 Stripe 价格。
|
||||
func (s *MigrationService) migrateProductStripePrice() {
|
||||
key := "migrate:product_stripe_price"
|
||||
if s.redisClient.Get(context.Background(), key).Val() == "1" {
|
||||
logger.Info("Product stripe_price 已迁移,跳过")
|
||||
return
|
||||
}
|
||||
|
||||
logger.Info("开始迁移 Product stripe_price...")
|
||||
|
||||
var products []model.Product
|
||||
if err := s.db.Where("stripe_price = 0 OR stripe_price IS NULL").Find(&products).Error; err != nil {
|
||||
logger.Errorf("查询 Product 数据失败: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
updatedCount := 0
|
||||
for _, product := range products {
|
||||
if product.Price <= 0 {
|
||||
continue
|
||||
}
|
||||
if err := s.db.Model(&model.Product{}).Where("id = ?", product.Id).Update("stripe_price", product.Price).Error; err != nil {
|
||||
logger.Errorf("回填 Product stripe_price 失败 (ID: %d): %v", product.Id, err)
|
||||
continue
|
||||
}
|
||||
updatedCount++
|
||||
}
|
||||
|
||||
logger.Infof("Product stripe_price 迁移完成,共更新 %d 条记录", updatedCount)
|
||||
s.redisClient.Set(context.Background(), key, "1", 0)
|
||||
}
|
||||
|
||||
@@ -45,7 +45,7 @@ func NewService(redisCli *redis.Client, db *gorm.DB, client *Client, manager *os
|
||||
func (s *Service) Run() {
|
||||
// 将数据库中未提交的人物加载到队列
|
||||
var jobs []model.MidJourneyJob
|
||||
s.db.Where("task_id", "").Where("progress", 0).Find(&jobs)
|
||||
s.db.Where("task_id", "").Where("status IN ?", []string{model.ImageStatusPending, model.ImageStatusInProgress}).Find(&jobs)
|
||||
for _, v := range jobs {
|
||||
var task types.MjTask
|
||||
err := utils.JsonDecode(v.TaskInfo, &task)
|
||||
@@ -80,6 +80,11 @@ func (s *Service) Run() {
|
||||
}
|
||||
|
||||
logger.Infof("handle a new MidJourney task: %+v", task)
|
||||
s.db.Model(&model.MidJourneyJob{Id: task.Id}).UpdateColumns(map[string]any{
|
||||
"status": model.ImageStatusInProgress,
|
||||
"progress": 1,
|
||||
"err_msg": "",
|
||||
})
|
||||
var res ImageRes
|
||||
switch task.Type {
|
||||
case types.TaskImage:
|
||||
@@ -111,7 +116,8 @@ func (s *Service) Run() {
|
||||
}
|
||||
|
||||
logger.Error("绘画任务执行失败:", errMsg)
|
||||
job.Progress = service.FailTaskProgress
|
||||
job.Status = model.ImageStatusFailed
|
||||
job.Progress = 0
|
||||
job.ErrMsg = errMsg
|
||||
// update the task progress
|
||||
s.db.Updates(&job)
|
||||
@@ -123,6 +129,7 @@ func (s *Service) Run() {
|
||||
job.TaskId = res.Result
|
||||
job.MessageId = res.Result
|
||||
job.ChannelId = res.Channel
|
||||
job.Status = model.ImageStatusInProgress
|
||||
s.db.Updates(&job)
|
||||
}
|
||||
}()
|
||||
@@ -158,7 +165,7 @@ func (s *Service) DownloadImages() {
|
||||
go func() {
|
||||
var items []model.MidJourneyJob
|
||||
for {
|
||||
res := s.db.Where("img_url = ? AND progress = ?", "", 100).Find(&items)
|
||||
res := s.db.Where("img_url = ? AND status = ?", "", model.ImageStatusDownloading).Find(&items)
|
||||
if res.Error != nil {
|
||||
continue
|
||||
}
|
||||
@@ -185,6 +192,7 @@ func (s *Service) DownloadImages() {
|
||||
}
|
||||
|
||||
v.ImgURL = imgURL
|
||||
v.Status = model.ImageStatusSuccess
|
||||
s.db.Updates(&v)
|
||||
}
|
||||
|
||||
@@ -206,7 +214,7 @@ func (s *Service) SyncTaskProgress() {
|
||||
go func() {
|
||||
var jobs []model.MidJourneyJob
|
||||
for {
|
||||
res := s.db.Where("progress < ?", 100).Where("channel_id <> ?", "").Find(&jobs)
|
||||
res := s.db.Where("status IN ?", []string{model.ImageStatusPending, model.ImageStatusInProgress, model.ImageStatusDownloading}).Where("channel_id <> ?", "").Find(&jobs)
|
||||
if res.Error != nil {
|
||||
continue
|
||||
}
|
||||
@@ -214,7 +222,8 @@ func (s *Service) SyncTaskProgress() {
|
||||
for _, job := range jobs {
|
||||
// 10 分钟还没完成的任务标记为失败
|
||||
if time.Since(job.CreatedAt) > time.Minute*10 {
|
||||
job.Progress = service.FailTaskProgress
|
||||
job.Status = model.ImageStatusFailed
|
||||
job.Progress = 0
|
||||
job.ErrMsg = "任务超时"
|
||||
s.db.Updates(&job)
|
||||
continue
|
||||
@@ -229,7 +238,8 @@ func (s *Service) SyncTaskProgress() {
|
||||
// 任务执行失败了
|
||||
if task.FailReason != "" {
|
||||
s.db.Model(&model.MidJourneyJob{Id: job.Id}).UpdateColumns(map[string]interface{}{
|
||||
"progress": service.FailTaskProgress,
|
||||
"status": model.ImageStatusFailed,
|
||||
"progress": 0,
|
||||
"err_msg": task.FailReason,
|
||||
})
|
||||
logger.Errorf("task failed: %v", task.FailReason)
|
||||
@@ -243,6 +253,11 @@ func (s *Service) SyncTaskProgress() {
|
||||
if task.ImageUrl != "" {
|
||||
job.OrgURL = task.ImageUrl
|
||||
}
|
||||
if job.Progress >= 100 {
|
||||
job.Status = model.ImageStatusDownloading
|
||||
} else {
|
||||
job.Status = model.ImageStatusInProgress
|
||||
}
|
||||
err = s.db.Updates(&job).Error
|
||||
if err != nil {
|
||||
logger.Errorf("error with update database: %v", err)
|
||||
@@ -250,13 +265,13 @@ func (s *Service) SyncTaskProgress() {
|
||||
}
|
||||
}
|
||||
|
||||
// 找出失败的任务,并恢复其扣减算力
|
||||
s.db.Where("progress", service.FailTaskProgress).Where("power > ?", 0).Find(&jobs)
|
||||
// 找出失败的任务,并恢复其扣减积分
|
||||
s.db.Where("status", model.ImageStatusFailed).Where("power > ?", 0).Find(&jobs)
|
||||
for _, job := range jobs {
|
||||
err := s.userService.IncreasePower(job.UserId, job.Power, model.PowerLog{
|
||||
Type: types.PowerRefund,
|
||||
Model: "mid-journey",
|
||||
Remark: fmt.Sprintf("任务失败,退回算力。任务ID:%d,Err: %s", job.Id, job.ErrMsg),
|
||||
Remark: fmt.Sprintf("任务失败,退回积分。任务ID:%d,Err: %s", job.Id, job.ErrMsg),
|
||||
})
|
||||
if err != nil {
|
||||
continue
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
package payment
|
||||
|
||||
// 支付渠道定义
|
||||
const PayChannelAL = "alipay" // 支付宝
|
||||
const PayChannelWX = "wxpay" // 微信支付
|
||||
const PayChannelEpay = "epay" // 易支付
|
||||
const PayChannelAL = "alipay" // 支付宝
|
||||
const PayChannelWX = "wxpay" // 微信支付
|
||||
const PayChannelEpay = "epay" // 易支付
|
||||
const PayChannelStripe = "stripe" // Stripe
|
||||
|
||||
// 支付方式
|
||||
const PayWayAL = "alipay"
|
||||
const PayWayWX = "wxpay"
|
||||
const PayWayStripe = "stripe"
|
||||
|
||||
const (
|
||||
Success = 0
|
||||
@@ -19,6 +21,7 @@ type PayRequest struct {
|
||||
OutTradeNo string // 商户订单号
|
||||
Subject string // 商品名称
|
||||
TotalFee string // 商品金额
|
||||
Currency string // 币种
|
||||
ReturnURL string // 回调地址
|
||||
NotifyURL string // 回调地址
|
||||
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
package payment
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"geekai/core/types"
|
||||
"io"
|
||||
"math"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/stripe/stripe-go/v81"
|
||||
"github.com/stripe/stripe-go/v81/checkout/session"
|
||||
"github.com/stripe/stripe-go/v81/webhook"
|
||||
)
|
||||
|
||||
type StripeService struct {
|
||||
config *types.StripeConfig
|
||||
}
|
||||
|
||||
func NewStripeService(sysConfig *types.SystemConfig) *StripeService {
|
||||
return &StripeService{config: &sysConfig.Payment.Stripe}
|
||||
}
|
||||
|
||||
func (s *StripeService) UpdateConfig(config *types.StripeConfig) {
|
||||
s.config = config
|
||||
}
|
||||
|
||||
func (s *StripeService) Pay(params PayRequest) (string, error) {
|
||||
if s.config == nil || !s.config.Enabled {
|
||||
return "", fmt.Errorf("stripe service disabled")
|
||||
}
|
||||
if s.config.SecretKey == "" {
|
||||
return "", fmt.Errorf("stripe secret key is empty")
|
||||
}
|
||||
|
||||
stripe.Key = s.config.SecretKey
|
||||
currency := strings.ToLower(s.config.Currency)
|
||||
if currency == "" {
|
||||
currency = "usd"
|
||||
}
|
||||
|
||||
amount := parseMoneyToMinorUnit(params.TotalFee, currency)
|
||||
if amount <= 0 {
|
||||
return "", fmt.Errorf("invalid stripe amount: %s", params.TotalFee)
|
||||
}
|
||||
|
||||
successURL := strings.TrimRight(params.ReturnURL, "/")
|
||||
if successURL == "" {
|
||||
successURL = strings.TrimRight(s.config.Domain, "/")
|
||||
}
|
||||
cancelURL := successURL
|
||||
sessionParams := &stripe.CheckoutSessionParams{
|
||||
Mode: stripe.String(string(stripe.CheckoutSessionModePayment)),
|
||||
LineItems: []*stripe.CheckoutSessionLineItemParams{
|
||||
{
|
||||
PriceData: &stripe.CheckoutSessionLineItemPriceDataParams{
|
||||
Currency: stripe.String(currency),
|
||||
ProductData: &stripe.CheckoutSessionLineItemPriceDataProductDataParams{
|
||||
Name: stripe.String(params.Subject),
|
||||
},
|
||||
UnitAmount: stripe.Int64(amount),
|
||||
},
|
||||
Quantity: stripe.Int64(1),
|
||||
},
|
||||
},
|
||||
SuccessURL: stripe.String(successURL + "?session_id={CHECKOUT_SESSION_ID}&order_no=" + params.OutTradeNo),
|
||||
CancelURL: stripe.String(cancelURL),
|
||||
Metadata: map[string]string{
|
||||
"order_no": params.OutTradeNo,
|
||||
},
|
||||
PaymentIntentData: &stripe.CheckoutSessionPaymentIntentDataParams{
|
||||
Metadata: map[string]string{
|
||||
"order_no": params.OutTradeNo,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
sess, err := session.New(sessionParams)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("error with create stripe checkout session: %w", err)
|
||||
}
|
||||
|
||||
return sess.URL, nil
|
||||
}
|
||||
|
||||
func (s *StripeService) Query(outTradeNo string) (OrderInfo, error) {
|
||||
return OrderInfo{}, fmt.Errorf("stripe orders are verified by webhook")
|
||||
}
|
||||
|
||||
func (s *StripeService) TradeVerify(request *http.Request) (OrderInfo, error) {
|
||||
if s.config == nil || !s.config.Enabled {
|
||||
return OrderInfo{}, fmt.Errorf("stripe service disabled")
|
||||
}
|
||||
payload, err := io.ReadAll(request.Body)
|
||||
if err != nil {
|
||||
return OrderInfo{}, fmt.Errorf("error with read stripe webhook body: %w", err)
|
||||
}
|
||||
body, err := webhook.ConstructEvent(payload, request.Header.Get("Stripe-Signature"), s.config.WebhookKey)
|
||||
if err != nil {
|
||||
return OrderInfo{}, fmt.Errorf("error with verify stripe webhook: %w", err)
|
||||
}
|
||||
|
||||
switch body.Type {
|
||||
case "checkout.session.completed":
|
||||
var sess stripe.CheckoutSession
|
||||
if err := json.Unmarshal(body.Data.Raw, &sess); err != nil {
|
||||
return OrderInfo{}, fmt.Errorf("error with parse stripe session: %w", err)
|
||||
}
|
||||
if sess.Metadata["order_no"] == "" {
|
||||
return OrderInfo{}, fmt.Errorf("missing order number in stripe metadata")
|
||||
}
|
||||
payTime := time.Unix(body.Created, 0).Format("2006-01-02 15:04:05")
|
||||
tradeID := ""
|
||||
if sess.PaymentIntent != nil {
|
||||
tradeID = sess.PaymentIntent.ID
|
||||
}
|
||||
return OrderInfo{
|
||||
Status: Success,
|
||||
OutTradeNo: sess.Metadata["order_no"],
|
||||
TradeId: strings.TrimSpace(tradeID),
|
||||
Amount: fmt.Sprintf("%.2f", float64(sess.AmountTotal)/100),
|
||||
PayTime: payTime,
|
||||
}, nil
|
||||
default:
|
||||
return OrderInfo{}, fmt.Errorf("unsupported stripe event: %s", body.Type)
|
||||
}
|
||||
}
|
||||
|
||||
func parseMoneyToMinorUnit(amount string, currency string) int64 {
|
||||
if amount == "" {
|
||||
return 0
|
||||
}
|
||||
if strings.EqualFold(currency, "jpy") {
|
||||
return int64(math.Round(parseFloat(amount)))
|
||||
}
|
||||
return int64(math.Round(parseFloat(amount) * 100))
|
||||
}
|
||||
|
||||
func parseFloat(s string) float64 {
|
||||
var v float64
|
||||
_, _ = fmt.Sscanf(s, "%f", &v)
|
||||
return v
|
||||
}
|
||||
|
||||
var _ PayService = (*StripeService)(nil)
|
||||
@@ -21,17 +21,33 @@ type SmtpService struct {
|
||||
config *types.SmtpConfig
|
||||
}
|
||||
|
||||
func NewSmtpService(appConfig *types.AppConfig) *SmtpService {
|
||||
return &SmtpService{
|
||||
config: &appConfig.SmtpConfig,
|
||||
// NewSmtpService 优先使用数据库中的 SMTP 配置(管理后台保存),否则回退到 config.toml。
|
||||
func NewSmtpService(appConfig *types.AppConfig, sysConfig *types.SystemConfig) *SmtpService {
|
||||
s := &SmtpService{}
|
||||
if sysConfig.SMTP.Host != "" && sysConfig.SMTP.Port > 0 {
|
||||
c := sysConfig.SMTP
|
||||
s.config = &c
|
||||
} else {
|
||||
s.config = &appConfig.SmtpConfig
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *SmtpService) UpdateConfig(config *types.SmtpConfig) {
|
||||
s.config = config
|
||||
}
|
||||
|
||||
func (s *SmtpService) smtpConfigured() error {
|
||||
if s.config == nil || s.config.Host == "" || s.config.Port <= 0 {
|
||||
return fmt.Errorf("SMTP 未配置或无效:请在管理后台填写邮件服务器地址和端口")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SmtpService) SendVerifyCode(to string, code int) error {
|
||||
if err := s.smtpConfigured(); err != nil {
|
||||
return err
|
||||
}
|
||||
subject := fmt.Sprintf("%s 注册验证码", s.config.AppName)
|
||||
body := fmt.Sprintf("【%s】:您的验证码为 %d,请不要告诉他人。如非本人操作,请忽略此邮件。", s.config.AppName, code)
|
||||
|
||||
|
||||
@@ -60,7 +60,7 @@ func (s *Service) PushTask(task types.SunoTask) {
|
||||
func (s *Service) Run() {
|
||||
// 将数据库中未提交的人物加载到队列
|
||||
var jobs []model.SunoJob
|
||||
s.db.Where("task_id", "").Where("progress", 0).Find(&jobs)
|
||||
s.db.Where("task_id", "").Where("status IN ?", []string{model.ImageStatusPending, model.ImageStatusInProgress}).Find(&jobs)
|
||||
for _, v := range jobs {
|
||||
// 从 Params 中提取字段构建 task
|
||||
task := types.SunoTask{
|
||||
@@ -91,6 +91,11 @@ func (s *Service) Run() {
|
||||
logger.Errorf("taking task with error: %v", err)
|
||||
continue
|
||||
}
|
||||
s.db.Model(&model.SunoJob{Id: task.Id}).UpdateColumns(map[string]any{
|
||||
"status": model.ImageStatusInProgress,
|
||||
"progress": 1,
|
||||
"err_msg": "",
|
||||
})
|
||||
var r RespVo
|
||||
if task.Type == 3 && task.SongId != "" { // 歌曲拼接
|
||||
r, err = s.Merge(task)
|
||||
@@ -103,7 +108,8 @@ func (s *Service) Run() {
|
||||
logger.Errorf("create task with error: %v", err)
|
||||
s.db.Model(&model.SunoJob{Id: task.Id}).UpdateColumns(map[string]interface{}{
|
||||
"err_msg": err.Error(),
|
||||
"progress": service.FailTaskProgress,
|
||||
"status": model.ImageStatusFailed,
|
||||
"progress": 101,
|
||||
})
|
||||
continue
|
||||
}
|
||||
@@ -112,8 +118,10 @@ func (s *Service) Run() {
|
||||
|
||||
// 更新任务信息
|
||||
s.db.Model(&model.SunoJob{Id: task.Id}).UpdateColumns(map[string]interface{}{
|
||||
"task_id": r.Data,
|
||||
"channel": r.Channel,
|
||||
"task_id": r.Data,
|
||||
"channel": r.Channel,
|
||||
"status": model.ImageStatusInProgress,
|
||||
"progress": 1,
|
||||
})
|
||||
}
|
||||
}()
|
||||
@@ -281,7 +289,7 @@ func (s *Service) DownloadFiles() {
|
||||
go func() {
|
||||
var items []model.SunoJob
|
||||
for {
|
||||
res := s.db.Where("progress", 102).Find(&items)
|
||||
res := s.db.Where("status", model.ImageStatusDownloading).Find(&items)
|
||||
if res.Error != nil {
|
||||
continue
|
||||
}
|
||||
@@ -304,6 +312,7 @@ func (s *Service) DownloadFiles() {
|
||||
v.CoverURL = coverURL
|
||||
v.AudioURL = audioURL
|
||||
v.Progress = 100
|
||||
v.Status = model.ImageStatusSuccess
|
||||
s.db.Updates(&v)
|
||||
}
|
||||
|
||||
@@ -317,7 +326,7 @@ func (s *Service) SyncTaskProgress() {
|
||||
go func() {
|
||||
var jobs []model.SunoJob
|
||||
for {
|
||||
res := s.db.Where("progress < ?", 100).Where("task_id <> ?", "").Find(&jobs)
|
||||
res := s.db.Where("status IN ?", []string{model.ImageStatusPending, model.ImageStatusInProgress, model.ImageStatusDownloading}).Where("task_id <> ?", "").Find(&jobs)
|
||||
if res.Error != nil {
|
||||
continue
|
||||
}
|
||||
@@ -342,7 +351,8 @@ func (s *Service) SyncTaskProgress() {
|
||||
tx := s.db.Begin()
|
||||
for _, v := range task.Data.Data {
|
||||
job.Id = 0
|
||||
job.Progress = 102 // 102 表示资源未下载完成
|
||||
job.Progress = 100
|
||||
job.Status = model.ImageStatusDownloading // 资源待下载
|
||||
job.Title = v.Title
|
||||
job.SongId = v.Id
|
||||
job.Duration = int(v.Metadata.Duration)
|
||||
@@ -384,20 +394,21 @@ func (s *Service) SyncTaskProgress() {
|
||||
}
|
||||
tx.Commit()
|
||||
} else if task.Data.FailReason != "" {
|
||||
job.Progress = service.FailTaskProgress
|
||||
job.Status = model.ImageStatusFailed
|
||||
job.Progress = 101
|
||||
job.ErrMsg = task.Data.FailReason
|
||||
s.db.Updates(&job)
|
||||
}
|
||||
}
|
||||
|
||||
// 找出失败的任务,并恢复其扣减算力
|
||||
// 找出失败的任务,并恢复其扣减积分
|
||||
s.db.Select("id", "user_id", "power", "task_id", "err_msg", "params").
|
||||
Where("progress", service.FailTaskProgress).Where("power > ?", 0).Find(&jobs)
|
||||
Where("status", model.ImageStatusFailed).Where("power > ?", 0).Find(&jobs)
|
||||
for _, job := range jobs {
|
||||
err := s.userService.IncreasePower(job.UserId, job.Power, model.PowerLog{
|
||||
Type: types.PowerRefund,
|
||||
Model: job.Params.Model,
|
||||
Remark: fmt.Sprintf("Suno 任务失败,退回算力。任务ID:%s,Err:%s", job.TaskId, job.ErrMsg),
|
||||
Remark: fmt.Sprintf("Suno 任务失败,退回积分。任务ID:%s,Err:%s", job.TaskId, job.ErrMsg),
|
||||
})
|
||||
if err != nil {
|
||||
continue
|
||||
@@ -2,7 +2,6 @@ package service
|
||||
|
||||
import "geekai/log"
|
||||
|
||||
const FailTaskProgress = 101
|
||||
const (
|
||||
TaskStatusRunning = "RUNNING"
|
||||
TaskStatusFinished = "FINISH"
|
||||
|
||||
@@ -13,13 +13,22 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"geekai/core/types"
|
||||
"geekai/utils"
|
||||
"image"
|
||||
"image/png"
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"geekai/utils"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
_ "image/gif"
|
||||
_ "image/jpeg"
|
||||
|
||||
"github.com/imroc/req/v3"
|
||||
"github.com/nfnt/resize"
|
||||
_ "golang.org/x/image/webp"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@@ -141,10 +150,10 @@ func (a *SoraAdapter) CreateTask(task types.VideoTask, videoConfig *types.VideoC
|
||||
|
||||
// 其他场景:保持原来的 JSON 调用,input_reference 继续传 URL 字符串
|
||||
reqBody := SoraCreateRequest{
|
||||
Model: model,
|
||||
Prompt: task.Prompt,
|
||||
Size: size,
|
||||
Seconds: seconds,
|
||||
Model: model,
|
||||
Prompt: task.Prompt,
|
||||
Size: size,
|
||||
Seconds: seconds,
|
||||
Watermark: watermark,
|
||||
}
|
||||
|
||||
@@ -201,6 +210,12 @@ func (a *SoraAdapter) createOfficialSoraTask(task types.VideoTask, videoConfig *
|
||||
if err != nil {
|
||||
return CreateTaskResponse{}, fmt.Errorf("下载参考图片失败:%v", err)
|
||||
}
|
||||
if size != "" {
|
||||
imgData, err = normalizeImageBytesForSize(imgData, size)
|
||||
if err != nil {
|
||||
return CreateTaskResponse{}, fmt.Errorf("处理参考图片尺寸失败:%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
writer := multipart.NewWriter(&buf)
|
||||
@@ -224,7 +239,7 @@ func (a *SoraAdapter) createOfficialSoraTask(task types.VideoTask, videoConfig *
|
||||
}
|
||||
|
||||
// 文件字段
|
||||
fileWriter, err := writer.CreateFormFile("input_reference", "image")
|
||||
fileWriter, err := writer.CreateFormFile("input_reference", "image.png")
|
||||
if err != nil {
|
||||
return CreateTaskResponse{}, err
|
||||
}
|
||||
@@ -281,6 +296,86 @@ func downloadImageBytes(imageURL string) ([]byte, error) {
|
||||
return body, err
|
||||
}
|
||||
|
||||
// normalizeImageBytesForSize 将参考图等比缩放并裁切为目标尺寸,确保与 Sora size 参数一致
|
||||
func normalizeImageBytesForSize(imageData []byte, size string) ([]byte, error) {
|
||||
targetWidth, targetHeight, err := parseVideoSize(size)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if targetWidth <= 0 || targetHeight <= 0 {
|
||||
return nil, fmt.Errorf("无效的图片尺寸: %s", size)
|
||||
}
|
||||
|
||||
srcImage, _, err := image.Decode(bytes.NewReader(imageData))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("解码参考图片失败:%w", err)
|
||||
}
|
||||
srcBounds := srcImage.Bounds()
|
||||
srcWidth := srcBounds.Dx()
|
||||
srcHeight := srcBounds.Dy()
|
||||
if srcWidth <= 0 || srcHeight <= 0 {
|
||||
return nil, fmt.Errorf("参考图片尺寸无效")
|
||||
}
|
||||
|
||||
if srcWidth == targetWidth && srcHeight == targetHeight {
|
||||
var exactBuf bytes.Buffer
|
||||
if err = png.Encode(&exactBuf, srcImage); err != nil {
|
||||
return nil, fmt.Errorf("编码参考图片失败:%w", err)
|
||||
}
|
||||
return exactBuf.Bytes(), nil
|
||||
}
|
||||
|
||||
scaleByWidth := float64(targetWidth) / float64(srcWidth)
|
||||
scaleByHeight := float64(targetHeight) / float64(srcHeight)
|
||||
scale := scaleByWidth
|
||||
if scaleByHeight > scale {
|
||||
scale = scaleByHeight
|
||||
}
|
||||
|
||||
resizedWidth := int(float64(srcWidth) * scale)
|
||||
resizedHeight := int(float64(srcHeight) * scale)
|
||||
if resizedWidth < targetWidth {
|
||||
resizedWidth = targetWidth
|
||||
}
|
||||
if resizedHeight < targetHeight {
|
||||
resizedHeight = targetHeight
|
||||
}
|
||||
|
||||
resizedImage := resize.Resize(uint(resizedWidth), uint(resizedHeight), srcImage, resize.Lanczos3)
|
||||
offsetX := (resizedWidth - targetWidth) / 2
|
||||
offsetY := (resizedHeight - targetHeight) / 2
|
||||
targetRect := image.Rect(0, 0, targetWidth, targetHeight)
|
||||
croppedImage := image.NewRGBA(targetRect)
|
||||
for y := 0; y < targetHeight; y++ {
|
||||
for x := 0; x < targetWidth; x++ {
|
||||
croppedImage.Set(x, y, resizedImage.At(x+offsetX, y+offsetY))
|
||||
}
|
||||
}
|
||||
|
||||
var out bytes.Buffer
|
||||
if err = png.Encode(&out, croppedImage); err != nil {
|
||||
return nil, fmt.Errorf("编码裁切后的参考图片失败:%w", err)
|
||||
}
|
||||
return out.Bytes(), nil
|
||||
}
|
||||
|
||||
// parseVideoSize 解析 "720x1280" 这样的尺寸字符串
|
||||
func parseVideoSize(size string) (int, int, error) {
|
||||
parts := strings.Split(strings.TrimSpace(size), "x")
|
||||
if len(parts) != 2 {
|
||||
return 0, 0, fmt.Errorf("非法尺寸格式: %s", size)
|
||||
}
|
||||
width, err := strconv.Atoi(strings.TrimSpace(parts[0]))
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("解析宽度失败: %w", err)
|
||||
}
|
||||
height, err := strconv.Atoi(strings.TrimSpace(parts[1]))
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("解析高度失败: %w", err)
|
||||
}
|
||||
return width, height, nil
|
||||
}
|
||||
|
||||
// downloadImageAsDataURL 下载远程图片并转为 data URL,避免向官方 Sora 直接传地址
|
||||
// QueryTask 查询任务状态
|
||||
func (a *SoraAdapter) QueryTask(taskId string, channel string, videoConfig *types.VideoConfig) (QueryTaskResponse, error) {
|
||||
|
||||
@@ -143,7 +143,7 @@ func (s *Service) PushTask(task types.VideoTask) {
|
||||
func (s *Service) Run() {
|
||||
// 将数据库中未提交的任务加载到队列
|
||||
var jobs []model.VideoJob
|
||||
s.db.Where("task_id", "").Where("progress", 0).Find(&jobs)
|
||||
s.db.Where("task_id", "").Where("status", types.VideoStatusPending).Find(&jobs)
|
||||
for _, v := range jobs {
|
||||
var task types.VideoTask
|
||||
err := utils.JsonDecode(v.Params, &task)
|
||||
@@ -234,7 +234,7 @@ func (s *Service) SyncTaskProgress() {
|
||||
logger.Info("[video] task status poller started")
|
||||
var jobs []model.VideoJob
|
||||
for {
|
||||
res := s.db.Where("status IN ?", []string{types.VideoStatusInProgress, types.VideoStatusPending}).Where("task_id <> ?", "").Find(&jobs)
|
||||
res := s.db.Where("status IN ?", []string{types.VideoStatusInProgress, types.VideoStatusPending, types.VideoStatusDownloading}).Find(&jobs)
|
||||
if res.Error != nil {
|
||||
continue
|
||||
}
|
||||
@@ -312,14 +312,14 @@ func (s *Service) SyncTaskProgress() {
|
||||
|
||||
}
|
||||
|
||||
// 找出失败的任务,并恢复其扣减算力
|
||||
// 找出失败的任务,并恢复其扣减积分
|
||||
s.db.Select("id", "user_id", "power", "task_id", "err_msg", "type").
|
||||
Where("status", types.VideoStatusFailed).Where("power > ?", 0).Find(&jobs)
|
||||
for _, job := range jobs {
|
||||
err := s.userService.IncreasePower(job.UserId, job.Power, model.PowerLog{
|
||||
Type: types.PowerRefund,
|
||||
Model: job.Type,
|
||||
Remark: fmt.Sprintf("%s 任务失败,退回算力。任务ID:%s,Err:%s", job.Type, job.TaskId, job.ErrMsg),
|
||||
Remark: fmt.Sprintf("%s 任务失败,退回积分。任务ID:%s,Err:%s", job.Type, job.TaskId, job.ErrMsg),
|
||||
})
|
||||
if err != nil {
|
||||
continue
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"geekai/core/types"
|
||||
"geekai/utils"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const wxGzhTokenSkew = 300 * time.Second
|
||||
|
||||
// GetClientCredentialToken 获取公众号 client_credential access_token(带进程内缓存)。
|
||||
func (s *WxGzhService) GetClientCredentialToken(ctx context.Context, appID, secret string) (string, error) {
|
||||
appID, secret = strings.TrimSpace(appID), strings.TrimSpace(secret)
|
||||
if appID == "" || secret == "" {
|
||||
return "", fmt.Errorf("微信公众号 AppID 或 AppSecret 未配置")
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
s.ccMu.Lock()
|
||||
if s.ccForApp == appID && s.ccToken != "" && now.Before(s.ccExpire) {
|
||||
tok := s.ccToken
|
||||
s.ccMu.Unlock()
|
||||
return tok, nil
|
||||
}
|
||||
s.ccMu.Unlock()
|
||||
|
||||
raw := fmt.Sprintf(
|
||||
"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s",
|
||||
url.QueryEscape(appID), url.QueryEscape(secret),
|
||||
)
|
||||
body, status, err := utils.FetchURLBytes(ctx, raw, "", 30*time.Second, 2, 2<<20)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("获取 access_token 失败: status=%d: %w", status, err)
|
||||
}
|
||||
|
||||
var parsed struct {
|
||||
AccessToken string `json:"access_token"`
|
||||
ExpiresIn int `json:"expires_in"`
|
||||
ErrCode int `json:"errcode"`
|
||||
ErrMsg string `json:"errmsg"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &parsed); err != nil {
|
||||
return "", fmt.Errorf("解析 token 响应失败: %w", err)
|
||||
}
|
||||
if parsed.ErrCode != 0 {
|
||||
return "", fmt.Errorf("微信 API 错误(%d): %s", parsed.ErrCode, parsed.ErrMsg)
|
||||
}
|
||||
if parsed.AccessToken == "" {
|
||||
return "", fmt.Errorf("微信未返回 access_token: %s", string(body))
|
||||
}
|
||||
|
||||
expireAt := now.Add(time.Duration(parsed.ExpiresIn) * time.Second).Add(-wxGzhTokenSkew)
|
||||
if parsed.ExpiresIn <= 0 {
|
||||
expireAt = now.Add(7000 * time.Second)
|
||||
}
|
||||
|
||||
s.ccMu.Lock()
|
||||
s.ccForApp = appID
|
||||
s.ccToken = parsed.AccessToken
|
||||
s.ccExpire = expireAt
|
||||
s.ccMu.Unlock()
|
||||
|
||||
return parsed.AccessToken, nil
|
||||
}
|
||||
|
||||
// PublishCustomMenu 调用 menu/create 全量覆盖公众号菜单。
|
||||
func (s *WxGzhService) PublishCustomMenu(ctx context.Context, appID, secret string, menu types.WxGzhMenuConfig) error {
|
||||
if err := ValidateWxGzhMenu(menu); err != nil {
|
||||
return err
|
||||
}
|
||||
token, err := s.GetClientCredentialToken(ctx, appID, secret)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
payload, err := json.Marshal(menu)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
api := fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/menu/create?access_token=%s", url.QueryEscape(token))
|
||||
body, status, err := utils.PostURLBytes(ctx, api, "application/json; charset=utf-8", payload, "", 45*time.Second, 2, 2<<20)
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建菜单请求失败: status=%d: %w", status, err)
|
||||
}
|
||||
var res struct {
|
||||
ErrCode int `json:"errcode"`
|
||||
ErrMsg string `json:"errmsg"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &res); err != nil {
|
||||
return fmt.Errorf("解析创建菜单响应失败: %w", err)
|
||||
}
|
||||
if res.ErrCode != 0 {
|
||||
return fmt.Errorf("微信创建菜单失败(%d): %s", res.ErrCode, res.ErrMsg)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetCustomMenuFromWechat 调用 menu/get 拉取线上菜单。
|
||||
func (s *WxGzhService) GetCustomMenuFromWechat(ctx context.Context, appID, secret string) (types.WxGzhMenuConfig, error) {
|
||||
var empty types.WxGzhMenuConfig
|
||||
token, err := s.GetClientCredentialToken(ctx, appID, secret)
|
||||
if err != nil {
|
||||
return empty, err
|
||||
}
|
||||
api := fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/menu/get?access_token=%s", url.QueryEscape(token))
|
||||
body, status, err := utils.FetchURLBytes(ctx, api, "", 30*time.Second, 2, 2<<20)
|
||||
if err != nil {
|
||||
return empty, fmt.Errorf("获取菜单失败: status=%d: %w", status, err)
|
||||
}
|
||||
|
||||
var envelope struct {
|
||||
Menu *struct {
|
||||
Button []types.WxGzhMenuButton `json:"button"`
|
||||
} `json:"menu"`
|
||||
ErrCode int `json:"errcode"`
|
||||
ErrMsg string `json:"errmsg"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &envelope); err != nil {
|
||||
return empty, fmt.Errorf("解析菜单响应失败: %w", err)
|
||||
}
|
||||
if envelope.ErrCode != 0 {
|
||||
return empty, fmt.Errorf("微信获取菜单失败(%d): %s", envelope.ErrCode, envelope.ErrMsg)
|
||||
}
|
||||
if envelope.Menu == nil || len(envelope.Menu.Button) == 0 {
|
||||
return types.WxGzhMenuConfig{Button: nil}, nil
|
||||
}
|
||||
return types.WxGzhMenuConfig{Button: envelope.Menu.Button}, nil
|
||||
}
|
||||
|
||||
// ValidateWxGzhMenu 校验自定义菜单是否符合微信公众平台规则(view / click / 子菜单)。
|
||||
func ValidateWxGzhMenu(m types.WxGzhMenuConfig) error {
|
||||
if len(m.Button) == 0 {
|
||||
return fmt.Errorf("请至少配置一个一级菜单")
|
||||
}
|
||||
if len(m.Button) > 3 {
|
||||
return fmt.Errorf("一级菜单最多 3 个")
|
||||
}
|
||||
for _, b := range m.Button {
|
||||
if err := validateTopButton(b); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateTopButton(b types.WxGzhMenuButton) error {
|
||||
if strings.TrimSpace(b.Name) == "" {
|
||||
return fmt.Errorf("菜单名称不能为空")
|
||||
}
|
||||
if len(b.SubButton) > 0 {
|
||||
if strings.TrimSpace(b.Type) != "" {
|
||||
return fmt.Errorf("含子菜单的一级项不应设置类型")
|
||||
}
|
||||
if len(b.SubButton) > 5 {
|
||||
return fmt.Errorf("子菜单最多 5 项")
|
||||
}
|
||||
for _, sb := range b.SubButton {
|
||||
if err := validateLeafButton(sb); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return validateLeafButton(b)
|
||||
}
|
||||
|
||||
func validateLeafButton(b types.WxGzhMenuButton) error {
|
||||
if strings.TrimSpace(b.Name) == "" {
|
||||
return fmt.Errorf("菜单名称不能为空")
|
||||
}
|
||||
if len(b.SubButton) > 0 {
|
||||
return fmt.Errorf("仅支持两级菜单,子菜单下不能再嵌套")
|
||||
}
|
||||
t := strings.ToLower(strings.TrimSpace(b.Type))
|
||||
switch t {
|
||||
case "view":
|
||||
if strings.TrimSpace(b.URL) == "" {
|
||||
return fmt.Errorf("跳转链接类型需要填写 URL")
|
||||
}
|
||||
case "click":
|
||||
if strings.TrimSpace(b.Key) == "" {
|
||||
return fmt.Errorf("点击事件类型需要填写 Key")
|
||||
}
|
||||
case "miniprogram":
|
||||
if strings.TrimSpace(b.AppID) == "" || strings.TrimSpace(b.PagePath) == "" || strings.TrimSpace(b.URL) == "" {
|
||||
return fmt.Errorf("小程序类型需要填写 appid、pagepath 与备用 url")
|
||||
}
|
||||
default:
|
||||
if t == "" {
|
||||
return fmt.Errorf("请选择菜单类型")
|
||||
}
|
||||
return fmt.Errorf("暂不支持的菜单类型: %s", b.Type)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"geekai/core/types"
|
||||
"geekai/store/model"
|
||||
"geekai/utils"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
@@ -16,10 +17,20 @@ import (
|
||||
type WxGzhService struct {
|
||||
config types.WxGzhConfig
|
||||
DB *gorm.DB
|
||||
|
||||
ccMu sync.Mutex
|
||||
ccToken string
|
||||
ccExpire time.Time
|
||||
ccForApp string
|
||||
}
|
||||
|
||||
func (s *WxGzhService) UpdateConfig(config types.WxGzhConfig) {
|
||||
s.config = config
|
||||
s.ccMu.Lock()
|
||||
s.ccToken = ""
|
||||
s.ccExpire = time.Time{}
|
||||
s.ccForApp = ""
|
||||
s.ccMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *WxGzhService) GetConfig() types.WxGzhConfig {
|
||||
|
||||
Reference in New Issue
Block a user