feat: 爬虫页取代抓取任务/入库任务两页，抓取改全量+覆盖率跳过
一、后台只留一个「爬虫」页（/admin/spider） 抓取与入库本就是同一条流水线的上下游，分两页看谁都看不全。 新页三块：状态条（在线/卡住 + 队列数 + 待入库数）、正在抓（含进度与心跳）、 最近入库结果（品牌/场次/结果/说明）。只读，取消与重试一并下线。 二、抓取一律全量 删除品牌页的场次上限下拉与 hidden maxShows=3，入队固定 0（全量）。 crawl_tasks.max_shows 列保留，历史任务照常显示。 三、跳过判定改覆盖率 - brand_runways 增加 expected_images（spider 上报的应有张数） - /shows/covered 返回 images + expected - 实际张数 >= 应有 - 5 才跳过；expected=0 时回落 10 张阈值 - 实体已存在时不再直接跳过：按 key 比对把缺的图补进库（repaired） 否则「判定不完整→重爬→后端跳过」会形成死循环，带宽白花、缺图永远补不上 四、租约 900→300 秒，且心跳算存活 只看 locked_at 会把跑了几小时的全量抓取误判成卡死； 加上 progress_at 后，真卡住 5 分钟才回收。
This commit is contained in:
toom1996
@ -47,12 +47,20 @@ type CrawlRepository interface {
|
||||
//
|
||||
// crawl_tasks 每品牌只有一行且会被覆盖，历史只存在 brand_crawl_logs 里。
|
||||
ListCrawlLogs(ctx context.Context, brandID uint32, limit int) ([]model.CrawlLog, error)
|
||||
// ShowCoverage 返回某品牌「已入库的秀 → 已有图片张数」，键是实体键（年份+季节+类型）。
|
||||
// ShowCoverage 返回某品牌「已入库的秀 → 覆盖率」，键是实体键（年份+季节+类型）。
|
||||
//
|
||||
// 用途：spider 抓取前预检，**整场跳过**已入库的秀。
|
||||
// 用途：spider 抓取前预检，整场跳过已经抓全的秀。
|
||||
// 没有它，重爬会把几百张已经下过的图再下一遍——图片 key 是内容 sha1，
|
||||
// 不下载就无法知道它其实已经存在，带宽全白烧（线上实测 62 分钟 0 新增）。
|
||||
ShowCoverage(ctx context.Context, brandID uint32) (map[ShowKey]int, error)
|
||||
ShowCoverage(ctx context.Context, brandID uint32) (map[ShowKey]ShowCoverageStat, error)
|
||||
}
|
||||
|
||||
// ShowCoverageStat 一场秀的图片覆盖情况。
|
||||
//
|
||||
// Expected 为 0 表示未知（该字段上线前入库的历史数据），调用方应退化为按张数阈值判断。
|
||||
type ShowCoverageStat struct {
|
||||
Images int // 库里实际图片张数
|
||||
Expected int // 应有张数（spider 上报，0=未知）
|
||||
}
|
||||
|
||||
// ShowKey 一场秀的业务主键，与 ingest 的实体键一致（品牌 + 年份 + 季节 + 类型）。
|
||||
@ -208,13 +216,19 @@ func (r *crawlRepository) Claim(ctx context.Context, worker string) (*model.Craw
|
||||
}
|
||||
}()
|
||||
|
||||
// 回收租约过期的 running（worker 崩溃 / 断电），否则任务永远卡着。
|
||||
// 回收「静默」的 running（worker 崩溃 / 断电），否则任务永远卡着。
|
||||
//
|
||||
// 存活判定用**两个信号取较晚者**：locked_at（领取时刻）与 progress_at（进度心跳）。
|
||||
// 只看 locked_at 会误杀长任务——一次全量抓取可能跑几个小时，而 locked_at 一写就不再更新；
|
||||
// 加上心跳后，正常在跑的任务永远不会被回收，只有真的 5 分钟没动静才落回队列。
|
||||
if now > model.CrawlStuckTimeoutSec {
|
||||
cutoff := 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 {
|
||||
WHERE status = 'running' AND locked_at > 0
|
||||
AND locked_at < ? AND coalesce(progress_at, 0) < ?`,
|
||||
now, cutoff, cutoff).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
@ -448,16 +462,17 @@ func (r *crawlRepository) Counts(ctx context.Context) (map[string]int64, error)
|
||||
// 只统计未软删的图：一张被删过的秀如果只剩几张，说明当初没抓全，
|
||||
// 调用方会按阈值判断是否值得重抓，而不是一律跳过。
|
||||
// JOIN 保证 image_count > 0，没有图的秀不会出现在结果里。
|
||||
func (r *crawlRepository) ShowCoverage(ctx context.Context, brandID uint32) (map[ShowKey]int, error) {
|
||||
func (r *crawlRepository) ShowCoverage(ctx context.Context, brandID uint32) (map[ShowKey]ShowCoverageStat, 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"`
|
||||
Expected int `gorm:"column:expected"`
|
||||
}
|
||||
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
|
||||
count(i.id) AS images, max(r.expected_images) AS expected
|
||||
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
|
||||
@ -465,9 +480,10 @@ func (r *crawlRepository) ShowCoverage(ctx context.Context, brandID uint32) (map
|
||||
brandID).Scan(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make(map[ShowKey]int, len(rows))
|
||||
out := make(map[ShowKey]ShowCoverageStat, len(rows))
|
||||
for _, row := range rows {
|
||||
out[ShowKey{Year: row.Year, Season: row.Season, CollectionType: row.CollectionType}] = row.Images
|
||||
out[ShowKey{Year: row.Year, Season: row.Season, CollectionType: row.CollectionType}] =
|
||||
ShowCoverageStat{Images: row.Images, Expected: row.Expected}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@ -65,6 +65,15 @@ type IngestRepository interface {
|
||||
BackfillRunwayPublishedAt(ctx context.Context, runwayID uint32, publishedAt uint32, year uint16, targetSeasonCode string) error
|
||||
// RepairRunwayDetailParents 入库后按 sort_order 回填走秀细节图的 parent_image_id（细节图挂到最近主图）。
|
||||
RepairRunwayDetailParents(ctx context.Context, runwayID uint32) error
|
||||
// RunwayImageState 返回某场秀已有图片的 key→id 映射与最大排序号，供「补齐缺图」比对。
|
||||
//
|
||||
// 图片 key 是内容 sha1，spider 直传时会一并上报，所以**不用下载**就能知道哪张缺。
|
||||
RunwayImageState(ctx context.Context, runwayID uint32) (RunwayImageState, error)
|
||||
// CreateRunwayImage 插入一张图片（回填自增 id，供细节图挂 parent_image_id）。
|
||||
CreateRunwayImage(ctx context.Context, img *model.BrandRunwayImage) error
|
||||
// SyncRunwayImageStats 重算 image_count（主图行数），并抬高 expected_images。
|
||||
// expected 只增不减：抓取中途失败导致张数变少时，不应把「应有张数」改小。
|
||||
SyncRunwayImageStats(ctx context.Context, runwayID uint32, expected int) error
|
||||
// StreetSnapEntityState 按实体键（city + year + title）查正式表既有行，返回其状态与是否已被软删。
|
||||
// 与 RunwayEntityState 同源：不再以 is_deleted = 0 过滤，软删实体跳过重爬（删除即永久）。
|
||||
// title 用 COALESCE 与唯一索引 uq_ss_entity 的表达式保持一致。
|
||||
@ -77,6 +86,13 @@ type IngestRepository interface {
|
||||
ListJobs(ctx context.Context, offset, limit int) ([]model.IngestJob, error)
|
||||
// CountJobs 返回 ingest_jobs 总条数（用于分页计算总页数）。
|
||||
CountJobs(ctx context.Context) (int64, error)
|
||||
// CountPendingJobs 返回尚未被 worker 处理的入库任务数（后台状态条）。
|
||||
CountPendingJobs(ctx context.Context) (int64, error)
|
||||
// RecentCrawlResults 取最近若干条爬虫入库结果，带品牌名与场次标题，供「最近入库结果」表格。
|
||||
//
|
||||
// 场次标题从 payload JSON 里取：入库表只存原始载荷，没有 title 列，
|
||||
// 为展示再冗余一列不值得——payload 里本来就有。
|
||||
RecentCrawlResults(ctx context.Context, limit int) ([]CrawlResultRow, error)
|
||||
// RetryJob 把一条 failed 任务重置回 pending，清 last_error/locked_at，等待 worker 重新处理。
|
||||
RetryJob(ctx context.Context, id uint32) error
|
||||
// EnqueueMediaCleanup 写入一条「清理S4孤儿图」任务（payload 为待清理 key 的 JSON）。
|
||||
@ -656,3 +672,124 @@ func indexOf(s, sub string) int {
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
// CrawlResultRow 一条最近的爬虫入库结果（后台展示用）。
|
||||
type CrawlResultRow struct {
|
||||
UpdatedAt uint32
|
||||
Status string
|
||||
Outcome string
|
||||
Note string
|
||||
BrandName string
|
||||
Title string // 场次英文标题（来自 payload）
|
||||
}
|
||||
|
||||
// CountPendingJobs 统计待处理的入库任务。
|
||||
func (r *ingestRepository) CountPendingJobs(ctx context.Context) (int64, error) {
|
||||
var n int64
|
||||
if err := r.db.WithContext(ctx).Raw(
|
||||
`SELECT count(*) FROM ingest_jobs WHERE status = ?`, model.IngestStatusPending).Scan(&n).Error; err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// RecentCrawlResults 取最近 N 条爬虫入库结果。
|
||||
//
|
||||
// payload 用 `LIKE '{%'` 先挡一道再转 jsonb：正常载荷一定是 JSON 对象，
|
||||
// 但历史脏数据/空串会让 `::jsonb` 抛错并整页 500——展示页不该被一条坏数据拖垮。
|
||||
func (r *ingestRepository) RecentCrawlResults(ctx context.Context, limit int) ([]CrawlResultRow, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
var rows []struct {
|
||||
UpdatedAt uint32 `gorm:"column:updated_at"`
|
||||
Status string `gorm:"column:status"`
|
||||
Outcome string `gorm:"column:outcome"`
|
||||
Note string `gorm:"column:note"`
|
||||
BrandName string `gorm:"column:brand_name"`
|
||||
Title string `gorm:"column:title"`
|
||||
}
|
||||
if err := r.db.WithContext(ctx).Raw(`
|
||||
SELECT j.updated_at, j.status, coalesce(j.outcome, '') AS outcome,
|
||||
coalesce(j.note, '') AS note,
|
||||
coalesce(b.name_en, '') AS brand_name,
|
||||
CASE WHEN j.payload LIKE '{%' THEN coalesce(j.payload::jsonb->>'title_en', '') ELSE '' END AS title
|
||||
FROM ingest_jobs j
|
||||
LEFT JOIN brands b ON b.id = j.brand_id
|
||||
WHERE j.kind = ?
|
||||
ORDER BY j.id DESC
|
||||
LIMIT ?`, model.IngestKindCrawl, limit).Scan(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]CrawlResultRow, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
out = append(out, CrawlResultRow{
|
||||
UpdatedAt: row.UpdatedAt, Status: row.Status, Outcome: row.Outcome,
|
||||
Note: row.Note, BrandName: row.BrandName, Title: row.Title,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// RunwayImageState 一场秀已有图片的快照：key→id 映射 + 最大排序号。
|
||||
type RunwayImageState struct {
|
||||
// Keys 图片存储 key（如 runway/<sha1>.jpg）→ 图片行 id。
|
||||
Keys map[string]uint32
|
||||
// MaxSort 当前最大 sort_order；补齐的新图从它往后排。
|
||||
MaxSort uint32
|
||||
}
|
||||
|
||||
// RunwayImageState 读取已有图片（只含未删除的行）。
|
||||
//
|
||||
// 用 key 而不是 URL 做比对：库里存的就是 key，spider 直传后上报的也是 key，
|
||||
// 于是「这张图有没有」变成一次纯内存查表，不用下载、不用请求存储。
|
||||
func (r *ingestRepository) RunwayImageState(ctx context.Context, runwayID uint32) (RunwayImageState, error) {
|
||||
var rows []struct {
|
||||
ID uint32 `gorm:"column:id"`
|
||||
Image string `gorm:"column:image"`
|
||||
SortOrder uint32 `gorm:"column:sort_order"`
|
||||
}
|
||||
if err := r.db.WithContext(ctx).Raw(
|
||||
`SELECT id, coalesce(image, '') AS image, coalesce(sort_order, 0) AS sort_order
|
||||
FROM brand_runway_images WHERE runway_id = ? AND is_deleted = 0`,
|
||||
runwayID).Scan(&rows).Error; err != nil {
|
||||
return RunwayImageState{}, err
|
||||
}
|
||||
st := RunwayImageState{Keys: make(map[string]uint32, len(rows))}
|
||||
for _, row := range rows {
|
||||
if row.Image != "" {
|
||||
st.Keys[row.Image] = row.ID
|
||||
}
|
||||
if row.SortOrder > st.MaxSort {
|
||||
st.MaxSort = row.SortOrder
|
||||
}
|
||||
}
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// CreateRunwayImage 插入一张图片并回填自增 id。
|
||||
// 细节图需要把父主图的 id 写进 parent_image_id，所以必须拿回 id。
|
||||
func (r *ingestRepository) CreateRunwayImage(ctx context.Context, img *model.BrandRunwayImage) error {
|
||||
now := uint32(time.Now().Unix())
|
||||
img.CreatedAt, img.UpdatedAt = now, now
|
||||
return r.db.WithContext(ctx).Create(img).Error
|
||||
}
|
||||
|
||||
// SyncRunwayImageStats 重算主图张数并抬高应有张数。
|
||||
//
|
||||
// image_count 用子查询从实际行数算，而不是传入累加值：
|
||||
// 并发补图或中途失败时，累加值容易和真实行数漂移，而 image_count 会直接影响
|
||||
// 公开卡片的「N 张」角标与热度排序，必须与真实行一致。
|
||||
func (r *ingestRepository) SyncRunwayImageStats(ctx context.Context, runwayID uint32, expected int) error {
|
||||
if expected < 0 {
|
||||
expected = 0
|
||||
}
|
||||
return r.db.WithContext(ctx).Exec(`
|
||||
UPDATE brand_runways SET
|
||||
image_count = (SELECT count(*) FROM brand_runway_images
|
||||
WHERE runway_id = ? AND is_deleted = 0 AND is_detail = 0),
|
||||
expected_images = greatest(expected_images, ?),
|
||||
updated_at = ?
|
||||
WHERE id = ?`,
|
||||
runwayID, expected, uint32(time.Now().Unix()), runwayID).Error
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user