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) // UpdateProgress 记录执行进度(spider 每处理完一场秀上报一次)。 // 只对 running 生效:任务已被取消/回收时,迟到的进度不应把状态写回去。 UpdateProgress(ctx context.Context, id uint32, done, total int, text string) error // ListActive 取 running + queued 的任务(后台进度轮询用)。 ListActive(ctx context.Context) ([]model.CrawlTaskRow, error) // BrandCrawlStates 返回这些品牌在抓取队列里的「最近状态 + 最近时间」。 // 品牌列表用它展示「抓取状态 / 上次抓取」;没有任务记录的品牌不会出现在 map 里。 BrandCrawlStates(ctx context.Context, brandIDs []uint32) (map[uint32]BrandCrawlState, error) // ListCrawlLogs 取某品牌最近的抓取历史(id 倒序),品牌详情页展示用。 // // crawl_tasks 每品牌只有一行且会被覆盖,历史只存在 brand_crawl_logs 里。 ListCrawlLogs(ctx context.Context, brandID uint32, limit int) ([]model.CrawlLog, error) // ShowCoverage 返回某品牌「已入库的秀 → 已有图片张数」,键是实体键(年份+季节+类型)。 // // 用途:spider 抓取前预检,**整场跳过**已入库的秀。 // 没有它,重爬会把几百张已经下过的图再下一遍——图片 key 是内容 sha1, // 不下载就无法知道它其实已经存在,带宽全白烧(线上实测 62 分钟 0 新增)。 ShowCoverage(ctx context.Context, brandID uint32) (map[ShowKey]int, error) } // ShowKey 一场秀的业务主键,与 ingest 的实体键一致(品牌 + 年份 + 季节 + 类型)。 type ShowKey struct { Year uint16 Season string CollectionType string } // 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.progress_done, t.progress_total, t.progress_text, t.progress_at, 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 } if err := 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; err != nil { return err } // 同事务写一条抓取历史:crawl_tasks 每品牌只有一行会被覆盖, // 不在这里留痕就永远查不到「上次爬是什么时候、抓到几场」。 log := model.CrawlLog{ BrandID: cur.BrandID, TaskID: cur.ID, Status: status, Attempt: cur.Attempts + 1, ShowsRequested: cur.MaxShows, ShowsSubmitted: int32(shows), Message: errMsg, Worker: cur.Worker, StartedAt: cur.LockedAt, FinishedAt: now, } if err := tx.Create(&log).Error; err != nil { return err } // 只保留该品牌最近 N 条:历史用于排障,不是归档,避免无限增长。 return tx.Exec( "DELETE FROM brand_crawl_logs WHERE brand_id = ? AND id < "+ "(SELECT id FROM brand_crawl_logs WHERE brand_id = ? ORDER BY id DESC OFFSET ? LIMIT 1)", cur.BrandID, cur.BrandID, model.CrawlLogKeepPerBrand).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 } // UpdateProgress 写入进度心跳。 // // 用 `AND status = 'running'` 而不是无条件更新: // 后台可能刚把任务取消,此时迟到的进度不该把它写活, // 也不该刷新 updated_at 掩盖「已经被取消」的事实。 func (r *crawlRepository) UpdateProgress(ctx context.Context, id uint32, done, total int, text string) error { if done < 0 { done = 0 } if total < 0 { total = 0 } if len(text) > 160 { text = text[:160] } now := uint32(time.Now().Unix()) return r.db.WithContext(ctx).Exec( `UPDATE crawl_tasks SET progress_done = ?, progress_total = ?, progress_text = ?, progress_at = ?, updated_at = ? WHERE id = ? AND status = 'running'`, done, total, text, now, now, id).Error } // ListActive 取正在跑与排队中的任务,供后台进度轮询(全量返回,不分页)。 // 只取 runnning/queued:已完成的任务进度不会再变,没必要每 4 秒重传一遍。 func (r *crawlRepository) ListActive(ctx context.Context) ([]model.CrawlTaskRow, error) { 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 WHERE t.status IN ('running', 'queued') ORDER BY CASE t.status WHEN 'running' THEN 0 ELSE 1 END, t.id DESC LIMIT 200`).Scan(&rows).Error if err != nil { return nil, err } return rows, 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 } // ListCrawlLogs 取某品牌最近的抓取历史。 func (r *crawlRepository) ListCrawlLogs(ctx context.Context, brandID uint32, limit int) ([]model.CrawlLog, error) { if brandID == 0 || limit <= 0 { return nil, nil } var rows []model.CrawlLog if err := r.db.WithContext(ctx). Where("brand_id = ?", brandID). Order("id DESC"). Limit(limit). Find(&rows).Error; err != nil { return nil, err } return rows, nil } 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 } // ShowCoverage 统计每个「(年份, 季节, 类型)」已入库的图片张数。 // // 只统计未软删的图:一张被删过的秀如果只剩几张,说明当初没抓全, // 调用方会按阈值判断是否值得重抓,而不是一律跳过。 // JOIN 保证 image_count > 0,没有图的秀不会出现在结果里。 func (r *crawlRepository) ShowCoverage(ctx context.Context, brandID uint32) (map[ShowKey]int, error) { var rows []struct { Year uint16 `gorm:"column:year"` Season string `gorm:"column:season"` CollectionType string `gorm:"column:collection_type"` Images int `gorm:"column:images"` } if err := r.db.WithContext(ctx).Raw( `SELECT r.year, coalesce(r.season, '') AS season, coalesce(r.collection_type, '') AS collection_type, count(i.id) AS images FROM brand_runways r JOIN brand_runway_images i ON i.runway_id = r.id AND i.is_deleted = 0 WHERE r.brand_id = ? AND r.is_deleted = 0 GROUP BY r.year, r.season, r.collection_type`, brandID).Scan(&rows).Error; err != nil { return nil, err } out := make(map[ShowKey]int, len(rows)) for _, row := range rows { out[ShowKey{Year: row.Year, Season: row.Season, CollectionType: row.CollectionType}] = row.Images } return out, nil }