c8a70760cc
背景：任务只有 status=done，分不清「新建」「已存在跳过」「已删除跳过」， 品牌更新历史只能在进程日志里翻，日志一滚就没了。 - ingest_jobs 新增 outcome / note / brand_id（迁移 2026-10-25-01） outcome 记结果类型，note 记人话细节（新建 N 张图 / 已存在(status=published)） - 提交任务时解出 brand_id 落库，品牌详情页按 (brand_id, id DESC) 直接查 - 处理时优先用 job.brand_id，payload 里的 hashid 只作老任务兜底： 避免「别的实例盐不同 → 解出错品牌 → 整场 brand not found」再发生 - 品牌编辑页新增「更新记录」区块：时间 / 场次 / 结果 / 明细（含失败原因与尝试次数） - 测试：fake 仓储补齐新接口，capture 仓储记录 outcome 供断言
639 lines
27 KiB
Go
639 lines
27 KiB
Go
package repository
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"fashionapi/internal/model"
|
||
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// IngestRepository 爬虫入库管线专属仓储：任务队列（ingest_jobs）+ nonce 防重放
|
||
// （ingest_nonces）+ 走秀正式表写入（brand_runways / brand_runway_images）。
|
||
//
|
||
// 实体键：走秀 = brand_id + season_code + collection_type，街拍 = city + year + title
|
||
// （同城同年可有多个专题，title 是数据里区分专题的字段）。
|
||
// 单表发布模型下入库直写正式表（status=pending），实体键命中后按既有行状态分三支：
|
||
// pending/published → 放弃本次入库；rejected → 复用该行重审。因此一个秀/街拍始终只留一行。
|
||
//
|
||
// 入队与领取用同一张 ingest_jobs 表，领取靠 PostgreSQL 的 FOR UPDATE SKIP LOCKED
|
||
// 实现「多 worker 安全并发」——同一条任务只会被一个 worker 拿到，其它 worker 跳过它。
|
||
type IngestRepository interface {
|
||
// Enqueue 写入一条待处理任务（payload 为原始 JSON）。
|
||
Enqueue(ctx context.Context, job *model.IngestJob) error
|
||
// Claim 原子领取最多 limit 条 pending 任务并置为 processing，返回这些任务。
|
||
// 用 SKIP LOCKED 保证多 worker 不抢同一条。
|
||
Claim(ctx context.Context, limit int) ([]model.IngestJob, error)
|
||
// MarkDone 标记任务成功，并记录结果类型与人可读细节。
|
||
//
|
||
// outcome 取 model.IngestOutcome* 常量，note 例如「新建 42 张图」「已存在(status=published)」。
|
||
// 二者是后台「品牌更新记录」的全部信息源：只有 status=done 看不出这次到底新建还是跳过。
|
||
MarkDone(ctx context.Context, id uint32, outcome, note string) error
|
||
// ListByBrand 取某品牌的入库记录（按 id 倒序），供品牌详情页展示更新历史。
|
||
ListByBrand(ctx context.Context, brandID uint32, limit int) ([]BrandIngestRecord, error)
|
||
// MarkFailed 标记任务失败并记录错误（attempts 自增）。
|
||
MarkFailed(ctx context.Context, id uint32, errMsg string) error
|
||
// ReserveNonce 写入一次性随机串；若已存在（重放）返回 ok=false。
|
||
ReserveNonce(ctx context.Context, nonce string) (ok bool, err error)
|
||
// RunwayEntityState 按实体键（brand_id + season_code + collection_type）查正式表既有行，
|
||
// 返回其状态与是否已被软删（isDeleted）。worker 据此分四支：
|
||
// - 命中 pending/published → 直接放弃本次入库（ingestSkip）
|
||
// - 命中 rejected → 复用该行重审（ingestReuseRejected）
|
||
// - 命中但已软删 → 跳过重爬（删除即永久，不再因重爬复活）
|
||
// - 未命中 → 新建（ingestCreate）
|
||
// 注意：本查询不再以 is_deleted = 0 为过滤条件（旧版口径），否则软删行会被当成「未命中」
|
||
// 而重新入库，违背「删除即永久」。优先取活着行（ORDER BY is_deleted ASC, id DESC）。
|
||
RunwayEntityState(ctx context.Context, brandID uint32, seasonCode, collectionType string) (id uint32, status string, found bool, isDeleted bool, err error)
|
||
// CreateRunwayWithImages 新建走秀正式行（status=pending）并写入图片，返回记录主键（事务内完成）。
|
||
CreateRunwayWithImages(ctx context.Context, rw *model.BrandRunway, imgs []model.BrandRunwayImage) (uint32, error)
|
||
// ReuseRejectedRunway 复用一条 rejected 走秀：覆盖内容字段、软删旧图、写入新图、置回 pending（事务内完成）。
|
||
// 驳回因此不是永久黑名单：重爬同一实体即重新送审，且始终「一个实体一行」。
|
||
ReuseRejectedRunway(ctx context.Context, id uint32, rw *model.BrandRunway, imgs []model.BrandRunwayImage) error
|
||
// BackfillRunwayPublishedAt 临时回填：给已存在走秀补 published_at，并在标题缺年份时
|
||
// 用发布日期年份补全 year / season_code（不重下图片）。回填为一次性操作，按主键 id 更新。
|
||
// publishedAt 为 Unix 时间戳（秒），0 表示未抽取。
|
||
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
|
||
// StreetSnapEntityState 按实体键（city + year + title）查正式表既有行，返回其状态与是否已被软删。
|
||
// 与 RunwayEntityState 同源：不再以 is_deleted = 0 过滤，软删实体跳过重爬（删除即永久）。
|
||
// title 用 COALESCE 与唯一索引 uq_ss_entity 的表达式保持一致。
|
||
StreetSnapEntityState(ctx context.Context, city string, year uint16, title string) (id uint32, status string, found bool, isDeleted bool, err error)
|
||
// CreateStreetSnapWithImages 新建街拍正式行（status=pending）并写入图片（事务内完成）。
|
||
CreateStreetSnapWithImages(ctx context.Context, snap *model.StreetSnap, imgs []model.StreetSnapImage) (uint32, error)
|
||
// ReuseRejectedStreetSnap 复用一条 rejected 街拍记录（事务内完成）。
|
||
ReuseRejectedStreetSnap(ctx context.Context, id uint32, snap *model.StreetSnap, imgs []model.StreetSnapImage) error
|
||
// ListJobs 按 id 倒序分页列出入库任务（用于后台监控页）。offset/limit 控制分页。
|
||
ListJobs(ctx context.Context, offset, limit int) ([]model.IngestJob, error)
|
||
// CountJobs 返回 ingest_jobs 总条数（用于分页计算总页数）。
|
||
CountJobs(ctx context.Context) (int64, error)
|
||
// RetryJob 把一条 failed 任务重置回 pending，清 last_error/locked_at，等待 worker 重新处理。
|
||
RetryJob(ctx context.Context, id uint32) error
|
||
// EnqueueMediaCleanup 写入一条「清理S4孤儿图」任务（payload 为待清理 key 的 JSON）。
|
||
// 由删除图集的服务调用，把同步的S4删除改为异步队列，避免阻塞删除请求。
|
||
EnqueueMediaCleanup(ctx context.Context, payload string) error
|
||
// ScheduleRetry 失败时调用：attempts+1，未达上限则退避后重置 pending，达上限则置 failed。
|
||
// 用于临时失败（网络抖动 / 单图下载失败）的自动重试，区别于永久失败（payload 解析错等）直接 MarkFailed。
|
||
ScheduleRetry(ctx context.Context, id uint32, errMsg string) error
|
||
}
|
||
|
||
type ingestRepository struct {
|
||
db *gorm.DB
|
||
}
|
||
|
||
// NewIngestRepository 创建入库管线仓储。
|
||
func NewIngestRepository(db *gorm.DB) IngestRepository {
|
||
return &ingestRepository{db: db}
|
||
}
|
||
|
||
func (r *ingestRepository) Enqueue(ctx context.Context, job *model.IngestJob) error {
|
||
now := uint32(time.Now().Unix())
|
||
job.CreatedAt = now
|
||
job.UpdatedAt = now
|
||
job.Status = model.IngestStatusPending
|
||
return r.db.WithContext(ctx).Create(job).Error
|
||
}
|
||
|
||
// Claim 在事务内 SELECT ... FOR UPDATE SKIP LOCKED 锁定 pending 行，
|
||
// 立即置为 processing，再返回这些行，保证领取与状态变更原子、且不被其它 worker 重复领取。
|
||
//
|
||
// 事务内先做两件事：
|
||
// 1. 回收卡死的 processing 任务——worker 崩溃/被杀会留下 processing 孤儿永久卡死，
|
||
// 锁定超时（IngestStuckTimeoutSec）后重置回 pending 并立即可领（next_attempt_at=now）。
|
||
// 2. 仅领取「已到重试时间」的 pending（next_attempt_at <= now），未到退避点的暂不领。
|
||
func (r *ingestRepository) Claim(ctx context.Context, limit int) ([]model.IngestJob, error) {
|
||
if limit < 1 {
|
||
limit = 10
|
||
}
|
||
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()
|
||
}
|
||
}()
|
||
|
||
// 1) 回收卡死的 processing 任务（仅当时间足够大，避免服务器启动初期把刚领取的任务误回收）。
|
||
if now > model.IngestStuckTimeoutSec {
|
||
if err := tx.Model(&model.IngestJob{}).
|
||
Where("status = ? AND locked_at > 0 AND locked_at < ?",
|
||
model.IngestStatusProcessing, now-model.IngestStuckTimeoutSec).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusPending,
|
||
"locked_at": 0,
|
||
"next_attempt_at": now,
|
||
"updated_at": now,
|
||
}).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
// 2) 领取 pending 且已到重试时间的任务（含刚回收的 + 新入队的 + 退避已到期的）。
|
||
// 优先处理 media_cleanup（删除图集时异步清理S4），避免被大批 crawl 任务排到后面、清理迟迟不触发。
|
||
var ids []uint32
|
||
if err := tx.Raw(
|
||
"SELECT id FROM ingest_jobs WHERE status = ? AND next_attempt_at <= ? "+
|
||
"ORDER BY CASE kind WHEN ? THEN 0 ELSE 1 END, id ASC LIMIT ? FOR UPDATE SKIP LOCKED",
|
||
model.IngestStatusPending, now, model.IngestKindMediaCleanup, limit,
|
||
).Scan(&ids).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
if len(ids) == 0 {
|
||
tx.Commit()
|
||
return nil, nil
|
||
}
|
||
if err := tx.Model(&model.IngestJob{}).
|
||
Where("id IN ?", ids).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusProcessing,
|
||
"locked_at": now,
|
||
"updated_at": now,
|
||
}).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
if err := tx.Commit().Error; err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
var jobs []model.IngestJob
|
||
if err := r.db.WithContext(ctx).Where("id IN ?", ids).Find(&jobs).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return jobs, nil
|
||
}
|
||
|
||
func (r *ingestRepository) MarkDone(ctx context.Context, id uint32, outcome, note string) error {
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.IngestJob{}).
|
||
Where("id = ?", id).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusDone,
|
||
"outcome": outcome,
|
||
"note": note,
|
||
"last_error": "",
|
||
"updated_at": uint32(time.Now().Unix()),
|
||
}).Error
|
||
}
|
||
|
||
// BrandIngestRecord 是品牌详情页需要的「一条更新记录」。
|
||
//
|
||
// 只取必要列：payload 可能几十 KB（含全部 look 的 key），一次拉 20 条会传上兆，
|
||
// 所以标题直接在 SQL 里从 JSON 取出，不回传整段 payload。
|
||
type BrandIngestRecord struct {
|
||
ID uint32 `gorm:"column:id"`
|
||
Title string `gorm:"column:title"`
|
||
Status string `gorm:"column:status"`
|
||
Outcome string `gorm:"column:outcome"`
|
||
Note string `gorm:"column:note"`
|
||
LastError string `gorm:"column:last_error"`
|
||
Attempts uint8 `gorm:"column:attempts"`
|
||
UpdatedAt uint32 `gorm:"column:updated_at"`
|
||
}
|
||
|
||
// ListByBrand 取某品牌的入库记录（按 id 倒序），供品牌详情页展示更新历史。
|
||
func (r *ingestRepository) ListByBrand(ctx context.Context, brandID uint32, limit int) ([]BrandIngestRecord, error) {
|
||
if brandID == 0 || limit <= 0 {
|
||
return nil, nil
|
||
}
|
||
var rows []BrandIngestRecord
|
||
if err := r.db.WithContext(ctx).Raw(
|
||
"SELECT id, status, coalesce(outcome, '') AS outcome, coalesce(note, '') AS note, "+
|
||
"coalesce(last_error, '') AS last_error, attempts, updated_at, "+
|
||
"coalesce(payload::json->>'title_en', '') AS title "+
|
||
"FROM ingest_jobs WHERE brand_id = ? ORDER BY id DESC LIMIT ?",
|
||
brandID, limit).Scan(&rows).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
func (r *ingestRepository) MarkFailed(ctx context.Context, id uint32, errMsg string) error {
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.IngestJob{}).
|
||
Where("id = ?", id).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusFailed,
|
||
"attempts": gorm.Expr("attempts + 1"),
|
||
"last_error": errMsg,
|
||
"updated_at": uint32(time.Now().Unix()),
|
||
}).Error
|
||
}
|
||
|
||
// ScheduleRetry 失败时调度自动重试：attempts+1，未达上限（IngestMaxAttempts）则按指数退避
|
||
// 重置为 pending 并写入 next_attempt_at（到点才可被 Claim 领取）；达上限则置 failed，需人工处理。
|
||
// 用于临时失败（网络抖动 / 单图下载失败），区别于永久失败（payload 解析错等）直接 MarkFailed。
|
||
func (r *ingestRepository) ScheduleRetry(ctx context.Context, id uint32, errMsg string) error {
|
||
now := uint32(time.Now().Unix())
|
||
var job model.IngestJob
|
||
if err := r.db.WithContext(ctx).Select("attempts").Where("id = ?", id).First(&job).Error; err != nil {
|
||
return err
|
||
}
|
||
attempts := int(job.Attempts) + 1
|
||
if attempts >= model.IngestMaxAttempts {
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.IngestJob{}).
|
||
Where("id = ?", id).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusFailed,
|
||
"attempts": attempts,
|
||
"last_error": errMsg,
|
||
"updated_at": now,
|
||
}).Error
|
||
}
|
||
delay := model.IngestRetryBackoff(attempts)
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.IngestJob{}).
|
||
Where("id = ?", id).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusPending,
|
||
"attempts": attempts,
|
||
"last_error": errMsg,
|
||
"next_attempt_at": now + uint32(delay),
|
||
"locked_at": 0,
|
||
"updated_at": now,
|
||
}).Error
|
||
}
|
||
|
||
// EnqueueMediaCleanup 写入一条「清理S4孤儿图」任务（kind=media_cleanup），
|
||
// payload 为待清理 key 的 JSON，由 worker 的 processMediaCleanup 按引用计数判定真孤儿后删除。
|
||
func (r *ingestRepository) EnqueueMediaCleanup(ctx context.Context, payload string) error {
|
||
now := uint32(time.Now().Unix())
|
||
job := &model.IngestJob{
|
||
Kind: model.IngestKindMediaCleanup,
|
||
Payload: payload,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
Status: model.IngestStatusPending,
|
||
}
|
||
return r.db.WithContext(ctx).Create(job).Error
|
||
}
|
||
|
||
// ReserveNonce 写入一次性随机串；依赖 ingest_nces.nonce 主键唯一约束，
|
||
// 重复插入触发 DuplicateEntry → 视为重放，返回 ok=false。
|
||
func (r *ingestRepository) ReserveNonce(ctx context.Context, nonce string) (bool, error) {
|
||
err := r.db.WithContext(ctx).Create(&model.IngestNonce{
|
||
Nonce: nonce,
|
||
CreatedAt: uint32(time.Now().Unix()),
|
||
}).Error
|
||
if err != nil {
|
||
// 唯一键冲突 → 重放
|
||
if errors.Is(err, gorm.ErrDuplicatedKey) || isDuplicateKey(err) {
|
||
return false, nil
|
||
}
|
||
return false, err
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// ---- 入库直写正式表（单表发布模型）----
|
||
|
||
// RunwayEntityState 按实体键（brand_id + season_code + collection_type）查正式表既有行。
|
||
// 不再过滤 is_deleted：软删行也会被查到，由调用方据此跳过重爬（删除即永久）。
|
||
// 优先取活着行（is_deleted ASC），保证「活着行 + 软删行同键共存」的存量数据下也返回活行。
|
||
func (r *ingestRepository) RunwayEntityState(ctx context.Context, brandID uint32, seasonCode, collectionType string) (uint32, string, bool, bool, error) {
|
||
var row struct {
|
||
ID uint32 `gorm:"column:id"`
|
||
Status string `gorm:"column:status"`
|
||
IsDeleted uint8 `gorm:"column:is_deleted"`
|
||
}
|
||
err := r.db.WithContext(ctx).
|
||
Model(&model.BrandRunway{}).
|
||
Select("id, status, is_deleted").
|
||
Where("brand_id = ? AND season_code = ? AND collection_type = ?", brandID, seasonCode, collectionType).
|
||
Order("is_deleted ASC, id DESC").
|
||
Limit(1).
|
||
Scan(&row).Error
|
||
if err != nil {
|
||
return 0, "", false, false, err
|
||
}
|
||
if row.ID == 0 {
|
||
return 0, "", false, false, nil
|
||
}
|
||
return row.ID, row.Status, true, row.IsDeleted == 1, nil
|
||
}
|
||
|
||
func (r *ingestRepository) CreateRunwayWithImages(ctx context.Context, rw *model.BrandRunway, imgs []model.BrandRunwayImage) (uint32, error) {
|
||
now := uint32(time.Now().Unix())
|
||
rw.CreatedAt, rw.UpdatedAt = now, now
|
||
if rw.Status == "" {
|
||
rw.Status = model.StatusPending
|
||
}
|
||
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if cErr := tx.Create(rw).Error; cErr != nil {
|
||
return cErr
|
||
}
|
||
if len(imgs) == 0 {
|
||
return nil
|
||
}
|
||
for i := range imgs {
|
||
imgs[i].RunwayID = rw.ID
|
||
imgs[i].CreatedAt, imgs[i].UpdatedAt = now, now
|
||
}
|
||
return tx.Create(&imgs).Error
|
||
})
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return rw.ID, nil
|
||
}
|
||
|
||
func (r *ingestRepository) ReuseRejectedRunway(ctx context.Context, id uint32, rw *model.BrandRunway, imgs []model.BrandRunwayImage) error {
|
||
now := uint32(time.Now().Unix())
|
||
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
// WHERE 带 status=rejected 守卫：状态是读阶段拿到的，而写库发生在整批图片下载/上传之后
|
||
// （可能数十秒窗口）。若无条件覆盖，期间已被改成 published 的行会被置回 pending，
|
||
// 等于把已发布内容从公开视图上撤下。
|
||
upd := tx.Model(&model.BrandRunway{}).Where("id = ? AND status = ? AND is_deleted = 0", id, model.StatusRejected)
|
||
if uErr := upd.Updates(map[string]any{
|
||
"title_en": rw.TitleEn,
|
||
"title_cn": rw.TitleCn,
|
||
"description_en": rw.DescriptionEn,
|
||
"description_cn": rw.DescriptionCn,
|
||
"year": rw.Year,
|
||
"season": rw.Season,
|
||
"collection_type": rw.CollectionType,
|
||
"season_code": rw.SeasonCode,
|
||
"cover": rw.Cover,
|
||
"image_count": rw.ImageCount,
|
||
"job_id": rw.JobID,
|
||
"status": model.StatusPending,
|
||
"reviewer": "",
|
||
"reject_reason": "",
|
||
"updated_at": now,
|
||
}).Error; uErr != nil {
|
||
return uErr
|
||
}
|
||
if upd.RowsAffected == 0 {
|
||
return fmt.Errorf("runway %d 复用前置条件失效：该行已不是 rejected 状态或已被软删", id)
|
||
}
|
||
if dErr := tx.Model(&model.BrandRunwayImage{}).
|
||
Where("runway_id = ? AND is_deleted = 0", id).
|
||
Updates(map[string]any{"is_deleted": 1, "updated_at": now}).Error; dErr != nil {
|
||
return dErr
|
||
}
|
||
if len(imgs) == 0 {
|
||
return nil
|
||
}
|
||
for i := range imgs {
|
||
imgs[i].RunwayID = id
|
||
imgs[i].CreatedAt, imgs[i].UpdatedAt = now, now
|
||
}
|
||
return tx.Create(&imgs).Error
|
||
})
|
||
}
|
||
|
||
// RepairRunwayDetailParents 入库后按 sort_order 顺序回填走秀细节图的 parent_image_id：
|
||
// 每张 is_detail=1 的图挂到它之前最近的一张 is_detail=0（主图）之下。
|
||
// 入库时细节图紧随其主图之后（同一 Look 先主图后细节），故「最近主图」即其归属主图。
|
||
func (r *ingestRepository) RepairRunwayDetailParents(ctx context.Context, runwayID uint32) error {
|
||
var imgs []model.BrandRunwayImage
|
||
if err := r.db.WithContext(ctx).
|
||
Where("runway_id = ? AND is_deleted = 0", runwayID).
|
||
Order("sort_order ASC, id ASC").
|
||
Find(&imgs).Error; err != nil {
|
||
return err
|
||
}
|
||
lastMain := uint32(0)
|
||
type fix struct{ id, parent uint32 }
|
||
fixes := make([]fix, 0)
|
||
for _, im := range imgs {
|
||
if im.IsDetail == 0 {
|
||
lastMain = im.ID
|
||
continue
|
||
}
|
||
if im.IsDetail == 1 && im.ParentImageID == 0 && lastMain != 0 {
|
||
fixes = append(fixes, fix{im.ID, lastMain})
|
||
}
|
||
}
|
||
for _, f := range fixes {
|
||
if err := r.db.WithContext(ctx).
|
||
Model(&model.BrandRunwayImage{}).
|
||
Where("id = ? AND runway_id = ? AND is_deleted = 0", f.id, runwayID).
|
||
Update("parent_image_id", f.parent).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// BackfillRunwayPublishedAt 临时回填（一次性）：按主键 id 给已存在走秀补 published_at；
|
||
// 并在标题缺年份（year=0）时，用发布日期年份补全 year / season_code。
|
||
//
|
||
// 设计要点：
|
||
// - published_at 总是写入（按 id，幂等，不会触犯唯一索引）。
|
||
// - year / season_code 仅在「当前 year=0」且「目标 season_code 不与同品牌其它行冲突」时补全，
|
||
// 避免触犯 (brand_id, season_code, collection_type) 唯一索引；冲突时只补 published_at，留待人工处理。
|
||
// - 不重下图片。
|
||
// - publishedAt 为 Unix 时间戳（秒），0 表示未抽取。
|
||
func (r *ingestRepository) BackfillRunwayPublishedAt(ctx context.Context, runwayID uint32, publishedAt uint32, year uint16, targetSeasonCode string) error {
|
||
now := time.Now().Unix()
|
||
if err := r.db.WithContext(ctx).
|
||
Model(&model.BrandRunway{}).
|
||
Where("id = ? AND is_deleted = 0", runwayID).
|
||
Updates(map[string]any{"published_at": publishedAt, "updated_at": now}).Error; err != nil {
|
||
return err
|
||
}
|
||
if year == 0 {
|
||
return nil
|
||
}
|
||
// 仅当目标 season_code 不与同品牌其它行冲突时才补全 year / season_code。
|
||
// 注意：占位符顺序必须是 (runwayID, targetSeasonCode) —— 本 Where 有两个 ?，
|
||
// 漏传 runwayID 会让 season_code 落到 id 槽位，生成 `WHERE id = 'FW27'`
|
||
// 这种必然语法错误的 SQL（该缺陷曾导致「重爬已存在走秀」的任务无限重试）。
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.BrandRunway{}).
|
||
Where("id = ? AND year = 0 AND is_deleted = 0 "+
|
||
"AND NOT EXISTS (SELECT 1 FROM brand_runways b "+
|
||
"WHERE b.brand_id = brand_runways.brand_id AND b.season_code = ? "+
|
||
"AND b.collection_type = brand_runways.collection_type AND b.id <> brand_runways.id AND b.is_deleted = 0)",
|
||
runwayID, targetSeasonCode).
|
||
Updates(map[string]any{
|
||
"year": year,
|
||
"season_code": targetSeasonCode,
|
||
"updated_at": now,
|
||
}).Error
|
||
}
|
||
|
||
// StreetSnapEntityState 按实体键（city + year + title）查正式表既有行，返回其状态与是否已被软删。
|
||
// title 用 COALESCE 与唯一索引 uq_ss_entity 的表达式保持一致：唯一索引里 NULL 互不冲突，
|
||
// 若这里写成裸 title，查重口径会比索引更严，导致重爬同城同年（title 为 NULL 的历史行）时误判未命中、
|
||
// 进而插入一行注定被索引拒绝的记录。
|
||
//
|
||
// 入参 title 会先经 NormalizeStreetTitle 归一化：与落库口径（processStreet 同样归一化）保持一致，
|
||
// 否则「空白 / 多空格」微调会被判为新实体，同城同年多出重复行（见设计规格 §12.4）。
|
||
//
|
||
// 不再过滤 is_deleted：软删行也会被查到，由调用方据此跳过重爬（删除即永久）。
|
||
// 与 RunwayEntityState 同源，优先取活着行（ORDER BY is_deleted ASC, id DESC）。
|
||
func (r *ingestRepository) StreetSnapEntityState(ctx context.Context, city string, year uint16, title string) (uint32, string, bool, bool, error) {
|
||
title = NormalizeStreetTitle(title)
|
||
var row struct {
|
||
ID uint32 `gorm:"column:id"`
|
||
Status string `gorm:"column:status"`
|
||
IsDeleted uint8 `gorm:"column:is_deleted"`
|
||
}
|
||
err := r.db.WithContext(ctx).
|
||
Model(&model.StreetSnap{}).
|
||
Select("id, status, is_deleted").
|
||
Where("city = ? AND year = ? AND COALESCE(title, '') = ?", city, year, title).
|
||
Order("is_deleted ASC, id DESC").
|
||
Limit(1).
|
||
Scan(&row).Error
|
||
if err != nil {
|
||
return 0, "", false, false, err
|
||
}
|
||
if row.ID == 0 {
|
||
return 0, "", false, false, nil
|
||
}
|
||
return row.ID, row.Status, true, row.IsDeleted == 1, nil
|
||
}
|
||
|
||
// NormalizeStreetTitle 街拍实体键标题归一化：去除首尾空白，并把内部任意空白（空格 / 制表 / 换行）
|
||
// 连续段折叠为单个空格。
|
||
//
|
||
// 为什么需要：实体键含 title（(city, year, COALESCE(title, ''))），同一专题标题只做空白 / 大小写 / 措辞
|
||
// 微调会被判为新实体（设计规格 §12.4）。本函数消除「纯空白」这一类可机械归一的差异，避免重复行。
|
||
// 措辞差异无法机械归一，仍按不同实体处理（符合预期）。调用方在落库与查重两侧都必须用同一函数，
|
||
// 才能保证索引（COALESCE(title, '')）与查重口径逐字一致。
|
||
func NormalizeStreetTitle(title string) string {
|
||
return strings.Join(strings.Fields(title), " ")
|
||
}
|
||
|
||
func (r *ingestRepository) CreateStreetSnapWithImages(ctx context.Context, snap *model.StreetSnap, imgs []model.StreetSnapImage) (uint32, error) {
|
||
now := uint32(time.Now().Unix())
|
||
snap.CreatedAt, snap.UpdatedAt = now, now
|
||
if snap.Status == "" {
|
||
snap.Status = model.StatusPending
|
||
}
|
||
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if cErr := tx.Create(snap).Error; cErr != nil {
|
||
return cErr
|
||
}
|
||
if len(imgs) == 0 {
|
||
return nil
|
||
}
|
||
for i := range imgs {
|
||
imgs[i].SnapID = snap.ID
|
||
imgs[i].CreatedAt, imgs[i].UpdatedAt = now, now
|
||
}
|
||
return tx.Create(&imgs).Error
|
||
})
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return snap.ID, nil
|
||
}
|
||
|
||
func (r *ingestRepository) ReuseRejectedStreetSnap(ctx context.Context, id uint32, snap *model.StreetSnap, imgs []model.StreetSnapImage) error {
|
||
now := uint32(time.Now().Unix())
|
||
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
// 同 runway：带 status=rejected 守卫 + 行数校验，避免把期间已发布的记录撤下公开视图。
|
||
upd := tx.Model(&model.StreetSnap{}).Where("id = ? AND status = ? AND is_deleted = 0", id, model.StatusRejected)
|
||
if uErr := upd.Updates(map[string]any{
|
||
"title": snap.Title,
|
||
"year": snap.Year,
|
||
"city": snap.City,
|
||
"cover": snap.Cover,
|
||
"image_count": snap.ImageCount,
|
||
"job_id": snap.JobID,
|
||
"status": model.StatusPending,
|
||
"reviewer": "",
|
||
"reject_reason": "",
|
||
"updated_at": now,
|
||
}).Error; uErr != nil {
|
||
return uErr
|
||
}
|
||
if upd.RowsAffected == 0 {
|
||
return fmt.Errorf("street snap %d 复用前置条件失效：该行已不是 rejected 状态或已被软删", id)
|
||
}
|
||
if dErr := tx.Model(&model.StreetSnapImage{}).
|
||
Where("snap_id = ? AND is_deleted = 0", id).
|
||
Updates(map[string]any{"is_deleted": 1, "updated_at": now}).Error; dErr != nil {
|
||
return dErr
|
||
}
|
||
if len(imgs) == 0 {
|
||
return nil
|
||
}
|
||
for i := range imgs {
|
||
imgs[i].SnapID = id
|
||
imgs[i].CreatedAt, imgs[i].UpdatedAt = now, now
|
||
}
|
||
return tx.Create(&imgs).Error
|
||
})
|
||
}
|
||
|
||
// ListJobs 按 id 倒序分页列出入库任务（后台监控页用）。offset/limit 控制分页区间。
|
||
func (r *ingestRepository) ListJobs(ctx context.Context, offset, limit int) ([]model.IngestJob, error) {
|
||
if limit < 1 {
|
||
limit = 50
|
||
}
|
||
if offset < 0 {
|
||
offset = 0
|
||
}
|
||
var jobs []model.IngestJob
|
||
if err := r.db.WithContext(ctx).Order("id DESC").Offset(offset).Limit(limit).Find(&jobs).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return jobs, nil
|
||
}
|
||
|
||
// CountJobs 返回 ingest_jobs 总条数（后台监控页分页用）。
|
||
func (r *ingestRepository) CountJobs(ctx context.Context) (int64, error) {
|
||
var n int64
|
||
if err := r.db.WithContext(ctx).Model(&model.IngestJob{}).Count(&n).Error; err != nil {
|
||
return 0, err
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// RetryJob 把一条 failed 任务重置回 pending，清掉 last_error / locked_at / next_attempt_at，
|
||
// 让 worker（每 3s 扫一次 pending）立即重新拉起处理。仅对 failed 生效，其它状态原样不动。
|
||
func (r *ingestRepository) RetryJob(ctx context.Context, id uint32) error {
|
||
now := uint32(time.Now().Unix())
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.IngestJob{}).
|
||
Where("id = ? AND status = ?", id, model.IngestStatusFailed).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusPending,
|
||
"locked_at": 0,
|
||
"last_error": "",
|
||
"next_attempt_at": 0,
|
||
"updated_at": now,
|
||
}).Error
|
||
}
|
||
|
||
// isDuplicateKey 兜底：gorm 的 ErrDuplicatedKey 在不同驱动下的封装不一定一致，
|
||
// 命中驱动返回的唯一约束冲突错误信息更稳（兼容 MySQL/PostgreSQL/SQLite 不同措辞）。
|
||
func isDuplicateKey(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
msg := err.Error()
|
||
return containsAny(msg, "Duplicate entry", "1062", "duplicate key", "violates unique constraint", "UNIQUE constraint failed")
|
||
}
|
||
|
||
func containsAny(s string, subs ...string) bool {
|
||
for _, sub := range subs {
|
||
if len(sub) > 0 && indexOf(s, sub) >= 0 {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func indexOf(s, sub string) int {
|
||
for i := 0; i+len(sub) <= len(s); i++ {
|
||
if s[i:i+len(sub)] == sub {
|
||
return i
|
||
}
|
||
}
|
||
return -1
|
||
}
|