feat: 抓取任务队列 + 媒体机直传 + 向量回填

crawl(品牌抓取任务队列)
- /admin/internal/crawl/tasks 领取 / :id/result 回报,spider 可常驻拉任务
- 后台品牌列表支持「加入抓取队列」,队列页可看状态与重试
- 租约超时的 running 自动回收为 queued,避免 worker 崩溃后任务卡死

media(媒体机直传)
- ingest 载荷新增 main_key/detail_keys/main_size/detail_sizes:
  有 key 就直接落库,跳过下载与上传;与 URL 形态可逐图混用
- 细节图按 Details/DetailKeys 两者较长者遍历(只给 key 时不再丢图)
- 媒体机上传的 key 不进失败回滚清单(内容寻址、可能与他人共享)
- /admin/internal/upload-url 签发 S4 PUT 预签名(spider 不持有 AK/SK)

embeddings(向量回填)
- GET /admin/internal/embeddings/pending 派发缺向量的图(含可直接下载的 url)
- POST /admin/internal/embeddings 回写向量并重算近邻;失败项计入
  image_embedding_failures,连续失败到上限后退出队列(坏图不占位)
- 近邻重算改为异步串行:同步会让 100 张的批次 POST 超时(实测 >60s)
- 新增 key 分支单测:混用 key/url、细节图只给 key、字节数、回滚清单、向量占位
This commit is contained in:
toom1996
2026-10-25 00:37:13 +08:00
parent e18c5a0815
commit 151e09ab30
22 changed files with 1535 additions and 28 deletions

View File

