222 lines
6.7 KiB
Go
222 lines
6.7 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"affiliate_dash/internal/model"
|
|
)
|
|
|
|
// PackageImportResult 是单个包的导入结果。
|
|
type PackageImportResult struct {
|
|
PackageID int64 `json:"package_id"`
|
|
ShareCode string `json:"share_code"`
|
|
Inserted int `json:"inserted"`
|
|
SkippedBound int `json:"skipped_bound"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// ImportAllResult 是全量导入的汇总结果。
|
|
type ImportAllResult struct {
|
|
Scanned int `json:"scanned"`
|
|
Skipped int `json:"skipped"` // 本地已有该包 Token，整包跳过
|
|
Empty int `json:"empty"` // 包内没有未绑定的 Token
|
|
Imported int `json:"imported"` // 成功导入了 Token 的包
|
|
Failed int `json:"failed"` // 拉详情或写库失败
|
|
TokensImported int `json:"tokens_imported"`
|
|
TokensSkippedBound int `json:"tokens_skipped_bound"`
|
|
Remaining int `json:"remaining"` // 超过单次上限未处理的包
|
|
Results []PackageImportResult `json:"results"`
|
|
}
|
|
|
|
// isUnboundRockToken 判断上游 Token 是否从未绑定过。
|
|
// 规则（与业务确认）：bind_changed_at 与 bind_change_type 同时为空才算未绑定。
|
|
func isUnboundRockToken(token RockPackageToken) bool {
|
|
if strings.TrimSpace(token.BindChangedAt) != "" {
|
|
return false
|
|
}
|
|
if strings.TrimSpace(token.BindChangeType) != "" {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// filterUnboundTokens 拆出未绑定 Token，并返回被跳过的已绑定数量。
|
|
func filterUnboundTokens(tokens []RockPackageToken) ([]RockPackageToken, int) {
|
|
kept := make([]RockPackageToken, 0, len(tokens))
|
|
skipped := 0
|
|
for _, token := range tokens {
|
|
if isUnboundRockToken(token) {
|
|
kept = append(kept, token)
|
|
continue
|
|
}
|
|
skipped++
|
|
}
|
|
return kept, skipped
|
|
}
|
|
|
|
// ImportAllUpstreamPackages 全量扫描上游包：已导入的跳过，只导入包内未绑定的 Token，
|
|
// 且导入后统一标记为 available。
|
|
func (s *RockService) ImportAllUpstreamPackages(ctx context.Context, maxPackages int, actorUserID uint) (*ImportAllResult, error) {
|
|
if maxPackages <= 0 {
|
|
maxPackages = 50
|
|
}
|
|
if maxPackages > 200 {
|
|
maxPackages = 200
|
|
}
|
|
|
|
cfg, err := s.GetConfig(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(cfg.BearerToken) == "" {
|
|
return nil, errors.New("请先配置 0666.pro Bearer Token")
|
|
}
|
|
|
|
// 1. 分页拉取全部上游包（最多 20 页，防止异常分页导致死循环）。
|
|
all := make([]RockPackage, 0, 64)
|
|
for page := 1; page <= 20; page++ {
|
|
resp, listErr := s.client.ListPackages(ctx, cfg.BearerToken, page, 100)
|
|
if listErr != nil {
|
|
return nil, listErr
|
|
}
|
|
all = append(all, resp.Packages...)
|
|
if resp.Pagination.TotalPages > 0 {
|
|
if page >= resp.Pagination.TotalPages {
|
|
break
|
|
}
|
|
} else if len(resp.Packages) == 0 {
|
|
break
|
|
}
|
|
if len(resp.Packages) == 0 {
|
|
break
|
|
}
|
|
}
|
|
|
|
result := &ImportAllResult{Scanned: len(all)}
|
|
if len(all) == 0 {
|
|
return result, nil
|
|
}
|
|
|
|
// 2. 查询本地已导入的 package_id 集合（以本地实际存在的 Token 为准）。
|
|
ids := make([]int64, 0, len(all))
|
|
for _, pkg := range all {
|
|
if pkg.ID > 0 {
|
|
ids = append(ids, pkg.ID)
|
|
}
|
|
}
|
|
importedIDs := make([]int64, 0, len(ids))
|
|
if len(ids) > 0 {
|
|
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
|
|
Where("package_id IN ?", ids).
|
|
Group("package_id").
|
|
Pluck("package_id", &importedIDs).Error; err != nil {
|
|
log.Printf("[RockService] 查询本地已导入包失败: %v", err)
|
|
}
|
|
}
|
|
importedSet := make(map[int64]struct{}, len(importedIDs))
|
|
for _, id := range importedIDs {
|
|
importedSet[id] = struct{}{}
|
|
}
|
|
|
|
// 3. 过滤出待处理的包，并按上限截断。
|
|
pending := make([]RockPackage, 0, len(all))
|
|
for _, pkg := range all {
|
|
if _, ok := importedSet[pkg.ID]; ok {
|
|
result.Skipped++
|
|
continue
|
|
}
|
|
pending = append(pending, pkg)
|
|
}
|
|
if len(pending) > maxPackages {
|
|
result.Remaining = len(pending) - maxPackages
|
|
pending = pending[:maxPackages]
|
|
}
|
|
|
|
// 4. 串行导入，单包失败不影响后续。
|
|
for _, pkg := range pending {
|
|
item := s.importAllOnePackage(ctx, pkg, actorUserID)
|
|
result.Results = append(result.Results, item)
|
|
switch {
|
|
case item.Error != "":
|
|
result.Failed++
|
|
case item.Inserted == 0:
|
|
result.Empty++
|
|
default:
|
|
result.Imported++
|
|
result.TokensImported += item.Inserted
|
|
}
|
|
result.TokensSkippedBound += item.SkippedBound
|
|
}
|
|
log.Printf("[RockService] 全量导入完成: %s", formatImportAllSummary(result))
|
|
return result, nil
|
|
}
|
|
|
|
func (s *RockService) importAllOnePackage(ctx context.Context, pkg RockPackage, actorUserID uint) PackageImportResult {
|
|
result := PackageImportResult{PackageID: pkg.ID, ShareCode: pkg.ShareCode}
|
|
detail, err := s.client.GetPackageDetail(ctx, pkg.ShareCode)
|
|
if err != nil {
|
|
result.Error = "拉取包详情失败: " + err.Error()
|
|
return result
|
|
}
|
|
if detail == nil || detail.Package.ID == 0 {
|
|
result.Error = "上游未返回 package.id"
|
|
return result
|
|
}
|
|
result.PackageID = detail.Package.ID
|
|
|
|
kept, skippedBound := filterUnboundTokens(detail.Tokens)
|
|
result.SkippedBound = skippedBound
|
|
if len(kept) == 0 {
|
|
return result
|
|
}
|
|
|
|
// 只把未绑定 Token 交给导入器，并强制写成 available。
|
|
filtered := *detail
|
|
filtered.Tokens = kept
|
|
forceIDs := make(map[string]struct{}, len(kept))
|
|
for _, token := range kept {
|
|
if token.ID > 0 {
|
|
forceIDs[strconv.FormatInt(token.ID, 10)] = struct{}{}
|
|
}
|
|
}
|
|
|
|
inserted, updated, skipped, err := s.importPackageDetail(ctx, &filtered, forceIDs)
|
|
if err != nil {
|
|
result.Error = "导入失败: " + err.Error()
|
|
return result
|
|
}
|
|
result.Inserted = inserted
|
|
|
|
if inserted > 0 {
|
|
actor := actorUserID
|
|
metadata := map[string]interface{}{
|
|
"package_id": detail.Package.ID,
|
|
"share_code": pkg.ShareCode,
|
|
"mode": "import_all",
|
|
"inserted": inserted,
|
|
"updated": updated,
|
|
"skipped": skipped,
|
|
"skipped_bound": skippedBound,
|
|
}
|
|
if auditErr := writeAudit(s.db.WithContext(ctx), nil, &actor, nil, "rock_package.import_all", "rock_package", pkg.ShareCode, metadata); auditErr != nil {
|
|
log.Printf("[RockService] 写入全量导入审计失败 share_code=%s: %v", pkg.ShareCode, auditErr)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
// formatImportAllSummary 生成便于日志排查的一行摘要。
|
|
func formatImportAllSummary(result *ImportAllResult) string {
|
|
if result == nil {
|
|
return ""
|
|
}
|
|
return fmt.Sprintf("扫描=%d 跳过=%d 空包=%d 导入=%d 失败=%d Token=%d 已绑定跳过=%d 剩余=%d",
|
|
result.Scanned, result.Skipped, result.Empty, result.Imported, result.Failed,
|
|
result.TokensImported, result.TokensSkippedBound, result.Remaining)
|
|
}
|