463a57c497
此前只要某场秀已入库 >=10 张就整场跳过，当初只成功抓了十几张的场次 剩下的图永远补不上。 - payload 增加 expected_images（本场应有张数，后端据此算覆盖率） - 跳过条件：images >= expected - 5；expected 未知时回落 10 张阈值 - 留 5 张余量：个别图在 Vogue 侧永久 404，要求一张不差会让这场秀 永远判不完整、每次全量抓都重下一遍
508 lines
15 KiB
Go
508 lines
15 KiB
Go
package ingest
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/rand"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
neturl "net/url"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// Client 向后台 ingest 接口发送 HMAC 签名上报的客户端。
|
||
type Client struct {
|
||
Endpoint string
|
||
Secret string
|
||
HTTPClient *http.Client
|
||
}
|
||
|
||
// NewClient 创建上报客户端。endpoint 形如 http://host:8092/admin/internal/ingest。
|
||
func NewClient(endpoint, secret string) *Client {
|
||
return &Client{
|
||
Endpoint: endpoint,
|
||
Secret: secret,
|
||
HTTPClient: &http.Client{Timeout: 30 * time.Second},
|
||
}
|
||
}
|
||
|
||
// Submit 把一条走秀上报签名后 POST 到后台，返回 job_id（202 入队即返回）。
|
||
// 失败返回 error（网络/校验/服务端拒绝），调用方应记日志并继续，不中断整轮抓取。
|
||
func (c *Client) Submit(ctx context.Context, p RunwayIngest) (uint32, error) {
|
||
if c == nil || c.Endpoint == "" {
|
||
return 0, fmt.Errorf("ingest client 未配置")
|
||
}
|
||
body, err := json.Marshal(p)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
ts := strconv.FormatInt(time.Now().Unix(), 10)
|
||
nonce, err := newNonce()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
sig := Sign(c.Secret, ts, nonce, string(body))
|
||
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.Endpoint, bytes.NewReader(body))
|
||
if err != nil {
|
||
return 0, 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 0, err
|
||
}
|
||
defer resp.Body.Close()
|
||
raw, _ := io.ReadAll(resp.Body)
|
||
if resp.StatusCode != http.StatusAccepted {
|
||
return 0, fmt.Errorf("ingest http %d: %s", resp.StatusCode, string(raw))
|
||
}
|
||
var out struct {
|
||
JobID uint32 `json:"job_id"`
|
||
Status string `json:"status"`
|
||
}
|
||
if err := json.Unmarshal(raw, &out); err != nil {
|
||
// 202 但响应结构非预期：任务已入队，记日志后当作成功
|
||
log.Printf("[Ingest] 响应解析失败但已被接受: %s", string(raw))
|
||
return 0, nil
|
||
}
|
||
return out.JobID, nil
|
||
}
|
||
|
||
// newNonce 生成 16 字节随机十六进制串，作为一次性 X-Nonce 防重放。
|
||
func newNonce() (string, error) {
|
||
b := make([]byte, 16)
|
||
if _, err := rand.Read(b); err != nil {
|
||
return "", err
|
||
}
|
||
return hex.EncodeToString(b), nil
|
||
}
|
||
|
||
// CrawlBrand 后端抓取任务接口返回的单个品牌（brand_uid 已编码，name 为英文名）。
|
||
type CrawlBrand struct {
|
||
BrandUID string `json:"brand_uid"`
|
||
Name string `json:"name"`
|
||
}
|
||
|
||
// 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, "/")
|
||
}
|
||
|
||
// 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 = 统计失败）。
|
||
// ShowCoverageKey 生成「秀覆盖」预检键：年份|季节|类型，与后端的实体键一一对应。
|
||
func ShowCoverageKey(year int, season, collectionType string) string {
|
||
return strconv.Itoa(year) + "|" + season + "|" + collectionType
|
||
}
|
||
|
||
// CoveredShows 取某品牌已入库的秀，返回「预检键 → 覆盖情况」。
|
||
//
|
||
// 用途：抓取前整场跳过已经抓全的秀。图片 key 是内容 sha1，不下载就无法知道
|
||
// 一张图是否已经存在，所以「跳过」只能在场这一级做。
|
||
func (c *Client) CoveredShows(ctx context.Context, brandUID string) (map[string]ShowCoverage, error) {
|
||
if c == nil || c.Endpoint == "" {
|
||
return nil, fmt.Errorf("ingest client 未配置")
|
||
}
|
||
url := c.internalBase() + "/shows/covered?brand_uid=" + neturl.QueryEscape(brandUID)
|
||
ts := NowTS()
|
||
nonce, err := newNonce()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||
if err != nil {
|
||
return nil, 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, err
|
||
}
|
||
defer resp.Body.Close()
|
||
raw, _ := io.ReadAll(resp.Body)
|
||
if resp.StatusCode != http.StatusOK {
|
||
return nil, fmt.Errorf("shows/covered http %d: %s", resp.StatusCode, string(raw))
|
||
}
|
||
var out struct {
|
||
Items []ShowCoverage `json:"items"`
|
||
}
|
||
if err := json.Unmarshal(raw, &out); err != nil {
|
||
return nil, fmt.Errorf("shows/covered 解析失败: %w", err)
|
||
}
|
||
covered := make(map[string]ShowCoverage, len(out.Items))
|
||
for _, it := range out.Items {
|
||
covered[ShowCoverageKey(int(it.Year), it.Season, it.CollectionType)] = it
|
||
}
|
||
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"`
|
||
Season string `json:"season"`
|
||
CollectionType string `json:"collection_type"`
|
||
// Images 库里实际图片张数。
|
||
Images int `json:"images"`
|
||
// Expected 应有张数（spider 上次上报）。0 表示历史数据没有记录，判定时退化为阈值。
|
||
Expected int `json:"expected"`
|
||
}
|
||
|
||
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，复用与上报相同的验签算法）。
|
||
// brandID>0 时只取该品牌（对应 --brand 单品牌调试）。返回 brand_uid（直接上送 ingest）+ 英文名。
|
||
func (c *Client) GetCrawlBrands(ctx context.Context, brandID uint) ([]CrawlBrand, error) {
|
||
if c == nil || c.Endpoint == "" {
|
||
return nil, fmt.Errorf("ingest client 未配置")
|
||
}
|
||
url := c.crawlBrandsURL()
|
||
if brandID > 0 {
|
||
url += fmt.Sprintf("?brand=%d", brandID)
|
||
}
|
||
// GET 无请求体，签名串的 body 部分为空（与后台 IngestAuth 验签一致）。
|
||
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("crawl brands http %d: %s", resp.StatusCode, string(raw))
|
||
}
|
||
var out struct {
|
||
Brands []CrawlBrand `json:"brands"`
|
||
}
|
||
if err := json.Unmarshal(raw, &out); err != nil {
|
||
return nil, fmt.Errorf("crawl brands 解析失败: %w", err)
|
||
}
|
||
return out.Brands, nil
|
||
}
|