From 151e09ab3063e8a95f70a2b6af5cb88c30b0e423 Mon Sep 17 00:00:00 2001 From: toom1996 <23cm.cn@gmail.com> Date: Sun, 25 Oct 2026 00:37:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8A=93=E5=8F=96=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E9=98=9F=E5=88=97=20+=20=E5=AA=92=E4=BD=93=E6=9C=BA=E7=9B=B4?= =?UTF-8?q?=E4=BC=A0=20+=20=E5=90=91=E9=87=8F=E5=9B=9E=E5=A1=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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、字节数、回滚清单、向量占位 --- cmd/server/main.go | 22 +- db/migrations/2026-10-24-01-crawl-tasks.sql | 23 ++ .../2026-10-24-01-embedding-failures.sql | 16 + internal/dto/brand.go | 5 + internal/dto/ingest.go | 15 + internal/handler/backstage_crawl.go | 347 ++++++++++++++++++ internal/handler/backstage_embeddings.go | 170 +++++++++ internal/handler/backstage_handler.go | 72 +++- internal/handler/backstage_handler_test.go | 5 + internal/handler/templates/pages/brands.html | 60 ++- internal/handler/templates/pages/crawl.html | 44 +++ internal/handler/templates/partials.html | 1 + internal/model/crawl_task.go | 57 +++ internal/pkg/storage/s4.go | 34 ++ internal/repository/brand_repository.go | 67 +++- internal/repository/crawl_repository.go | 338 +++++++++++++++++ internal/repository/embedding_repository.go | 96 +++++ internal/router/backstage.go | 18 + internal/router/backstage_test.go | 5 + internal/service/brand_service.go | 7 + internal/service/ingest_keypath_test.go | 83 +++++ internal/service/ingest_service.go | 78 +++- 22 files changed, 1535 insertions(+), 28 deletions(-) create mode 100644 db/migrations/2026-10-24-01-crawl-tasks.sql create mode 100644 db/migrations/2026-10-24-01-embedding-failures.sql create mode 100644 internal/handler/backstage_crawl.go create mode 100644 internal/handler/backstage_embeddings.go create mode 100644 internal/handler/templates/pages/crawl.html create mode 100644 internal/model/crawl_task.go create mode 100644 internal/repository/crawl_repository.go create mode 100644 internal/service/ingest_keypath_test.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 2799f33..3903006 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -13,6 +13,8 @@ import ( "os" "os/signal" "path/filepath" + "strconv" + "strings" "syscall" "time" @@ -217,6 +219,12 @@ func main() { // 后台页面因此只需读一次小查询（原先是页面内对每张图实时检索，一个 116 张图的图集要 5 秒）。 ingestSvc.SetSimilar(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，事务内完成）与同一套存储上传 // （imgUp = S4/本地主上传器，sha1 内容寻址），保证与爬虫入库产出的 key / 行结构一致。 backstageHandler.SetManual(service.NewManualArticleService(ingestRepo, brandRepo, imgUp, localUp, visionClient, embedRepo)) @@ -275,8 +283,18 @@ func main() { // 7.6 启动爬虫入库 worker（异步处理 ingest_jobs 队列）。 // 用独立可取消 ctx，优雅关闭时随主流程一起退出。 workerCtx, workerCancel := context.WithCancel(context.Background()) - go ingestSvc.Run(workerCtx, 2, 3*time.Second) - slog.Info("入库 worker 已启动", "queue", "ingest_jobs") + // 并发数可配：批量重爬时单进程 2 个 worker 是吞吐瓶颈（配合 INGEST_IMAGE_CONCURRENCY 决定总并发）。 + // 默认仍是 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{ Addr: ":" + cfg.Server.BackstagePort, Handler: backstageEngine, diff --git a/db/migrations/2026-10-24-01-crawl-tasks.sql b/db/migrations/2026-10-24-01-crawl-tasks.sql new file mode 100644 index 0000000..c72c6bb --- /dev/null +++ b/db/migrations/2026-10-24-01-crawl-tasks.sql @@ -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); diff --git a/db/migrations/2026-10-24-01-embedding-failures.sql b/db/migrations/2026-10-24-01-embedding-failures.sql new file mode 100644 index 0000000..648f1db --- /dev/null +++ b/db/migrations/2026-10-24-01-embedding-failures.sql @@ -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) +); diff --git a/internal/dto/brand.go b/internal/dto/brand.go index cc0aacf..e629b0b 100644 --- a/internal/dto/brand.go +++ b/internal/dto/brand.go @@ -16,6 +16,11 @@ type BrandQuery struct { // Featured 仅保留"代表品牌"精选集合（按指标排名的前 N 个）。 Featured bool Metric string // images（默认）| shows + + // Sort 后台列表排序：空 / "name"（默认，按英文名）；"updated" 按「最近有动静」倒序 + // （取该品牌下走秀图集的最大 updated_at 与抓取任务的最大 updated_at，两者取大）。 + // 公开侧忽略此字段，避免改变前台 A-Z 索引的既定顺序。 + Sort string Limit int // 精选集合大小，默认 200 // Locale 语言：cn | en。空时由 service 回落默认 en。 diff --git a/internal/dto/ingest.go b/internal/dto/ingest.go index ad60ae0..51dfa0d 100644 --- a/internal/dto/ingest.go +++ b/internal/dto/ingest.go @@ -37,7 +37,22 @@ type RunwayIngest struct { // RunwayLook 一场秀中的一个 look：1 张主图 + 0..N 张细节图。 // 主图即对外默认展示的走秀图；细节图挂靠到对应主图，由「细节入口」按需展开。 +// +// 图片有两种来源，二选一（同一条 payload 内可混用）： +// +// ① 旧路径：Main / Details 给原始 URL，worker 自己下载 + 上传存储（默认行为）； +// ② 新路径（媒体机直传）：MainKey / DetailKeys 给「已上传到 S4 的对象 key」， +// worker 直接落库，不再下载、不再上传。 +// +// 新路径的意义：抓取与入库分离后，下载/上传这类带宽密集的活由部署在快网络里的 +// spider 完成，后端只做业务写库，吞吐不再被后端机器的出口带宽卡住。 +// key 由 spider 按内容 sha1 生成（runway/.jpg），因此重复上传天然幂等。 type RunwayLook struct { Main string `json:"main"` // 主图（look）原始 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=未知） } diff --git a/internal/handler/backstage_crawl.go b/internal/handler/backstage_crawl.go new file mode 100644 index 0000000..56e0c67 --- /dev/null +++ b/internal/handler/backstage_crawl.go @@ -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}) +} diff --git a/internal/handler/backstage_embeddings.go b/internal/handler/backstage_embeddings.go new file mode 100644 index 0000000..bc8d344 --- /dev/null +++ b/internal/handler/backstage_embeddings.go @@ -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}) +} diff --git a/internal/handler/backstage_handler.go b/internal/handler/backstage_handler.go index ed21ad6..7704402 100644 --- a/internal/handler/backstage_handler.go +++ b/internal/handler/backstage_handler.go @@ -16,6 +16,7 @@ import ( "fashionapi/internal/pkg/hashid" "fashionapi/internal/pkg/pagination" "fashionapi/internal/pkg/season" + "fashionapi/internal/pkg/storage" "fashionapi/internal/repository" "fashionapi/internal/service" @@ -45,8 +46,25 @@ type BackstageHandler struct { // manual 后台「手工新建文章」（可为 nil：未装配时新建入口提示不可用）。 // 用 SetManual 注入，避免改构造签名（同 SetSimilar 的取舍）。 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 注入手工建稿服务（装配期调用；未注入时新建页提示不可用）。 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", Page: atoiDefault(c.Query("page"), 1), Locale: "en", + // 后台可切换「按最近更新排序」（走秀或抓取任务的最新时间） + Sort: c.Query("sort"), } items, total, err := h.brand.AdminList(c.Request.Context(), q) if err != nil { @@ -157,13 +177,53 @@ func (h *BackstageHandler) Brands(c *gin.Context) { p.Page = 1 } 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{ - "Items": items, - "Total": total, - "Keyword": q.Keyword, - "Letter": q.Letter, - "All": q.All, - "Letters": lettersAtoZ(), + "Items": items, + "Total": total, + "Keyword": q.Keyword, + "Letter": q.Letter, + "All": q.All, + "Sort": q.Sort, + "Letters": lettersAtoZ(), + "CrawlStates": crawlStates, + "Stats": stats, + "Added": atoiDefault(c.Query("added"), 0), "Pager": pagination.Pager{ PrevURL: pagination.PageLink(c, meta.PrevPage), NextURL: pagination.PageLink(c, meta.NextPage), diff --git a/internal/handler/backstage_handler_test.go b/internal/handler/backstage_handler_test.go index eca7d17..be7b7e9 100644 --- a/internal/handler/backstage_handler_test.go +++ b/internal/handler/backstage_handler_test.go @@ -28,6 +28,7 @@ func TestBackstageTemplatesParsed(t *testing.T) { "articles", "brand-form", "brands", + "crawl", "dashboard", "error", "ingest-jobs", @@ -73,6 +74,10 @@ func TestBackstageTemplatesRender(t *testing.T) { "Images": []any{}, // 列表数据 "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 覆盖） "Article": &dto.PublicArticleDetail{UID: "abc123"}, "Snap": &dto.PublicStreetSnapDetail{UID: "abc123"}, diff --git a/internal/handler/templates/pages/brands.html b/internal/handler/templates/pages/brands.html index 552ccff..ea5d938 100644 --- a/internal/handler/templates/pages/brands.html +++ b/internal/handler/templates/pages/brands.html @@ -8,21 +8,69 @@
{{if .All}}{{end}} +{{if .Sort}}{{end}} {{if .Keyword}}清除{{end}} {{if .All}}只看有图集的品牌{{else}}显示全部品牌（含无图集）{{end}}
-

默认只显示「已有图集」的品牌；品牌目录共 2 万余条，其余请用上面的搜索或开关查看。

+ +

+ 默认只显示「已有图集」的品牌；品牌目录共 2 万余条，其余请用上面的搜索或开关查看。 + 排序： + {{if eq .Sort "updated"}}最近更新{{else}}最近更新{{end}} + · + {{if eq .Sort "updated"}}按名字{{else}}按名字{{end}} +

+ +{{if .Added}}

✅ 本次已加入抓取队列 {{.Added}} 个品牌（已在队列中的不会重复添加）。

{{end}} +
-全部 -中文 -{{range $l := .Letters}}{{$l}}{{end}} +全部 +中文 +{{range $l := .Letters}}{{$l}}{{end}}
+ +{{/* 批量操作表单：勾选框用 form="bulk" 关联，避免与行内表单嵌套 */}} +
+
+ + + （勾选本页任意行，先点复选框列头可全选本页） +
+ - + + + + -{{range .Items}}{{end}} +{{range .Items}}{{$cs := index $.CrawlStates .UID}}{{$st := index $.Stats .UID}} + + + + + + + + + +{{end}}
编码 ID英文名中文名操作
编码 ID英文名中文名图集图片抓取状态上次抓取操作
{{.UID}}{{.NameEn}}{{.NameCn}}编辑
{{.UID}}{{.NameEn}}{{.NameCn}}{{if $st}}{{$st.Runways}}{{else}}0{{end}}{{if $st}}{{$st.Images}}{{else}}0{{end}}{{if $cs}}{{$cs.Label}}{{else}}—{{end}}{{if $cs}}{{$cs.LastAt}}{{else}}—{{end}} + 编辑 +
+ + +
+
+ +
+
{{template "pager" .Pager}} diff --git a/internal/handler/templates/pages/crawl.html b/internal/handler/templates/pages/crawl.html new file mode 100644 index 0000000..eb263ee --- /dev/null +++ b/internal/handler/templates/pages/crawl.html @@ -0,0 +1,44 @@ +{{define "title"}}抓取任务 · 管理后台{{end}} +{{define "barLabel"}}管理后台 · 品牌抓取任务{{end}} +{{define "barRight"}}去品牌管理加任务 · 退出{{end}} +{{define "content"}} +{{template "bar" .}} +
+

抓取任务

+

+ 任务是「从品牌管理页加入」的：勾选品牌 → 选场次上限 → 加入抓取队列。 + 这里看的是执行状态与失败原因。真正干活的是 spider 服务（可在另一台机器）。 +

+

+ 待领取 {{.Queued}} · 执行中 {{.Running}} · 已完成 {{.Done}} · 失败 {{.Failed}}，共 {{.Total}} 条。 + 「已提交 0 场」不是错误——Vogue 上很多品牌没有走秀页面（实测 A 开头里约 75%）。 +

+ + + + + + +{{range .Rows}} + + + + + + + + + + +{{end}} + +
ID品牌状态场次上限已提交尝试Worker更新时间错误操作
{{.ID}}{{.BrandName}}{{.StatusLabel}}{{if eq .MaxShows 0}}全量{{else}}{{.MaxShows}}{{end}}{{.Shows}}{{.Attempts}}{{if .Worker}}{{.Worker}}{{else}}—{{end}}{{.UpdatedAt}}{{.LastError}} +
+ {{if .CanRetry}}
{{end}} + {{if .CanCancel}}
{{end}} +
+
+
+{{template "pager" .Pager}} +
+{{end}} diff --git a/internal/handler/templates/partials.html b/internal/handler/templates/partials.html index 86cae5a..465da9f 100644 --- a/internal/handler/templates/partials.html +++ b/internal/handler/templates/partials.html @@ -18,6 +18,7 @@ 首页 文章{{if gt .PendingReviews 0}}{{.PendingReviews}}{{end}} 品牌 + 抓取任务 入库任务 用户 diff --git a/internal/model/crawl_task.go b/internal/model/crawl_task.go new file mode 100644 index 0000000..52882d3 --- /dev/null +++ b/internal/model/crawl_task.go @@ -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"` +} diff --git a/internal/pkg/storage/s4.go b/internal/pkg/storage/s4.go index 4a7b1ba..ee24a80 100644 --- a/internal/pkg/storage/s4.go +++ b/internal/pkg/storage/s4.go @@ -3,8 +3,10 @@ package storage import ( "bytes" "context" + "errors" "fmt" "strings" + "time" "github.com/aws/aws-sdk-go-v2/aws" 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 推导虚拟主机风格访问域名： // endpoint=https://s3.bitiful.net, bucket=toom → https://toom.s3.bitiful.net func deriveS4BaseURL(endpoint, bucket string) string { diff --git a/internal/repository/brand_repository.go b/internal/repository/brand_repository.go index 36febf3..770dca3 100644 --- a/internal/repository/brand_repository.go +++ b/internal/repository/brand_repository.go @@ -52,6 +52,16 @@ type BrandRepository interface { // 规则：先按品牌总图片数取前 limit 个热门品牌，再为每个品牌取最新一篇走秀。 // 返回的是仓储层裸数据（数字主键），对外编码由 service 层负责。 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。 @@ -174,11 +184,30 @@ func (r *brandRepository) list(ctx context.Context, q dto.BrandQuery, restrictID return []model.Brand{}, 0, nil } - var items []model.Brand - if err := r.db.WithContext(ctx). + // 后台可按「最近有动静」排序：走秀图集的 updated_at 与抓取任务的 updated_at 取大。 + // + // 性能注意：不要写成逐行的相关子查询（2.3 万品牌 × 2 个子查询实测 2.7 秒）。 + // 这里先把两张小表各自按品牌聚合（brand_runways / crawl_tasks 都只有几百行）， + // 再 LEFT JOIN 到品牌表排序 —— 一次哈希连接 + 一次排序，量级降到百毫秒。 + // 公开侧不传 Sort，保持原有 name_en ASC，前台顺序不变。 + itemsQ := r.db.WithContext(ctx). Model(&model.Brand{}). - Scopes(scope). - Order("name_en ASC"). + Scopes(scope) + 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()). Limit(q.Size). Find(&items).Error; err != nil { @@ -319,6 +348,36 @@ func (r *brandRepository) CrawlBrands(ctx context.Context, brandID uint32) ([]mo // 3. 结果按热门品牌排名顺序返回 // // 注意：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) { if limit < 1 { limit = 20 diff --git a/internal/repository/crawl_repository.go b/internal/repository/crawl_repository.go new file mode 100644 index 0000000..a0104d0 --- /dev/null +++ b/internal/repository/crawl_repository.go @@ -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 +} diff --git a/internal/repository/embedding_repository.go b/internal/repository/embedding_repository.go index 1cd7d78..1fe667e 100644 --- a/internal/repository/embedding_repository.go +++ b/internal/repository/embedding_repository.go @@ -58,7 +58,24 @@ type DupStat struct { // // 近邻结果走 image_similar_pairs 预计算表：入库时（或 dbtool similarbackfill 回填时） // 算一次存表，后台页面只读结果。原因见 ReplaceSimilarPairs 注释。 +// PendingEmbedding 是一张「缺向量」的图片，供回填 worker 拉取。 +// 带 Key 而不是完整 URL：worker 自己按存储基址拼（后端可能换了域名，不该把地址写死在队列里）。 +type PendingEmbedding struct { + Kind string // runway | street + ImageID uint32 + Key string // 存储 key（S4）或本地相对路径（/uploads/...） +} + 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(ctx context.Context, kind string, imageID uint32, vec []float32) error // 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}) } +// 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 批量：日入库仅几百张，量级极小， // 换来的是可读性与低出错概率。ON CONFLICT 保证同一图重复入库不会写出多行。 func (r *embeddingRepository) BulkUpsertEmbedding(ctx context.Context, kind string, ids []uint32, vecs [][]float32) error { diff --git a/internal/router/backstage.go b/internal/router/backstage.go index 3bd8d6e..19202aa 100644 --- a/internal/router/backstage.go +++ b/internal/router/backstage.go @@ -65,6 +65,16 @@ func NewBackstage(opt BackstageOptions) *gin.Engine { internal.Use(middleware.IngestAuth(*opt.IngestAuth)) internal.POST("/ingest", opt.Ingest.Submit) 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 { r.POST("/admin/internal/ingest", func(c *gin.Context) { 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/cover", opt.Backstage.StreetSnapSetCover) 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.POST("/ingest-jobs/:id/retry", opt.Backstage.IngestJobRetry) diff --git a/internal/router/backstage_test.go b/internal/router/backstage_test.go index c7fc5ae..c842789 100644 --- a/internal/router/backstage_test.go +++ b/internal/router/backstage_test.go @@ -85,6 +85,11 @@ func (f *fakeBrand) CrawlTasks(ctx context.Context, brandID uint32) ([]dto.Crawl 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 ( _ service.ArticleService = (*fakeArticle)(nil) diff --git a/internal/service/brand_service.go b/internal/service/brand_service.go index 4043b9c..dfc24d9 100644 --- a/internal/service/brand_service.go +++ b/internal/service/brand_service.go @@ -20,6 +20,8 @@ type BrandService interface { // AdminList 后台品牌列表：字母索引不要求走秀已发布， // 使只有待审 / 驳回走秀的品牌仍出现在后台（公开 List 读视图，会挡掉它们）。 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 个）并附带各品牌档案数。 // 供 SSG 构建期（RunwayLooks 侧栏 / 品牌弹窗 HOT 标签）一次性取前 N 个热门品牌； // 复用 List 的 featured 收窄逻辑（FeaturedIDs("images", limit)），与首页精选口径一致。 @@ -201,6 +203,11 @@ func (s *brandService) Delete(ctx context.Context, uid string) error { // CrawlTasks 返回爬虫取任务所需的品牌列表：编码 brand_uid + 英文品牌名。 // 爬虫直接拿 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) { brands, err := s.brands.CrawlBrands(ctx, brandID) if err != nil { diff --git a/internal/service/ingest_keypath_test.go b/internal/service/ingest_keypath_test.go new file mode 100644 index 0000000..93da154 --- /dev/null +++ b/internal/service/ingest_keypath_test.go @@ -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) + } + } +} diff --git a/internal/service/ingest_service.go b/internal/service/ingest_service.go index 24c658f..61c0f21 100644 --- a/internal/service/ingest_service.go +++ b/internal/service/ingest_service.go @@ -5,6 +5,7 @@ import ( "crypto/sha1" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "log" @@ -271,7 +272,15 @@ func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob, return } if _, err := s.brandRepo.FindByID(ctx, brandID); err != nil { - _ = s.repo.MarkFailed(ctx, job.ID, "brand not found") + // 必须区分「品牌真不存在」与「查询本身失败」：前者永久失败，后者应退避重试。 + // 曾把两者一律记为 "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") + return + } + s.failOrRetry(ctx, job.ID, "brand lookup: "+err.Error()) return } @@ -454,22 +463,48 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL failed := false // 摊平：顺序 = 逐个 look 先主图、再其细节图，与原串行遍历顺序完全一致。 + // 每张图优先取「媒体机已上传的 key」，没有 key 才回落到「下载原始 URL」。 tasks := make([]lookImageTask, 0, len(looks)) for li, look := range looks { 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}) } - for di, d := range look.Details { - if d == "" { + // 细节图数量取两个数组的较大者：媒体机直传时可能只给 DetailKeys 而不给 URL， + // 只按 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 } - 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 { + 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) }) @@ -484,7 +519,12 @@ func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayL failed = true continue } - keys = append(keys, r.key) + // 只把「本次任务自己上传的 key」纳入回滚清单。 + // 媒体机直传的 key 不进清单：它是内容寻址的，可能被别的图集共享， + // 失败时删掉会连带弄坏别人已入库的图；这类孤儿由存储侧清理策略负责。 + if task.key == "" { + keys = append(keys, r.key) + } if seen[r.sha1] { continue } @@ -830,9 +870,27 @@ func concurrentFetch(ctx context.Context, n int, fn func(context.Context, int) d // 是主图还是第几张细节图，供并发下载完成后重建 Name / 主副图归属 / IsDetail。 type lookImageTask struct { url string - lookIdx int // 1-based look 序号 - isDetail bool // true=细节图 - detailNo int // 细节图序号（1-based），主图为 0 + key string // 非空 = 媒体机已上传，worker 跳过下载/上传，直接落库 + size int64 // key 路径下由 spider 上报的字节数（0=未知） + lookIdx int // 1-based look 序号 + isDetail bool // true=细节图 + detailNo int // 细节图序号（1-based），主图为 0 +} + +// sha1FromKey 从内容寻址 key（runway/.jpg 或 street/.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优先，失败兜底本地）。