@ -13,6 +13,8 @@ import (
"os" "os"
"os/signal" "os/signal"
"path/filepath" "path/filepath"
"strconv"
"strings"
"syscall" "syscall"
"time" "time"
@ -217,6 +219,12 @@ func main() {
// 后台页面因此只需读一次小查询(原先是页面内对每张图实时检索,一个 116 张图的图集要 5 秒)。 // 后台页面因此只需读一次小查询(原先是页面内对每张图实时检索,一个 116 张图的图集要 5 秒)。
ingestSvc.SetSimilar(similarSvc) ingestSvc.SetSimilar(similarSvc)
backstageHandler := handler.NewBackstageHandler(authSvc, brandSvc, articleSvc, snapSvc, reviewSvc, ingestSvc, adminRepo, translateSvc, similarSvc) backstageHandler := handler.NewBackstageHandler(authSvc, brandSvc, articleSvc, snapSvc, reviewSvc, ingestSvc, adminRepo, translateSvc, similarSvc)
// 品牌抓取任务(crawl_tasks):后台决定爬哪些品牌,spider 服务(可跨机)领取执行。
backstageHandler.SetCrawl(repository.NewCrawlRepository(db))
// 向量回填接口的仓储(媒体机直传后,向量只能由外部 worker 回传)。
backstageHandler.SetEmbeddingRepo(embedRepo)
// S4 上传器:给媒体机签发直传 URL(密钥只留在后端)。
backstageHandler.SetUploader(s4Up)
// 手工建稿:复用入库仓储的写入方法(Create*WithImages,事务内完成)与同一套存储上传 // 手工建稿:复用入库仓储的写入方法(Create*WithImages,事务内完成)与同一套存储上传
// (imgUp = S4/本地主上传器,sha1 内容寻址),保证与爬虫入库产出的 key / 行结构一致。 // (imgUp = S4/本地主上传器,sha1 内容寻址),保证与爬虫入库产出的 key / 行结构一致。
backstageHandler.SetManual(service.NewManualArticleService(ingestRepo, brandRepo, imgUp, localUp, visionClient, embedRepo)) backstageHandler.SetManual(service.NewManualArticleService(ingestRepo, brandRepo, imgUp, localUp, visionClient, embedRepo))
@ -275,8 +283,18 @@ func main() {
// 7.6 启动爬虫入库 worker(异步处理 ingest_jobs 队列)。 // 7.6 启动爬虫入库 worker(异步处理 ingest_jobs 队列)。
// 用独立可取消 ctx,优雅关闭时随主流程一起退出。 // 用独立可取消 ctx,优雅关闭时随主流程一起退出。
workerCtx, workerCancel := context.WithCancel(context.Background()) workerCtx, workerCancel := context.WithCancel(context.Background())
go ingestSvc.Run(workerCtx, 2, 3*time.Second) // 并发数可配:批量重爬时单进程 2 个 worker 是吞吐瓶颈(配合 INGEST_IMAGE_CONCURRENCY 决定总并发)。
slog.Info("入库 worker 已启动", "queue", "ingest_jobs") // 默认仍是 2,行为与之前一致;1~16 之外的值忽略并回落默认。
ingestWorkers := 2
if v := strings.TrimSpace(os.Getenv("INGEST_WORKERS")); v != "" {
if n, err := strconv.Atoi(v); err == nil && n >= 1 && n <= 16 {
ingestWorkers = n
} else {
slog.Warn("INGEST_WORKERS 非法(应为 1~16 的整数),回退默认 2", "value", v)
}
}
go ingestSvc.Run(workerCtx, ingestWorkers, 3*time.Second)
slog.Info("入库 worker 已启动", "queue", "ingest_jobs", "workers", ingestWorkers)
srvBackstage := &http.Server{ srvBackstage := &http.Server{
Addr: ":" + cfg.Server.BackstagePort, Addr: ":" + cfg.Server.BackstagePort,
Handler: backstageEngine, Handler: backstageEngine,

View File

@ -0,0 +1,23 @@
-- 品牌抓取任务表:后台表达「要爬哪些品牌」,spider 服务来领取执行。
-- 与 ingest_jobs 分层:crawl_tasks 对外(跨机器),ingest_jobs 对内(后端 worker)。
CREATE TABLE IF NOT EXISTS crawl_tasks (
id bigserial PRIMARY KEY,
brand_id bigint NOT NULL,
status varchar(16) NOT NULL DEFAULT 'queued',
priority smallint NOT NULL DEFAULT 0,
max_shows smallint NOT NULL DEFAULT 0,
worker varchar(64) NOT NULL DEFAULT '',
locked_at bigint NOT NULL DEFAULT 0,
attempts smallint NOT NULL DEFAULT 0,
shows_submitted integer NOT NULL DEFAULT 0,
last_error text NOT NULL DEFAULT '',
created_at bigint NOT NULL DEFAULT 0,
updated_at bigint NOT NULL DEFAULT 0
);
-- 一个品牌同时只允许一条任务:后台反复点「加入队列」不会产生重复任务,
-- 而是把已完成/失败的记录重置为 queued。
CREATE UNIQUE INDEX IF NOT EXISTS uq_crawl_tasks_brand ON crawl_tasks (brand_id);
-- 领取用索引:按优先级 + 入队顺序。
CREATE INDEX IF NOT EXISTS idx_crawl_tasks_pick ON crawl_tasks (status, priority DESC, id);

View File

@ -0,0 +1,16 @@
-- 向量回填的失败计数表(媒体拆分后新增)。
--
-- 背景:媒体机直传后,后端拿到的是 key 而不是图片字节,向量只能由「回填 worker」
-- (有 GPU 的机器)从存储拉图、编码、再回传。若某张图永远编不出向量
-- (文件损坏、格式不支持、被删),没有计数就会每次回填都重试,把队列堵死。
--
-- 规则:同一 (kind, image_id) 连续失败 attempts 次后不再出现在待回填队列里;
-- 回填成功时删除该行(这张图已经不需要再试了)。
CREATE TABLE IF NOT EXISTS image_embedding_failures (
kind varchar(16) NOT NULL,
image_id bigint NOT NULL,
attempts smallint NOT NULL DEFAULT 0,
last_error text NOT NULL DEFAULT '',
updated_at bigint NOT NULL DEFAULT 0,
PRIMARY KEY (kind, image_id)
);

View File

@ -16,6 +16,11 @@ type BrandQuery struct {
// Featured 仅保留"代表品牌"精选集合(按指标排名的前 N 个)。 // Featured 仅保留"代表品牌"精选集合(按指标排名的前 N 个)。
Featured bool Featured bool
Metric string // images(默认)| shows Metric string // images(默认)| shows
// Sort 后台列表排序:空 / "name"(默认,按英文名);"updated" 按「最近有动静」倒序
// (取该品牌下走秀图集的最大 updated_at 与抓取任务的最大 updated_at,两者取大)。
// 公开侧忽略此字段,避免改变前台 A-Z 索引的既定顺序。
Sort string
Limit int // 精选集合大小,默认 200 Limit int // 精选集合大小,默认 200
// Locale 语言:cn | en。空时由 service 回落默认 en。 // Locale 语言:cn | en。空时由 service 回落默认 en。

View File

@ -37,7 +37,22 @@ type RunwayIngest struct {
// RunwayLook 一场秀中的一个 look:1 张主图 + 0..N 张细节图。 // RunwayLook 一场秀中的一个 look:1 张主图 + 0..N 张细节图。
// 主图即对外默认展示的走秀图;细节图挂靠到对应主图,由「细节入口」按需展开。 // 主图即对外默认展示的走秀图;细节图挂靠到对应主图,由「细节入口」按需展开。
//
// 图片有两种来源,二选一(同一条 payload 内可混用):
//
// ① 旧路径:Main / Details 给原始 URL,worker 自己下载 + 上传存储(默认行为);
// ② 新路径(媒体机直传):MainKey / DetailKeys 给「已上传到 S4 的对象 key」,
// worker 直接落库,不再下载、不再上传。
//
// 新路径的意义:抓取与入库分离后,下载/上传这类带宽密集的活由部署在快网络里的
// spider 完成,后端只做业务写库,吞吐不再被后端机器的出口带宽卡住。
// key 由 spider 按内容 sha1 生成(runway/<sha1>.jpg),因此重复上传天然幂等。
type RunwayLook struct { type RunwayLook struct {
Main string `json:"main"` // 主图(look)原始 URL Main string `json:"main"` // 主图(look)原始 URL
Details []string `json:"details"` // 细节图原始 URL 列表 Details []string `json:"details"` // 细节图原始 URL 列表
MainKey string `json:"main_key,omitempty"` // 主图已上传的存储 key(新路径)
DetailKeys []string `json:"detail_keys,omitempty"` // 细节图已上传的存储 key(新路径)
MainSize int64 `json:"main_size,omitempty"` // 主图字节数(新路径,0=未知)
DetailSizes []int64 `json:"detail_sizes,omitempty"` // 细节图字节数(与 DetailKeys 对齐,0=未知)
} }

View File

@ -0,0 +1,347 @@
package handler
import (
"fmt"
"net/http"
"strings"
"time"
"github.com/gin-gonic/gin"
"fashionapi/internal/model"
"fashionapi/internal/pkg/hashid"
"fashionapi/internal/pkg/pagination"
)
// 本文件收口「品牌抓取任务」的控制面:
// - 后台页面:/admin/crawl(列表 + 批量入队 + 取消/重试/删除)
// - 品牌页行内动作:/admin/crawl/brand/:uid(把单个品牌加入队列)
// - spider 服务接口(HMAC 验签,不走 cookie):
// GET /admin/internal/crawl/tasks?worker=xxx → 领取一个任务
// POST /admin/internal/crawl/tasks/:id/result → 回报执行结果
//
// 与入库任务(ingest_jobs)的分工见 model.CrawlTask 的注释。
// crawlRowView 列表展示结构(时间已格式化为字符串,模板里没有日期函数)。
type crawlRowView struct {
ID uint32
BrandUID string
BrandName string
Status string
StatusLabel string
MaxShows int16
Shows int32
Attempts int16
Worker string
UpdatedAt string
LastError string
CanRetry bool
CanCancel bool
}
// crawlStatusClass 把抓取状态映射到后台已有的徽标配色类(admin.css 只有
// pending / processing / done / failed 四套配色,这里做一次语义映射)。
func crawlStatusClass(s string) string {
switch s {
case model.CrawlQueued:
return "pending"
case model.CrawlRunning:
return "processing"
case model.CrawlDone:
return "done"
case model.CrawlFailed:
return "failed"
}
return ""
}
func crawlStatusLabel(s string) string {
switch s {
case model.CrawlQueued:
return "待领取"
case model.CrawlRunning:
return "执行中"
case model.CrawlDone:
return "已完成"
case model.CrawlFailed:
return "失败"
case model.CrawlCanceled:
return "已取消"
}
return s
}
// CrawlList 渲染抓取任务列表页。
func (h *BackstageHandler) CrawlList(c *gin.Context) {
if h.crawl == nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": "抓取任务未装配(crawl repository 为空)"})
return
}
p := pagination.Parse(c, "page", "pageSize", 1, 30, 200)
rows, total, err := h.crawl.List(c.Request.Context(), p.Offset(), p.PageSize)
if err != nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": err.Error()})
return
}
views := make([]crawlRowView, 0, len(rows))
for _, r := range rows {
ua := "-"
if r.UpdatedAt > 0 {
ua = time.Unix(int64(r.UpdatedAt), 0).Format("2006-01-02 15:04")
}
views = append(views, crawlRowView{
ID: r.ID,
BrandUID: hashid.Encode(r.BrandID),
BrandName: r.BrandName,
Status: r.Status,
StatusLabel: crawlStatusLabel(r.Status),
MaxShows: r.MaxShows,
Shows: r.ShowsSubmitted,
Attempts: r.Attempts,
Worker: r.Worker,
UpdatedAt: ua,
LastError: r.LastError,
CanRetry: r.Status == model.CrawlFailed || r.Status == model.CrawlDone || r.Status == model.CrawlCanceled,
CanCancel: r.Status == model.CrawlQueued || r.Status == model.CrawlRunning,
})
}
counts, _ := h.crawl.Counts(c.Request.Context())
meta := p.NewMeta(total)
renderPage(c, http.StatusOK, "crawl", gin.H{
"Rows": views,
"Total": total,
"Pager": pagination.Pager{
PrevURL: pagination.PageLinkWith(c, meta.PrevPage, "pageSize", p.PageSize),
NextURL: pagination.PageLinkWith(c, meta.NextPage, "pageSize", p.PageSize),
Label: fmt.Sprintf("第 %d / %d 页", meta.Page, meta.TotalPages),
HasPrev: meta.HasPrev,
HasNext: meta.HasNext,
},
"Queued": counts[model.CrawlQueued],
"Running": counts[model.CrawlRunning],
"Done": counts[model.CrawlDone],
"Failed": counts[model.CrawlFailed],
// 批量入队后回显「本次加入了多少个」
"Added": atoiDefault(c.Query("added"), 0),
})
}
// CrawlEnqueue 把单个品牌加入抓取队列(品牌列表页的行内按钮)。
// 品牌 id 用 hashid(与品牌页其它动作一致),避免暴露自增主键。
func (h *BackstageHandler) CrawlEnqueue(c *gin.Context) {
if h.crawl == nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": "抓取任务未装配"})
return
}
brandID, err := hashid.Decode(c.Param("uid"))
if err != nil {
renderPage(c, http.StatusBadRequest, "error", gin.H{"Error": "品牌 ID 非法"})
return
}
maxShows := atoiDefault(c.PostForm("maxShows"), 0)
if _, err := h.crawl.Enqueue(c.Request.Context(), brandID, maxShows, 10); err != nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": err.Error()})
return
}
c.Redirect(http.StatusFound, c.PostForm("back"))
}
// CrawlBulk 把「品牌管理页勾选的品牌」批量加入抓取队列。
//
// 这是批量操作的唯一入口(原先的「按品牌名前缀批量」已下线):
// 勾选哪几个就加哪几个,作用域明确,不会因为前缀写错而误抓几千个品牌。
func (h *BackstageHandler) CrawlBulk(c *gin.Context) {
if h.crawl == nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": "抓取任务未装配"})
return
}
uids := c.PostFormArray("uid")
if len(uids) == 0 {
renderPage(c, http.StatusBadRequest, "error", gin.H{"Error": "没有勾选任何品牌"})
return
}
maxShows := atoiDefault(c.PostForm("maxShows"), 3)
added := 0
for _, uid := range uids {
id, derr := hashid.Decode(strings.TrimSpace(uid))
if derr != nil {
continue
}
ok, err := h.crawl.Enqueue(c.Request.Context(), id, maxShows, 5)
if err != nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": err.Error()})
return
}
if ok {
added++
}
}
c.Redirect(http.StatusFound, fmt.Sprintf("/admin/brands?added=%d", added))
}
// CrawlRetry 把任务重置为待领取(失败 / 取消 / 已完成后想再跑一遍)。
func (h *BackstageHandler) CrawlRetry(c *gin.Context) {
id := atoiDefault(c.Param("id"), 0)
if id <= 0 || h.crawl == nil {
renderPage(c, http.StatusBadRequest, "error", gin.H{"Error": "任务 ID 非法"})
return
}
if err := h.crawl.SetStatus(c.Request.Context(), uint32(id), model.CrawlQueued); err != nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": err.Error()})
return
}
c.Redirect(http.StatusFound, "/admin/crawl")
}
// CrawlCancel 取消任务(执行中的任务无法真正打断 spider,只保证不再被领取)。
func (h *BackstageHandler) CrawlCancel(c *gin.Context) {
id := atoiDefault(c.Param("id"), 0)
if id <= 0 || h.crawl == nil {
renderPage(c, http.StatusBadRequest, "error", gin.H{"Error": "任务 ID 非法"})
return
}
if err := h.crawl.SetStatus(c.Request.Context(), uint32(id), model.CrawlCanceled); err != nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": err.Error()})
return
}
c.Redirect(http.StatusFound, "/admin/crawl")
}
// CrawlDelete 删除任务记录(已入库的数据不受影响)。
func (h *BackstageHandler) CrawlDelete(c *gin.Context) {
id := atoiDefault(c.Param("id"), 0)
if id <= 0 || h.crawl == nil {
renderPage(c, http.StatusBadRequest, "error", gin.H{"Error": "任务 ID 非法"})
return
}
if err := h.crawl.Delete(c.Request.Context(), uint32(id)); err != nil {
renderPage(c, http.StatusInternalServerError, "error", gin.H{"Error": err.Error()})
return
}
c.Redirect(http.StatusFound, "/admin/crawl")
}
// ---------------------------------------------------------------------------
// spider 服务接口(HMAC 验签,由 router 的 /admin/internal 组挂载)
// ---------------------------------------------------------------------------
// crawlTaskPayload 领取接口返回给 spider 的最小信息:品牌 uid(直接用于上报)+ 英文名(拼 Vogue URL)+ 场次上限。
type crawlTaskPayload struct {
ID uint32 `json:"id"`
BrandUID string `json:"brand_uid"`
BrandName string `json:"brand_name"`
MaxShows int16 `json:"max_shows"`
}
// WorkerTask 领取一个抓取任务;无任务时返回 {"task": null}(spider 据此退避轮询)。
func (h *BackstageHandler) WorkerTask(c *gin.Context) {
if h.crawl == nil {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "crawl repository not configured"})
return
}
worker := c.Query("worker")
if worker == "" {
worker = "spider"
}
row, err := h.crawl.Claim(c.Request.Context(), worker)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "claim failed"})
return
}
if row == nil {
c.JSON(http.StatusOK, gin.H{"task": nil})
return
}
c.JSON(http.StatusOK, gin.H{"task": crawlTaskPayload{
ID: row.ID,
BrandUID: hashid.Encode(row.BrandID),
BrandName: row.BrandName,
MaxShows: row.MaxShows,
}})
}
// WorkerUploadURL 给媒体机签发 S4 直传 URL(PUT 预签名)。
//
// 设计取舍:不在 spider 里配 AK/SK、也不引入 S3 SDK —— 密钥只留在后端,
// 媒体机拿到有时效的 URL 直接 PUT 即可。上传成功后把 key 一起上送 ingest。
// 请求:{"items":[{"key":"runway/ab12.jpg","content_type":"image/jpeg"}]}
// 响应:{"items":[{"key":"...","url":"https://..."}],"expires_in":900}
func (h *BackstageHandler) WorkerUploadURL(c *gin.Context) {
if h.uploader == nil || !h.uploader.Enabled() {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "S4 未启用,无法签发直传 URL"})
return
}
var body struct {
Items []struct {
Key string `json:"key"`
ContentType string `json:"content_type"`
} `json:"items"`
}
if err := c.ShouldBindJSON(&body); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid json: " + err.Error()})
return
}
if len(body.Items) == 0 || len(body.Items) > 200 {
c.JSON(http.StatusBadRequest, gin.H{"error": "items 数量应为 1~200"})
return
}
const ttl = 15 * time.Minute
out := make([]gin.H, 0, len(body.Items))
for _, it := range body.Items {
// key 由调用方生成(内容 sha1),这里只做前缀与路径穿越校验,避免被拿来写任意对象。
if !strings.HasPrefix(it.Key, "runway/") && !strings.HasPrefix(it.Key, "street/") {
c.JSON(http.StatusBadRequest, gin.H{"error": "key 必须以 runway/ 或 street/ 开头: " + it.Key})
return
}
if strings.Contains(it.Key, "..") {
c.JSON(http.StatusBadRequest, gin.H{"error": "key 非法: " + it.Key})
return
}
u, err := h.uploader.PresignPut(c.Request.Context(), it.Key, it.ContentType, ttl)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
out = append(out, gin.H{"key": it.Key, "url": u})
}
c.JSON(http.StatusOK, gin.H{"items": out, "expires_in": int(ttl.Seconds())})
}
// WorkerReport spider 回报执行结果。
// status: done(执行完成,哪怕 Vogue 上没有该品牌的秀)/ failed(进程级失败,可重试)。
func (h *BackstageHandler) WorkerReport(c *gin.Context) {
if h.crawl == nil {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "crawl repository not configured"})
return
}
id := atoiDefault(c.Param("id"), 0)
if id <= 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid task id"})
return
}
var body struct {
Status string `json:"status"`
ShowsSubmitted int `json:"shows_submitted"`
Error string `json:"error"`
}
if err := c.ShouldBindJSON(&body); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid json: " + err.Error()})
return
}
if body.Status != model.CrawlDone && body.Status != model.CrawlFailed {
c.JSON(http.StatusBadRequest, gin.H{"error": "status must be done or failed"})
return
}
if body.Error != "" && len(body.Error) > 1000 {
body.Error = body.Error[:1000]
}
if err := h.crawl.Report(c.Request.Context(), uint32(id), body.Status, body.ShowsSubmitted, body.Error); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "report failed"})
return
}
c.JSON(http.StatusOK, gin.H{"ok": true})
}

View File

