fix: 抓取前预检已入库的秀,避免重复下载(重爬 62 分钟 0 新增)
问题:图片 key 是内容 sha1,不下载就无法判断是否已存在; 后端又没存 source_url,无法做 URL 级判重。 于是每次重爬都把几百张已下过的图再下一遍 —— 线上实测 Celine 跑了 62 分钟、约 600MB,入库 0 新增。 修复: - 后端新增 GET /admin/internal/shows/covered?brand_uid= 按 (年份, 季节, 类型) 聚合已入库图片张数 - spider 抓取前预检,整场已入库(>=10 张)就跳过,一张图都不下载 - 加 --force 可强制重抓(默认跳过) - 媒体直传并发改为全进程共享信号量:此前秀级 2 × 每场 6 = 12 路 同时打 Vogue,被限流后单张从 9.8s 掉到 ~90s - 批次下载失败时打印第一条错误(此前只统计个数,查不出原因)
This commit is contained in:
@ -124,6 +124,9 @@ func executeBrand(ctx context.Context, brandUID, brandName string, maxShows int)
|
|||||||
|
|
||||||
s := spider.NewVogueSpider(serveMaxCo, 0, maxShows, IngestClient)
|
s := spider.NewVogueSpider(serveMaxCo, 0, maxShows, IngestClient)
|
||||||
s.Debug = debug
|
s.Debug = debug
|
||||||
|
// 默认跳过已入库的秀:图片 key 是内容 sha1,不下载无法判重,
|
||||||
|
// 场次级预检是唯一能在下载前省掉整场带宽的手段。
|
||||||
|
s.Force = forceAll
|
||||||
// 媒体机直传:spider 自己下载并上传图片,后端只写库。
|
// 媒体机直传:spider 自己下载并上传图片,后端只写库。
|
||||||
if config.GlobalConfig.Media.Upload && IngestClient != nil {
|
if config.GlobalConfig.Media.Upload && IngestClient != nil {
|
||||||
s.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency)
|
s.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency)
|
||||||
@ -150,4 +153,6 @@ func init() {
|
|||||||
serveCmd.Flags().IntVarP(&serveWorkers, "workers", "w", 1, "并发执行的品牌数(每路一个 goroutine)")
|
serveCmd.Flags().IntVarP(&serveWorkers, "workers", "w", 1, "并发执行的品牌数(每路一个 goroutine)")
|
||||||
serveCmd.Flags().IntVarP(&serveInterval, "interval", "i", 5, "队列为空时的轮询间隔(秒)")
|
serveCmd.Flags().IntVarP(&serveInterval, "interval", "i", 5, "队列为空时的轮询间隔(秒)")
|
||||||
serveCmd.Flags().IntVarP(&serveMaxCo, "max-co", "m", 5, "单个品牌内部的抓取并发数")
|
serveCmd.Flags().IntVarP(&serveMaxCo, "max-co", "m", 5, "单个品牌内部的抓取并发数")
|
||||||
|
// 与 vogue 子命令共用同一个开关:默认跳过已入库的秀,确需刷新时加 --force。
|
||||||
|
serveCmd.Flags().BoolVarP(&forceAll, "force", "f", false, "忽略「已入库」预检,强制重抓全部场次")
|
||||||
}
|
}
|
||||||
|
|||||||
@ -12,6 +12,7 @@ var (
|
|||||||
brandID int64
|
brandID int64
|
||||||
maxCo int
|
maxCo int
|
||||||
maxShows int
|
maxShows int
|
||||||
|
forceAll bool // --force:忽略「已入库」预检,强制重抓
|
||||||
)
|
)
|
||||||
|
|
||||||
var vogueCmd = &cobra.Command{
|
var vogueCmd = &cobra.Command{
|
||||||
@ -24,6 +25,8 @@ var vogueCmd = &cobra.Command{
|
|||||||
if config.GlobalConfig.Media.Upload && IngestClient != nil {
|
if config.GlobalConfig.Media.Upload && IngestClient != nil {
|
||||||
vogueSpider.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency)
|
vogueSpider.Media = media.New(IngestClient, "runway", config.GlobalConfig.Media.Concurrency)
|
||||||
}
|
}
|
||||||
|
// --force:跳过「已入库的秀」预检,强制重抓(默认跳过,避免重复下载)。
|
||||||
|
vogueSpider.Force = forceAll
|
||||||
|
|
||||||
// 2. 修复方法名:将 Execute() 改为 Run()
|
// 2. 修复方法名:将 Execute() 改为 Run()
|
||||||
vogueSpider.Run()
|
vogueSpider.Run()
|
||||||
@ -41,4 +44,7 @@ func init() {
|
|||||||
|
|
||||||
// 小批量试爬:仅抓取单个品牌的前 N 场发布会(0=不限制)。用于控制入库量 / 先看效果。
|
// 小批量试爬:仅抓取单个品牌的前 N 场发布会(0=不限制)。用于控制入库量 / 先看效果。
|
||||||
vogueCmd.Flags().IntVarP(&maxShows, "max-shows", "s", 0, "单个品牌最多抓取的发布会场次(0=不限制,小批量试爬用)")
|
vogueCmd.Flags().IntVarP(&maxShows, "max-shows", "s", 0, "单个品牌最多抓取的发布会场次(0=不限制,小批量试爬用)")
|
||||||
|
|
||||||
|
// 默认跳过已入库的秀(省带宽);确需刷新元数据时加 --force。
|
||||||
|
vogueCmd.Flags().BoolVarP(&forceAll, "force", "f", false, "忽略「已入库」预检,强制重抓全部场次")
|
||||||
}
|
}
|
||||||
|
|||||||
@ -195,6 +195,63 @@ type EmbeddingFailure struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// FetchPendingEmbeddings 拉一批缺向量的图片。remaining 为后端统计的剩余量(-1 = 统计失败)。
|
// 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]int, 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]int, len(out.Items))
|
||||||
|
for _, it := range out.Items {
|
||||||
|
covered[ShowCoverageKey(int(it.Year), it.Season, it.CollectionType)] = it.Images
|
||||||
|
}
|
||||||
|
return covered, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ShowCoverage 一场秀在库里的覆盖情况。
|
||||||
|
type ShowCoverage struct {
|
||||||
|
Year uint16 `json:"year"`
|
||||||
|
Season string `json:"season"`
|
||||||
|
CollectionType string `json:"collection_type"`
|
||||||
|
Images int `json:"images"`
|
||||||
|
}
|
||||||
|
|
||||||
func (c *Client) FetchPendingEmbeddings(ctx context.Context, kind string, limit int) ([]PendingEmbedding, int, error) {
|
func (c *Client) FetchPendingEmbeddings(ctx context.Context, kind string, limit int) ([]PendingEmbedding, int, error) {
|
||||||
if c == nil || c.Endpoint == "" {
|
if c == nil || c.Endpoint == "" {
|
||||||
return nil, 0, fmt.Errorf("ingest client 未配置")
|
return nil, 0, fmt.Errorf("ingest client 未配置")
|
||||||
|
|||||||
@ -31,8 +31,27 @@ import (
|
|||||||
type Uploader struct {
|
type Uploader struct {
|
||||||
Client *ingest.Client // 复用它来签发预签名 URL(HMAC)
|
Client *ingest.Client // 复用它来签发预签名 URL(HMAC)
|
||||||
Prefix string // 存储前缀:runway / street
|
Prefix string // 存储前缀:runway / street
|
||||||
Concurrency int // 下载与上传的并发数
|
Concurrency int // **全进程共享**的下载/上传并发数
|
||||||
HTTP *http.Client
|
HTTP *http.Client
|
||||||
|
|
||||||
|
// sem 全进程共享的信号量,避免「每场秀各开 N 路」叠乘。
|
||||||
|
//
|
||||||
|
// 早先每个 UploadAll 调用各自 make 一个信号量:秀级并发 2 × 每场 6 = 12 路
|
||||||
|
// 同时打 Vogue,被对方限流,单张从 9.8s 掉到 ~90s,还伴随 50% 失败。
|
||||||
|
once sync.Once
|
||||||
|
sem chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
// semaphore 惰性初始化共享信号量(并发数非法时兜底为 1,否则无缓冲 channel 会死锁)。
|
||||||
|
func (u *Uploader) semaphore() chan struct{} {
|
||||||
|
u.once.Do(func() {
|
||||||
|
c := u.Concurrency
|
||||||
|
if c <= 0 {
|
||||||
|
c = 1
|
||||||
|
}
|
||||||
|
u.sem = make(chan struct{}, c)
|
||||||
|
})
|
||||||
|
return u.sem
|
||||||
}
|
}
|
||||||
|
|
||||||
// Result 单张图的直传结果;Err 非空表示这张失败(调用方应回退为上报 URL)。
|
// Result 单张图的直传结果;Err 非空表示这张失败(调用方应回退为上报 URL)。
|
||||||
@ -94,7 +113,19 @@ func (u *Uploader) uploadBatch(ctx context.Context, urls []string, results []Res
|
|||||||
}
|
}
|
||||||
data[i] = body
|
data[i] = body
|
||||||
})
|
})
|
||||||
log.Printf("[media] 下载完成 %d 张(失败 %d)", countOK(data), n-countOK(data))
|
// 失败时把**第一条**错误打出来:失败通常是同一个原因(限流 / 连接重置),
|
||||||
|
// 全打会把日志刷爆,而一条就足以定位。此前只统计个数,导致 50% 失败却查不出原因。
|
||||||
|
if ok := countOK(data); ok < n {
|
||||||
|
for i := 0; i < n; i++ {
|
||||||
|
if errs[i] != nil {
|
||||||
|
log.Printf("[media][warn] 下载失败 %s: %v", clipURL(urls[start+i]), errs[i])
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
log.Printf("[media] 下载完成 %d 张(失败 %d)", ok, n-ok)
|
||||||
|
} else {
|
||||||
|
log.Printf("[media] 下载完成 %d 张(失败 0)", ok)
|
||||||
|
}
|
||||||
|
|
||||||
// ② 生成本批 key(内容 sha1 + 原扩展名)
|
// ② 生成本批 key(内容 sha1 + 原扩展名)
|
||||||
for i := 0; i < n; i++ {
|
for i := 0; i < n; i++ {
|
||||||
@ -164,11 +195,10 @@ func (u *Uploader) parallel(n int, fn func(i int)) {
|
|||||||
if n == 0 {
|
if n == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if u.Concurrency <= 0 {
|
// 共享信号量:并发上限是**整个进程**的,不是每次调用各自一份。
|
||||||
// 兜底:并发为 0 会造成无缓冲 channel 死锁(曾把整条抓取卡住)。
|
// 早先每次 UploadAll 各建一个:秀级并发 2 × 每场 6 = 12 路同时打 Vogue,
|
||||||
u.Concurrency = 1
|
// 被对方限流后单张从 9.8s 掉到 ~90s,并伴随大量失败。
|
||||||
}
|
sem := u.semaphore()
|
||||||
sem := make(chan struct{}, u.Concurrency)
|
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
for i := 0; i < n; i++ {
|
for i := 0; i < n; i++ {
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
@ -224,6 +254,14 @@ func (u *Uploader) put(ctx context.Context, putURL, originalURL string, data []b
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// clipURL 截短 URL 只用于日志(完整 URL 太长,会淹没真正有用的错误信息)。
|
||||||
|
func clipURL(u string) string {
|
||||||
|
if len(u) <= 90 {
|
||||||
|
return u
|
||||||
|
}
|
||||||
|
return u[:90] + "..."
|
||||||
|
}
|
||||||
|
|
||||||
// countOK 数一下成功下载的张数(用于日志定位卡在哪一步)。
|
// countOK 数一下成功下载的张数(用于日志定位卡在哪一步)。
|
||||||
func countOK(data [][]byte) int {
|
func countOK(data [][]byte) int {
|
||||||
n := 0
|
n := 0
|
||||||
|
|||||||
@ -42,6 +42,10 @@ type VogueSpider struct {
|
|||||||
BrandID uint // 客户端指定的品牌 ID(--brand,单品牌调试用)
|
BrandID uint // 客户端指定的品牌 ID(--brand,单品牌调试用)
|
||||||
MaxShows int // 小批量模式:单品牌最多抓取的发布会场次(0=不限制)
|
MaxShows int // 小批量模式:单品牌最多抓取的发布会场次(0=不限制)
|
||||||
Debug bool // Debug 模式:仅输出不入库
|
Debug bool // Debug 模式:仅输出不入库
|
||||||
|
// Force 忽略「已入库」预检,强制重抓全部场次(--force)。
|
||||||
|
// 默认 false:整场已入库就跳过,一张图都不下载——图片 key 是内容 sha1,
|
||||||
|
// 不下载无法判断是否已存在,只能靠这一层预检来省掉整场带宽。
|
||||||
|
Force bool
|
||||||
// Media 媒体机直传器(可为 nil:为 nil 时走旧路径,上报 URL 由后端下载)。
|
// Media 媒体机直传器(可为 nil:为 nil 时走旧路径,上报 URL 由后端下载)。
|
||||||
Media *media.Uploader
|
Media *media.Uploader
|
||||||
// submitted 本次品牌执行中成功上报的场次数(服务化模式回报给后台用)。
|
// submitted 本次品牌执行中成功上报的场次数(服务化模式回报给后台用)。
|
||||||
@ -206,6 +210,53 @@ func (s *VogueSpider) filterUncrawled(showsList []gjson.Result) []gjson.Result {
|
|||||||
return showsList
|
return showsList
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// coveredMinImages 一场秀至少已入库多少张,才算「抓全了、可以整场跳过」。
|
||||||
|
//
|
||||||
|
// 用阈值而不是「>0」:只存了两三张说明当初大面积失败,这时重抓补齐
|
||||||
|
// 比跳过更有价值;而几百张的秀几乎不可能再变,跳过无损失。
|
||||||
|
const coveredMinImages = 10
|
||||||
|
|
||||||
|
// skipCoveredShows 询问后端「这个品牌哪些秀已经入库了」,把命中的整场剔除。
|
||||||
|
//
|
||||||
|
// 这是唯一能在下载**之前**判重的地方:图片 key 是内容 sha1,
|
||||||
|
// 必须先下载才能算出 key,所以「这张图有没有」无法在下载前回答,
|
||||||
|
// 但「这场秀有没有」可以——秀的身份是 品牌+年份+季节+类型。
|
||||||
|
//
|
||||||
|
// 预检失败不阻断抓取:宁可多下一遍,也不能因为一次查询失败把整场丢掉。
|
||||||
|
func (s *VogueSpider) skipCoveredShows(brandUID, brandName string, list []gjson.Result) []gjson.Result {
|
||||||
|
if s.Ingest == nil || s.Force || len(list) == 0 {
|
||||||
|
return list
|
||||||
|
}
|
||||||
|
covered, err := s.Ingest.CoveredShows(context.Background(), brandUID)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("[skip][warn] 预检已入库秀失败,按全量抓取: %v", err)
|
||||||
|
return list
|
||||||
|
}
|
||||||
|
if len(covered) == 0 {
|
||||||
|
return list
|
||||||
|
}
|
||||||
|
|
||||||
|
out := make([]gjson.Result, 0, len(list))
|
||||||
|
skipped := 0
|
||||||
|
for _, info := range list {
|
||||||
|
hed := info.Get("hed").String()
|
||||||
|
ct, season := ingest.ParseCollection(hed)
|
||||||
|
// 年份取不到时不做跳过:键对不上,宁可重抓也不错判成「已有」。
|
||||||
|
if year := s.parseYear(hed); year > 0 {
|
||||||
|
if n, ok := covered[ingest.ShowCoverageKey(year, season, ct)]; ok && n >= coveredMinImages {
|
||||||
|
skipped++
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out = append(out, info)
|
||||||
|
}
|
||||||
|
if skipped > 0 {
|
||||||
|
log.Printf("[skip] [%s] 跳过 %d 场已入库的秀(各 ≥%d 张),剩余 %d 场待抓",
|
||||||
|
brandName, skipped, coveredMinImages, len(out))
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
// SpiderStart 针对单个品牌开启抓取
|
// SpiderStart 针对单个品牌开启抓取
|
||||||
func (s *VogueSpider) SpiderStart(task ingest.CrawlBrand) {
|
func (s *VogueSpider) SpiderStart(task ingest.CrawlBrand) {
|
||||||
brandName := s.getTaskName(task.Name)
|
brandName := s.getTaskName(task.Name)
|
||||||
@ -220,6 +271,9 @@ func (s *VogueSpider) SpiderStart(task ingest.CrawlBrand) {
|
|||||||
|
|
||||||
// 抓取前不再做预检(source_url 预检已下线);直接抓取,去重由后端 worker 按实体键幂等收敛。
|
// 抓取前不再做预检(source_url 预检已下线);直接抓取,去重由后端 worker 按实体键幂等收敛。
|
||||||
showsList = s.filterUncrawled(showsList)
|
showsList = s.filterUncrawled(showsList)
|
||||||
|
// 场次级预检:整场已入库的秀直接跳过,一张图都不下载。
|
||||||
|
// 这是唯一能在「下载之前」判重的位置(图片 key 是内容 sha1,下载后才能算出来)。
|
||||||
|
showsList = s.skipCoveredShows(task.BrandUID, brandName, showsList)
|
||||||
if len(showsList) == 0 {
|
if len(showsList) == 0 {
|
||||||
log.Printf("[Info] 品牌 [%s] 的图集均已爬取过,全部跳过", brandName)
|
log.Printf("[Info] 品牌 [%s] 的图集均已爬取过,全部跳过", brandName)
|
||||||
return
|
return
|
||||||
|
|||||||
Reference in New Issue
Block a user