577 lines
23 KiB
Go
577 lines
23 KiB
Go
package repository
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"math"
|
||
"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 标记任务成功。
|
||
MarkDone(ctx context.Context, id uint32) 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）查正式表既有行，返回其状态。
|
||
// worker 据此分三支：pending/published → 直接放弃本次入库；rejected → 复用该行重审；未命中 → 新建。
|
||
RunwayEntityState(ctx context.Context, brandID uint32, seasonCode, collectionType string) (id uint32, status string, found 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
|
||
// RepairRunwayDetailParents 入库后按 sort_order 回填走秀细节图的 parent_image_id（细节图挂到最近主图）。
|
||
RepairRunwayDetailParents(ctx context.Context, runwayID uint32) error
|
||
// StreetSnapEntityState 按实体键（city + year + title）查正式表既有行，返回其状态。
|
||
// title 用 COALESCE 与唯一索引 uq_ss_entity 的表达式保持一致。
|
||
StreetSnapEntityState(ctx context.Context, city string, year uint16, title string) (id uint32, status string, found 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
|
||
// FindNearDuplicateImage 在给定图片表中按 dHash 汉明距离检索近重复，返回命中行 id（仅取最近一条）。
|
||
FindNearDuplicateImage(ctx context.Context, tables []string, phashBits string, threshold int) (dupID uint32, found bool, err 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) error {
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.IngestJob{}).
|
||
Where("id = ?", id).
|
||
Updates(map[string]any{
|
||
"status": model.IngestStatusDone,
|
||
"last_error": "",
|
||
"updated_at": uint32(time.Now().Unix()),
|
||
}).Error
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
// ---- 入库直写正式表（单表发布模型）----
|
||
|
||
func (r *ingestRepository) RunwayEntityState(ctx context.Context, brandID uint32, seasonCode, collectionType string) (uint32, string, bool, error) {
|
||
var row struct {
|
||
ID uint32 `gorm:"column:id"`
|
||
Status string `gorm:"column:status"`
|
||
}
|
||
err := r.db.WithContext(ctx).
|
||
Model(&model.BrandRunway{}).
|
||
Select("id, status").
|
||
Where("brand_id = ? AND season_code = ? AND collection_type = ? AND is_deleted = 0", brandID, seasonCode, collectionType).
|
||
Limit(1).
|
||
Scan(&row).Error
|
||
if err != nil {
|
||
return 0, "", false, err
|
||
}
|
||
if row.ID == 0 {
|
||
return 0, "", false, nil
|
||
}
|
||
return row.ID, row.Status, true, 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
|
||
}
|
||
|
||
// StreetSnapEntityState 按实体键（city + year + title）查正式表既有行，返回其状态。
|
||
// title 用 COALESCE 与唯一索引 uq_ss_entity 的表达式保持一致：唯一索引里 NULL 互不冲突，
|
||
// 若这里写成裸 title，查重口径会比索引更严，导致重爬同城同年（title 为 NULL 的历史行）时误判未命中、
|
||
// 进而插入一行注定被索引拒绝的记录。
|
||
//
|
||
// 入参 title 会先经 NormalizeStreetTitle 归一化：与落库口径（processStreet 同样归一化）保持一致，
|
||
// 否则「空白 / 多空格」微调会被判为新实体，同城同年多出重复行（见设计规格 §12.4）。
|
||
func (r *ingestRepository) StreetSnapEntityState(ctx context.Context, city string, year uint16, title string) (uint32, string, bool, error) {
|
||
title = NormalizeStreetTitle(title)
|
||
var row struct {
|
||
ID uint32 `gorm:"column:id"`
|
||
Status string `gorm:"column:status"`
|
||
}
|
||
err := r.db.WithContext(ctx).
|
||
Model(&model.StreetSnap{}).
|
||
Select("id, status").
|
||
Where("city = ? AND year = ? AND COALESCE(title, '') = ? AND is_deleted = 0", city, year, title).
|
||
Limit(1).
|
||
Scan(&row).Error
|
||
if err != nil {
|
||
return 0, "", false, err
|
||
}
|
||
if row.ID == 0 {
|
||
return 0, "", false, nil
|
||
}
|
||
return row.ID, row.Status, true, 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
|
||
}
|
||
|
||
// FindNearDuplicateImage 在给定图片表中按 dHash 汉明距离检索近重复，取距离 ≤ threshold 的最近一条。
|
||
// 算子用 pgvector 的 L2（<->）；因 phash 是 {0,1}^64 向量，L2² == 汉明距离，故 L2 阈值 = sqrt(threshold)。
|
||
// phashBits 为 vector(64) 二进制向量串；NULL 的 phash 不参与比较。
|
||
//
|
||
// 只比对未软删的行（is_deleted = 0）：已被软删（审核剔除 / 复用重审）的旧图不应再作为
|
||
// 「重复」的参照。注意复用驳回行的时序 —— 去重发生在软删旧图**之前**，那一刻旧行仍是
|
||
// is_deleted = 0、会被正常命中；dup_of 只是人工留痕，不承担可见性语义。
|
||
// 本方法被 runway / street 两条入库路径共用，过滤对两者语义一致。
|
||
func (r *ingestRepository) FindNearDuplicateImage(ctx context.Context, tables []string, phashBits string, threshold int) (uint32, bool, error) {
|
||
// 汉明阈值转 L2 阈值：phash 为 {0,1}^64 向量，L2² == 汉明距离，故 L2 阈值 = sqrt(汉明阈值)。
|
||
l2Limit := math.Sqrt(float64(threshold))
|
||
for _, t := range tables {
|
||
var row struct {
|
||
ID uint32 `gorm:"column:id"`
|
||
Dist float64 `gorm:"column:dist"`
|
||
}
|
||
err := r.db.WithContext(ctx).Table(t).
|
||
Select("id, (phash <-> ?::vector) AS dist", phashBits).
|
||
Where("(phash <-> ?::vector) <= ?", phashBits, l2Limit).
|
||
Where("is_deleted = 0").
|
||
Order("dist ASC").
|
||
Limit(1).
|
||
Scan(&row).Error
|
||
if err != nil {
|
||
return 0, false, err
|
||
}
|
||
if row.ID != 0 {
|
||
return row.ID, true, nil
|
||
}
|
||
}
|
||
return 0, false, nil
|
||
}
|
||
|
||
// 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
|
||
}
|