@ -0,0 +1,170 @@
package handler
import (
"context"
"log"
"net/http"
"strconv"
"strings"
"sync"
"time"
"fashionapi/internal/service"
"github.com/gin-gonic/gin"
)
// pairMu 串行化「近邻重算」:见 WorkerEmbeddings 里的说明。
var pairMu sync.Mutex
// precomputePairsAsync 在后台重算近邻,失败只记日志(向量已写入,重算可由 similarbackfill 补)。
func precomputePairsAsync(similar *service.SimilarService, kind string, ids []uint32) {
pairMu.Lock()
defer pairMu.Unlock()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
if err := similar.PrecomputeForGallery(ctx, kind, ids); err != nil {
log.Printf("[similar] 回填后近邻重算失败 kind=%s 数量=%d err=%v", kind, len(ids), err)
}
}
// embeddingMaxAttempts 同一张图最多回填几次。
//
// 为什么必须有上限:坏图(文件损坏 / 格式不支持 / 已从存储删掉)是**永久失败**,
// 没有计数的话它每轮都会重新出现在队列头部,把后面的好图永远挤在后面。
// 5 次足够覆盖网络抖动与视觉服务重启,又不至于让坏图长期占位。
const embeddingMaxAttempts = 5
// embeddingMaxDim 允许的最大向量维度(FashionSigLIP 是 768,留足余量防误配)。
const embeddingMaxDim = 4096
// WorkerPendingEmbeddings 给回填 worker 派发「缺向量」的图片。
//
// GET /admin/internal/embeddings/pending?kind=runway&limit=32
// → {"items":[{"kind":"runway","image_id":123,"key":"runway/ab.jpg","url":"https://..."}],
// "remaining": 812}
//
// 返回 url 而不只是 key:worker 直接下载,不必知道存储域名怎么拼
// (域名换了、换了存储、本地兜底,都在后端一处收敛)。
func (h *BackstageHandler) WorkerPendingEmbeddings(c *gin.Context) {
if h.emb == nil {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "embedding repository not configured"})
return
}
kind := c.DefaultQuery("kind", "runway")
if kind != "runway" && kind != "street" {
c.JSON(http.StatusBadRequest, gin.H{"error": "kind 只能是 runway / street"})
return
}
limit := 32
if v := c.Query("limit"); v != "" {
n, err := strconv.Atoi(v)
if err != nil || n <= 0 || n > 200 {
c.JSON(http.StatusBadRequest, gin.H{"error": "limit 应为 1~200"})
return
}
limit = n
}
rows, err := h.emb.ListPendingEmbeddings(c.Request.Context(), kind, limit, embeddingMaxAttempts)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "query pending failed: " + err.Error()})
return
}
items := make([]gin.H, 0, len(rows))
for _, r := range rows {
items = append(items, gin.H{
"kind": r.Kind,
"image_id": r.ImageID,
"key": r.Key,
"url": imgComposer.Compose(r.Key),
})
}
// remaining 用独立 count:worker 靠它判断「还有没有活儿」,不必取到空批次才知道。
remaining, err := h.emb.CountPendingEmbeddings(c.Request.Context(), kind, embeddingMaxAttempts)
if err != nil {
remaining = -1 // -1 = 统计失败,worker 应继续拉取后续批次
}
c.JSON(http.StatusOK, gin.H{"items": items, "remaining": remaining})
}
// WorkerEmbeddings 接收回填结果:成功的写向量并重算近邻,失败的累加失败计数。
//
// POST /admin/internal/embeddings
// {"items":[{"kind":"runway","image_id":123,"vector":[...768 个数...]}],
// "failures":[{"kind":"runway","image_id":124,"error":"下载 404"}]}
//
// 为什么近邻重算放在这里:image_similar_pairs 是入库时预计算的(页面只读结果)。
// 回填补上的向量如果不重算,后台「疑似同款」就永远看不到新入库的图。
func (h *BackstageHandler) WorkerEmbeddings(c *gin.Context) {
if h.emb == nil {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "embedding repository not configured"})
return
}
var body struct {
Items []struct {
Kind string `json:"kind"`
ImageID uint32 `json:"image_id"`
Vector []float32 `json:"vector"`
} `json:"items"`
Failures []struct {
Kind string `json:"kind"`
ImageID uint32 `json:"image_id"`
Error string `json:"error"`
} `json:"failures"`
}
if err := c.ShouldBindJSON(&body); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid json: " + err.Error()})
return
}
if len(body.Items) == 0 && len(body.Failures) == 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "items 与 failures 不能同时为空"})
return
}
// 按 kind 分组批量写:一次往返写一批,避免 32 张图 32 次事务。
idsByKind := map[string][]uint32{}
vecsByKind := map[string][][]float32{}
failed := len(body.Failures)
for _, it := range body.Items {
kind := strings.TrimSpace(it.Kind)
if kind != "runway" && kind != "street" {
failed++
_ = h.emb.RecordEmbeddingFailure(c.Request.Context(), kind, it.ImageID, "kind 非法: "+it.Kind)
continue
}
if n := len(it.Vector); n == 0 || n > embeddingMaxDim {
// 维度异常一律拒写:写进去会污染向量列,让近邻检索结果全错。
failed++
_ = h.emb.RecordEmbeddingFailure(c.Request.Context(), kind, it.ImageID, "向量维度异常: "+strconv.Itoa(n))
continue
}
idsByKind[kind] = append(idsByKind[kind], it.ImageID)
vecsByKind[kind] = append(vecsByKind[kind], it.Vector)
}
saved := 0
for kind, ids := range idsByKind {
if err := h.emb.BulkUpsertEmbedding(c.Request.Context(), kind, ids, vecsByKind[kind]); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "write embedding failed: " + err.Error()})
return
}
saved += len(ids)
if h.similar != nil {
// 近邻重算**必须异步**:它要对每张图各跑一次向量近邻查询(20KB payload/次),
// 一批 100 张就是 100 次往返。同步做会让 worker 的 POST 直接超时
// (实测:库在远端时 100 张超过 60s),于是向量其实已经写入、worker 却判失败并重试。
//
// 用互斥串行而不是每请求一个 goroutine:多个回填 worker 并发时,
// 并发跑近邻查询只会互相拖慢,还容易把连接池占满。
go precomputePairsAsync(h.similar, kind, ids)
}
}
for _, f := range body.Failures {
_ = h.emb.RecordEmbeddingFailure(c.Request.Context(), f.Kind, f.ImageID, f.Error)
}
c.JSON(http.StatusOK, gin.H{"saved": saved, "failed": failed})
}

View File

