update
This commit is contained in:
toom1996
@ -3,7 +3,6 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha1"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
@ -11,18 +10,20 @@ import (
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"fashionapi/internal/dto"
|
||||
"fashionapi/internal/model"
|
||||
"fashionapi/internal/pkg/hashid"
|
||||
"fashionapi/internal/pkg/phash"
|
||||
"fashionapi/internal/pkg/season"
|
||||
"fashionapi/internal/pkg/storage"
|
||||
"fashionapi/internal/repository"
|
||||
"fashionapi/internal/vision"
|
||||
)
|
||||
|
||||
// IngestService 爬虫入库管线的业务逻辑层：Submit 入队，worker 异步处理。
|
||||
@ -39,14 +40,26 @@ type IngestService struct {
|
||||
local *storage.LocalUploader // S4失败时的兜底落地
|
||||
httpClient *http.Client
|
||||
translate *TranslateService // 标题中文翻译（可为 nil：测试/未启用时跳过自动补全）
|
||||
// 视觉服务与向量仓储（均可为 nil：未启用视觉服务时跳过 embed，不影响任何入库行为）。
|
||||
// 只服务于后台「疑似同款」提示，不参与去重。
|
||||
vision *vision.Client
|
||||
embedRep repository.EmbeddingRepository
|
||||
// similar 「疑似同款」预计算服务（可为 nil：装配期未注入或视觉服务未启用时跳过）。
|
||||
// 写完向量后顺手把该图集的跨图集近邻算好落库，使后台页面只需读一次小查询。
|
||||
// 注入用 SetSimilar，避免改动构造签名（该服务在主装配期晚于本服务创建）。
|
||||
similar *SimilarService
|
||||
}
|
||||
|
||||
// SetSimilar 注入「疑似同款」服务，供写向量后顺手预计算近邻。
|
||||
// 装配期调用；未注入（nil）时完全跳过预计算，不影响入库。
|
||||
func (s *IngestService) SetSimilar(sv *SimilarService) { s.similar = sv }
|
||||
|
||||
// NewIngestService 创建入库服务。
|
||||
//
|
||||
// uploader: 主上传器（S4或本地）
|
||||
// del: 删除器（S4或本地），用于 media_cleanup 真删S4/本地孤儿文件
|
||||
// local: 本地兜底上传器（S4上传失败时回退，避免图片完全丢失）
|
||||
func NewIngestService(repo repository.IngestRepository, brandRepo repository.BrandRepository, media repository.MediaRepository, uploader storage.Uploader, del storage.Deleter, local *storage.LocalUploader, downloadProxy string, translateSvc *TranslateService) *IngestService {
|
||||
func NewIngestService(repo repository.IngestRepository, brandRepo repository.BrandRepository, media repository.MediaRepository, uploader storage.Uploader, del storage.Deleter, local *storage.LocalUploader, downloadProxy string, translateSvc *TranslateService, vis *vision.Client, embedRep repository.EmbeddingRepository) *IngestService {
|
||||
// 下载代理：download_proxy（或 INGEST_DOWNLOAD_PROXY）非空时，下载客户端经该代理访问源站；
|
||||
// 仅作用于图片下载，S4 上传走独立 s3 客户端、pgx 数据库连接不走 http，均不受影响。
|
||||
// 留空回退 http.ProxyFromEnvironment（默认直连，仍可被系统 HTTP(S)_PROXY 影响）。
|
||||
@ -76,6 +89,8 @@ func NewIngestService(repo repository.IngestRepository, brandRepo repository.Bra
|
||||
del: del,
|
||||
local: local,
|
||||
translate: translateSvc,
|
||||
vision: vis,
|
||||
embedRep: embedRep,
|
||||
// 图片下载是并发执行的（见 imageFetchConcurrency），连接池须匹配并发度：
|
||||
// Go 默认 MaxIdleConnsPerHost=2，多余连接会在每轮下载时反复重建、白付 TLS 握手开销。
|
||||
httpClient: &http.Client{
|
||||
@ -319,12 +334,14 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
|
||||
// 内容哈希（sha1）key 保证重爬不产生孤儿文件：失败回滚删本批 key 即可。
|
||||
var cover string
|
||||
var images []model.BrandRunwayImage
|
||||
// vecs 与 images 下标一一对应（nil = 该张未产出向量）；落库拿到图片 id 后写入 image_embeddings。
|
||||
var vecs [][]float32
|
||||
var keys []string
|
||||
var imgFailed bool
|
||||
var imageCount uint16
|
||||
var timing *fetchTiming
|
||||
if len(p.Looks) > 0 {
|
||||
cover, images, keys, imgFailed, timing = s.fetchLookImages(ctx, p.Looks, "runway")
|
||||
cover, images, vecs, keys, imgFailed, timing = s.fetchLookImages(ctx, p.Looks, "runway")
|
||||
// image_count 必须等于**实际落库的主图行数**，不能是 len(p.Looks)：
|
||||
// fetchLookImages 会跳过 look.Main == "" 的 look 与 sha1 重复的行，实际主图行可能更少。
|
||||
// 旧流程靠晋升时按实际行重算自愈，该路径已随单表发布模型删除 —— 一旦用 len(Looks)，
|
||||
@ -332,19 +349,17 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
|
||||
imageCount = countMainImages(images)
|
||||
} else {
|
||||
var fimgs []fetchedImage
|
||||
cover, fimgs, keys, imgFailed, timing = s.fetchImages(ctx, p.Images, "runway", dto.IngestKindRunway)
|
||||
cover, fimgs, keys, imgFailed, timing = s.fetchImages(ctx, p.Images, "runway")
|
||||
imageCount = uint16(len(fimgs))
|
||||
for i, fi := range fimgs {
|
||||
images = append(images, model.BrandRunwayImage{
|
||||
Image: fi.url,
|
||||
Name: fmt.Sprintf("Look %d", i+1),
|
||||
SortOrder: uint32(i + 1),
|
||||
IsDetail: 0,
|
||||
Phash: sqlNull(fi.phash),
|
||||
IsDuplicate: fi.isDup,
|
||||
DupOf: strconv.FormatUint(uint64(fi.dupID), 10),
|
||||
ImageSize: fi.size,
|
||||
Image: fi.url,
|
||||
Name: fmt.Sprintf("Look %d", i+1),
|
||||
SortOrder: uint32(i + 1),
|
||||
IsDetail: 0,
|
||||
ImageSize: fi.size,
|
||||
})
|
||||
vecs = append(vecs, fi.vec)
|
||||
}
|
||||
}
|
||||
log.Printf("[ingest] job=%d runway 拉图 总耗时=%v %s", job.ID, timing.total.Round(time.Millisecond), timing)
|
||||
@ -396,6 +411,15 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
|
||||
s.failOrRetry(ctx, job.ID, "save runway: "+saveErr.Error())
|
||||
return
|
||||
}
|
||||
// 落库成功后写入语义向量（仅后台「疑似同款」用）。GORM 批量 Create 会把自增主键回填到
|
||||
// images，因此此处 images[i].ID 即新建图片 id；写失败只记日志，绝不影响入库结果。
|
||||
embedIDs := make([]uint32, 0, len(images))
|
||||
for i := range images {
|
||||
embedIDs = append(embedIDs, images[i].ID)
|
||||
}
|
||||
s.saveEmbeddings(ctx, dto.IngestKindRunway, embedIDs, vecs)
|
||||
// 顺手把本场的「疑似同款」近邻算好落库（失败不影响入库）。
|
||||
s.precomputeSimilar(ctx, dto.IngestKindRunway, embedIDs)
|
||||
// 入库后按 sort_order 回填细节图的 parent_image_id，使新走秀自动成组（与街拍一致）。
|
||||
runwayID := rw.ID
|
||||
if runwayID == 0 {
|
||||
@ -414,16 +438,17 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
|
||||
// failed 置 true，调用方据此把整条任务判失败并回滚本批已上传的 key，符合「单图失败=整任务失败」策略。
|
||||
// cover 取首个成功下载的主图；image_count 由调用方按返回行中 is_detail == 0 的数量计
|
||||
//（countMainImages），而不是 len(Looks)：跳过空主图 / sha1 重复后实际行数可能更少。
|
||||
// 每张图入库前做去重：命中 dHash 近重复则仍入库但标记留痕。
|
||||
//
|
||||
// 实现：先把全部「主图 + 细节图」按原始顺序摊平成任务列表，用 concurrentFetch 并发完成
|
||||
// 「下载 → phash → 上传」这段 IO 密集操作；随后再按下标顺序串行去重与组装。
|
||||
// 「下载 → 上传」这段 IO 密集操作；随后再按下标顺序串行做批次内去重与组装。
|
||||
// 这样既拿到并发收益，又保证 cover / sort_order / 批次内去重与原串行实现一致。
|
||||
func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayLook, prefix string) (string, []model.BrandRunwayImage, []string, bool, *fetchTiming) {
|
||||
func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayLook, prefix string) (string, []model.BrandRunwayImage, [][]float32, []string, bool, *fetchTiming) {
|
||||
start := time.Now()
|
||||
timing := &fetchTiming{}
|
||||
cover := ""
|
||||
rows := make([]model.BrandRunwayImage, 0)
|
||||
// vecs 与 rows 下标一一对应（nil 表示该张未产出向量），供落库后写入 image_embeddings。
|
||||
vecs := make([][]float32, 0)
|
||||
keys := make([]string, 0)
|
||||
seen := make(map[string]bool)
|
||||
failed := false
|
||||
@ -443,7 +468,7 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
|
||||
}
|
||||
}
|
||||
|
||||
// 并发段：下载 + phash + 上传（耗时大头）。
|
||||
// 并发段：下载 + 上传（耗时大头）。
|
||||
results := concurrentFetch(ctx, len(tasks), func(c context.Context, i int) downloadResult {
|
||||
return s.downloadAndUpload(c, tasks[i].url, prefix)
|
||||
})
|
||||
@ -453,8 +478,8 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
|
||||
for i, r := range results {
|
||||
task := tasks[i]
|
||||
timing.download += r.download
|
||||
timing.phash += r.phash
|
||||
timing.upload += r.upload
|
||||
timing.embed += r.embed
|
||||
if r.err != nil {
|
||||
failed = true
|
||||
continue
|
||||
@ -464,7 +489,6 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
|
||||
continue
|
||||
}
|
||||
seen[r.sha1] = true
|
||||
dupID, isDup := s.dedupImage(ctx, dto.IngestKindRunway, r.phashBits, timing)
|
||||
order++
|
||||
if cover == "" && !task.isDetail {
|
||||
cover = r.url
|
||||
@ -476,19 +500,17 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
|
||||
isDetail = 1
|
||||
}
|
||||
rows = append(rows, model.BrandRunwayImage{
|
||||
Image: r.url,
|
||||
Name: name,
|
||||
SortOrder: uint32(order),
|
||||
IsDetail: isDetail,
|
||||
Phash: sqlNull(r.phashBits),
|
||||
IsDuplicate: isDup,
|
||||
DupOf: strconv.FormatUint(uint64(dupID), 10),
|
||||
ImageSize: r.size,
|
||||
Image: r.url,
|
||||
Name: name,
|
||||
SortOrder: uint32(order),
|
||||
IsDetail: isDetail,
|
||||
ImageSize: r.size,
|
||||
})
|
||||
vecs = append(vecs, r.vec)
|
||||
}
|
||||
timing.images = len(rows)
|
||||
timing.total = time.Since(start)
|
||||
return cover, rows, keys, failed, timing
|
||||
return cover, rows, vecs, keys, failed, timing
|
||||
}
|
||||
|
||||
// countMainImages 数一批待落库图片行中的主图（is_detail = 0）数量，即 image_count 的口径。
|
||||
@ -530,7 +552,7 @@ func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob,
|
||||
}
|
||||
|
||||
// 2) 下载图片并上传到存储（S4优先，失败兜底本地）。
|
||||
cover, fimgs, keys, imgFailed, timing := s.fetchImages(ctx, p.Images, "street", dto.IngestKindStreet)
|
||||
cover, fimgs, keys, imgFailed, timing := s.fetchImages(ctx, p.Images, "street")
|
||||
log.Printf("[ingest] job=%d street 拉图 总耗时=%v %s", job.ID, timing.total.Round(time.Millisecond), timing)
|
||||
if len(p.Images) > 0 && imgFailed {
|
||||
// 单图失败=整任务失败：先回滚本批已上传的图（S4 + 本地兜底），避免孤儿文件永远堆在存储里，
|
||||
@ -562,16 +584,16 @@ func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob,
|
||||
Status: model.StatusPending,
|
||||
}
|
||||
rows := make([]model.StreetSnapImage, 0, len(fimgs))
|
||||
// vecs 与 rows 下标一一对应（nil = 该张未产出向量）；落库拿到图片 id 后写入 image_embeddings。
|
||||
vecs := make([][]float32, 0, len(fimgs))
|
||||
for i, fi := range fimgs {
|
||||
rows = append(rows, model.StreetSnapImage{
|
||||
Image: fi.url,
|
||||
Name: fmt.Sprintf("Look %d", i+1),
|
||||
SortOrder: uint32(i + 1),
|
||||
Phash: sqlNull(fi.phash),
|
||||
IsDuplicate: fi.isDup,
|
||||
DupOf: strconv.FormatUint(uint64(fi.dupID), 10),
|
||||
ImageSize: fi.size,
|
||||
Image: fi.url,
|
||||
Name: fmt.Sprintf("Look %d", i+1),
|
||||
SortOrder: uint32(i + 1),
|
||||
ImageSize: fi.size,
|
||||
})
|
||||
vecs = append(vecs, fi.vec)
|
||||
}
|
||||
var saveErr error
|
||||
if action == ingestReuseRejected {
|
||||
@ -583,19 +605,27 @@ func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob,
|
||||
s.failOrRetry(ctx, job.ID, "save street snap: "+saveErr.Error())
|
||||
return
|
||||
}
|
||||
// 落库成功后写入语义向量（同走秀：GORM 批量 Create 回填自增主键，故 rows[i].ID 即新建图片 id）。
|
||||
embedIDs := make([]uint32, 0, len(rows))
|
||||
for i := range rows {
|
||||
embedIDs = append(embedIDs, rows[i].ID)
|
||||
}
|
||||
s.saveEmbeddings(ctx, dto.IngestKindStreet, embedIDs, vecs)
|
||||
// 顺手把本专辑的「疑似同款」近邻算好落库（失败不影响入库）。
|
||||
s.precomputeSimilar(ctx, dto.IngestKindStreet, embedIDs)
|
||||
log.Printf("[ingest] job=%d street 写正式表 耗时=%v 图片行=%d 复用驳回=%v",
|
||||
job.ID, time.Since(writeStart).Round(time.Millisecond), len(rows), action == ingestReuseRejected)
|
||||
_ = s.repo.MarkDone(ctx, job.ID)
|
||||
}
|
||||
|
||||
// fetchImages 下载图片并上传到存储，返回 (cover 地址, 已下载图结构, 已成功上传对象的 key 列表, 是否有任意一张失败)。
|
||||
// prefix 为S4 key 前缀（runway/ 或 street/）。kind 用于选择去重比对表。
|
||||
// prefix 为S4 key 前缀（runway/ 或 street/）。
|
||||
// 只要任意一张下载/上传失败，failed 即置 true，调用方据此把整条任务判为失败并回滚本批已上传的对象，
|
||||
// 符合「单图失败=整任务失败」策略。每张图入库前做去重（近重复标记）。
|
||||
// 符合「单图失败=整任务失败」策略。
|
||||
//
|
||||
// 实现：先用 concurrentFetch 并发完成全部图片的「下载 → phash → 上传」，再按下标顺序串行
|
||||
// 做去重与组装，保证 cover / 顺序 / 批次内去重与原串行实现一致。
|
||||
func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix, kind string) (string, []fetchedImage, []string, bool, *fetchTiming) {
|
||||
// 实现：先用 concurrentFetch 并发完成全部图片的「下载 → 上传」，再按下标顺序串行
|
||||
// 做批次内去重与组装，保证 cover / 顺序 / 批次内去重与原串行实现一致。
|
||||
func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix string) (string, []fetchedImage, []string, bool, *fetchTiming) {
|
||||
start := time.Now()
|
||||
timing := &fetchTiming{}
|
||||
cover := ""
|
||||
@ -604,7 +634,7 @@ func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix,
|
||||
seen := make(map[string]bool)
|
||||
failed := false
|
||||
|
||||
// 并发段：下载 + phash + 上传（耗时大头）。
|
||||
// 并发段：下载 + 上传（耗时大头）。
|
||||
results := concurrentFetch(ctx, len(urls), func(c context.Context, i int) downloadResult {
|
||||
return s.downloadAndUpload(c, urls[i], prefix)
|
||||
})
|
||||
@ -612,8 +642,8 @@ func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix,
|
||||
// 串行段：按原始顺序累计耗时、做批次内去重、组装结果。
|
||||
for _, r := range results {
|
||||
timing.download += r.download
|
||||
timing.phash += r.phash
|
||||
timing.upload += r.upload
|
||||
timing.embed += r.embed
|
||||
if r.err != nil {
|
||||
failed = true
|
||||
continue
|
||||
@ -623,17 +653,14 @@ func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix,
|
||||
continue
|
||||
}
|
||||
seen[r.sha1] = true
|
||||
dupID, isDup := s.dedupImage(ctx, kind, r.phashBits, timing)
|
||||
if cover == "" {
|
||||
cover = r.url
|
||||
}
|
||||
out = append(out, fetchedImage{
|
||||
url: r.url,
|
||||
key: r.key,
|
||||
phash: r.phashBits,
|
||||
dupID: dupID,
|
||||
isDup: isDup,
|
||||
size: r.size,
|
||||
url: r.url,
|
||||
key: r.key,
|
||||
size: r.size,
|
||||
vec: r.vec,
|
||||
})
|
||||
}
|
||||
timing.images = len(out)
|
||||
@ -641,6 +668,41 @@ func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix,
|
||||
return cover, out, keys, failed, timing
|
||||
}
|
||||
|
||||
// precomputeSimilar 写完向量后顺手预计算「疑似同款」近邻并落库。
|
||||
//
|
||||
// 为什么放在这里：原先是在**打开图集页时**对每张图实时做向量近邻检索，一个 116 张图的
|
||||
// 图集要发 116 次查询、每次约 20KB SQL；本机 PostgreSQL 跑在 WSL2 里、Windows 经 wslrelay
|
||||
// 访问，该链路大 payload 吞吐实测约 0.47MB/s，合计 5 秒。挪到入库时算一次（每个图集只付一次），
|
||||
// 页面就退化成一次小查询。
|
||||
//
|
||||
// 失败只记日志：预计算是增值数据，绝不能让「算不出同款」影响入库结果。
|
||||
// 注意：这里统计的是本次新入库的图片；它们与**其他图集存量图**的配对由预计算内部双向写入，
|
||||
// 因此对方图集的页面也能立刻看到提示，无需重算对方。
|
||||
func (s *IngestService) precomputeSimilar(ctx context.Context, kind string, imageIDs []uint32) {
|
||||
if s.similar == nil || len(imageIDs) == 0 {
|
||||
return
|
||||
}
|
||||
start := time.Now()
|
||||
if err := s.similar.PrecomputeForGallery(ctx, kind, imageIDs); err != nil {
|
||||
log.Printf("[ingest] 「疑似同款」预计算失败 kind=%s 图=%d err=%v（不影响入库）", kind, len(imageIDs), err)
|
||||
return
|
||||
}
|
||||
log.Printf("[ingest] 「疑似同款」预计算耗时 kind=%s 图=%d 耗时=%v", kind, len(imageIDs), time.Since(start).Round(time.Millisecond))
|
||||
}
|
||||
|
||||
// saveEmbeddings 落库成功后写入一批图片的语义向量（仅后台「疑似同款」提示用）。
|
||||
//
|
||||
// 向量是增值数据、不是入库的必要条件：仓储未注入（未启用视觉服务）、无向量、
|
||||
// 或写入失败，都只记日志并返回，绝不影响入库结果。ids[i] 与 vecs[i] 一一对应。
|
||||
func (s *IngestService) saveEmbeddings(ctx context.Context, kind string, ids []uint32, vecs [][]float32) {
|
||||
if s.embedRep == nil || len(ids) == 0 || len(vecs) == 0 {
|
||||
return
|
||||
}
|
||||
if err := s.embedRep.BulkUpsertEmbedding(ctx, kind, ids, vecs); err != nil {
|
||||
log.Printf("[ingest] 写向量失败 kind=%s 数量=%d err=%v（不影响入库）", kind, len(ids), err)
|
||||
}
|
||||
}
|
||||
|
||||
// cleanupUploads 删除一批本批次成功上传的对象（S4 + 本地兜底），用于任务失败回滚：
|
||||
// 「单图失败=整任务失败」时，前面已成功上传的图若放任不管就成了孤儿，永远堆在存储里。
|
||||
// 对两种存储都尝试删除（任一不存在即按幂等成功处理）；删除失败仅告警，不阻断任务置失败。
|
||||
@ -660,14 +722,12 @@ func (s *IngestService) cleanupUploads(ctx context.Context, keys []string) {
|
||||
}
|
||||
}
|
||||
|
||||
// fetchedImage 是单张已下载图片的暂存结构，携带去重所需的指纹信息。
|
||||
// fetchedImage 是单张已下载图片的暂存结构。
|
||||
type fetchedImage struct {
|
||||
url string
|
||||
key string
|
||||
phash string // dHash 的 pgvector 二进制向量串，空串表示无法解码
|
||||
dupID uint32 // 命中近重复时的参考图 id
|
||||
isDup uint8 // 是否标记为近重复（供审核留痕）
|
||||
size int64 // 原图源文件字节数，随结果带回供落库写入 image_size
|
||||
url string
|
||||
key string
|
||||
size int64 // 原图源文件字节数，随结果带回供落库写入 image_size
|
||||
vec []float32 // FashionSigLIP 向量（nil = 未产出），供落库后写入 image_embeddings
|
||||
}
|
||||
|
||||
// slowImageThreshold 单张图各阶段合计超过该阈值时单独打一条告警，便于从大量图里定位异常慢图。
|
||||
@ -675,90 +735,69 @@ const slowImageThreshold = 2 * time.Second
|
||||
|
||||
// imageFetchConcurrency 单条入库任务内「下载 + 上传」的并发度。
|
||||
//
|
||||
// 实测结论（theimpression.com，59 张图）：
|
||||
// - 并发 1：单张下载 ~3.7s，整批 59 张约 3m54s，任务成功。
|
||||
// - 并发 5：单张下载涨到 ~23s（约 6 倍），5m24s 仅完成 12/59 张便因 30s 超时失败。
|
||||
// 默认 1（等效串行），可用环境变量 INGEST_IMAGE_CONCURRENCY（1~16）覆盖。
|
||||
//
|
||||
// 即源站对同 IP 的并发连接有强限流：并发不仅不提速，反而把吞吐从 ~0.27 张/秒
|
||||
// 打到 ~0.04 张/秒并直接拖垮任务。因此默认取 1（等效串行）。
|
||||
// 换到不限流的源站时，可把这个值调大（机制已就绪），但务必先做小批量实测。
|
||||
const imageFetchConcurrency = 1
|
||||
// 之所以默认 1：早期实测 theimpression.com（59 张图）显示源站对同 IP 并发有强限流——
|
||||
// 并发 1 时单张 ~3.7s / 整批 3m54s；并发 5 时单张涨到 ~23s，5m24s 只完成 12/59 张
|
||||
// 就因超时失败，吞吐从 ~0.27 张/秒掉到 ~0.04 张/秒。结论：并发会把任务拖垮，故取 1。
|
||||
//
|
||||
// 但该结论**只对 theimpression 成立，不能外推到所有源站**。2026-10-13 对 assets.vogue.com
|
||||
// 的冷图（未命中 CDN 边缘）实测：串行 1582ms/张，8 并发降到 661ms/张（2.4x），并发是有收益的。
|
||||
// 冷图慢是因为 CDN 要回源，属延迟受限而非带宽受限——正好是并发能掩盖的那一类。
|
||||
//
|
||||
// 因此把它做成可配置项：换源站时先小批量实测再调（本文件顶部注释记录了两组实测数据）。
|
||||
var imageFetchConcurrency = func() int {
|
||||
v := strings.TrimSpace(os.Getenv("INGEST_IMAGE_CONCURRENCY"))
|
||||
if v == "" {
|
||||
return 1
|
||||
}
|
||||
n, err := strconv.Atoi(v)
|
||||
if err != nil || n < 1 || n > 16 {
|
||||
log.Printf("[ingest] INGEST_IMAGE_CONCURRENCY=%q 非法（应为 1~16 的整数），回退并发 1", v)
|
||||
return 1
|
||||
}
|
||||
return n
|
||||
}()
|
||||
|
||||
// fetchTiming 汇总一次图集拉取各阶段的累计耗时，用于定位「慢在哪一步」。
|
||||
// 下载 / 上传是网络 IO，phash 是 CPU（解码 + 哈希），近重复是 DB（HNSW 检索）——
|
||||
// 三类瓶颈成因不同，分开计时才能对症下药。
|
||||
// 下载与上传是网络 IO，embed 是视觉服务调用（HTTP + 模型推理），分开计时才能对症下药。
|
||||
type fetchTiming struct {
|
||||
images int // 成功入行的图片数
|
||||
total time.Duration // 整个拉取过程的墙钟耗时
|
||||
download time.Duration // 各张 HTTP 下载耗时之和
|
||||
phash time.Duration // 各张 dHash 指纹计算耗时之和
|
||||
upload time.Duration // 各张上传存储耗时之和
|
||||
dedup time.Duration // 近重复查询累计
|
||||
embed time.Duration // 各张视觉服务编码耗时之和
|
||||
}
|
||||
|
||||
// String 输出各阶段耗时（毫秒精度），一条日志看清瓶颈。
|
||||
// 注意：下载 / phash / 上传是「各张图耗时之和」，并发执行下其总和会大于墙钟总耗时，
|
||||
// 注意：下载 / 上传 / embed 是「各张图耗时之和」，并发执行下其总和会大于墙钟总耗时，
|
||||
// 因此三者与「总耗时」的比值不再等于时间占比——它们用于横向对比哪个阶段最重。
|
||||
func (t *fetchTiming) String() string {
|
||||
if t == nil {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("图片=%d 并发=%d 下载=%v phash=%v 上传=%v 近重复=%v（下载/phash/上传为各张累计值）",
|
||||
return fmt.Sprintf("图片=%d 并发=%d 下载=%v 上传=%v embed=%v（各张累计值）",
|
||||
t.images,
|
||||
imageFetchConcurrency,
|
||||
t.download.Round(time.Millisecond),
|
||||
t.phash.Round(time.Millisecond),
|
||||
t.upload.Round(time.Millisecond),
|
||||
t.dedup.Round(time.Millisecond),
|
||||
t.embed.Round(time.Millisecond),
|
||||
)
|
||||
}
|
||||
|
||||
// sqlNull 把 phash 字符串转成可空向量字段：空串 → NULL（不参与近邻检索）。
|
||||
func sqlNull(ph string) sql.NullString {
|
||||
return sql.NullString{String: ph, Valid: ph != ""}
|
||||
}
|
||||
|
||||
// dedupImage 判断单张图是否近重复（dHash 汉明距离 ≤ 阈值）：命中则仍入库，但标记 is_duplicate + dup_of。
|
||||
func (s *IngestService) dedupImage(ctx context.Context, kind, phashBits string, timing *fetchTiming) (dupID uint32, isDup uint8) {
|
||||
// repo 未注入（如离线单测）时跳过去重，不阻断主流程。生产环境 repo 必不为空。
|
||||
if s.repo == nil {
|
||||
return 0, 0
|
||||
}
|
||||
var tables []string
|
||||
switch kind {
|
||||
case dto.IngestKindRunway:
|
||||
tables = []string{"brand_runway_images"}
|
||||
case dto.IngestKindStreet:
|
||||
tables = []string{"street_snap_images"}
|
||||
default:
|
||||
return 0, 0
|
||||
}
|
||||
if phashBits != "" {
|
||||
start := time.Now()
|
||||
id, found, err := s.repo.FindNearDuplicateImage(ctx, tables, phashBits, phash.DefaultThreshold)
|
||||
if timing != nil {
|
||||
timing.dedup += time.Since(start)
|
||||
}
|
||||
if err == nil && found {
|
||||
return id, 1
|
||||
}
|
||||
}
|
||||
return 0, 0
|
||||
}
|
||||
|
||||
// downloadResult 是单张图「下载 → phash → 上传」的结果，供并发执行后按下标串行组装。
|
||||
// downloadResult 是单张图「下载 → 上传」的结果，供并发执行后按下标串行组装。
|
||||
// 各阶段耗时随结果一并带回、由调用方在串行段汇总（而非直接累加到共享 timing），
|
||||
// 这样并发路径上无需加锁即可保证计时准确。
|
||||
type downloadResult struct {
|
||||
url string // 存储访问地址（失败为空）
|
||||
key string // 对象 key（内容寻址）
|
||||
sha1 string // 内容 sha1，用于批次内去重
|
||||
phashBits string // dHash 向量串，空串表示无法解码
|
||||
download time.Duration // 本张下载耗时
|
||||
phash time.Duration // 本张指纹计算耗时
|
||||
upload time.Duration // 本张上传耗时
|
||||
size int64 // 原图源文件字节数（io.ReadAll 读到的下载响应体长度）
|
||||
err error // 下载或上传失败原因（非空即该张失败）
|
||||
url string // 存储访问地址（失败为空）
|
||||
key string // 对象 key（内容寻址）
|
||||
sha1 string // 内容 sha1，用于批次内去重
|
||||
download time.Duration // 本张下载耗时
|
||||
upload time.Duration // 本张上传耗时
|
||||
size int64 // 原图源文件字节数（io.ReadAll 读到的下载响应体长度）
|
||||
vec []float32 // FashionSigLIP 向量（nil = 未产出：服务未启用/不可用/失败）
|
||||
embed time.Duration // 本张向量编码耗时
|
||||
err error // 下载或上传失败原因（非空即该张失败）
|
||||
}
|
||||
|
||||
// concurrentFetch 以 imageFetchConcurrency 路并发执行 fn(ctx, i)（i ∈ [0,n)），返回长度 n、
|
||||
@ -836,10 +875,19 @@ func (s *IngestService) downloadAndUpload(ctx context.Context, u, prefix string)
|
||||
res.key = fmt.Sprintf("%s/%s%s", prefix, res.sha1, ext)
|
||||
ct := resp.Header.Get("Content-Type")
|
||||
|
||||
// ② dHash 指纹（仅用于近重复检索；解码失败返回空串 → NULL，不参与检索）。
|
||||
phStart := time.Now()
|
||||
res.phashBits = phash.ToVectorBits(phash.Of(data))
|
||||
res.phash = time.Since(phStart)
|
||||
// ② 视觉服务编码（可选）：产出后台「疑似同款」用的语义向量。
|
||||
// 未启用 / 不可用 / 超时 / 维度不符一律留空 vec —— 绝不阻断入库，也不在此重试
|
||||
// （重试会拖慢整批；熔断由客户端统一处理）。
|
||||
if s.vision != nil && s.vision.Enabled() {
|
||||
embedStart := time.Now()
|
||||
vecs, _, eerr := s.vision.Embed(ctx, [][]byte{data})
|
||||
res.embed = time.Since(embedStart)
|
||||
if eerr == nil && len(vecs) > 0 {
|
||||
res.vec = vecs[0]
|
||||
} else if eerr != nil {
|
||||
log.Printf("[ingest] embed 失败 url=%s err=%v（不写向量，继续入库）", u, eerr)
|
||||
}
|
||||
}
|
||||
|
||||
// ③ 上传（主存储 S4，失败兜底本地）
|
||||
upStart := time.Now()
|
||||
@ -858,10 +906,11 @@ func (s *IngestService) downloadAndUpload(ctx context.Context, u, prefix string)
|
||||
res.upload = time.Since(upStart)
|
||||
|
||||
// 单图异常慢（合计超阈值）单独告警：便于从几十上百张图里一眼锁定是哪张、卡在哪一段。
|
||||
if total := res.download + res.phash + res.upload; total > slowImageThreshold {
|
||||
log.Printf("[ingest] 慢图 url=%s 合计=%v 下载=%v phash=%v 上传=%v",
|
||||
if total := res.download + res.embed + res.upload; total > slowImageThreshold {
|
||||
log.Printf("[ingest] 慢图 url=%s 合计=%v 下载=%v embed=%v 上传=%v",
|
||||
u, total.Round(time.Millisecond),
|
||||
res.download.Round(time.Millisecond), res.phash.Round(time.Millisecond), res.upload.Round(time.Millisecond))
|
||||
res.download.Round(time.Millisecond), res.embed.Round(time.Millisecond),
|
||||
res.upload.Round(time.Millisecond))
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user