feat: rebind mobile

2026-10-21 13:29:03 +08:00
parent 113756fd46
commit 6107ac8d9d
47 changed files with 2210 additions and 289 deletions
@@ -0,0 +1,56 @@
+package callback
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+
+	"lone-services/pkg/utils"
+)
+
+func BuildDownloadURL(object string) (string, error) {
+	endpoint := utils.GetConfigString("oss.internalEndpoint")
+	if endpoint == utils.StringEmpty {
+		endpoint = utils.GetConfigString("oss.endpoint")
+	}
+	if endpoint == utils.StringEmpty {
+		return utils.StringEmpty, fmt.Errorf("oss endpoint empty")
+	}
+	return url.JoinPath(endpoint, object)
+}
+
+func DownloadFile(fileURL, destPath string) error {
+	resp, err := http.Get(fileURL)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("http status %d", resp.StatusCode)
+	}
+
+	tmpPath := destPath + ".tmp"
+	out, err := os.Create(tmpPath)
+	if err != nil {
+		return err
+	}
+
+	_, copyErr := io.Copy(out, resp.Body)
+	closeErr := out.Close()
+	if copyErr != nil {
+		_ = os.Remove(tmpPath)
+		return copyErr
+	}
+	if closeErr != nil {
+		_ = os.Remove(tmpPath)
+		return closeErr
+	}
+	if err = os.Rename(tmpPath, destPath); err != nil {
+		_ = os.Remove(tmpPath)
+		return err
+	}
+	return nil
+}
@@ -0,0 +1,60 @@
+package callback
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"lone-services/pkg/redis"
+	"lone-services/pkg/utils"
+	pkgws "lone-services/pkg/ws"
+)
+
+func CacheUploadedLights(taskID, finalDir string) error {
+	files, err := os.ReadDir(finalDir)
+	if err != nil {
+		return err
+	}
+
+	seen := make(map[string]struct{})
+	lights := make([]interface{}, 0)
+	for _, file := range files {
+		if file.IsDir() {
+			continue
+		}
+		light := lightingTypeFromFileName(file.Name())
+		if light == utils.StringEmpty {
+			continue
+		}
+		if _, ok := seen[light]; ok {
+			continue
+		}
+		seen[light] = struct{}{}
+		lights = append(lights, light)
+	}
+	if len(lights) == 0 {
+		return nil
+	}
+
+	ctx := context.Background()
+	key := pkgws.RedisKeyTaskLights + taskID
+	if err = redis.Client.SAdd(ctx, key, lights...).Err(); err != nil {
+		return err
+	}
+	return redis.Client.Expire(ctx, key, time.Hour).Err()
+}
+
+func lightingTypeFromFileName(fileName string) string {
+	ext := strings.ToLower(filepath.Ext(fileName))
+	if !isImageExt(ext) {
+		return utils.StringEmpty
+	}
+	base := strings.TrimSuffix(fileName, filepath.Ext(fileName))
+	parts := strings.Split(base, "_")
+	if len(parts) < 2 {
+		return utils.StringEmpty
+	}
+	return strings.ToLower(parts[0])
+}
@@ -0,0 +1,40 @@
+package callback
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"lone-services/pkg/utils"
+
+	jsoniter "github.com/json-iterator/go"
+)
+
+func NotifyPythonService(taskId, subTaskId, folderPath string) error {
+	if err := MarkTaskProcessing(taskId); err != nil {
+		return err
+	}
+
+	host := utils.GetConfigString("services.skin_py_analysis_host")
+	apiPath := utils.GetConfigString("services.skin_py_analysis_path")
+	if host == utils.StringEmpty || apiPath == utils.StringEmpty {
+		return fmt.Errorf("python analysis service config empty")
+	}
+
+	url := strings.TrimRight(host, "/") + apiPath
+	body, _ := jsoniter.Marshal(map[string]interface{}{
+		"task_id":     subTaskId,
+		"folder_name": folderPath,
+	})
+
+	resp, err := http.Post(url, "application/json", bytes.NewBuffer(body))
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("python service error: %s", resp.Status)
+	}
+	return nil
+}
@@ -0,0 +1,74 @@
+package callback
+
+import (
+	"strconv"
+
+	"lone-services/pkg/modelbase"
+	"lone-services/pkg/utils"
+	"lone-services/services/analysis/internal/dao"
+	"lone-services/services/analysis/internal/model"
+
+	"gorm.io/gorm"
+)
+
+func SaveTask(ts dao.TaskTimeStamp, userId int64, path *dao.HandleZipResource, coverPath string) error {
+	m := model.ImageAnalysisModel{}.Init()
+	var task dao.ImageAnalysis
+	w := modelbase.Params{Eq: map[string]string{"task_id": ts.TaskId}}
+	if err := m.GetOne(w, &task); err != nil {
+		return err
+	}
+
+	now := utils.Now()
+	if task.Id < utils.NumberOne {
+		return m.Create(&dao.ImageAnalysis{
+			TaskId:      ts.TaskId,
+			UserId:      userId,
+			FolderPath:  path.TargetDir,
+			Status:      dao.AnalysisStatusPending,
+			UploadCount: 1,
+			Cover:       coverPath,
+			CreatedAt:   now,
+			UpdatedAt:   now,
+		})
+	}
+
+	updateData := map[string]interface{}{
+		"upload_count": gorm.Expr("upload_count + ?", 1),
+		"folder_path":  path.TargetDir,
+		"updated_at":   now,
+	}
+	if coverPath != utils.StringEmpty {
+		updateData["cover"] = coverPath
+	}
+	_, err := m.Edit(w, updateData)
+	return err
+}
+
+func SaveSubTask(ts dao.TaskTimeStamp, path *dao.HandleZipResource) error {
+	now := utils.Now()
+	return model.ImageAnalysisSubModel{}.Init().Create(&dao.ImageAnalysisSub{
+		TaskId:         ts.TaskId,
+		SubTaskId:      ts.SubTaskId,
+		Status:         dao.AnalysisStatusProcessing,
+		FilePath:       path.FilePath,
+		ImageDir:       path.FinalDir,
+		UploadStartTs:  ts.UploadStartTs,
+		UploadEndTs:    ts.UploadEndTs,
+		ProcessStartTs: ts.ProcessStartTs,
+		CreatedAt:      now,
+		UpdatedAt:      now,
+	})
+}
+
+func MarkTaskProcessing(taskId string) error {
+	_, err := model.ImageAnalysisModel{}.Init().Edit(modelbase.Params{
+		Eq: map[string]string{
+			"task_id": taskId,
+			"status":  strconv.Itoa(int(dao.AnalysisStatusPending)),
+		},
+	}, map[string]interface{}{
+		"status": dao.AnalysisStatusProcessing,
+	})
+	return err
+}
@@ -0,0 +1,81 @@
+package callback
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	"lone-services/pkg/utils"
+	"lone-services/services/analysis/internal/dao"
+	analysisoss "lone-services/services/analysis/internal/oss"
+
+	"github.com/google/uuid"
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+func ProcessUpload(ctx context.Context, payload analysisoss.CallbackPayload) error {
+	if payload.TaskId == utils.StringEmpty || payload.Object == utils.StringEmpty {
+		return fmt.Errorf("callback missing task_id or object")
+	}
+
+	sess, err := analysisoss.LoadUploadSession(ctx, payload.TaskId)
+	if err != nil {
+		return fmt.Errorf("load upload session: %w", err)
+	}
+	if sess.DateDir == utils.StringEmpty {
+		return fmt.Errorf("upload session date_dir empty")
+	}
+
+	downloadURL, err := BuildDownloadURL(payload.Object)
+	if err != nil {
+		return err
+	}
+
+	subTaskId := uuid.New().String()
+	subDirName := strconv.FormatInt(time.Now().UnixMilli(), 10)
+	targetDir := filepath.Join(utils.GetConfigString("file.analysis_path"), sess.DateDir, payload.TaskId)
+	if err = os.MkdirAll(targetDir, os.ModePerm); err != nil {
+		return fmt.Errorf("mkdir analysis path: %w", err)
+	}
+	zipPath := filepath.Join(targetDir, subDirName+".zip")
+
+	if err = DownloadFile(downloadURL, zipPath); err != nil {
+		return fmt.Errorf("download zip: %w", err)
+	}
+
+	pathRes, coverPath, err := HandleTaskZip(zipPath)
+	if err != nil {
+		return fmt.Errorf("handle zip: %w", err)
+	}
+
+	if err = CacheUploadedLights(payload.TaskId, pathRes.FinalDir); err != nil {
+		logx.WithContext(ctx).Errorf("cache uploaded lights failed: %v", err)
+	}
+
+	if utils.GetConfigInt("switch.facepp") == utils.NumberOne {
+		logx.WithContext(ctx).Infof("facepp switch on but not migrated yet, skip task_id=%s", payload.TaskId)
+	}
+
+	if utils.GetConfigInt("switch.analysis") == utils.NumberOne {
+		if err = NotifyPythonService(payload.TaskId, subTaskId, pathRes.FinalDir); err != nil {
+			logx.WithContext(ctx).Errorf("notify python failed: %v", err)
+		}
+	}
+
+	ts := dao.TaskTimeStamp{
+		TaskId:         payload.TaskId,
+		SubTaskId:      subTaskId,
+		ProcessStartTs: strconv.FormatInt(time.Now().UnixMilli(), 10),
+	}
+	userId := sess.SkinUserId()
+	if userId < utils.NumberOne {
+		userId = sess.UserId
+	}
+	if err = SaveTask(ts, userId, pathRes, coverPath); err != nil {
+		return err
+	}
+	return SaveSubTask(ts, pathRes)
+}
@@ -0,0 +1,88 @@
+package callback
+
+import (
+	"archive/zip"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"lone-services/pkg/utils"
+	"lone-services/services/analysis/internal/dao"
+)
+
+func HandleTaskZip(zipPath string) (*dao.HandleZipResource, string, error) {
+	targetDir := filepath.Dir(zipPath)
+	subDirName := strings.TrimSuffix(filepath.Base(zipPath), filepath.Ext(zipPath))
+	finalDir := filepath.Join(targetDir, subDirName)
+
+	coverPath, err := unzipImages(zipPath, finalDir)
+	if err != nil {
+		return nil, utils.StringEmpty, err
+	}
+
+	return &dao.HandleZipResource{
+		FilePath:  zipPath,
+		FinalDir:  finalDir,
+		TargetDir: targetDir,
+	}, coverPath, nil
+}
+
+func unzipImages(zipPath, targetDir string) (string, error) {
+	if err := os.MkdirAll(targetDir, os.ModePerm); err != nil {
+		return utils.StringEmpty, err
+	}
+
+	r, err := zip.OpenReader(zipPath)
+	if err != nil {
+		return utils.StringEmpty, err
+	}
+	defer r.Close()
+
+	var crossMidPath string
+	for _, f := range r.File {
+		ext := strings.ToLower(filepath.Ext(f.Name))
+		if !isImageExt(ext) {
+			continue
+		}
+
+		rc, err := f.Open()
+		if err != nil {
+			continue
+		}
+
+		filename := filepath.Base(f.Name)
+		if filename == utils.StringEmpty {
+			_ = rc.Close()
+			continue
+		}
+
+		fullPath := filepath.Join(targetDir, filename)
+		dst, err := os.Create(fullPath)
+		if err != nil {
+			_ = rc.Close()
+			continue
+		}
+
+		_, copyErr := io.Copy(dst, rc)
+		_ = dst.Close()
+		_ = rc.Close()
+		if copyErr != nil {
+			continue
+		}
+
+		if strings.EqualFold(filename, "cross_mid.jpg") {
+			crossMidPath = fullPath
+		}
+	}
+	return crossMidPath, nil
+}
+
+func isImageExt(ext string) bool {
+	switch ext {
+	case ".jpg", ".jpeg", ".png", ".gif":
+		return true
+	default:
+		return false
+	}
+}