toom1996/backend_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
toom1996
2026-09-17 00:38:01 +08:00
parent 7dd3fda73c
commit 6c31b18e0f
10 changed files with 389 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
internal/service/ingest_service.go
View File

@ -15,6 +15,7 @@ import (
"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"
@ -234,17 +235,24 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
// 4) 下载图片并上传到存储（结构化 Looks 优先：主图+细节图分组；否则回退 Images 全部视为主图）。
// 内容哈希（sha1）key 保证重爬不产生孤儿文件：失败回滚删本批 key 即可。
// 同时算出每张图的 pHash（downloadOne 内基于原始字节），供后续全局近似去重标记。
var cover string
var draftImages []model.BrandRunwayDraftImage
var keys []string
var imgFailed bool
var imageCount uint16
var phashList []uint64
if len(p.Looks) > 0 {
cover, draftImages, keys, imgFailed = s.fetchLookImages(ctx, p.Looks, "runway")
imageCount = uint16(len(p.Looks))
phashList = make([]uint64, len(draftImages))
for i, d := range draftImages {
phashList[i] = d.Phash
}
} else {
var imgs []string
cover, imgs, keys, imgFailed = s.fetchImages(ctx, p.Images, "runway")
var phs []uint64
cover, imgs, phs, keys, imgFailed = s.fetchImages(ctx, p.Images, "runway")
imageCount = uint16(len(imgs))
for i, img := range imgs {
draftImages = append(draftImages, model.BrandRunwayDraftImage{
@ -253,8 +261,10 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
SortOrder: uint32(i + 1),
LookIndex: uint32(i + 1),
IsDetail: 0,
Phash: phs[i],
})
}
phashList = phs
}
if imgFailed {
// 单图失败=整任务失败：先回滚本批已上传的图（S4 + 本地兜底），避免孤儿文件永远堆在存储里，
@ -264,6 +274,13 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
return
}
// 4.5) 全局近似去重标记：与「已晋升图片」比对汉明距离，命中则留痕（is_duplicate=1 + dup_of），
// 不丢弃、交后台人工裁决（只拦新增，不碰存量）。比对失败仅告警，不阻断入库。
s.tagDuplicates(ctx, phashList, func(i int, dupOf string) {
draftImages[i].IsDuplicate = 1
draftImages[i].DupOf = dupOf
})
// 5) 写草稿表（status=pending），等待后台审核通过后再晋升正式表
draft := &model.BrandRunwayDraft{
JobID: job.ID,
@ -309,7 +326,7 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
lookIdx := li + 1
if look.Main != "" {
order++
url, key, err := s.downloadOne(ctx, look.Main, order, prefix)
url, key, ph, err := s.downloadOne(ctx, look.Main, order, prefix)
if err != nil {
failed = true
} else {
@ -323,6 +340,7 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
SortOrder: uint32(order),
LookIndex: uint32(lookIdx),
IsDetail: 0,
Phash: ph,
})
}
}
@ -331,7 +349,7 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
continue
}
order++
url, key, err := s.downloadOne(ctx, d, order, prefix)
url, key, ph, err := s.downloadOne(ctx, d, order, prefix)
if err != nil {
failed = true
continue
@ -343,6 +361,7 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
SortOrder: uint32(order),
LookIndex: uint32(lookIdx),
IsDetail: 1,
Phash: ph,
})
}
}
@ -358,8 +377,8 @@ func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob,
return
}
// 2) 下载图片并上传到存储（七牛优先，失败兜底本地）
cover, imgs, keys, imgFailed := s.fetchImages(ctx, p.Images, "street")
// 2) 下载图片并上传到存储（七牛优先，失败兜底本地）；同时算 pHash 供近似去重。
cover, imgs, phs, keys, imgFailed := s.fetchImages(ctx, p.Images, "street")
if len(p.Images) > 0 && imgFailed {
// 单图失败=整任务失败：先回滚本批已上传的图（七牛 + 本地兜底），避免孤儿文件永远堆在存储里，
// 然后按指数退避自动重试，达上限才置 failed 等后台手动重试。
@ -390,8 +409,16 @@ func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob,
Image: img,
Name: fmt.Sprintf("Look %d", i+1),
SortOrder: uint32(i + 1),
Phash: phs[i],
})
}
// 3.5) 全局近似去重标记（与走秀同逻辑，只拦新增、留痕不删）。
s.tagDuplicates(ctx, phs, func(i int, dupOf string) {
rows[i].IsDuplicate = 1
rows[i].DupOf = dupOf
})
if err := s.repo.CreateStreetSnapDraftImages(ctx, rows); err != nil {
s.failOrRetry(ctx, job.ID, "create street draft images: "+err.Error())
return
@ -399,16 +426,18 @@ func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob,
_ = s.repo.MarkDone(ctx, job.ID)
}
// fetchImages 下载图片并上传到存储，返回 (cover 地址, 全部图片地址, 已成功上传对象的 key 列表, 是否有任意一张失败)。
// fetchImages 下载图片并上传到存储，返回 (cover 地址, 全部图片地址, 各图 pHash, 已成功上传对象的 key 列表, 是否有任意一张失败)。
// prefix 为七牛 key 前缀（runway/ 或 street/）。只要任意一张下载/上传失败，failed 即置 true，
// 调用方据此把整条任务判为失败（不再写草稿），并拿 keys 回滚本批已上传的对象，符合「单图失败=整任务失败」策略。
func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix string) (string, []string, []string, bool) {
// 返回的 pHash 与图片地址按索引对齐（phash=0 表示解码失败未计算，比对时跳过）。
func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix string) (string, []string, []uint64, []string, bool) {
cover := ""
out := make([]string, 0, len(urls))
phashes := make([]uint64, 0, len(urls))
keys := make([]string, 0, len(urls))
failed := false
for i, u := range urls {
url, key, err := s.downloadOne(ctx, u, i, prefix)
url, key, ph, err := s.downloadOne(ctx, u, i, prefix)
if err != nil {
failed = true
continue
@ -417,9 +446,10 @@ func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix s
cover = url
}
out = append(out, url)
phashes = append(phashes, ph)
keys = append(keys, key)
}
return cover, out, keys, failed
return cover, out, phashes, keys, failed
}
// cleanupUploads 删除一批本批次成功上传的对象（七牛 + 本地兜底），用于任务失败回滚：
@ -442,19 +472,20 @@ func (s *IngestService) cleanupUploads(ctx context.Context, keys []string) {
}
// downloadOne 把单张远程图下载后上传到存储（七牛优先，失败兜底本地），
// 返回可直接写入数据库的访问地址（七牛为完整 https URL，本地为相对 /uploads 路径）。
func (s *IngestService) downloadOne(ctx context.Context, u string, idx int, prefix string) (string, string, error) {
// 返回 (访问地址, 对象 key, 感知哈希, error)。访问地址可直接写入数据库
// （七牛为完整 https URL，本地为相对 /uploads 路径）；感知哈希基于原始字节算一次，供入库时近似去重。
func (s *IngestService) downloadOne(ctx context.Context, u string, idx int, prefix string) (string, string, uint64, error) {
resp, err := s.httpClient.Get(u)
if err != nil {
return "", "", err
return "", "", 0, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", "", fmt.Errorf("status %d", resp.StatusCode)
return "", "", 0, fmt.Errorf("status %d", resp.StatusCode)
}
data, err := io.ReadAll(resp.Body)
if err != nil {
return "", "", err
return "", "", 0, err
}
ext := path.Ext(u)
if ext == "" || len(ext) > 5 {
@ -467,16 +498,74 @@ func (s *IngestService) downloadOne(ctx context.Context, u string, idx int, pref
key := fmt.Sprintf("%s/%s%s", prefix, hash, ext)
ct := resp.Header.Get("Content-Type")
// 感知哈希：同一份字节在此算一次（decode 失败时 ph=0，比对时跳过，不阻断入库）。
ph := phash.Of(data)
if ph == 0 {
log.Printf("[warn] ingest pHash 未计算（解码失败或非 jpeg/png/gif 格式） url=%s", u)
}
// 主上传器（七牛）
if url, err := s.uploader.Upload(ctx, key, data, ct); err == nil {
return url, key, nil
return url, key, ph, nil
} else if s.local != nil {
// 兜底本地，避免图片完全丢失
if lurl, lerr := s.local.Upload(ctx, key, data, ct); lerr == nil {
return lurl, key, nil
return lurl, key, ph, nil
} else {
return "", "", err
return "", "", 0, err
}
}
return "", "", err
return "", "", 0, err
}
// tagDuplicates 对一批图片（phashList 与 apply 下标对齐）做全局近似去重标记：
// 与「已晋升图片」的 phash 库比汉明距离（≤ phash.DefaultThreshold 即近似重复），
// 命中则调用 apply(i, dupOf) 由调用方把第 i 张图标 IsDuplicate=1、DupOf=命中图 uid。
// 不丢弃任何图、仅留痕（交后台人工裁决），契合「只拦新增、全局跨所有图、标记不删」。
// 比对库拉取失败仅告警并跳过（不阻断入库）；phash=0 的图不比对、不误杀。
func (s *IngestService) tagDuplicates(ctx context.Context, phashList []uint64, apply func(i int, dupOf string)) {
uids, err := s.matchDuplicates(ctx, phashList)
if err != nil {
log.Printf("[warn] ingest 近似去重比对失败 err=%v（跳过标记，不阻断入库）", err)
return
}
for i, u := range uids {
if u != "" {
apply(i, u)
}
}
}
// matchDuplicates 返回与 phashList 等长的 dup_of uid 切片（""=未命中近似重复）。
// 对每张新图，在已晋升图片库里找汉明距离 ≤ 阈值的命中，取距离最近者，
// 按其 kind（runway/street）编码成对应 hashid 类型作为 dup_of。
func (s *IngestService) matchDuplicates(ctx context.Context, phashList []uint64) ([]string, error) {
refs, err := s.repo.ListImagePHashes(ctx)
if err != nil {
return nil, err
}
out := make([]string, len(phashList))
for i, ph := range phashList {
if ph == 0 {
continue // 未计算 phash 的图不比对（避免误杀）
}
bestUID := ""
bestDist := phash.DefaultThreshold + 1
for _, ref := range refs {
if ref.Phash == 0 {
continue
}
d := phash.Hamming(ph, ref.Phash)
if d <= phash.DefaultThreshold && d < bestDist {
var typ byte = hashid.TypeRunwayImage
if ref.Kind == "street" {
typ = hashid.TypeSnapImage
}
bestUID = hashid.EncodeWithType(ref.ID, typ)
bestDist = d
}
}
out[i] = bestUID
}
return out, nil
}