From 3fe0b86e112bd50c76dd06cf4c7115a55e67adee Mon Sep 17 00:00:00 2001 From: toom1996 <23cm.cn@gmail.com> Date: Sun, 25 Oct 2026 00:15:22 +0800 Subject: [PATCH] =?UTF-8?q?feat:=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 - media: spider 下载图片后按内容 sha1 生成 key，经后端预签名直传 S4，上报只带 key （spider 不持有 AK/SK；单张失败自动回退为上报 URL） - embed: 新增 embed 子命令，从后端领取缺向量的图片→下载→本地 CLIP 编码→回传 （媒体拆分后后端拿不到图片字节，向量只能靠回填补齐） - serve: 服务化常驻模式（领取 crawl 任务 → 抓取 → 回报） - config: 新增 media.upload/concurrency、vision.endpoint/concurrency --- README.md | 132 +++++++++++++++++++ cmd/embed.go | 113 ++++++++++++++++ cmd/serve.go | 153 +++++++++++++++++++++ cmd/vogue.go | 6 + config.yml | 13 +- internal/config/config.go | 18 +++ internal/embed/worker.go | 255 +++++++++++++++++++++++++++++++++++ internal/ingest/client.go | 264 ++++++++++++++++++++++++++++++++++++- internal/ingest/payload.go | 9 ++ internal/media/upload.go | 262 ++++++++++++++++++++++++++++++++++++ internal/spider/vogue.go | 82 ++++++++++++ 11 files changed, 1305 insertions(+), 2 deletions(-) create mode 100644 cmd/embed.go create mode 100644 cmd/serve.go create mode 100644 internal/embed/worker.go create mode 100644 internal/media/upload.go diff --git a/README.md b/README.md index e69de29..8ea23a1 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,132 @@ +# spider + +Vogue 走秀 / The Impression 街拍采集器。抓到的元数据经 HMAC 签名上报后端 `ingest` 接口，由后端异步下载图片、写库、编码向量——**spider 不直连数据库**。 + +## 两种运行模式 + +| 模式 | 命令 | 适用 | +|---|---|---| +| 单次抓取 | `spider vogue -b <品牌id> [-s 场次上限]` | 临时补某个品牌、调试 | +| **常驻服务** | `spider serve -w 4 -i 5 -m 5` | **生产用法**：从后台领取任务 | + +服务模式不再由人指定品牌：**后台决定爬谁**（`/admin/crawl`），spider 领取并执行，执行结果回报后台。 + +## 配置（config.yml） + +```yaml +ingest: + enabled: true # serve 模式必须为 true + endpoint: "http://<后端IP>:8092/admin/internal/ingest" + secret: "<与后端 INGEST_SECRET 完全一致>" + hashid_secret: "" # 服务模式不需要（brand_uid 由后端下发） +``` + +> `endpoint` 末尾是 `/ingest`，客户端会自己推导出 `/crawl/tasks`（领取任务）和 `/crawl/tasks/:id/result`（回报结果）。 + +## 媒体机直传（把下载/上传从前端搬出来） + +```yaml +media: + upload: true # 打开：spider 自己下载图并直传对象存储，后端只写库 + concurrency: 6 # 下载/上传并发 +``` + +打开后 spider 会：下载图片 → 按内容 sha1 生成 key（`runway/.jpg`）→ 向后端 +`/admin/internal/upload-url` 申请一次性 PUT 地址 → 直接上传 → 上报时只带 key。 + +**关键点**： + +- 密钥只在后端：spider 拿的是 15 分钟有效的预签名 URL，不持有存储 AK/SK +- key 是内容寻址，重复上传幂等，重爬不产生孤儿文件 +- 单张失败不影响整场：失败的图仍带原始 URL，后端会兜底下载 +- 关掉 `upload` 就完全回到旧行为（后端下载），可随时回滚 + +**什么时候值得打开**：spider 部署在**带宽比后端机器好**的地方时收益明显； +如果两者在同一台机器上，下载带宽是同一份，拆开不会更快（只是把活儿换个进程干）。 + +## 服务模式怎么跑 + +```bash +# 编译 +go build -o spider . + +# 启动：4 路并发、空队列每 5 秒轮询、单品牌内部并发 5 +./spider serve -w 4 -i 5 -m 5 +``` + +| 参数 | 含义 | 建议 | +|---|---|---| +| `-w/--workers` | 同时跑几个品牌 | 带宽/CPU 够就 2~4；每个品牌内部还有 `-m` 并发 | +| `-i/--interval` | 队列空时的轮询间隔（秒） | 5~10，别设 1（会把后台接口打满） | +| `-m/--max-co` | 单品牌内部的抓取并发 | 3~8；太大会被 Vogue 限流 | + +优雅退出：`Ctrl+C` 会停止领新任务，等手上的品牌跑完再退。 + +日志里每个任务有两行关键输出： + +``` +[serve] host-1234-0 领取任务 task=7 品牌=Acne Studios(uid=001AJGov) 场次上限=3 +[serve] host-1234-0 任务 7 完成：status=done 提交 3 场，耗时 42s +``` + +## 单次模式（调试用） + +```bash +./spider vogue -b 7 # 只抓 Gucci，不限场次 +./spider vogue -b 7 -s 2 # 只抓最近 2 场 +./spider vogue -d # debug：只打印，不上报 +./spider theimpression -n 10 # 街拍，最多 10 条任务 +``` + +`-b` 是**数据库里的数字 id**（`brands.id`），不是 hashid。 + +## 后台怎么操作 + +访问 `http://<后端>:8092/admin/crawl`（需先登录后台）。 + +### 1. 加任务 + +| 场景 | 操作 | +|---|---| +| 抓单个品牌 | 去 `/admin/brands` 找到品牌 → 点「加入抓取队列」（默认每品牌 3 场） | +| 批量抓一批 | 在 `/admin/crawl` 顶部「批量加入」：前缀 `A` + 场次上限 `3` + 数量 `50` → 就是把 A 开头的 50 个品牌各抓最近 3 场 | +| 重爬某个品牌 | 在任务列表该行点「重试」（会把状态重置为待领取） | + +「场次上限」填 `0` 表示不限场次——**批量时不要填 0**，否则一个老品牌可能有上百场，一次灌进来几万张图。 + +### 2. 看状态 + +| 状态 | 含义 | +|---|---| +| 待领取 | 已入队，等 spider 来拿 | +| 执行中 | 某个 spider 已领取（列表里会显示 worker 标识） | +| 已完成 | 跑完了。`已提交` 列是本次提交了几场走秀 | +| 失败 | spider 侧进程级失败，会自动重试最多 3 次；耗尽后需手动「重试」 | + +**`已提交 = 0` 不一定是错误**：Vogue 上很多品牌没有走秀页面（实测 A 开头品牌里约 75% 没有），这种情况任务是「已完成、提交 0 场」。 + +### 3. 取消 / 删除 + +- **取消**：只是不再被领取，**无法中断正在执行的品牌**（spider 手里还在跑） +- **删除**：只删任务记录，已入库的图集不受影响 + +### 4. 任务卡在「执行中」怎么办 + +正常情况会被自动回收：后台对 `执行中` 的任务有 **15 分钟租约**，超过仍未回报就自动放回「待领取」。所以 spider 断电/被杀不会导致任务永久丢失，重启后会被重新领走。 + +## 抓完之后 + +任务完成只代表**元数据已提交**。图片下载、写库、编码向量由后端 worker 异步完成，看进度去： + +- `/admin/ingest-jobs` — 入库任务（pending / processing / done / failed） +- `/admin/articles` — 已入库的图集（默认 `pending` 状态，前台不可见，需在列表里通过） + +## 常见问题 + +| 现象 | 原因 / 处理 | +|---|---| +| 启动即报「ingest 未启用」 | `config.yml` 的 `ingest.enabled` 不是 true | +| 一直「无任务」 | 后台还没加任务，或全部被其他 spider 领走了 | +| 领取报 401 | `ingest.secret` 与后端 `INGEST_SECRET` 不一致 | +| 任务反复失败 | 看后台列表的「错误」列；常见是网络不通/被 Vogue 限流 | +| 抓到了但前台看不到 | 图集是 `pending`，去 `/admin/articles` 通过审核 | diff --git a/cmd/embed.go b/cmd/embed.go new file mode 100644 index 0000000..d4967fc --- /dev/null +++ b/cmd/embed.go @@ -0,0 +1,113 @@ +package cmd + +import ( + "context" + "log" + "net/http" + "strings" + "time" + + "github.com/spf13/cobra" + + "my-spiders/internal/config" + "my-spiders/internal/embed" +) + +var ( + embedKinds string + embedLimit int + embedOnce bool + embedSleep int + embedVision string + embedCo int +) + +var embedCmd = &cobra.Command{ + Use: "embed", + Short: "向量回填：拉取缺向量的图片，本地 CLIP 编码后回传后端", + Long: `媒体拆分后由 spider 直接传图，后端拿不到图片字节、算不出向量。 +本命令跑在有 GPU 的机器上，把缺向量的图补上： + + 后端领取（缺向量的图） → 从存储下载 → 本地 CLIP 编码 → 回传向量 + +常驻运行（默认）：spider embed +只跑一轮： spider embed --once`, + Run: func(cmd *cobra.Command, args []string) { + if IngestClient == nil { + log.Fatal("[embed] 入库管线未启用（config.ingest.enabled=false），无法领取回填任务") + } + vision := embedVision + if vision == "" { + vision = config.GlobalConfig.Vision.Endpoint + } + if vision == "" { + vision = "http://127.0.0.1:8900" + } + // 视觉服务没起来就白跑：先探活，明确报错比反复超时好排查。 + if err := visionReady(vision); err != nil { + log.Fatalf("[embed] 视觉服务不可用（%s）: %v\n 先把 CLIP 服务起来：app-cli\\vision\\vision.ps1 start", vision, err) + } + + kinds := []string{"runway", "street"} + if strings.TrimSpace(embedKinds) != "" { + kinds = nil + for _, k := range strings.Split(embedKinds, ",") { + if k = strings.TrimSpace(k); k != "" { + kinds = append(kinds, k) + } + } + } + conc := embedCo + if conc <= 0 { + conc = config.GlobalConfig.Vision.Concurrency + } + w := embed.New(IngestClient, vision, conc, 8) + log.Printf("[embed] 启动：视觉=%s 类别=%v 单批=%d 并发=%d", vision, kinds, embedLimit, conc) + + ctx := context.Background() + for { + idle := 0 + for _, kind := range kinds { + st, err := w.RunOnce(ctx, kind, embedLimit) + if err != nil { + log.Printf("[embed] %s 本轮失败: %v", kind, err) + continue + } + if st.Fetched == 0 { + idle++ + } + } + if embedOnce { + return + } + if idle == len(kinds) { + // 没有待办：睡一会儿再看，避免空转打后端 + time.Sleep(time.Duration(embedSleep) * time.Second) + } + } + }, +} + +// visionReady 探活视觉服务（/health），失败返回错误。 +func visionReady(endpoint string) error { + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Get(endpoint + "/health") + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return err + } + return nil +} + +func init() { + rootCmd.AddCommand(embedCmd) + embedCmd.Flags().StringVar(&embedKinds, "kind", "", "只处理指定类别（runway,street），默认两类都做") + embedCmd.Flags().IntVar(&embedLimit, "limit", 32, "每轮向后端领取的张数（1~200）") + embedCmd.Flags().BoolVar(&embedOnce, "once", false, "只跑一轮就退出（用于验证/定时任务）") + embedCmd.Flags().IntVar(&embedSleep, "sleep", 30, "空闲时休眠秒数（常驻模式）") + embedCmd.Flags().StringVar(&embedVision, "vision", "", "视觉服务地址（默认取 config.vision.endpoint，再兜底 127.0.0.1:8900）") + embedCmd.Flags().IntVar(&embedCo, "concurrency", 0, "下载并发（默认取 config.vision.concurrency）") +} diff --git a/cmd/serve.go b/cmd/serve.go new file mode 100644 index 0000000..ee9c1d7 --- /dev/null +++ b/cmd/serve.go @@ -0,0 +1,153 @@ +package cmd + +import ( + "context" + "fmt" + "log" + "os" + "os/signal" + "strings" + "sync" + "syscall" + "time" + + "github.com/spf13/cobra" + + "my-spiders/internal/config" + "my-spiders/internal/media" + "my-spiders/internal/spider" +) + +var ( + serveWorkers int + serveInterval int + serveMaxCo int +) + +// serveCmd 把爬虫变成「常驻服务」：不再由人决定爬谁，而是从后台领取任务。 +// +// 与 vogue 子命令的区别： +// - vogue：一次性的，参数指定品牌（--brand），跑完退出； +// - serve：常驻，循环「领任务 → 执行 → 回报」，没有任务就空转等待。 +// +// 这样后台（/admin/crawl）就能控制爬哪些品牌，爬虫部署在哪台机器都不影响。 +var serveCmd = &cobra.Command{ + Use: "serve", + Short: "以服务方式运行：从后台领取品牌抓取任务并执行", + Run: func(cmd *cobra.Command, args []string) { + if IngestClient == nil { + log.Fatal("[致命] 未启用入库管线：config.yml 里 ingest.enabled 必须为 true（serve 模式依赖后台接口领取任务）") + } + + host, _ := os.Hostname() + if strings.TrimSpace(host) == "" { + host = "spider" + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Ctrl+C / SIGTERM 优雅退出：不再领新任务，等在跑的品牌结束。 + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGTERM) + var wg sync.WaitGroup + + log.Printf("[serve] 启动 %d 个 worker，轮询间隔 %ds，单品牌并发 %d", serveWorkers, serveInterval, serveMaxCo) + + for i := 0; i < serveWorkers; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + workerID := fmt.Sprintf("%s-%d-%d", host, os.Getpid(), idx) + runServeWorker(ctx, workerID) + }(i) + } + + <-sig + log.Println("[serve] 收到退出信号，等待进行中的任务结束…") + cancel() + wg.Wait() + log.Println("[serve] 已退出") + }, +} + +func runServeWorker(ctx context.Context, workerID string) { + for { + select { + case <-ctx.Done(): + return + default: + } + + task, err := IngestClient.FetchTask(ctx, workerID) + if err != nil { + log.Printf("[serve] %s 领取任务失败: %v", workerID, err) + sleepOrDone(ctx, serveInterval) + continue + } + if task == nil { + // 队列空：退避轮询，避免打满后台接口。 + sleepOrDone(ctx, serveInterval) + continue + } + + log.Printf("[serve] %s 领取任务 task=%d 品牌=%s(uid=%s) 场次上限=%d", + workerID, task.ID, task.BrandName, task.BrandUID, task.MaxShows) + + start := time.Now() + shows, errMsg := executeBrand(ctx, task.BrandUID, task.BrandName, int(task.MaxShows)) + elapsed := time.Since(start).Round(time.Second) + + status := "done" + if errMsg != "" { + status = "failed" + } + if err := IngestClient.ReportTask(ctx, task.ID, status, shows, errMsg); err != nil { + log.Printf("[serve] %s 回报任务 %d 失败: %v", workerID, task.ID, err) + // 回报失败不要紧：租约（CrawlStuckTimeoutSec）到期后后台会把任务重新放回队列。 + sleepOrDone(ctx, serveInterval) + continue + } + log.Printf("[serve] %s 任务 %d 完成：status=%s 提交 %d 场，耗时 %s", workerID, task.ID, status, shows, elapsed) + } +} + +// executeBrand 执行单个品牌，异常（panic）也转成错误返回，避免一个坏品牌把整个 serve 进程带崩。 +func executeBrand(ctx context.Context, brandUID, brandName string, maxShows int) (shows int, errMsg string) { + defer func() { + if r := recover(); r != nil { + shows = 0 + errMsg = fmt.Sprintf("panic: %v", r) + log.Printf("[serve] 执行品牌 %s 时 panic: %v", brandName, r) + } + }() + + s := spider.NewVogueSpider(serveMaxCo, 0, maxShows, IngestClient) + s.Debug = debug + // 媒体机直传：spider 自己下载并上传图片，后端只写库。 + if config.GlobalConfig.Media.Upload && IngestClient != nil { + s.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency) + } + shows = s.RunBrand(brandUID, brandName, maxShows) + return shows, "" +} + +// sleepOrDone 可被 ctx 取消打断的 sleep。 +func sleepOrDone(ctx context.Context, seconds int) { + if seconds < 1 { + seconds = 5 + } + t := time.NewTimer(time.Duration(seconds) * time.Second) + defer t.Stop() + select { + case <-ctx.Done(): + case <-t.C: + } +} + +func init() { + rootCmd.AddCommand(serveCmd) + serveCmd.Flags().IntVarP(&serveWorkers, "workers", "w", 1, "并发执行的品牌数（每路一个 goroutine）") + serveCmd.Flags().IntVarP(&serveInterval, "interval", "i", 5, "队列为空时的轮询间隔（秒）") + serveCmd.Flags().IntVarP(&serveMaxCo, "max-co", "m", 5, "单个品牌内部的抓取并发数") +} diff --git a/cmd/vogue.go b/cmd/vogue.go index e5daaf6..e5c4a51 100644 --- a/cmd/vogue.go +++ b/cmd/vogue.go @@ -3,6 +3,8 @@ package cmd import ( "github.com/spf13/cobra" + "my-spiders/internal/config" + "my-spiders/internal/media" "my-spiders/internal/spider" ) @@ -18,6 +20,10 @@ var vogueCmd = &cobra.Command{ Run: func(cmd *cobra.Command, args []string) { // 1. 修复参数数量 & 类型强转 uint(brandID) vogueSpider := spider.NewVogueSpider(maxCo, uint(brandID), maxShows, IngestClient) + // 媒体机直传：由 spider 下载并上传图片，后端只写库（config.media.upload 控制）。 + if config.GlobalConfig.Media.Upload && IngestClient != nil { + vogueSpider.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency) + } // 2. 修复方法名：将 Execute() 改为 Run() vogueSpider.Run() diff --git a/config.yml b/config.yml index 4ea6d21..8fdab96 100644 --- a/config.yml +++ b/config.yml @@ -11,4 +11,15 @@ ingest: enabled: true endpoint: "http://localhost:8092/admin/internal/ingest" secret: "dev-ingest-secret-2026" - hashid_secret: "" \ No newline at end of file + hashid_secret: "" + +# 媒体机直传：spider 自己下载图片并直传对象存储，后端只写库（不再下载/上传）。 +# 打开后数据库里存的仍是 key，展示链路完全不变；关闭则回退到「后端下载」的旧行为。 +media: + upload: true # 需要后端已配置 S4，否则 /upload-url 返回 503（会自动回退） + concurrency: 6 # 下载/上传并发 + +# 向量回填（spider embed）：跑在有 GPU 的机器上，把缺向量的图补上。 +vision: + endpoint: "http://127.0.0.1:8900" + concurrency: 4 # 下载并发 \ No newline at end of file diff --git a/internal/config/config.go b/internal/config/config.go index edd7f7d..25b0914 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,6 +12,24 @@ var GlobalConfig Config type Config struct { App AppConfig `mapstructure:"app"` Ingest IngestConfig `mapstructure:"ingest"` + Media MediaConfig `mapstructure:"media"` + Vision VisionConfig `mapstructure:"vision"` +} + +// VisionConfig 视觉服务（FashionSigLIP）配置，供 `spider embed` 回填向量使用。 +type VisionConfig struct { + Endpoint string `mapstructure:"endpoint"` // 如 http://127.0.0.1:8900 + Concurrency int `mapstructure:"concurrency"` // 下载并发，默认 4 +} + +// MediaConfig 媒体机直传配置（把下载/上传从后端搬到 spider 所在机器）。 +// +// 打开后 spider 会：下载图片 → 内容 sha1 生成 key → 向后端申请预签名 PUT 地址 +// → 直接传到对象存储 → 上报时只带 key。后端 worker 则跳过下载与上传，只写库。 +// 关闭时行为与旧版完全一致（上报 URL，由后端下载）。 +type MediaConfig struct { + Upload bool `mapstructure:"upload"` // 是否启用直传 + Concurrency int `mapstructure:"concurrency"` // 下载/上传并发，默认 6 } // IngestConfig 入库管线配置：爬虫 → 后台 ingest 接口（HMAC 签名）。 diff --git a/internal/embed/worker.go b/internal/embed/worker.go new file mode 100644 index 0000000..24549fa --- /dev/null +++ b/internal/embed/worker.go @@ -0,0 +1,255 @@ +// Package embed 是「向量回填」的执行端。 +// +// 为什么需要它：媒体拆分后 spider 直接传图到对象存储，后端拿到的是 key 而不是图片字节， +// 于是入库时算不出向量（image_embeddings 缺行）。这个 worker 补上这一段： +// +// 向后端要「缺向量的图片」 → 从存储下载 → 送本地 CLIP 编码 → 把向量回传给后端 +// +// 它跑在**有 GPU 的机器**上，所以放在 spider 仓库里（和爬虫同一份部署物）， +// 但逻辑上只在用 GPU，与抓取完全独立：不抓 Vogue、不写业务库。 +package embed + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "sync" + "time" + + "my-spiders/internal/ingest" +) + +// Worker 向量回填执行端。 +type Worker struct { + Client *ingest.Client // 取任务 + 回传（HMAC） + Vision string // 视觉服务地址，如 http://127.0.0.1:8900 + Concurrency int // 下载并发 + Batch int // 单次送编码的张数 + HTTP *http.Client +} + +// Stats 一轮处理的统计。 +type Stats struct { + Fetched int // 后端派发的张数 + Saved int // 后端确认写入的张数 + Failed int // 失败张数（已计入失败计数） + Remaining int // 后端统计的剩余量（-1 = 未知） +} + +// New 创建 worker。并发/批量传 0 时取默认值（4 / 8）。 +func New(client *ingest.Client, vision string, concurrency, batch int) *Worker { + if concurrency <= 0 { + concurrency = 4 + } + if batch <= 0 { + batch = 8 + } + return &Worker{ + Client: client, + Vision: vision, + Concurrency: concurrency, + Batch: batch, + // 图片可能几百 KB~几 MB，60s 足够；向量 POST 走同一个 client。 + HTTP: &http.Client{Timeout: 60 * time.Second}, + } +} + +// RunOnce 处理一轮：拉取 → 下载 → 编码 → 回传。 +// +// 不会因为单张失败而中断：下载失败/编码跳过的图进 failures，由后端累加计数， +// 连续失败到上限后自然退出队列（坏图不会无限占位）。 +func (w *Worker) RunOnce(ctx context.Context, kind string, limit int) (Stats, error) { + var st Stats + pending, remaining, err := w.Client.FetchPendingEmbeddings(ctx, kind, limit) + if err != nil { + return st, err + } + st.Fetched = len(pending) + st.Remaining = remaining + if len(pending) == 0 { + log.Printf("[embed] %s 无待回填图片（取时剩余 %d）", kind, remaining) + return st, nil + } + + // ① 并发下载：后端给的是可直接访问的 url，worker 不关心存储实现。 + data := make([][]byte, len(pending)) + errs := make([]error, len(pending)) + w.parallel(len(pending), func(i int) { + body, derr := w.download(ctx, pending[i].URL) + if derr != nil { + errs[i] = derr + return + } + data[i] = body + }) + + // ② 编码：只把下载成功的送进去，按 Batch 分批（视觉服务对单次张数有上限）。 + type job struct { + pos int + img []byte + } + jobs := make([]job, 0, len(pending)) + for i := range pending { + if errs[i] == nil { + jobs = append(jobs, job{pos: i, img: data[i]}) + } + } + + items := make([]ingest.EmbeddingVector, 0, len(jobs)) + failures := make([]ingest.EmbeddingFailure, 0, len(pending)-len(jobs)) + + for start := 0; start < len(jobs); start += w.Batch { + end := start + w.Batch + if end > len(jobs) { + end = len(jobs) + } + batch := jobs[start:end] + imgs := make([][]byte, len(batch)) + for i := range batch { + imgs[i] = batch[i].img + } + vecs, idxs, skipped, eerr := w.embed(ctx, imgs) + if eerr != nil { + for _, b := range batch { + p := pending[b.pos] + failures = append(failures, ingest.EmbeddingFailure{Kind: kind, ImageID: p.ImageID, Error: eerr.Error()}) + } + continue + } + // indexes 缺省时按位置对应；存在时严格按它映射（视觉服务可能跳过坏图）。 + for j := range vecs { + posInBatch := j + if len(idxs) == len(vecs) { + posInBatch = idxs[j] + } + if posInBatch < 0 || posInBatch >= len(batch) { + continue + } + p := pending[batch[posInBatch].pos] + items = append(items, ingest.EmbeddingVector{Kind: kind, ImageID: p.ImageID, Vector: vecs[j]}) + } + for _, s := range skipped { + if s < 0 || s >= len(batch) { + continue + } + p := pending[batch[s].pos] + failures = append(failures, ingest.EmbeddingFailure{Kind: kind, ImageID: p.ImageID, Error: "视觉服务跳过（图片无法解码）"}) + } + } + + // ③ 下载失败的也一并上报，让后端计数 + for i := range pending { + if errs[i] != nil { + failures = append(failures, ingest.EmbeddingFailure{Kind: kind, ImageID: pending[i].ImageID, Error: errs[i].Error()}) + } + } + + if len(items) == 0 && len(failures) == 0 { + return st, nil + } + saved, failed, serr := w.Client.SubmitEmbeddings(ctx, items, failures) + if serr != nil { + return st, serr + } + st.Saved = saved + st.Failed = failed + log.Printf("[embed] %s 本轮：取到 %d 张 → 写入 %d，失败 %d（剩余 %d）", + kind, st.Fetched, st.Saved, st.Failed, st.Remaining) + return st, nil +} + +// embed 调本地视觉服务编码一批图片。 +// 返回的 idxs 是「vectors 对应入参下标」；skipped 是无法解码的下标。 +func (w *Worker) embed(ctx context.Context, imgs [][]byte) ([][]float32, []int, []int, error) { + payload := struct { + Images []string `json:"images"` + }{Images: make([]string, len(imgs))} + for i, b := range imgs { + payload.Images[i] = base64.StdEncoding.EncodeToString(b) + } + body, err := json.Marshal(payload) + if err != nil { + return nil, nil, nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, w.Vision+"/embed", bytes.NewReader(body)) + if err != nil { + return nil, nil, nil, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := w.HTTP.Do(req) + if err != nil { + return nil, nil, nil, fmt.Errorf("视觉服务不可用: %w", err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return nil, nil, nil, fmt.Errorf("视觉服务返回 %d: %s", resp.StatusCode, truncate(string(raw), 120)) + } + var out struct { + Dim int `json:"dim"` + Vectors [][]float32 `json:"vectors"` + Indexes []int `json:"indexes"` + Skipped []int `json:"skipped"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return nil, nil, nil, fmt.Errorf("视觉服务响应解析失败: %w", err) + } + return out.Vectors, out.Indexes, out.Skipped, nil +} + +func (w *Worker) download(ctx context.Context, url string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + resp, err := w.HTTP.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("下载失败 status=%d", resp.StatusCode) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if len(body) == 0 { + return nil, fmt.Errorf("下载到空文件") + } + return body, nil +} + +// parallel 以 Concurrency 路并发执行 fn(0..n-1)。 +func (w *Worker) parallel(n int, fn func(i int)) { + if n == 0 { + return + } + if w.Concurrency <= 0 { + w.Concurrency = 1 + } + sem := make(chan struct{}, w.Concurrency) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + fn(i) + }(i) + } + wg.Wait() +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "..." +} diff --git a/internal/ingest/client.go b/internal/ingest/client.go index 7298dab..ddfbf49 100644 --- a/internal/ingest/client.go +++ b/internal/ingest/client.go @@ -10,6 +10,7 @@ import ( "io" "log" "net/http" + neturl "net/url" "strconv" "strings" "time" @@ -95,11 +96,272 @@ type CrawlBrand struct { // crawlBrandsURL 由 ingest endpoint 推导抓取任务接口地址（把末尾 /ingest 换成 /crawl/brands）。 func (c *Client) crawlBrandsURL() string { + return c.internalBase() + "/crawl/brands" +} + +// crawlTasksURL 服务化模式的任务接口：领取 / 回报。 +func (c *Client) crawlTasksURL() string { + return c.internalBase() + "/crawl/tasks" +} + +// internalBase 把 endpoint 的结尾 /ingest 去掉，得到内部接口的公共前缀。 +func (c *Client) internalBase() string { base := c.Endpoint if i := strings.LastIndex(base, "/ingest"); i != -1 { base = base[:i] } - return strings.TrimRight(base, "/") + "/crawl/brands" + return strings.TrimRight(base, "/") +} + +// UploadItem / UploadURL 媒体机直传用：向后端要一次性 PUT 地址（媒体机不持有 S4 密钥）。 +type UploadItem struct { + Key string `json:"key"` + ContentType string `json:"content_type"` +} + +// UploadURL 后端签发的预签名地址（有效期默认 15 分钟）。 +type UploadURL struct { + Key string `json:"key"` + URL string `json:"url"` +} + +// RequestUploadURLs 批量申请预签名 PUT 地址。单次上限 200（后端限制），调用方自行分片。 +func (c *Client) RequestUploadURLs(ctx context.Context, items []UploadItem) ([]UploadURL, error) { + if c == nil || c.Endpoint == "" { + return nil, fmt.Errorf("ingest client 未配置") + } + if len(items) == 0 { + return nil, nil + } + body, err := json.Marshal(map[string]any{"items": items}) + if err != nil { + return nil, err + } + ts := NowTS() + nonce, err := newNonce() + if err != nil { + return nil, err + } + sig := Sign(c.Secret, ts, nonce, string(body)) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.internalBase()+"/upload-url", bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Signature", sig) + req.Header.Set("X-Timestamp", ts) + req.Header.Set("X-Nonce", nonce) + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("upload-url http %d: %s", resp.StatusCode, string(raw)) + } + var out struct { + Items []UploadURL `json:"items"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("upload-url 解析失败: %w", err) + } + return out.Items, nil +} + +// PendingEmbedding 一张「缺向量」的图片（向量回填用）。 +// URL 由后端拼好，worker 不需要知道存储域名怎么配。 +type PendingEmbedding struct { + Kind string `json:"kind"` + ImageID uint32 `json:"image_id"` + Key string `json:"key"` + URL string `json:"url"` +} + +// EmbeddingVector 回传成功编码的向量。 +type EmbeddingVector struct { + Kind string `json:"kind"` + ImageID uint32 `json:"image_id"` + Vector []float32 `json:"vector"` +} + +// EmbeddingFailure 回传失败项；后端累加计数，连续失败到上限后该图不再进队列。 +type EmbeddingFailure struct { + Kind string `json:"kind"` + ImageID uint32 `json:"image_id"` + Error string `json:"error"` +} + +// FetchPendingEmbeddings 拉一批缺向量的图片。remaining 为后端统计的剩余量（-1 = 统计失败）。 +func (c *Client) FetchPendingEmbeddings(ctx context.Context, kind string, limit int) ([]PendingEmbedding, int, error) { + if c == nil || c.Endpoint == "" { + return nil, 0, fmt.Errorf("ingest client 未配置") + } + url := fmt.Sprintf("%s/embeddings/pending?kind=%s&limit=%d", c.internalBase(), neturl.QueryEscape(kind), limit) + ts := NowTS() + nonce, err := newNonce() + if err != nil { + return nil, 0, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, 0, err + } + req.Header.Set("X-Signature", Sign(c.Secret, ts, nonce, "")) + req.Header.Set("X-Timestamp", ts) + req.Header.Set("X-Nonce", nonce) + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return nil, 0, err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return nil, 0, fmt.Errorf("pending embeddings http %d: %s", resp.StatusCode, string(raw)) + } + var out struct { + Items []PendingEmbedding `json:"items"` + Remaining int `json:"remaining"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return nil, 0, fmt.Errorf("pending embeddings 解析失败: %w", err) + } + return out.Items, out.Remaining, nil +} + +// SubmitEmbeddings 回传向量与失败项，返回后端实际写入与失败计数。 +func (c *Client) SubmitEmbeddings(ctx context.Context, items []EmbeddingVector, failures []EmbeddingFailure) (int, int, error) { + if c == nil || c.Endpoint == "" { + return 0, 0, fmt.Errorf("ingest client 未配置") + } + body, err := json.Marshal(map[string]any{"items": items, "failures": failures}) + if err != nil { + return 0, 0, err + } + ts := NowTS() + nonce, err := newNonce() + if err != nil { + return 0, 0, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.internalBase()+"/embeddings", bytes.NewReader(body)) + if err != nil { + return 0, 0, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Signature", Sign(c.Secret, ts, nonce, string(body))) + req.Header.Set("X-Timestamp", ts) + req.Header.Set("X-Nonce", nonce) + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return 0, 0, err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return 0, 0, fmt.Errorf("submit embeddings http %d: %s", resp.StatusCode, string(raw)) + } + var out struct { + Saved int `json:"saved"` + Failed int `json:"failed"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return 0, 0, fmt.Errorf("submit embeddings 解析失败: %w", err) + } + return out.Saved, out.Failed, nil +} + +// CrawlTask 后台派发的品牌抓取任务（服务化模式）。 +// brand_uid 是后端已编码好的串，直接原样上送 ingest 即可；brand_name 用于拼 Vogue 抓取 URL。 +type CrawlTask struct { + ID uint32 `json:"id"` + BrandUID string `json:"brand_uid"` + BrandName string `json:"brand_name"` + MaxShows int16 `json:"max_shows"` +} + +// FetchTask 领取一个抓取任务；队列为空时返回 nil, nil（调用方据此退避轮询）。 +func (c *Client) FetchTask(ctx context.Context, worker string) (*CrawlTask, error) { + if c == nil || c.Endpoint == "" { + return nil, fmt.Errorf("ingest client 未配置") + } + url := c.crawlTasksURL() + "?worker=" + neturl.QueryEscape(worker) + ts := NowTS() + nonce, err := newNonce() + if err != nil { + return nil, err + } + sig := Sign(c.Secret, ts, nonce, "") + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + req.Header.Set("X-Signature", sig) + req.Header.Set("X-Timestamp", ts) + req.Header.Set("X-Nonce", nonce) + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("fetch task http %d: %s", resp.StatusCode, string(raw)) + } + var out struct { + Task *CrawlTask `json:"task"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("fetch task 解析失败: %w", err) + } + return out.Task, nil +} + +// ReportTask 回报任务结果：status 取 done（执行完毕，哪怕 Vogue 上没有这个品牌）/ failed（进程级失败，后端会自动重试）。 +func (c *Client) ReportTask(ctx context.Context, id uint32, status string, shows int, errMsg string) error { + if c == nil || c.Endpoint == "" { + return fmt.Errorf("ingest client 未配置") + } + body, err := json.Marshal(map[string]any{ + "status": status, + "shows_submitted": shows, + "error": errMsg, + }) + if err != nil { + return err + } + url := fmt.Sprintf("%s/%d/result", c.crawlTasksURL(), id) + ts := NowTS() + nonce, err := newNonce() + if err != nil { + return err + } + sig := Sign(c.Secret, ts, nonce, string(body)) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Signature", sig) + req.Header.Set("X-Timestamp", ts) + req.Header.Set("X-Nonce", nonce) + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("report task http %d: %s", resp.StatusCode, string(raw)) + } + return nil } // GetCrawlBrands 拉取爬虫要抓取的品牌任务列表（HMAC 签名 GET，复用与上报相同的验签算法）。 diff --git a/internal/ingest/payload.go b/internal/ingest/payload.go index a737133..ce1d259 100644 --- a/internal/ingest/payload.go +++ b/internal/ingest/payload.go @@ -32,9 +32,18 @@ type RunwayIngest struct { } // RunwayLook 一场秀中的一个 look：1 张主图 + 0..N 张细节图。 +// +// 两种上报形态（媒体机直传开关打开时用 key，否则用 URL）： +// - 旧：Main/Details 给原始 URL，后端自己下载上传； +// - 新：MainKey/DetailKeys 给已上传的存储 key，后端跳过下载直接落库。 type RunwayLook struct { Main string `json:"main"` // 主图（look）原始 URL Details []string `json:"details"` // 细节图原始 URL 列表 + + MainKey string `json:"main_key,omitempty"` // 主图已上传 key（runway/.jpg） + DetailKeys []string `json:"detail_keys,omitempty"` // 细节图已上传 key + MainSize int64 `json:"main_size,omitempty"` // 主图字节数 + DetailSizes []int64 `json:"detail_sizes,omitempty"` // 细节图字节数（与 DetailKeys 对齐） } diff --git a/internal/media/upload.go b/internal/media/upload.go new file mode 100644 index 0000000..2bd871d --- /dev/null +++ b/internal/media/upload.go @@ -0,0 +1,262 @@ +// Package media 负责「媒体机直传」：把图片下载下来、算内容 sha1、 +// 向后端要预签名 PUT 地址、直接传到对象存储，最后只把 key 交给后端。 +// +// 为什么由 spider 做这件事： +// - spider 部署在带宽好的机器上，下载 Vogue 图 + 上传存储是带宽密集的活； +// - 后端机器只需写库，不再被出口带宽卡住； +// - 爬虫**不持有**存储密钥：预签名地址由后端签发（15 分钟有效）， +// 密钥只留在后端，减少泄露面。 +// +// key 采用内容寻址（runway/.jpg），同一张图重复上传是幂等的， +// 重爬不会产生孤儿文件。 +package media + +import ( + "bytes" + "context" + "crypto/sha1" + "encoding/hex" + "fmt" + "io" + "log" + "net/http" + "strings" + "sync" + "time" + + "my-spiders/internal/ingest" +) + +// Uploader 媒体直传器。 +type Uploader struct { + Client *ingest.Client // 复用它来签发预签名 URL（HMAC） + Prefix string // 存储前缀：runway / street + Concurrency int // 下载与上传的并发数 + HTTP *http.Client +} + +// Result 单张图的直传结果；Err 非空表示这张失败（调用方应回退为上报 URL）。 +type Result struct { + Key string + Size int64 + Err error +} + +// New 创建直传器。concurrency <= 0 时取 6。 +func New(client *ingest.Client, prefix string, concurrency int) *Uploader { + if concurrency <= 0 { + concurrency = 6 + } + return &Uploader{ + Client: client, + Prefix: prefix, + Concurrency: concurrency, + HTTP: &http.Client{Timeout: 60 * time.Second}, + } +} + +// UploadAll 并发完成「下载 → 生成 key → 预签名 → PUT」，返回与 urls 下标一一对应的结果。 +// 单张失败不影响其他张：该下标 Err 非空，调用方据此对这张回退成 URL 上报。 +func (u *Uploader) UploadAll(ctx context.Context, urls []string) []Result { + results := make([]Result, len(urls)) + if len(urls) == 0 { + return results + } + + // ① 并发下载 + data := make([][]byte, len(urls)) + errs := make([]error, len(urls)) + u.parallel(len(urls), func(i int) { + body, err := u.download(ctx, urls[i]) + if err != nil { + errs[i] = err + return + } + data[i] = body + }) + log.Printf("[media] 下载完成 %d 张（失败 %d）", countOK(data), len(urls)-countOK(data)) + + // ② 生成本批 key（内容 sha1 + 原扩展名） + for i := range urls { + if errs[i] != nil { + results[i] = Result{Err: errs[i]} + continue + } + results[i] = Result{ + Key: u.Prefix + "/" + sha1Hex(data[i]) + extOf(urls[i]), + Size: int64(len(data[i])), + } + } + + // ③ 批量申请预签名地址（同一张图可能在同一场里出现多次，按 key 去重） + want := map[string]ingest.UploadItem{} + for i := range urls { + if results[i].Key == "" { + continue + } + if _, ok := want[results[i].Key]; ok { + continue + } + want[results[i].Key] = ingest.UploadItem{Key: results[i].Key, ContentType: contentTypeOf(urls[i])} + } + items := make([]ingest.UploadItem, 0, len(want)) + for _, it := range want { + items = append(items, it) + } + + presigned := make(map[string]string, len(items)) + const chunk = 100 // 后端单次上限 200，这里留一半余量 + for start := 0; start < len(items); start += chunk { + end := start + chunk + if end > len(items) { + end = len(items) + } + got, err := u.Client.RequestUploadURLs(ctx, items[start:end]) + if err != nil { + // 申请失败：整批回退成 URL 上报（后端自己下载），不中断整场抓取。 + for i := range results { + if results[i].Key != "" { + results[i] = Result{Err: fmt.Errorf("申请上传地址失败: %w", err)} + } + } + return results + } + for _, g := range got { + presigned[g.Key] = g.URL + } + } + + log.Printf("[media] 已拿到 %d 个预签名地址，开始上传", len(presigned)) + + // ④ 并发 PUT（内容已在内存里，直接传字节） + u.parallel(len(urls), func(i int) { + if results[i].Key == "" { + return + } + putURL, ok := presigned[results[i].Key] + if !ok { + results[i] = Result{Err: fmt.Errorf("未拿到 %s 的上传地址", results[i].Key)} + return + } + if err := u.put(ctx, putURL, urls[i], data[i]); err != nil { + results[i] = Result{Err: err} + } + }) + return results +} + +// parallel 以 Concurrency 路并发执行 fn(0..n-1)。 +func (u *Uploader) parallel(n int, fn func(i int)) { + if n == 0 { + return + } + if u.Concurrency <= 0 { + // 兜底：并发为 0 会造成无缓冲 channel 死锁（曾把整条抓取卡住）。 + u.Concurrency = 1 + } + sem := make(chan struct{}, u.Concurrency) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + fn(i) + }(i) + } + wg.Wait() +} + +func (u *Uploader) download(ctx context.Context, url string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + resp, err := u.HTTP.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("下载失败 status=%d", resp.StatusCode) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if len(body) == 0 { + return nil, fmt.Errorf("下载到空文件") + } + return body, nil +} + +func (u *Uploader) put(ctx context.Context, putURL, originalURL string, data []byte) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPut, putURL, bytes.NewReader(data)) + if err != nil { + return err + } + // Content-Type 必须与申请预签名时一致，否则签名校验失败（403）。 + req.Header.Set("Content-Type", contentTypeOf(originalURL)) + resp, err := u.HTTP.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 300)) + return fmt.Errorf("PUT 失败 status=%d %s", resp.StatusCode, string(raw)) + } + return nil +} + +// countOK 数一下成功下载的张数（用于日志定位卡在哪一步）。 +func countOK(data [][]byte) int { + n := 0 + for i := range data { + if len(data[i]) > 0 { + n++ + } + } + return n +} + +func sha1Hex(b []byte) string { + sum := sha1.Sum(b) + return hex.EncodeToString(sum[:]) +} + +// extOf 从图片 URL 推断扩展名（取路径最后一段，忽略查询串）。 +func extOf(rawURL string) string { + p := rawURL + if i := strings.IndexAny(p, "?#"); i >= 0 { + p = p[:i] + } + if i := strings.LastIndex(p, "."); i >= 0 && len(p)-i <= 6 { + ext := strings.ToLower(p[i:]) + switch ext { + case ".jpg", ".jpeg", ".png", ".gif", ".webp", ".avif": + if ext == ".jpeg" { + return ".jpg" + } + return ext + } + } + return ".jpg" +} + +// contentTypeOf 与 extOf 保持一致，保证预签名与 PUT 的 Content-Type 相同。 +func contentTypeOf(rawURL string) string { + switch extOf(rawURL) { + case ".png": + return "image/png" + case ".gif": + return "image/gif" + case ".webp": + return "image/webp" + case ".avif": + return "image/avif" + default: + return "image/jpeg" + } +} diff --git a/internal/spider/vogue.go b/internal/spider/vogue.go index 3299327..7f4782a 100644 --- a/internal/spider/vogue.go +++ b/internal/spider/vogue.go @@ -7,12 +7,14 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/tidwall/gjson" "my-spiders/internal/fetch" "my-spiders/internal/ingest" + "my-spiders/internal/media" ) const ( @@ -40,6 +42,10 @@ type VogueSpider struct { BrandID uint // 客户端指定的品牌 ID（--brand，单品牌调试用） MaxShows int // 小批量模式：单品牌最多抓取的发布会场次（0=不限制） Debug bool // Debug 模式：仅输出不入库 + // Media 媒体机直传器（可为 nil：为 nil 时走旧路径，上报 URL 由后端下载）。 + Media *media.Uploader + // submitted 本次品牌执行中成功上报的场次数（服务化模式回报给后台用）。 + submitted int32 } // NewVogueSpider 创建 Vogue 爬虫。spider 不再直连数据库：品牌任务经 ingest 接口从后端拉取， @@ -64,6 +70,68 @@ func NewVogueSpider(maxCo int, brandID uint, maxShows int, ingestClient *ingest. } } +// RunBrand 服务化模式下执行「一个品牌」，返回本次成功上报的场次数。 +// maxShows=0 表示不限场次；返回 0 有两种含义：Vogue 上没有该品牌，或该品牌的图集已全部存在（幂等跳过）。 +func (s *VogueSpider) RunBrand(brandUID, brandName string, maxShows int) int { + s.MaxShows = maxShows + atomic.StoreInt32(&s.submitted, 0) + s.SpiderStart(ingest.CrawlBrand{BrandUID: brandUID, Name: brandName}) + return int(atomic.LoadInt32(&s.submitted)) +} + +// uploadLookMedia 把一批 look 的图片直传到对象存储，回填 main_key / detail_keys。 +// +// 返回 true 表示全部成功；false 表示有失败项 —— 失败的图仍保留原始 URL， +// 后端见到「没有 key」就用 URL 自己下载，所以整场不会因为几张图失败而报废。 +func (s *VogueSpider) uploadLookMedia(ctx context.Context, looks []ingest.RunwayLook) bool { + type ref struct{ look, detail int } // detail = -1 表示主图 + + urls := make([]string, 0, len(looks)*2) + refs := make([]ref, 0, len(looks)*2) + for i := range looks { + if looks[i].Main != "" { + urls = append(urls, looks[i].Main) + refs = append(refs, ref{i, -1}) + } + for j, d := range looks[i].Details { + if d == "" { + continue + } + urls = append(urls, d) + refs = append(refs, ref{i, j}) + } + } + if len(urls) == 0 { + return true + } + + res := s.Media.UploadAll(ctx, urls) + okAll := true + for k, r := range res { + if r.Err != nil || r.Key == "" { + okAll = false + if r.Err != nil { + log.Printf("[media][warn] 直传失败 %s: %v", urls[k], r.Err) + } + continue + } + lk := &looks[refs[k].look] + if refs[k].detail < 0 { + lk.MainKey = r.Key + lk.MainSize = r.Size + continue + } + // DetailKeys 必须与 Details 下标对齐：缺口补空串，后端据此回落到 URL。 + for len(lk.DetailKeys) <= refs[k].detail { + lk.DetailKeys = append(lk.DetailKeys, "") + lk.DetailSizes = append(lk.DetailSizes, 0) + } + lk.DetailKeys[refs[k].detail] = r.Key + lk.DetailSizes[refs[k].detail] = r.Size + } + return okAll +} + // Run 爬虫启动总入口 func (s *VogueSpider) Run() { tasks, err := s.getTask() @@ -268,6 +336,19 @@ func (s *VogueSpider) getDetail(brandUID string, info gjson.Result) { // 入库分支：把走秀元数据 HMAC 签名 POST 到后台 ingest，由后台 worker 异步下载图、 // 补 season_code、写正式表（替代直写 brand_runway）。brand_uid 直接来自任务接口。 if s.Ingest != nil { + // 媒体机直传：先把图片传到对象存储，payload 里只带 key， + // 后端 worker 于是跳过「下载 + 上传」，只做写库。 + // 任一张失败不影响整场：失败的图仍带原始 URL，由后端兜底下载。 + if s.Media != nil && len(looks) > 0 { + log.Printf("[media] 开始直传 %d 个 look / %d 张图", len(looks), totalImages) + okAll := s.uploadLookMedia(context.Background(), looks) + if okAll { + log.Printf("[media] 已直传 %d 个 look / %d 张图，后端将跳过下载", len(looks), totalImages) + } else { + log.Printf("[media][warn] 部分图片直传失败，失败的将回退为 URL 上报（后端下载）") + } + } + collectionType, season := ingest.ParseCollection(hed) p := ingest.RunwayIngest{ Kind: ingest.KindRunway, @@ -284,6 +365,7 @@ func (s *VogueSpider) getDetail(brandUID string, info gjson.Result) { if _, err := s.Ingest.Submit(context.Background(), p); err != nil { log.Printf("[错误] 入库上报失败 [%s]: %v", hed, err) } else { + atomic.AddInt32(&s.submitted, 1) log.Printf("[成功] 已上报待入库: %s (共 %d 个 look / %d 张图, 发布时间 %s)", hed, len(looks), totalImages, publishedAt) } return