@ -16,6 +16,7 @@ import (
"fashionapi/internal/pkg/hashid" "fashionapi/internal/pkg/hashid"
"fashionapi/internal/pkg/pagination" "fashionapi/internal/pkg/pagination"
"fashionapi/internal/pkg/season" "fashionapi/internal/pkg/season"
"fashionapi/internal/pkg/storage"
"fashionapi/internal/repository" "fashionapi/internal/repository"
"fashionapi/internal/service" "fashionapi/internal/service"
@ -45,8 +46,25 @@ type BackstageHandler struct {
// manual 后台「手工新建文章」(可为 nil:未装配时新建入口提示不可用)。 // manual 后台「手工新建文章」(可为 nil:未装配时新建入口提示不可用)。
// 用 SetManual 注入,避免改构造签名(同 SetSimilar 的取舍)。 // 用 SetManual 注入,避免改构造签名(同 SetSimilar 的取舍)。
manual *service.ManualArticleService manual *service.ManualArticleService
// crawl 品牌抓取任务(crawl_tasks)。用 SetCrawl 注入,避免改构造签名。
crawl repository.CrawlRepository
// uploader S4 上传器(可为 nil:未配置 S4 时不提供预签名直传)。
// 媒体机(spider)用它拿一次性 PUT URL,自己上传图片,不必持有 AK/SK。
uploader *storage.S4Uploader
// emb 向量仓储(可为 nil:未注入时回填接口返回 503)。
// 媒体机直传后后端拿不到图片字节,向量只能靠回填 worker 从存储拉图编码再回传。
emb repository.EmbeddingRepository
} }
// SetUploader 注入 S4 上传器(装配期调用),用于给媒体机签发直传 URL。
func (h *BackstageHandler) SetUploader(up *storage.S4Uploader) { h.uploader = up }
// SetEmbeddingRepo 注入向量仓储(装配期调用),用于向量回填的取件与回写。
func (h *BackstageHandler) SetEmbeddingRepo(repo repository.EmbeddingRepository) { h.emb = repo }
// SetCrawl 注入抓取任务仓储(装配期调用;未注入时 /admin/crawl 提示不可用)。
func (h *BackstageHandler) SetCrawl(repo repository.CrawlRepository) { h.crawl = repo }
// SetManual 注入手工建稿服务(装配期调用;未注入时新建页提示不可用)。 // SetManual 注入手工建稿服务(装配期调用;未注入时新建页提示不可用)。
func (h *BackstageHandler) SetManual(svc *service.ManualArticleService) { h.manual = svc } func (h *BackstageHandler) SetManual(svc *service.ManualArticleService) { h.manual = svc }
@ -145,6 +163,8 @@ func (h *BackstageHandler) Brands(c *gin.Context) {
All: c.Query("all") == "1", All: c.Query("all") == "1",
Page: atoiDefault(c.Query("page"), 1), Page: atoiDefault(c.Query("page"), 1),
Locale: "en", Locale: "en",
// 后台可切换「按最近更新排序」(走秀或抓取任务的最新时间)
Sort: c.Query("sort"),
} }
items, total, err := h.brand.AdminList(c.Request.Context(), q) items, total, err := h.brand.AdminList(c.Request.Context(), q)
if err != nil { if err != nil {
@ -157,13 +177,53 @@ func (h *BackstageHandler) Brands(c *gin.Context) {
p.Page = 1 p.Page = 1
} }
meta := p.NewMeta(total) meta := p.NewMeta(total)
// 当前页品牌的数字 id:下面的抓取状态与数据量统计都要用。
ids := make([]uint32, 0, len(items))
for _, it := range items {
if id, derr := hashid.Decode(it.UID); derr == nil {
ids = append(ids, id)
}
}
// 装饰「抓取状态 / 上次抓取」:从 crawl_tasks 取每个品牌最近一条任务。
// 没有任务记录的品牌不在 map 里,模板显示「—」(= 从没加入过抓取队列)。
crawlStates := map[string]gin.H{}
if h.crawl != nil {
if m, lerr := h.crawl.BrandCrawlStates(c.Request.Context(), ids); lerr == nil {
for id, st := range m {
last := ""
if !st.UpdatedAt.IsZero() {
last = st.UpdatedAt.Format("2006-01-02 15:04")
}
crawlStates[hashid.Encode(id)] = gin.H{
"Label": crawlStatusLabel(st.Status),
"Class": crawlStatusClass(st.Status),
"LastAt": last,
}
}
}
}
// 数据量:图集数 / 图片数(只查当前页的 30 个品牌)
stats := map[string]gin.H{}
if m, serr := h.brand.StatsByIDs(c.Request.Context(), ids); serr == nil {
for id, st := range m {
stats[hashid.Encode(id)] = gin.H{"Runways": st.Runways, "Images": st.Images}
}
}
renderPage(c, http.StatusOK, "brands", gin.H{ renderPage(c, http.StatusOK, "brands", gin.H{
"Items": items, "Items": items,
"Total": total, "Total": total,
"Keyword": q.Keyword, "Keyword": q.Keyword,
"Letter": q.Letter, "Letter": q.Letter,
"All": q.All, "All": q.All,
"Sort": q.Sort,
"Letters": lettersAtoZ(), "Letters": lettersAtoZ(),
"CrawlStates": crawlStates,
"Stats": stats,
"Added": atoiDefault(c.Query("added"), 0),
"Pager": pagination.Pager{ "Pager": pagination.Pager{
PrevURL: pagination.PageLink(c, meta.PrevPage), PrevURL: pagination.PageLink(c, meta.PrevPage),
NextURL: pagination.PageLink(c, meta.NextPage), NextURL: pagination.PageLink(c, meta.NextPage),

View File

@ -28,6 +28,7 @@ func TestBackstageTemplatesParsed(t *testing.T) {
"articles", "articles",
"brand-form", "brand-form",
"brands", "brands",
"crawl",
"dashboard", "dashboard",
"error", "error",
"ingest-jobs", "ingest-jobs",
@ -73,6 +74,10 @@ func TestBackstageTemplatesRender(t *testing.T) {
"Images": []any{}, "Images": []any{},
// 列表数据 // 列表数据
"Items": []any{}, "Users": []any{}, "Kinds": []any{}, "Records": []any{}, "TabCounts": map[string]int64{}, "Items": []any{}, "Users": []any{}, "Kinds": []any{}, "Records": []any{}, "TabCounts": map[string]int64{},
// 抓取任务页(crawl.html):列表 + 四个状态计数
"Rows": []any{}, "Queued": int64(0), "Running": int64(0), "Done": int64(0), "Failed": int64(0),
// 品牌页的「抓取状态 / 上次抓取」列(按 hashid 索引)
"CrawlStates": map[string]any{}, "Stats": map[string]any{}, "Added": 0,
// 详情页实体(.Snap 在两个页面里类型不同,见下面的 perPage 覆盖) // 详情页实体(.Snap 在两个页面里类型不同,见下面的 perPage 覆盖)
"Article": &dto.PublicArticleDetail{UID: "abc123"}, "Article": &dto.PublicArticleDetail{UID: "abc123"},
"Snap": &dto.PublicStreetSnapDetail{UID: "abc123"}, "Snap": &dto.PublicStreetSnapDetail{UID: "abc123"},

View File

@ -8,21 +8,69 @@
<form class="toolbar" method="get" action="/admin/brands"> <form class="toolbar" method="get" action="/admin/brands">
<input name="keyword" placeholder="搜索品牌名…" value="{{.Keyword}}"> <input name="keyword" placeholder="搜索品牌名…" value="{{.Keyword}}">
{{if .All}}<input type="hidden" name="all" value="1">{{end}} {{if .All}}<input type="hidden" name="all" value="1">{{end}}
{{if .Sort}}<input type="hidden" name="sort" value="{{.Sort}}">{{end}}
<button type="submit">搜索</button> <button type="submit">搜索</button>
{{if .Keyword}}<a href="/admin/brands{{if .All}}?all=1{{end}}">清除</a>{{end}} {{if .Keyword}}<a href="/admin/brands{{if .All}}?all=1{{end}}">清除</a>{{end}}
<span class="spacer"></span> <span class="spacer"></span>
{{if .All}}<a class="view" href="/admin/brands?keyword={{.Keyword}}">只看有图集的品牌</a>{{else}}<a class="view" href="/admin/brands?keyword={{.Keyword}}&all=1">显示全部品牌(含无图集)</a>{{end}} {{if .All}}<a class="view" href="/admin/brands?keyword={{.Keyword}}">只看有图集的品牌</a>{{else}}<a class="view" href="/admin/brands?keyword={{.Keyword}}&all=1">显示全部品牌(含无图集)</a>{{end}}
</form> </form>
<p class="muted">默认只显示「已有图集」的品牌;品牌目录共 2 万余条,其余请用上面的搜索或开关查看。</p>
<p class="muted">
默认只显示「已有图集」的品牌;品牌目录共 2 万余条,其余请用上面的搜索或开关查看。
排序:
{{if eq .Sort "updated"}}<strong>最近更新</strong>{{else}}<a href="/admin/brands?keyword={{.Keyword}}{{if .All}}&all=1{{end}}&sort=updated">最近更新</a>{{end}}
·
{{if eq .Sort "updated"}}<a href="/admin/brands?keyword={{.Keyword}}{{if .All}}&all=1{{end}}">按名字</a>{{else}}<strong>按名字</strong>{{end}}
</p>
{{if .Added}}<p class="muted">✅ 本次已加入抓取队列 <strong>{{.Added}}</strong> 个品牌(已在队列中的不会重复添加)。</p>{{end}}
<div class="idx"> <div class="idx">
<a href="/admin/brands?keyword={{.Keyword}}{{if .All}}&all=1{{end}}" class="{{if eq .Letter ""}}on{{end}}">全部</a> <a href="/admin/brands?keyword={{.Keyword}}{{if .All}}&all=1{{end}}{{if .Sort}}&sort={{.Sort}}{{end}}" class="{{if eq .Letter ""}}on{{end}}">全部</a>
<a href="/admin/brands?letter=OTHER&keyword={{.Keyword}}{{if .All}}&all=1{{end}}" class="{{if eq .Letter "OTHER"}}on{{end}}">中文</a> <a href="/admin/brands?letter=OTHER&keyword={{.Keyword}}{{if .All}}&all=1{{end}}{{if .Sort}}&sort={{.Sort}}{{end}}" class="{{if eq .Letter "OTHER"}}on{{end}}">中文</a>
{{range $l := .Letters}}<a href="/admin/brands?letter={{$l}}&keyword={{$.Keyword}}{{if $.All}}&all=1{{end}}" class="{{if eq $l $.Letter}}on{{end}}">{{$l}}</a>{{end}} {{range $l := .Letters}}<a href="/admin/brands?letter={{$l}}&keyword={{$.Keyword}}{{if $.All}}&all=1{{end}}{{if $.Sort}}&sort={{$.Sort}}{{end}}" class="{{if eq $l $.Letter}}on{{end}}">{{$l}}</a>{{end}}
</div> </div>
{{/* 批量操作表单:勾选框用 form="bulk" 关联,避免与行内表单嵌套 */}}
<form id="bulk" method="post" action="/admin/crawl/bulk"></form>
<div class="ops" style="margin:10px 0">
<label class="muted">每品牌场次上限
<select name="maxShows" form="bulk">
<option value="1">1 场(最快探有无数据)</option>
<option value="3" selected>3 场(最近,推荐)</option>
<option value="0">全量(所有历史场次)</option>
</select>
</label>
<button type="submit" form="bulk" class="op">把勾选的品牌加入抓取队列</button>
<span class="muted">(勾选本页任意行,先点复选框列头可全选本页)</span>
</div>
<table> <table>
<thead><tr><th>编码 ID</th><th>英文名</th><th>中文名</th><th>操作</th></tr></thead> <thead><tr>
<th><input type="checkbox" title="全选本页" onclick="document.querySelectorAll('input[name=uid]').forEach(function(c){c.checked=this.checked}.bind(this))"></th>
<th>编码 ID</th><th>英文名</th><th>中文名</th><th>图集</th><th>图片</th><th>抓取状态</th><th>上次抓取</th><th>操作</th>
</tr></thead>
<tbody> <tbody>
{{range .Items}}<tr><td><code>{{.UID}}</code></td><td>{{.NameEn}}</td><td>{{.NameCn}}</td><td><a class="view" href="/admin/brands/{{.UID}}">编辑</a><form class="inline" method="post" action="/admin/brands/{{.UID}}/delete" onsubmit="return confirm('确认删除品牌「{{.NameEn}}」?该品牌下若还有图集会拒绝删除。')"><button type="submit" class="op op-warn">删除</button></form></td></tr>{{end}} {{range .Items}}{{$cs := index $.CrawlStates .UID}}{{$st := index $.Stats .UID}}<tr>
<td><input type="checkbox" name="uid" value="{{.UID}}" form="bulk"></td>
<td><code>{{.UID}}</code></td>
<td>{{.NameEn}}</td>
<td>{{.NameCn}}</td>
<td>{{if $st}}{{$st.Runways}}{{else}}0{{end}}</td>
<td>{{if $st}}{{$st.Images}}{{else}}0{{end}}</td>
<td>{{if $cs}}<span class="badge {{$cs.Class}}">{{$cs.Label}}</span>{{else}}<span class="muted">—</span>{{end}}</td>
<td class="muted">{{if $cs}}{{$cs.LastAt}}{{else}}—{{end}}</td>
<td>
<a class="view" href="/admin/brands/{{.UID}}">编辑</a>
<form class="inline" method="post" action="/admin/crawl/brands/{{.UID}}" onsubmit="return confirm('把「{{.NameEn}}」加入抓取队列?默认只抓最近 3 场。')">
<input type="hidden" name="back" value="/admin/brands"><input type="hidden" name="maxShows" value="3">
<button type="submit" class="op">加入队列</button>
</form>
<form class="inline" method="post" action="/admin/brands/{{.UID}}/delete" onsubmit="return confirm('确认删除品牌「{{.NameEn}}」?该品牌下若还有图集会拒绝删除。')">
<button type="submit" class="op op-warn">删除</button>
</form>
</td>
</tr>{{end}}
</tbody> </tbody>
</table> </table>
{{template "pager" .Pager}} {{template "pager" .Pager}}

View File

@ -0,0 +1,44 @@
{{define "title"}}抓取任务 · 管理后台{{end}}
{{define "barLabel"}}管理后台 · 品牌抓取任务{{end}}
{{define "barRight"}}<a href="/admin/brands">去品牌管理加任务</a> · <a href="/admin/logout">退出</a>{{end}}
{{define "content"}}
{{template "bar" .}}
<div class="main">
<h1>抓取任务</h1>
<p class="muted">
任务是「从品牌管理页加入」的:勾选品牌 → 选场次上限 → 加入抓取队列。
这里看的是执行状态与失败原因。真正干活的是 spider 服务(可在另一台机器)。
</p>
<p class="muted">
待领取 {{.Queued}} · 执行中 {{.Running}} · 已完成 {{.Done}} · 失败 {{.Failed}},共 {{.Total}} 条。
<strong>「已提交 0 场」不是错误</strong>——Vogue 上很多品牌没有走秀页面(实测 A 开头里约 75%)。
</p>
<table>
<thead><tr>
<th>ID</th><th>品牌</th><th>状态</th><th>场次上限</th><th>已提交</th><th>尝试</th><th>Worker</th><th>更新时间</th><th>错误</th><th>操作</th>
</tr></thead>
<tbody>
{{range .Rows}}<tr>
<td><code>{{.ID}}</code></td>
<td><a class="view" href="/admin/brands/{{.BrandUID}}">{{.BrandName}}</a></td>
<td><span class="badge">{{.StatusLabel}}</span></td>
<td>{{if eq .MaxShows 0}}<strong>全量</strong>{{else}}{{.MaxShows}}{{end}}</td>
<td>{{.Shows}}</td>
<td>{{.Attempts}}</td>
<td class="muted">{{if .Worker}}{{.Worker}}{{else}}—{{end}}</td>
<td class="muted">{{.UpdatedAt}}</td>
<td class="err break">{{.LastError}}</td>
<td>
<div class="ops">
{{if .CanRetry}}<form class="inline" method="post" action="/admin/crawl/tasks/{{.ID}}/retry"><button class="op" type="submit">重试</button></form>{{end}}
{{if .CanCancel}}<form class="inline" method="post" action="/admin/crawl/tasks/{{.ID}}/cancel"><button class="op" type="submit">取消</button></form>{{end}}
<form class="inline" method="post" action="/admin/crawl/tasks/{{.ID}}/delete" onsubmit="return confirm('删除任务记录?已入库的数据不受影响。')"><button class="op" type="submit">删除</button></form>
</div>
</td>
</tr>{{end}}
</tbody>
</table>
{{template "pager" .Pager}}
</div>
{{end}}

View File

@ -18,6 +18,7 @@
<a href="/admin/" class="{{if eq .NavKey "dashboard"}}on{{end}}">首页</a> <a href="/admin/" class="{{if eq .NavKey "dashboard"}}on{{end}}">首页</a>
<a href="/admin/articles" class="{{if or (eq .NavKey "articles") (eq .NavKey "runway-edit") (eq .NavKey "snap-edit") (eq .NavKey "runway-detail") (eq .NavKey "street-snap-detail")}}on{{end}}">文章{{if gt .PendingReviews 0}}<span class="bar-badge">{{.PendingReviews}}</span>{{end}}</a> <a href="/admin/articles" class="{{if or (eq .NavKey "articles") (eq .NavKey "runway-edit") (eq .NavKey "snap-edit") (eq .NavKey "runway-detail") (eq .NavKey "street-snap-detail")}}on{{end}}">文章{{if gt .PendingReviews 0}}<span class="bar-badge">{{.PendingReviews}}</span>{{end}}</a>
<a href="/admin/brands" class="{{if or (eq .NavKey "brands") (eq .NavKey "brand-form")}}on{{end}}">品牌</a> <a href="/admin/brands" class="{{if or (eq .NavKey "brands") (eq .NavKey "brand-form")}}on{{end}}">品牌</a>
<a href="/admin/crawl" class="{{if eq .NavKey "crawl"}}on{{end}}">抓取任务</a>
<a href="/admin/ingest-jobs" class="{{if eq .NavKey "ingest-jobs"}}on{{end}}">入库任务</a> <a href="/admin/ingest-jobs" class="{{if eq .NavKey "ingest-jobs"}}on{{end}}">入库任务</a>
<a href="/admin/users" class="{{if eq .NavKey "users"}}on{{end}}">用户</a> <a href="/admin/users" class="{{if eq .NavKey "users"}}on{{end}}">用户</a>
<span class="sep"></span> <span class="sep"></span>

View File

@ -0,0 +1,57 @@
package model
// CrawlTask 后台派发给爬虫服务的「品牌抓取任务」。
//
// 与 ingest_jobs 的分工:
// - ingest_jobs:一场走秀的入库任务,由后端 worker 消费(下载图 / 写库 / 编码);
// - crawl_tasks:一个品牌的抓取任务,由 spider 服务领取执行(可能在另一台机器)。
//
// 两层解耦的好处:后台只需表达「我要爬哪些品牌」,不必关心 spider 在哪、跑几个。
// spider 拉不到任务就空转,任务积压也不会影响已有数据的读写。
const (
CrawlQueued = "queued" // 待领取
CrawlRunning = "running" // 已被某个 worker 领取
CrawlDone = "done" // 执行完成(不代表每场都入库成功,见 last_error)
CrawlFailed = "failed" // 执行失败(spider 进程报错 / 品牌在 Vogue 无页面等)
CrawlCanceled = "canceled" // 后台手动取消
)
// CrawlStuckTimeoutSec 是 running 任务的租约时长:超过该时长仍未回报,
// 视为 worker 已崩溃(断电 / 被杀),任务可被重新领取,避免永久卡在 running。
const CrawlStuckTimeoutSec = 900
// CrawlMaxAttempts 自动重试上限;达到后置 failed,需后台手动「重试」。
const CrawlMaxAttempts = 3
// CrawlTask 品牌抓取任务。
// 时间字段与项目其它表一致,统一用 Unix 秒(uint32)。
type CrawlTask struct {
ID uint32 `gorm:"primaryKey;column:id" json:"id"`
BrandID uint32 `gorm:"column:brand_id;not null" json:"brand_id"`
// Status 见上方常量;默认 queued。
Status string `gorm:"column:status;size:16;not null;default:'queued'" json:"status"`
// Priority 越大越先被领取;用于「手动加的」优先于「批量加的」。
Priority int16 `gorm:"column:priority;not null;default:0" json:"priority"`
// MaxShows 单品牌最多抓几场(0=不限)。批量重爬历史品牌时用它控制入库量。
MaxShows int16 `gorm:"column:max_shows;not null;default:0" json:"max_shows"`
// Worker 领取者标识(主机名 + 进程号),便于排查「谁跑的」。
Worker string `gorm:"column:worker;size:64;not null;default:''" json:"worker"`
// LockedAt 领取时间,配合 CrawlStuckTimeoutSec 做租约回收。
LockedAt uint32 `gorm:"column:locked_at;not null;default:0" json:"locked_at"`
// Attempts 已被领取次数;用于自动重试与放弃判定。
Attempts int16 `gorm:"column:attempts;not null;default:0" json:"attempts"`
// ShowsSubmitted 本任务向 ingest 提交了多少场走秀(spider 回报)。
ShowsSubmitted int32 `gorm:"column:shows_submitted;not null;default:0" json:"shows_submitted"`
LastError string `gorm:"column:last_error;type:text" json:"last_error"`
CreatedAt uint32 `gorm:"column:created_at;not null;default:0" json:"created_at"`
UpdatedAt uint32 `gorm:"column:updated_at;not null;default:0" json:"updated_at"`
}
// TableName 指定表名。
func (CrawlTask) TableName() string { return "crawl_tasks" }
// CrawlTaskRow 是列表/领取接口用的投影行,附带品牌名(列表展示与拼 Vogue URL 都用得上)。
type CrawlTaskRow struct {
CrawlTask
BrandName string `gorm:"column:brand_name"`
}

View File

@ -3,8 +3,10 @@ package storage
import ( import (
"bytes" "bytes"
"context" "context"
"errors"
"fmt" "fmt"
"strings" "strings"
"time"
"github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/aws"
awsconfig "github.com/aws/aws-sdk-go-v2/config" awsconfig "github.com/aws/aws-sdk-go-v2/config"
@ -67,6 +69,38 @@ func NewS4Uploader(ak, sk, bucket, endpoint, baseURL, region string) *S4Uploader
} }
} }
// PresignPut 为指定 key 生成有时效的 PUT 预签名 URL。
//
// 用途:让**不持有 S4 密钥**的客户端(例如另部署的 spider 媒体机)直传对象。
// 密钥只留在后端,客户端拿到 URL 直接 HTTP PUT 即可 —— 不必在爬虫里引入 S3 SDK,
// 也不用把 AK/SK 分发到更多机器。
//
// contentType 必须与 PUT 请求头一致,否则 S4 会因签名不匹配拒绝(403)。
func (u *S4Uploader) PresignPut(ctx context.Context, key, contentType string, ttl time.Duration) (string, error) {
if !u.enabled || u.client == nil {
return "", errors.New("S4 未启用(缺少 AK/SK 或 bucket)")
}
if key == "" {
return "", errors.New("key 不能为空")
}
if contentType == "" {
contentType = "application/octet-stream"
}
if ttl <= 0 {
ttl = 15 * time.Minute
}
pc := s3.NewPresignClient(u.client)
out, err := pc.PresignPutObject(ctx, &s3.PutObjectInput{
Bucket: aws.String(u.bucket),
Key: aws.String(key),
ContentType: aws.String(contentType),
}, s3.WithPresignExpires(ttl))
if err != nil {
return "", err
}
return out.URL, nil
}
// deriveS4BaseURL 由 endpoint + bucket 推导虚拟主机风格访问域名: // deriveS4BaseURL 由 endpoint + bucket 推导虚拟主机风格访问域名:
// endpoint=https://s3.bitiful.net, bucket=toom → https://toom.s3.bitiful.net // endpoint=https://s3.bitiful.net, bucket=toom → https://toom.s3.bitiful.net
func deriveS4BaseURL(endpoint, bucket string) string { func deriveS4BaseURL(endpoint, bucket string) string {

View File

@ -52,6 +52,16 @@ type BrandRepository interface {
// 规则:先按品牌总图片数取前 limit 个热门品牌,再为每个品牌取最新一篇走秀。 // 规则:先按品牌总图片数取前 limit 个热门品牌,再为每个品牌取最新一篇走秀。
// 返回的是仓储层裸数据(数字主键),对外编码由 service 层负责。 // 返回的是仓储层裸数据(数字主键),对外编码由 service 层负责。
PopularWithCover(ctx context.Context, limit int) ([]PopularBrandRow, error) PopularWithCover(ctx context.Context, limit int) ([]PopularBrandRow, error)
// StatsByIDs 批量统计若干品牌的「图集数 / 图片数」,供后台列表展示数据量。
// 只查当前页的 id,避免为了两列统计把整张品牌表都聚合一遍。
StatsByIDs(ctx context.Context, ids []uint32) (map[uint32]BrandStats, error)
}
// BrandStats 品牌数据量统计(后台列表用)。
type BrandStats struct {
Runways int64 // 未删除的图集数
Images int64 // 未删除的图片数
} }
// PopularBrandRow 热门品牌裸数据:品牌数字主键 + 中英文名称 + 代表走秀封面/中英文标题/文章 ID。 // PopularBrandRow 热门品牌裸数据:品牌数字主键 + 中英文名称 + 代表走秀封面/中英文标题/文章 ID。
@ -174,11 +184,30 @@ func (r *brandRepository) list(ctx context.Context, q dto.BrandQuery, restrictID
return []model.Brand{}, 0, nil return []model.Brand{}, 0, nil
} }
var items []model.Brand // 后台可按「最近有动静」排序:走秀图集的 updated_at 与抓取任务的 updated_at 取大。
if err := r.db.WithContext(ctx). //
// 性能注意:不要写成逐行的相关子查询(2.3 万品牌 × 2 个子查询实测 2.7 秒)。
// 这里先把两张小表各自按品牌聚合(brand_runways / crawl_tasks 都只有几百行),
// 再 LEFT JOIN 到品牌表排序 —— 一次哈希连接 + 一次排序,量级降到百毫秒。
// 公开侧不传 Sort,保持原有 name_en ASC,前台顺序不变。
itemsQ := r.db.WithContext(ctx).
Model(&model.Brand{}). Model(&model.Brand{}).
Scopes(scope). Scopes(scope)
Order("name_en ASC"). if admin && q.Sort == "updated" {
itemsQ = itemsQ.
Joins(`LEFT JOIN (SELECT brand_id, max(updated_at) AS u
FROM brand_runways WHERE is_deleted = 0 GROUP BY brand_id) br
ON br.brand_id = brands.id`).
Joins(`LEFT JOIN (SELECT brand_id, max(updated_at) AS u
FROM crawl_tasks GROUP BY brand_id) ct
ON ct.brand_id = brands.id`).
Order("GREATEST(coalesce(br.u, 0), coalesce(ct.u, 0)) DESC, name_en ASC")
} else {
itemsQ = itemsQ.Order("name_en ASC")
}
var items []model.Brand
if err := itemsQ.
Offset(q.Offset()). Offset(q.Offset()).
Limit(q.Size). Limit(q.Size).
Find(&items).Error; err != nil { Find(&items).Error; err != nil {
@ -319,6 +348,36 @@ func (r *brandRepository) CrawlBrands(ctx context.Context, brandID uint32) ([]mo
// 3. 结果按热门品牌排名顺序返回 // 3. 结果按热门品牌排名顺序返回
// //
// 注意:created_at/updated_at 是数据导入时间,不能代表走秀新旧,因此必须用 year + season_code。 // 注意:created_at/updated_at 是数据导入时间,不能代表走秀新旧,因此必须用 year + season_code。
// StatsByIDs 批量统计图集数 / 图片数(只针对传入的 id,通常是一页 30 个)。
func (r *brandRepository) StatsByIDs(ctx context.Context, ids []uint32) (map[uint32]BrandStats, error) {
out := make(map[uint32]BrandStats, len(ids))
if len(ids) == 0 {
return out, nil
}
var rows []struct {
BrandID uint32
Runways int64
Images int64
}
// 一次查询同时拿两个计数:按品牌聚合图集,再左连图片聚合,避免 N+1。
err := r.db.WithContext(ctx).Raw(`
SELECT r.brand_id,
count(DISTINCT r.id) AS runways,
count(i.id) AS images
FROM brand_runways r
LEFT JOIN brand_runway_images i
ON i.runway_id = r.id AND i.is_deleted = 0
WHERE r.is_deleted = 0 AND r.brand_id IN ?
GROUP BY r.brand_id`, ids).Scan(&rows).Error
if err != nil {
return out, err
}
for _, row := range rows {
out[row.BrandID] = BrandStats{Runways: row.Runways, Images: row.Images}
}
return out, nil
}
func (r *brandRepository) PopularWithCover(ctx context.Context, limit int) ([]PopularBrandRow, error) { func (r *brandRepository) PopularWithCover(ctx context.Context, limit int) ([]PopularBrandRow, error) {
if limit < 1 { if limit < 1 {
limit = 20 limit = 20

View File

@ -0,0 +1,338 @@
package repository
import (
"context"
"strings"
"time"
"gorm.io/gorm"
"fashionapi/internal/model"
)
// CrawlRepository 品牌抓取任务(crawl_tasks)的数据访问。
//
// 任务表同时被两个方向使用:
// - 后台(cookie 会话)往里加任务 / 取消 / 重试;
// - spider 服务(HMAC 验签)来领取 / 回报结果。
//
// 领取用 `FOR UPDATE SKIP LOCKED`,多个 spider 实例并发领取不会抢到同一条。
type CrawlRepository interface {
// Enqueue 把单个品牌加入队列。
// 返回 true 表示新入队或把已完成/失败的任务重置为待领取;false 表示该品牌已在队列中/正在跑。
Enqueue(ctx context.Context, brandID uint32, maxShows int, priority int) (bool, error)
// EnqueueBulk 按「范围 + 前缀」批量入队,返回本次实际新入队(或重置)的数量。
EnqueueBulk(ctx context.Context, f BulkFilter) (int, error)
// Claim 领取一个待执行任务并置为 running;无任务时返回 nil, nil。
Claim(ctx context.Context, worker string) (*model.CrawlTaskRow, error)
// Report spider 回报结果。failed 且未达重试上限时自动回到 queued。
Report(ctx context.Context, id uint32, status string, shows int, errMsg string) error
// List 后台列表(按 running → queued → 其它 排序)。
List(ctx context.Context, offset, limit int) ([]model.CrawlTaskRow, int64, error)
// SetStatus 后台手动改状态(取消 / 重试)。
SetStatus(ctx context.Context, id uint32, status string) error
// Delete 删除任务记录(不影响已入库的数据)。
Delete(ctx context.Context, id uint32) error
// Counts 各状态任务数,供导航角标使用。
Counts(ctx context.Context) (map[string]int64, error)
// BrandCrawlStates 返回这些品牌在抓取队列里的「最近状态 + 最近时间」。
// 品牌列表用它展示「抓取状态 / 上次抓取」;没有任务记录的品牌不会出现在 map 里。
BrandCrawlStates(ctx context.Context, brandIDs []uint32) (map[uint32]BrandCrawlState, error)
}
// BrandCrawlState 品牌最近一次抓取任务的状态快照。
type BrandCrawlState struct {
Status string // queued / running / done / failed / canceled
UpdatedAt time.Time // 该任务最后一次更新时间
}
type crawlRepository struct {
db *gorm.DB
}
// NewCrawlRepository 创建爬取任务仓储。
func NewCrawlRepository(db *gorm.DB) CrawlRepository {
return &crawlRepository{db: db}
}
const crawlColumns = `t.id, t.brand_id, t.status, t.priority, t.max_shows, t.worker,
t.locked_at, t.attempts, t.shows_submitted, t.last_error, t.created_at, t.updated_at,
coalesce(b.name_en, '') AS brand_name`
func (r *crawlRepository) Enqueue(ctx context.Context, brandID uint32, maxShows int, priority int) (bool, error) {
now := uint32(time.Now().Unix())
// INSERT ... SELECT 保证品牌不存在(或已软删)时不会凭空建任务;
// ON CONFLICT 的 WHERE 让「已在排队/正在跑」的任务不被重置,避免打断正在执行的工作。
res := r.db.WithContext(ctx).Exec(`
INSERT INTO crawl_tasks (brand_id, status, priority, max_shows, created_at, updated_at)
SELECT b.id, ?, ?, ?, ?, ?
FROM brands b
WHERE b.id = ? AND b.is_deleted = 0
ON CONFLICT (brand_id) DO UPDATE
SET status = 'queued',
priority = EXCLUDED.priority,
max_shows = EXCLUDED.max_shows,
attempts = 0,
locked_at = 0,
worker = '',
shows_submitted = 0,
last_error = '',
updated_at = EXCLUDED.updated_at
WHERE crawl_tasks.status IN ('done', 'failed', 'canceled')`,
model.CrawlQueued, priority, maxShows, now, now, brandID)
if res.Error != nil {
return false, res.Error
}
return res.RowsAffected > 0, nil
}
// BulkScope 批量入队的品牌范围。
type BulkScope string
const (
// ScopeAll 所有品牌(首次全站导入用,量大)。
ScopeAll BulkScope = "all"
// ScopeWithData 只挑「已有图集」的品牌 —— 用于增量更新:只补最新场次。
ScopeWithData BulkScope = "with-data"
// ScopeWithoutData 只挑「还没有任何图集」的品牌 —— 用于首次补全。
ScopeWithoutData BulkScope = "without-data"
)
// BulkFilter 批量入队条件。
type BulkFilter struct {
Prefix string // 品牌英文名前缀,空 = 不限
Scope BulkScope // 范围(见上方常量)
MaxShows int // 0 = 不限场次(全量),>0 = 只抓最近 N 场
Limit int // 本次最多入队多少个品牌(防止一次灌爆)
Priority int16 // 越大越先被领取
}
func (r *crawlRepository) EnqueueBulk(ctx context.Context, f BulkFilter) (int, error) {
now := uint32(time.Now().Unix())
if f.Limit < 1 || f.Limit > 5000 {
f.Limit = 500
}
if f.Scope == "" {
f.Scope = ScopeAll
}
where := []string{"b.is_deleted = 0", "b.name_en <> ''"}
// 注意:参数顺序必须与 SQL 里 ? 出现的顺序严格一致(INSERT 的 4 个 + 每个 WHERE 条件 + LIMIT)。
args := []any{model.CrawlQueued, f.Priority, f.MaxShows, now, now}
if f.Prefix != "" {
where = append(where, `b.name_en ILIKE ? ESCAPE '\'`)
args = append(args, escapeLike(f.Prefix)+"%")
}
switch f.Scope {
case ScopeWithData:
where = append(where, "EXISTS (SELECT 1 FROM brand_runways r WHERE r.brand_id = b.id AND r.is_deleted = 0)")
case ScopeWithoutData:
where = append(where, "NOT EXISTS (SELECT 1 FROM brand_runways r WHERE r.brand_id = b.id AND r.is_deleted = 0)")
}
args = append(args, f.Limit)
sql := `
INSERT INTO crawl_tasks (brand_id, status, priority, max_shows, created_at, updated_at)
SELECT b.id, ?, ?, ?, ?, ?
FROM brands b
WHERE ` + strings.Join(where, " AND ") + `
ORDER BY EXISTS (SELECT 1 FROM crawl_tasks t WHERE t.brand_id = b.id) ASC, b.id ASC
LIMIT ?
ON CONFLICT (brand_id) DO UPDATE
SET status = 'queued',
priority = EXCLUDED.priority,
max_shows = EXCLUDED.max_shows,
attempts = 0,
locked_at = 0,
worker = '',
shows_submitted = 0,
last_error = '',
updated_at = EXCLUDED.updated_at
WHERE crawl_tasks.status IN ('done', 'failed', 'canceled')`
res := r.db.WithContext(ctx).Exec(sql, args...)
if res.Error != nil {
return 0, res.Error
}
return int(res.RowsAffected), nil
}
// escapeLike 转义 LIKE 的通配符,让用户输入按字面量匹配。
func escapeLike(s string) string {
out := make([]rune, 0, len(s))
for _, c := range s {
switch c {
case '%', '_', '\\':
out = append(out, '\\')
}
out = append(out, c)
}
return string(out)
}
func (r *crawlRepository) Claim(ctx context.Context, worker string) (*model.CrawlTaskRow, error) {
now := uint32(time.Now().Unix())
tx := r.db.WithContext(ctx).Begin()
if tx.Error != nil {
return nil, tx.Error
}
defer func() {
if tx.Error != nil {
tx.Rollback()
}
}()
// 回收租约过期的 running(worker 崩溃 / 断电),否则任务永远卡着。
if now > model.CrawlStuckTimeoutSec {
if err := tx.Exec(`
UPDATE crawl_tasks
SET status = 'queued', locked_at = 0, worker = '', updated_at = ?
WHERE status = 'running' AND locked_at > 0 AND locked_at < ?`,
now, now-model.CrawlStuckTimeoutSec).Error; err != nil {
return nil, err
}
}
var id uint32
if err := tx.Raw(`
SELECT id FROM crawl_tasks
WHERE status = 'queued'
ORDER BY priority DESC, id ASC
LIMIT 1
FOR UPDATE SKIP LOCKED`).Scan(&id).Error; err != nil {
return nil, err
}
if id == 0 {
if err := tx.Commit().Error; err != nil {
return nil, err
}
return nil, nil
}
if err := tx.Exec(`
UPDATE crawl_tasks
SET status = 'running', worker = ?, locked_at = ?, attempts = attempts + 1, updated_at = ?
WHERE id = ?`, worker, now, now, id).Error; err != nil {
return nil, err
}
if err := tx.Commit().Error; err != nil {
return nil, err
}
var row model.CrawlTaskRow
if err := r.db.WithContext(ctx).Raw(`
SELECT `+crawlColumns+`
FROM crawl_tasks t LEFT JOIN brands b ON b.id = t.brand_id
WHERE t.id = ?`, id).Scan(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (r *crawlRepository) Report(ctx context.Context, id uint32, status string, shows int, errMsg string) error {
now := uint32(time.Now().Unix())
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var cur model.CrawlTask
if err := tx.Where("id = ?", id).First(&cur).Error; err != nil {
return err
}
next := status
// 失败但还有重试额度 → 回到队列,由任意 worker 重试(网络抖动不必人工介入)。
if status == model.CrawlFailed && cur.Attempts < model.CrawlMaxAttempts {
next = model.CrawlQueued
}
return tx.Model(&model.CrawlTask{}).Where("id = ?", id).
Updates(map[string]any{
"status": next,
"shows_submitted": shows,
"last_error": errMsg,
"locked_at": 0,
"worker": "",
"updated_at": now,
}).Error
})
}
func (r *crawlRepository) List(ctx context.Context, offset, limit int) ([]model.CrawlTaskRow, int64, error) {
if limit < 1 {
limit = 30
}
var total int64
if err := r.db.WithContext(ctx).Table("crawl_tasks").Count(&total).Error; err != nil {
return nil, 0, err
}
var rows []model.CrawlTaskRow
err := r.db.WithContext(ctx).Raw(`
SELECT `+crawlColumns+`
FROM crawl_tasks t LEFT JOIN brands b ON b.id = t.brand_id
ORDER BY CASE t.status WHEN 'running' THEN 0 WHEN 'queued' THEN 1 ELSE 2 END,
t.priority DESC, t.id DESC
LIMIT ? OFFSET ?`, limit, offset).Scan(&rows).Error
if err != nil {
return nil, 0, err
}
return rows, total, nil
}
func (r *crawlRepository) SetStatus(ctx context.Context, id uint32, status string) error {
return r.db.WithContext(ctx).Model(&model.CrawlTask{}).Where("id = ?", id).
Updates(map[string]any{
"status": status,
"locked_at": 0,
"worker": "",
"attempts": 0,
"last_error": "",
"updated_at": uint32(time.Now().Unix()),
}).Error
}
func (r *crawlRepository) Delete(ctx context.Context, id uint32) error {
return r.db.WithContext(ctx).Exec(`DELETE FROM crawl_tasks WHERE id = ?`, id).Error
}
func (r *crawlRepository) BrandCrawlStates(ctx context.Context, brandIDs []uint32) (map[uint32]BrandCrawlState, error) {
out := make(map[uint32]BrandCrawlState, len(brandIDs))
if len(brandIDs) == 0 {
return out, nil
}
var rows []struct {
BrandID uint32
Status string
UpdatedAt int64
}
// DISTINCT ON 取每个品牌最近一条任务(按 updated_at 倒序),一次查询拿全状态与时间。
if err := r.db.WithContext(ctx).Raw(`
SELECT DISTINCT ON (brand_id) brand_id, status, updated_at
FROM crawl_tasks
WHERE brand_id IN ?
ORDER BY brand_id, updated_at DESC`, brandIDs).Scan(&rows).Error; err != nil {
return out, err
}
for _, row := range rows {
st := BrandCrawlState{Status: row.Status}
if row.UpdatedAt > 0 {
st.UpdatedAt = time.Unix(row.UpdatedAt, 0)
}
out[row.BrandID] = st
}
return out, nil
}
func (r *crawlRepository) Counts(ctx context.Context) (map[string]int64, error) {
var rows []struct {
Status string
N int64
}
if err := r.db.WithContext(ctx).Raw(
`SELECT status, count(*) AS n FROM crawl_tasks GROUP BY status`).Scan(&rows).Error; err != nil {
return nil, err
}
out := make(map[string]int64, len(rows))
for _, r := range rows {
out[r.Status] = r.N
}
return out, nil
}

View File

@ -58,7 +58,24 @@ type DupStat struct {
// //
// 近邻结果走 image_similar_pairs 预计算表:入库时(或 dbtool similarbackfill 回填时) // 近邻结果走 image_similar_pairs 预计算表:入库时(或 dbtool similarbackfill 回填时)
// 算一次存表,后台页面只读结果。原因见 ReplaceSimilarPairs 注释。 // 算一次存表,后台页面只读结果。原因见 ReplaceSimilarPairs 注释。
// PendingEmbedding 是一张「缺向量」的图片,供回填 worker 拉取。
// 带 Key 而不是完整 URL:worker 自己按存储基址拼(后端可能换了域名,不该把地址写死在队列里)。
type PendingEmbedding struct {
Kind string // runway | street
ImageID uint32
Key string // 存储 key(S4)或本地相对路径(/uploads/...)
}
type EmbeddingRepository interface { type EmbeddingRepository interface {
// ListPendingEmbeddings 取一批「还活着、没有向量、失败次数未达上限」的图片,按 id 升序。
//
// 媒体拆分后这条查询是向量回填的唯一入口:媒体机直传只写 key,后端不碰图片字节,
// 向量必须靠外部 worker 回传,而 worker 需要知道「还有哪些图缺向量」。
ListPendingEmbeddings(ctx context.Context, kind string, limit, maxAttempts int) ([]PendingEmbedding, error)
// CountPendingEmbeddings 统计剩余待回填数量(worker 据此决定是否继续拉)。
CountPendingEmbeddings(ctx context.Context, kind string, maxAttempts int) (int, error)
// RecordEmbeddingFailure 记一次失败(attempts+1)。达到 maxAttempts 后该图不再进队列。
RecordEmbeddingFailure(ctx context.Context, kind string, imageID uint32, reason string) error
// UpsertEmbedding 写入或更新一张图的向量((kind, image_id) 唯一)。 // UpsertEmbedding 写入或更新一张图的向量((kind, image_id) 唯一)。
UpsertEmbedding(ctx context.Context, kind string, imageID uint32, vec []float32) error UpsertEmbedding(ctx context.Context, kind string, imageID uint32, vec []float32) error
// BulkUpsertEmbedding 批量写入;vecs[i] 对应 ids[i],vecs[i] 为 nil 则跳过该行。 // BulkUpsertEmbedding 批量写入;vecs[i] 对应 ids[i],vecs[i] 为 nil 则跳过该行。
@ -175,6 +192,85 @@ func (r *embeddingRepository) UpsertEmbedding(ctx context.Context, kind string,
return r.BulkUpsertEmbedding(ctx, kind, []uint32{imageID}, [][]float32{vec}) return r.BulkUpsertEmbedding(ctx, kind, []uint32{imageID}, [][]float32{vec})
} }
// ListPendingEmbeddings 取缺失向量的图片。
//
// 三个 LEFT JOIN 各有明确目的:
// - e 判「已有没有向量」;
// - f 判「失败次数是否已到上限」——没有它,一张永远编不出来的坏图会让每轮回填都空转;
// - i.is_deleted = 0 跳过软删图,避免为已删内容白算向量。
//
// 表名/列名来自 ownerOf 常量,kind 走参数绑定,无注入面。
func (r *embeddingRepository) ListPendingEmbeddings(ctx context.Context, kind string, limit, maxAttempts int) ([]PendingEmbedding, error) {
owner, ok := ownerOf(kind)
if !ok {
return nil, nil
}
if limit <= 0 {
limit = 50
}
if maxAttempts <= 0 {
maxAttempts = 5
}
var rows []struct {
ImageID uint32 `gorm:"column:image_id"`
Image string `gorm:"column:image"`
}
if err := r.db.WithContext(ctx).Raw(
"SELECT i.id AS image_id, i.image AS image "+
"FROM "+owner.table+" AS i "+
"LEFT JOIN image_embeddings AS e ON e.kind = ? AND e.image_id = i.id "+
"LEFT JOIN image_embedding_failures AS f ON f.kind = ? AND f.image_id = i.id "+
"WHERE i.is_deleted = 0 AND i.image <> '' AND e.image_id IS NULL "+
" AND coalesce(f.attempts, 0) < ? "+
"ORDER BY i.id LIMIT ?",
kind, kind, maxAttempts, limit).Scan(&rows).Error; err != nil {
return nil, err
}
out := make([]PendingEmbedding, 0, len(rows))
for _, row := range rows {
out = append(out, PendingEmbedding{Kind: kind, ImageID: row.ImageID, Key: row.Image})
}
return out, nil
}
// CountPendingEmbeddings 统计剩余量(与 ListPendingEmbeddings 同口径,避免两边不一致)。
func (r *embeddingRepository) CountPendingEmbeddings(ctx context.Context, kind string, maxAttempts int) (int, error) {
owner, ok := ownerOf(kind)
if !ok {
return 0, nil
}
if maxAttempts <= 0 {
maxAttempts = 5
}
var n int
if err := r.db.WithContext(ctx).Raw(
"SELECT count(*) FROM "+owner.table+" AS i "+
"LEFT JOIN image_embeddings AS e ON e.kind = ? AND e.image_id = i.id "+
"LEFT JOIN image_embedding_failures AS f ON f.kind = ? AND f.image_id = i.id "+
"WHERE i.is_deleted = 0 AND i.image <> '' AND e.image_id IS NULL "+
" AND coalesce(f.attempts, 0) < ?",
kind, kind, maxAttempts).Scan(&n).Error; err != nil {
return 0, err
}
return n, nil
}
// RecordEmbeddingFailure 累加失败次数。
// 用 upsert 而不是先查后写:worker 可能并发回传,两步写法会丢计数。
func (r *embeddingRepository) RecordEmbeddingFailure(ctx context.Context, kind string, imageID uint32, reason string) error {
if len(reason) > 300 {
reason = reason[:300]
}
return r.db.WithContext(ctx).Exec(`
INSERT INTO image_embedding_failures (kind, image_id, attempts, last_error, updated_at)
VALUES (?, ?, 1, ?, ?)
ON CONFLICT (kind, image_id) DO UPDATE
SET attempts = image_embedding_failures.attempts + 1,
last_error = EXCLUDED.last_error,
updated_at = EXCLUDED.updated_at`,
kind, imageID, reason, uint32(time.Now().Unix())).Error
}
// BulkUpsertEmbedding 逐行写入而非 unnest 批量:日入库仅几百张,量级极小, // BulkUpsertEmbedding 逐行写入而非 unnest 批量:日入库仅几百张,量级极小,
// 换来的是可读性与低出错概率。ON CONFLICT 保证同一图重复入库不会写出多行。 // 换来的是可读性与低出错概率。ON CONFLICT 保证同一图重复入库不会写出多行。
func (r *embeddingRepository) BulkUpsertEmbedding(ctx context.Context, kind string, ids []uint32, vecs [][]float32) error { func (r *embeddingRepository) BulkUpsertEmbedding(ctx context.Context, kind string, ids []uint32, vecs [][]float32) error {

View File

@ -65,6 +65,16 @@ func NewBackstage(opt BackstageOptions) *gin.Engine {
internal.Use(middleware.IngestAuth(*opt.IngestAuth)) internal.Use(middleware.IngestAuth(*opt.IngestAuth))
internal.POST("/ingest", opt.Ingest.Submit) internal.POST("/ingest", opt.Ingest.Submit)
internal.GET("/crawl/brands", opt.Ingest.CrawlBrands) internal.GET("/crawl/brands", opt.Ingest.CrawlBrands)
// spider 服务化:领取品牌抓取任务 / 回报结果(与 ingest 共用同一套 HMAC 验签)。
internal.GET("/crawl/tasks", opt.Backstage.WorkerTask)
internal.POST("/crawl/tasks/:id/result", opt.Backstage.WorkerReport)
// 媒体机直传:签发 S4 PUT 预签名 URL(媒体机不持有 AK/SK)。
internal.POST("/upload-url", opt.Backstage.WorkerUploadURL)
// 向量回填:媒体机直传后后端不碰图片字节,向量由外部 worker 拉图编码后回传。
// GET /embeddings/pending → 缺向量的图片(含可直接下载的 url)
// POST /embeddings → 回传向量(写库 + 重算近邻)与失败(累加计数)
internal.GET("/embeddings/pending", opt.Backstage.WorkerPendingEmbeddings)
internal.POST("/embeddings", opt.Backstage.WorkerEmbeddings)
} else { } else {
r.POST("/admin/internal/ingest", func(c *gin.Context) { r.POST("/admin/internal/ingest", func(c *gin.Context) {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "ingest disabled"}) c.JSON(http.StatusServiceUnavailable, gin.H{"error": "ingest disabled"})
@ -118,6 +128,14 @@ func NewBackstage(opt BackstageOptions) *gin.Engine {
admin.POST("/street-snaps/:id/images/:img/delete", opt.Backstage.StreetSnapImageDelete) admin.POST("/street-snaps/:id/images/:img/delete", opt.Backstage.StreetSnapImageDelete)
admin.POST("/street-snaps/:id/cover", opt.Backstage.StreetSnapSetCover) admin.POST("/street-snaps/:id/cover", opt.Backstage.StreetSnapSetCover)
admin.POST("/street-snaps/:id/toggle", opt.Backstage.StreetSnapToggleDeleted) admin.POST("/street-snaps/:id/toggle", opt.Backstage.StreetSnapToggleDeleted)
// 品牌抓取任务控制面:后台决定爬哪些品牌,spider 服务来领取执行。
// 列表 + 批量按前缀入队 + 单品牌入队 + 取消/重试/删除。
admin.GET("/crawl", opt.Backstage.CrawlList)
admin.POST("/crawl/bulk", opt.Backstage.CrawlBulk)
admin.POST("/crawl/brands/:uid", opt.Backstage.CrawlEnqueue)
admin.POST("/crawl/tasks/:id/retry", opt.Backstage.CrawlRetry)
admin.POST("/crawl/tasks/:id/cancel", opt.Backstage.CrawlCancel)
admin.POST("/crawl/tasks/:id/delete", opt.Backstage.CrawlDelete)
// 爬虫入库任务监控:列出处理进度 + 失败重试 // 爬虫入库任务监控:列出处理进度 + 失败重试
admin.GET("/ingest-jobs", opt.Backstage.IngestJobs) admin.GET("/ingest-jobs", opt.Backstage.IngestJobs)
admin.POST("/ingest-jobs/:id/retry", opt.Backstage.IngestJobRetry) admin.POST("/ingest-jobs/:id/retry", opt.Backstage.IngestJobRetry)

View File

@ -85,6 +85,11 @@ func (f *fakeBrand) CrawlTasks(ctx context.Context, brandID uint32) ([]dto.Crawl
return []dto.CrawlBrand{{BrandUID: "001DESke", Name: "Acne Studios"}}, nil return []dto.CrawlBrand{{BrandUID: "001DESke", Name: "Acne Studios"}}, nil
} }
// StatsByIDs 后台品牌列表的数据量统计(测试里返回空表即可)。
func (f *fakeBrand) StatsByIDs(ctx context.Context, ids []uint32) (map[uint32]repository.BrandStats, error) {
return map[uint32]repository.BrandStats{}, nil
}
// 编译期接口满足性检查 // 编译期接口满足性检查
var ( var (
_ service.ArticleService = (*fakeArticle)(nil) _ service.ArticleService = (*fakeArticle)(nil)

View File

@ -20,6 +20,8 @@ type BrandService interface {
// AdminList 后台品牌列表:字母索引不要求走秀已发布, // AdminList 后台品牌列表:字母索引不要求走秀已发布,
// 使只有待审 / 驳回走秀的品牌仍出现在后台(公开 List 读视图,会挡掉它们)。 // 使只有待审 / 驳回走秀的品牌仍出现在后台(公开 List 读视图,会挡掉它们)。
AdminList(ctx context.Context, q dto.BrandQuery) ([]dto.PublicBrand, int64, error) AdminList(ctx context.Context, q dto.BrandQuery) ([]dto.PublicBrand, int64, error)
// StatsByIDs 批量统计品牌的图集数 / 图片数(后台列表展示数据量)。
StatsByIDs(ctx context.Context, ids []uint32) (map[uint32]repository.BrandStats, error)
// Hot 返回热门品牌(按走秀图片数排名前 limit 个)并附带各品牌档案数。 // Hot 返回热门品牌(按走秀图片数排名前 limit 个)并附带各品牌档案数。
// 供 SSG 构建期(RunwayLooks 侧栏 / 品牌弹窗 HOT 标签)一次性取前 N 个热门品牌; // 供 SSG 构建期(RunwayLooks 侧栏 / 品牌弹窗 HOT 标签)一次性取前 N 个热门品牌;
// 复用 List 的 featured 收窄逻辑(FeaturedIDs("images", limit)),与首页精选口径一致。 // 复用 List 的 featured 收窄逻辑(FeaturedIDs("images", limit)),与首页精选口径一致。
@ -201,6 +203,11 @@ func (s *brandService) Delete(ctx context.Context, uid string) error {
// CrawlTasks 返回爬虫取任务所需的品牌列表:编码 brand_uid + 英文品牌名。 // CrawlTasks 返回爬虫取任务所需的品牌列表:编码 brand_uid + 英文品牌名。
// 爬虫直接拿 brand_uid 上送 ingest,无需自己编码;name 用于拼 vogue 抓取 URL。 // 爬虫直接拿 brand_uid 上送 ingest,无需自己编码;name 用于拼 vogue 抓取 URL。
// StatsByIDs 批量统计品牌的图集数 / 图片数(后台列表用)。
func (s *brandService) StatsByIDs(ctx context.Context, ids []uint32) (map[uint32]repository.BrandStats, error) {
return s.brands.StatsByIDs(ctx, ids)
}
func (s *brandService) CrawlTasks(ctx context.Context, brandID uint32) ([]dto.CrawlBrand, error) { func (s *brandService) CrawlTasks(ctx context.Context, brandID uint32) ([]dto.CrawlBrand, error) {
brands, err := s.brands.CrawlBrands(ctx, brandID) brands, err := s.brands.CrawlBrands(ctx, brandID)
if err != nil { if err != nil {

View File

@ -0,0 +1,83 @@
package service
import (
"context"
"testing"
"fashionapi/internal/dto"
)
// TestFetchLookImagesUsesProvidedKeys 验证「媒体机直传」分支:
// payload 里带 main_key / detail_keys 时,worker 不应下载、不应上传,
// 而是直接把 key 当作要落库的存储地址,且 cover 取首张主图。
//
// 这条路径是媒体拆分的核心保证:一旦这里退化成下载 URL,
// 后端又会重新背上带宽瓶颈,而错误会非常隐蔽(只是变慢,不报错)。
func TestFetchLookImagesUsesProvidedKeys(t *testing.T) {
s := &IngestService{} // 不装配 uploader/vision:走 key 分支时它们都不该被用到
looks := []dto.RunwayLook{
{
MainKey: "runway/aaaa.jpg",
MainSize: 100,
DetailKeys: []string{"runway/bbbb.jpg"},
DetailSizes: []int64{200},
},
{
// 第二张混用旧形态:没有 key,带 URL(这里只验主图,不触发网络)
MainKey: "runway/cccc.jpg",
MainSize: 300,
},
}
cover, rows, vecs, keys, failed, _ := s.fetchLookImages(context.Background(), looks, "runway")
if failed {
t.Fatal("key 分支不应被判定为失败")
}
if cover != "runway/aaaa.jpg" {
t.Fatalf("cover 应取首张主图 key,实际 %q", cover)
}
if len(rows) != 3 {
t.Fatalf("应落 3 行(2 主图 + 1 细节图),实际 %d", len(rows))
}
if rows[0].Image != "runway/aaaa.jpg" || rows[0].IsDetail != 0 {
t.Fatalf("第 1 行应为 aaaa 主图,实际 %+v", rows[0])
}
if rows[1].Image != "runway/bbbb.jpg" || rows[1].IsDetail != 1 {
t.Fatalf("第 2 行应为 bbbb 细节图,实际 %+v", rows[1])
}
if rows[1].ImageSize != 200 || rows[0].ImageSize != 100 {
t.Fatalf("字节数应来自 payload,实际 %d / %d", rows[0].ImageSize, rows[1].ImageSize)
}
// 媒体机上传的 key 不进回滚清单(内容寻址、可能与他人共享)
if len(keys) != 0 {
t.Fatalf("key 分支不应产生回滚清单,实际 %v", keys)
}
if len(vecs) != len(rows) {
t.Fatalf("vecs 应与 rows 对齐,实际 %d vs %d", len(vecs), len(rows))
}
for i, v := range vecs {
if v != nil {
t.Fatalf("第 %d 张不应带向量(向量由 CLIP 回填),实际 %v", i, v)
}
}
if got := countMainImages(rows); got != 2 {
t.Fatalf("主图数应为 2,实际 %d", got)
}
}
// TestSha1FromKey 锁住「从内容寻址 key 反推 sha1」的行为:批次内去重依赖它,
// 若解析退化(例如带上目录前缀),同一张图会被重复落库。
func TestSha1FromKey(t *testing.T) {
cases := map[string]string{
"runway/4080435af23d8f80bbbcf9e6f55a5f1d00c49685.jpg": "4080435af23d8f80bbbcf9e6f55a5f1d00c49685",
"street/abc123.png": "abc123",
"plain-key": "plain-key",
}
for in, want := range cases {
if got := sha1FromKey(in); got != want {
t.Fatalf("sha1FromKey(%q) = %q,期望 %q", in, got, want)
}
}
}

View File

@ -5,6 +5,7 @@ import (
"crypto/sha1" "crypto/sha1"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"log" "log"
@ -271,9 +272,17 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob,
return return
} }
if _, err := s.brandRepo.FindByID(ctx, brandID); err != nil { if _, err := s.brandRepo.FindByID(ctx, brandID); err != nil {
// 必须区分「品牌真不存在」与「查询本身失败」:前者永久失败,后者应退避重试。
// 曾把两者一律记为 "brand not found",导致数据库抖动时一批任务被误判为永久失败,
// 且真实错误(连接超时 / 池耗尽)在队列里完全看不到,排查时无线索。
log.Printf("[ingest][diag] job=%d brand_uid=%q decoded=%d err=%v", job.ID, p.BrandUID, brandID, err)
if errors.Is(err, repository.ErrNotFound) {
_ = s.repo.MarkFailed(ctx, job.ID, "brand not found") _ = s.repo.MarkFailed(ctx, job.ID, "brand not found")
return return
} }
s.failOrRetry(ctx, job.ID, "brand lookup: "+err.Error())
return
}
// 2) 实体键查重并决定分支(单表模型) // 2) 实体键查重并决定分支(单表模型)
// 标题缺年份(p.Year=0)时,用发布日期的年份补全年份,使 year / season_code 不再为空, // 标题缺年份(p.Year=0)时,用发布日期的年份补全年份,使 year / season_code 不再为空,
@ -454,22 +463,48 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
failed := false failed := false
// 摊平:顺序 = 逐个 look 先主图、再其细节图,与原串行遍历顺序完全一致。 // 摊平:顺序 = 逐个 look 先主图、再其细节图,与原串行遍历顺序完全一致。
// 每张图优先取「媒体机已上传的 key」,没有 key 才回落到「下载原始 URL」。
tasks := make([]lookImageTask, 0, len(looks)) tasks := make([]lookImageTask, 0, len(looks))
for li, look := range looks { for li, look := range looks {
lookIdx := li + 1 lookIdx := li + 1
if look.Main != "" { if look.MainKey != "" {
tasks = append(tasks, lookImageTask{key: look.MainKey, size: look.MainSize, lookIdx: lookIdx})
} else if look.Main != "" {
tasks = append(tasks, lookImageTask{url: look.Main, lookIdx: lookIdx}) tasks = append(tasks, lookImageTask{url: look.Main, lookIdx: lookIdx})
} }
for di, d := range look.Details { // 细节图数量取两个数组的较大者:媒体机直传时可能只给 DetailKeys 而不给 URL,
if d == "" { // 只按 Details 遍历会把这些细节图静默丢掉。
detailN := len(look.Details)
if len(look.DetailKeys) > detailN {
detailN = len(look.DetailKeys)
}
for di := 0; di < detailN; di++ {
var d string
if di < len(look.Details) {
d = look.Details[di]
}
hasKey := di < len(look.DetailKeys) && look.DetailKeys[di] != ""
if d == "" && !hasKey {
continue continue
} }
tasks = append(tasks, lookImageTask{url: d, lookIdx: lookIdx, isDetail: true, detailNo: di + 1}) t := lookImageTask{lookIdx: lookIdx, isDetail: true, detailNo: di + 1}
if hasKey {
t.key = look.DetailKeys[di]
if di < len(look.DetailSizes) {
t.size = look.DetailSizes[di]
}
} else {
t.url = d
}
tasks = append(tasks, t)
} }
} }
// 并发段:下载 + 上传(耗时大头)。 // 并发段:下载 + 上传(耗时大头)。key 路径的图已在存储里,直接构造结果跳过 IO。
results := concurrentFetch(ctx, len(tasks), func(c context.Context, i int) downloadResult { results := concurrentFetch(ctx, len(tasks), func(c context.Context, i int) downloadResult {
if k := tasks[i].key; k != "" {
return downloadResult{url: k, key: k, sha1: sha1FromKey(k), size: tasks[i].size}
}
return s.downloadAndUpload(c, tasks[i].url, prefix) return s.downloadAndUpload(c, tasks[i].url, prefix)
}) })
@ -484,7 +519,12 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL
failed = true failed = true
continue continue
} }
// 只把「本次任务自己上传的 key」纳入回滚清单。
// 媒体机直传的 key 不进清单:它是内容寻址的,可能被别的图集共享,
// 失败时删掉会连带弄坏别人已入库的图;这类孤儿由存储侧清理策略负责。
if task.key == "" {
keys = append(keys, r.key) keys = append(keys, r.key)
}
if seen[r.sha1] { if seen[r.sha1] {
continue continue
} }
@ -830,11 +870,29 @@ func concurrentFetch(ctx context.Context, n int, fn func(context.Context, int) d
// 是主图还是第几张细节图,供并发下载完成后重建 Name / 主副图归属 / IsDetail。 // 是主图还是第几张细节图,供并发下载完成后重建 Name / 主副图归属 / IsDetail。
type lookImageTask struct { type lookImageTask struct {
url string url string
key string // 非空 = 媒体机已上传,worker 跳过下载/上传,直接落库
size int64 // key 路径下由 spider 上报的字节数(0=未知)
lookIdx int // 1-based look 序号 lookIdx int // 1-based look 序号
isDetail bool // true=细节图 isDetail bool // true=细节图
detailNo int // 细节图序号(1-based),主图为 0 detailNo int // 细节图序号(1-based),主图为 0
} }
// sha1FromKey 从内容寻址 key(runway/<sha1>.jpg 或 street/<sha1>.png)反推 sha1,
// 供批次内去重用;key 不符合该形态时退化为 key 本身(仍能起到去重作用)。
func sha1FromKey(key string) string {
base := key
if i := strings.LastIndex(base, "/"); i >= 0 {
base = base[i+1:]
}
if i := strings.LastIndex(base, "."); i > 0 {
base = base[:i]
}
if base == "" {
return key
}
return base
}
// downloadAndUpload 把单张远程图下载后上传到存储(S4优先,失败兜底本地)。 // downloadAndUpload 把单张远程图下载后上传到存储(S4优先,失败兜底本地)。
// 内容 sha1 用于生成内容寻址的存储 key(重爬幂等)与批次内去重;访问地址可直接写入数据库 // 内容 sha1 用于生成内容寻址的存储 key(重爬幂等)与批次内去重;访问地址可直接写入数据库
// (S4为完整 https URL,本地为相对 /uploads 路径)。 // (S4为完整 https URL,本地为相对 /uploads 路径)。