From 7fce365d758c399441497047e9149b479d6b565d Mon Sep 17 00:00:00 2001 From: toom1996 <23cm.cn@gmail.com> Date: Sun, 25 Oct 2026 20:33:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=B8=8A=E6=8A=A5=E6=8A=93=E5=8F=96?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=EF=BC=88=E5=90=8E=E5=8F=B0=E6=98=BE=E7=A4=BA?= =?UTF-8?q?=203/25=20=E5=9C=BA=20=C2=B7=20=E6=AD=A3=E5=9C=A8=E6=8A=93?= =?UTF-8?q?=E4=BB=80=E4=B9=88=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一次品牌抓取可能跑几十分钟，此前后台只有 running/done 两个状态， 看不到「在干什么」，也分不清「慢」和「卡死」。 - getDetail 进/出一场各打一次点（含 404 等失败场次，避免进度永远差几场） - 总数在「预检过滤 + 场次上限」之后定格，与实际要抓的场次一致 - 只对 TaskID != 0（serve 模式）上报，单机 vogue 模式不上报 --- cmd/serve.go | 6 ++++-- internal/ingest/client.go | 38 ++++++++++++++++++++++++++++++++++++++ internal/spider/vogue.go | 33 +++++++++++++++++++++++++++++++++ 3 files changed, 75 insertions(+), 2 deletions(-) diff --git a/cmd/serve.go b/cmd/serve.go index db835c1..2968cd6 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -95,7 +95,7 @@ func runServeWorker(ctx context.Context, workerID string) { workerID, task.ID, task.BrandName, task.BrandUID, task.MaxShows) start := time.Now() - shows, errMsg := executeBrand(ctx, task.BrandUID, task.BrandName, int(task.MaxShows)) + shows, errMsg := executeBrand(ctx, task.ID, task.BrandUID, task.BrandName, int(task.MaxShows)) elapsed := time.Since(start).Round(time.Second) status := "done" @@ -113,7 +113,7 @@ func runServeWorker(ctx context.Context, workerID string) { } // executeBrand 执行单个品牌，异常（panic）也转成错误返回，避免一个坏品牌把整个 serve 进程带崩。 -func executeBrand(ctx context.Context, brandUID, brandName string, maxShows int) (shows int, errMsg string) { +func executeBrand(ctx context.Context, taskID uint32, brandUID, brandName string, maxShows int) (shows int, errMsg string) { defer func() { if r := recover(); r != nil { shows = 0 @@ -127,6 +127,8 @@ func executeBrand(ctx context.Context, brandUID, brandName string, maxShows int) // 默认跳过已入库的秀：图片 key 是内容 sha1，不下载无法判重， // 场次级预检是唯一能在下载前省掉整场带宽的手段。 s.Force = forceAll + // 任务 ID 用于回报执行进度（后台「进度」列）。 + s.TaskID = taskID // 媒体机直传：spider 自己下载并上传图片，后端只写库。 if config.GlobalConfig.Media.Upload && IngestClient != nil { s.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency) diff --git a/internal/ingest/client.go b/internal/ingest/client.go index f62ed7f..bad60ce 100644 --- a/internal/ingest/client.go +++ b/internal/ingest/client.go @@ -244,6 +244,44 @@ func (c *Client) CoveredShows(ctx context.Context, brandUID string) (map[string] return covered, nil } +// ReportProgress 上报执行进度（每处理完一场秀调用一次）。 +// +// 失败只是观测信息丢失：调用方记日志即可，不要中断抓取。 +func (c *Client) ReportProgress(ctx context.Context, taskID uint32, done, total int, text string) error { + if c == nil || c.Endpoint == "" || taskID == 0 { + return nil + } + body, err := json.Marshal(map[string]any{"done": done, "total": total, "text": text}) + if err != nil { + return err + } + ts := NowTS() + nonce, err := newNonce() + if err != nil { + return err + } + url := fmt.Sprintf("%s/crawl/tasks/%d/progress", c.internalBase(), taskID) + 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", 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 err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + raw, _ := io.ReadAll(resp.Body) + return fmt.Errorf("progress http %d: %s", resp.StatusCode, string(raw)) + } + return nil +} + // ShowCoverage 一场秀在库里的覆盖情况。 type ShowCoverage struct { Year uint16 `json:"year"` diff --git a/internal/spider/vogue.go b/internal/spider/vogue.go index dde9506..99d48e0 100644 --- a/internal/spider/vogue.go +++ b/internal/spider/vogue.go @@ -46,6 +46,12 @@ type VogueSpider struct { // 默认 false：整场已入库就跳过，一张图都不下载——图片 key 是内容 sha1， // 不下载无法判断是否已存在，只能靠这一层预检来省掉整场带宽。 Force bool + // TaskID 当前抓取任务 ID（serve 模式由后台派发）；0 表示不是任务驱动，不上报进度。 + TaskID uint32 + // progressDone / progressTotal 本任务的进度计数，供后台列表页显示 + // 「3/25 场 · Fall 2020 Ready-to-Wear」。并发抓取，故用原子操作。 + progressDone int32 + progressTotal int32 // Media 媒体机直传器（可为 nil：为 nil 时走旧路径，上报 URL 由后端下载）。 Media *media.Uploader // submitted 本次品牌执行中成功上报的场次数（服务化模式回报给后台用）。 @@ -295,6 +301,9 @@ func (s *VogueSpider) SpiderStart(task ingest.CrawlBrand) { var wg sync.WaitGroup showSem := make(chan struct{}, maxShowConcurrency) + // 进度总数在这里定格：这是「预检过滤 + 场次上限」之后真正要抓的数量。 + atomic.StoreInt32(&s.progressTotal, int32(len(showsList))) + for _, list := range showsList { wg.Add(1) @@ -328,10 +337,34 @@ func (s *VogueSpider) getShowsList(url string) []gjson.Result { return nil } +// reportProgress 打一次进度点。done=false 表示「正在抓这一场」，true 表示「这一场处理完了」。 +// +// 单独抽成函数，是为了让调用点读起来像打点，而不是把上报逻辑塞进抓取主流程里。 +// 上报失败只记日志：进度是观测信息，丢了顶多后台显示旧一点，不该影响抓取。 +func (s *VogueSpider) reportProgress(title string, done bool) { + if s.Ingest == nil || s.TaskID == 0 { + return + } + n := int(atomic.LoadInt32(&s.progressDone)) + if done { + n = int(atomic.AddInt32(&s.progressDone, 1)) + } + total := int(atomic.LoadInt32(&s.progressTotal)) + if err := s.Ingest.ReportProgress(context.Background(), s.TaskID, n, total, title); err != nil { + log.Printf("[进度] 上报失败: %v", err) + } +} + // getDetail 获取并解析单个发布会详情 func (s *VogueSpider) getDetail(brandUID string, info gjson.Result) { hed := info.Get("hed").String() + // 进度打点：进这一场就报「正在抓 X」，退出时（含 404 等提前 return）再报「已完成 n 场」。 + // 用 defer 而不是在成功分支里报，是为了让失败场次也计入进度， + // 否则界面上会出现「25 场永远差几场」的错觉。 + s.reportProgress(hed, false) + defer s.reportProgress(hed, true) + // 判重交由后端 worker 处理（按实体键幂等去重），spider 不再直连 brand_runway 判重。 // 获取图片（保持你原本的 URL 拼接与请求逻辑）