package service

import (
	"context"
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"

	"fashionapi/internal/dto"
	"fashionapi/internal/model"
	"fashionapi/internal/pkg/hashid"
	"fashionapi/internal/pkg/season"
	"fashionapi/internal/pkg/storage"
	"fashionapi/internal/repository"
	"fashionapi/internal/vision"
)

// IngestService 爬虫入库管线的业务逻辑层：Submit 入队，worker 异步处理。
//
// 队列复用一张 ingest_jobs 表，靠 kind 列分流两类任务：
//   - crawl：爬虫上报的走秀/街拍入库（下载图 → 写草稿）
//   - media_cleanup：删除图集时异步清理S4孤儿图（引用计数归零才真删）
type IngestService struct {
	repo       repository.IngestRepository
	brandRepo  repository.BrandRepository
	media      repository.MediaRepository // 跨表图片引用计数（media_cleanup 删孤儿用）
	uploader   storage.Uploader           // 主上传器（S4启用时为S4，否则本地）
	del        storage.Deleter            // 删除器（S4或本地，media_cleanup 真删用）
	local      *storage.LocalUploader     // S4失败时的兜底落地
	httpClient *http.Client
	translate  *TranslateService // 标题中文翻译（可为 nil：测试/未启用时跳过自动补全）
	// 视觉服务与向量仓储（均可为 nil：未启用视觉服务时跳过 embed，不影响任何入库行为）。
	// 只服务于后台「疑似同款」提示，不参与去重。
	vision   *vision.Client
	embedRep repository.EmbeddingRepository
	// similar 「疑似同款」预计算服务（可为 nil：装配期未注入或视觉服务未启用时跳过）。
	// 写完向量后顺手把该图集的跨图集近邻算好落库，使后台页面只需读一次小查询。
	// 注入用 SetSimilar，避免改动构造签名（该服务在主装配期晚于本服务创建）。
	similar *SimilarService
}

// SetSimilar 注入「疑似同款」服务，供写向量后顺手预计算近邻。
// 装配期调用；未注入（nil）时完全跳过预计算，不影响入库。
func (s *IngestService) SetSimilar(sv *SimilarService) { s.similar = sv }

// NewIngestService 创建入库服务。
//
//	uploader: 主上传器（S4或本地）
//	del:      删除器（S4或本地），用于 media_cleanup 真删S4/本地孤儿文件
//	local:    本地兜底上传器（S4上传失败时回退，避免图片完全丢失）
func NewIngestService(repo repository.IngestRepository, brandRepo repository.BrandRepository, media repository.MediaRepository, uploader storage.Uploader, del storage.Deleter, local *storage.LocalUploader, downloadProxy string, translateSvc *TranslateService, vis *vision.Client, embedRep repository.EmbeddingRepository) *IngestService {
	// 下载代理：download_proxy（或 INGEST_DOWNLOAD_PROXY）非空时，下载客户端经该代理访问源站；
	// 仅作用于图片下载，S4 上传走独立 s3 客户端、pgx 数据库连接不走 http，均不受影响。
	// 留空回退 http.ProxyFromEnvironment（默认直连，仍可被系统 HTTP(S)_PROXY 影响）。
	dlTransport := &http.Transport{
		MaxIdleConns:        imageFetchConcurrency * 4,
		MaxIdleConnsPerHost: imageFetchConcurrency * 2,
		IdleConnTimeout:     90 * time.Second,
		TLSHandshakeTimeout: 10 * time.Second,
	}
	if downloadProxy != "" {
		if pu, perr := url.Parse(downloadProxy); perr == nil {
			dlTransport.Proxy = http.ProxyURL(pu)
			log.Printf("[ingest] 图片下载走代理: %s", downloadProxy)
		} else {
			log.Printf("[ingest] download_proxy 非法，已忽略并回退 ProxyFromEnvironment: %q", downloadProxy)
			dlTransport.Proxy = http.ProxyFromEnvironment
		}
	} else {
		dlTransport.Proxy = http.ProxyFromEnvironment
	}

	return &IngestService{
		repo:      repo,
		brandRepo: brandRepo,
		media:     media,
		uploader:  uploader,
		del:       del,
		local:     local,
		translate: translateSvc,
		vision:    vis,
		embedRep:  embedRep,
		// 图片下载是并发执行的（见 imageFetchConcurrency），连接池须匹配并发度：
		// Go 默认 MaxIdleConnsPerHost=2，多余连接会在每轮下载时反复重建、白付 TLS 握手开销。
		httpClient: &http.Client{
			// 超时须显著大于单张图的正常下载耗时（实测 4~22s，且随源站波动剧烈）。
			// 原 30s 余量太薄：一次抖动就会让「单图失败=整任务失败」把整批打回重试。
			Timeout:   60 * time.Second,
			Transport: dlTransport,
		},
	}
}

// Submit 把一条上报写入队列（payload 原样存 JSON），立即返回任务 id。
// 真正处理由 worker 异步完成，因此本方法本身很快、不阻塞爬虫。
func (s *IngestService) Submit(ctx context.Context, payload dto.RunwayIngest) (uint32, error) {
	if payload.Kind == "" {
		payload.Kind = dto.IngestKindRunway
	}
	raw, err := json.Marshal(payload)
	if err != nil {
		return 0, NewError(http.StatusBadRequest, "payload invalid: "+err.Error())
	}
	// runway 必须带品牌；street 不关联品牌，brand_uid 可选
	if payload.Kind == dto.IngestKindRunway && payload.BrandUID == "" {
		return 0, NewError(http.StatusBadRequest, "brand_uid required")
	}
	job := &model.IngestJob{Payload: string(raw), Kind: model.IngestKindCrawl}
	if err := s.repo.Enqueue(ctx, job); err != nil {
		return 0, internalErr(err.Error())
	}
	return job.ID, nil
}

// ListJobs 分页列出入库任务（后台监控页用）。返回本页任务、总条数与错误。
func (s *IngestService) ListJobs(ctx context.Context, offset, limit int) ([]model.IngestJob, int64, error) {
	jobs, err := s.repo.ListJobs(ctx, offset, limit)
	if err != nil {
		return nil, 0, err
	}
	total, err := s.repo.CountJobs(ctx)
	if err != nil {
		return nil, 0, err
	}
	return jobs, total, nil
}

// RetryJob 把一条 failed 任务重新入队，等待 worker 重新处理。
func (s *IngestService) RetryJob(ctx context.Context, id uint32) error {
	return s.repo.RetryJob(ctx, id)
}

// Run 启动 worker 循环：定时领取 pending 任务并处理，直到 ctx 取消。
// 多实例安全：Claim 用 SKIP LOCKED，互不抢同一条。
func (s *IngestService) Run(ctx context.Context, batch int, interval time.Duration) {
	if batch < 1 {
		batch = 2
	}
	if interval <= 0 {
		interval = 3 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.drain(ctx, batch)
		}
	}
}

// drain 一次性领取并处理一批任务。
func (s *IngestService) drain(ctx context.Context, batch int) {
	jobs, err := s.repo.Claim(ctx, batch)
	if err != nil {
		return
	}
	for i := range jobs {
		s.process(ctx, jobs[i])
	}
}

// process 处理单条任务：先按 job.Kind（DB 列）分流到 crawl / media_cleanup 两条管线。
// 旧爬虫任务（迁移前）kind 为空，回落 crawl 走秀/街拍入库管线。
func (s *IngestService) process(ctx context.Context, job model.IngestJob) {
	start := time.Now()
	switch job.Kind {
	case model.IngestKindMediaCleanup:
		s.processMediaCleanup(ctx, job)
		return
	case "", model.IngestKindCrawl:
		// 走秀/街拍入库管线，按 payload 内 Kind 再分流 runway / street。
	default:
		// 未知 job.Kind 绝不静默处理，标记失败避免脏数据。
		_ = s.repo.MarkFailed(ctx, job.ID, "unknown job kind: "+job.Kind)
		return
	}

	var p dto.RunwayIngest
	if err := json.Unmarshal([]byte(job.Payload), &p); err != nil {
		_ = s.repo.MarkFailed(ctx, job.ID, "payload parse: "+err.Error())
		return
	}
	if p.Kind == "" {
		p.Kind = dto.IngestKindRunway
	}
	log.Printf("[ingest] job=%d kind=%s 开始处理（图片=%d 张）", job.ID, p.Kind, len(p.Images))
	switch p.Kind {
	case dto.IngestKindRunway:
		s.processRunway(ctx, job, p)
	case dto.IngestKindStreet:
		s.processStreet(ctx, job, p)
	default:
		// 未知 kind 绝不静默当成走秀处理；所有爬虫数据都必须落到已注册的审核模块，
		// 否则标记任务失败，避免出现「未审核就入库」的脏数据。
		_ = s.repo.MarkFailed(ctx, job.ID, "unknown ingest kind: "+p.Kind)
		return
	}
	log.Printf("[ingest] job=%d kind=%s 处理结束 总耗时=%v", job.ID, p.Kind, time.Since(start).Round(time.Millisecond))
}

// MediaCleanupPayload 是「清理S4孤儿图」任务的 payload：待清理的图片 key 列表。
type MediaCleanupPayload struct {
	Keys []string `json:"keys"`
}

// EnqueueMediaCleanup 把一批待清理的S4 key 异步入队；真正删除由 worker 的
// processMediaCleanup 按引用计数判定，仅当 key 在所有图集/草稿表中引用归零才真删
// （内容寻址共享 key 不会被误删）。
func (s *IngestService) EnqueueMediaCleanup(ctx context.Context, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	raw, err := json.Marshal(MediaCleanupPayload{Keys: keys})
	if err != nil {
		return err
	}
	return s.repo.EnqueueMediaCleanup(ctx, string(raw))
}

// processMediaCleanup 清理S4孤儿图任务：解析 key 列表，按跨表引用计数删除真孤儿。
// 删除/计数失败一律保守跳过（宁可留文件），因此本任务几乎总是成功置 done。
func (s *IngestService) processMediaCleanup(ctx context.Context, job model.IngestJob) {
	if s.media == nil || s.del == nil {
		_ = s.repo.MarkFailed(ctx, job.ID, "media cleanup not configured")
		return
	}
	var p MediaCleanupPayload
	if err := json.Unmarshal([]byte(job.Payload), &p); err != nil {
		_ = s.repo.MarkFailed(ctx, job.ID, "media cleanup payload parse: "+err.Error())
		return
	}
	if len(p.Keys) == 0 {
		_ = s.repo.MarkDone(ctx, job.ID)
		return
	}
	purgeOrphanImages(ctx, s.media, s.del, p.Keys)
	_ = s.repo.MarkDone(ctx, job.ID)
}

// failOrRetry 把临时失败（如网络抖动 / 单图下载失败）按指数退避自动重试：
// attempts 未达上限则重置为 pending 并推迟 next_attempt_at，worker 到点再领；
// 达上限才置 failed，等待后台手动「重试」。永久失败（payload 解析错 / 未知 kind）直接
// MarkFailed，不进重试。
func (s *IngestService) failOrRetry(ctx context.Context, id uint32, errMsg string) {
	if err := s.repo.ScheduleRetry(ctx, id, errMsg); err != nil {
		// 调度失败兜底硬失败，避免任务卡在 processing 无人收。
		_ = s.repo.MarkFailed(ctx, id, errMsg)
	}
}

// processRunway 走秀入库：品牌校验 → 实体键分三支（新建 / 放弃 / 复用驳回）→ 下载图 → 直写 brand_runways。
func (s *IngestService) processRunway(ctx context.Context, job model.IngestJob, p dto.RunwayIngest) {
	// 1) 品牌必须存在（爬虫负责先建/复用品牌）
	brandID, err := hashid.Decode(p.BrandUID)
	if err != nil {
		_ = s.repo.MarkFailed(ctx, job.ID, "unknown brand_uid")
		return
	}
	if _, err := s.brandRepo.FindByID(ctx, brandID); err != nil {
		_ = s.repo.MarkFailed(ctx, job.ID, "brand not found")
		return
	}

	// 2) 实体键查重并决定分支（单表模型）
	// 标题缺年份（p.Year=0）时，用发布日期的年份补全年份，使 year / season_code 不再为空，
	// 避免这些秀在「按年份/季节筛选」与热门排序（按 year + season_code）中丢失。
	effYear := p.Year
	if effYear == 0 && p.PublishedAt != 0 {
		effYear = uint16(time.Unix(int64(p.PublishedAt), 0).UTC().Year())
	}
	seasonCode := season.Derive(effYear, p.CollectionType, p.Season)
	reuseID, existStatus, exist, isDeleted, err := s.repo.RunwayEntityState(ctx, brandID, seasonCode, p.CollectionType)
	if err != nil {
		// 读失败绝不能当成「未命中」：误判为新建会走一次注定冲突的写入——实体键唯一索引
		//（2026-09-22-03）会把它拒成一次可重试的写库错误，重试时实体键已能查到该行（pending）
		// 而直接跳过。但唯一索引是最后一道防线，不该靠它来替代这里的前置判断，
		// 故仍 fail-closed + 退避重试，不做无谓的写尝试。
		s.failOrRetry(ctx, job.ID, "entity lookup: "+err.Error())
		return
	}
	// 删除即永久：已软删的实体键不再因重爬复活（原设计靠部分唯一索引释放键、允许重爬，已改为禁止）。
	if exist && isDeleted {
		log.Printf("[ingest] job=%d runway 实体已删除(is_deleted=1)，跳过重爬", job.ID)
		_ = s.repo.MarkDone(ctx, job.ID)
		return
	}
	if !exist && p.Year == 0 {
		// 标题无年份的历史行（year=0、season_code 为空）待回填：按旧键再查一次，
		// 命中后走跳过分支按主键 id 回填（回填后season_code变为effYear码，后续重跑即可直接命中，幂等）。
		var rid uint32
		rid, existStatus, exist, isDeleted, err = s.repo.RunwayEntityState(ctx, brandID, "", p.CollectionType)
		if err != nil {
			s.failOrRetry(ctx, job.ID, "entity lookup: "+err.Error())
			return
		}
		reuseID = rid
		if exist && isDeleted {
			log.Printf("[ingest] job=%d runway 实体已删除(is_deleted=1)，跳过重爬", job.ID)
			_ = s.repo.MarkDone(ctx, job.ID)
			return
		}
	}
	action := decideIngestAction(exist, existStatus)
	if action == ingestSkip {
		log.Printf("[ingest] job=%d runway 实体已存在(status=%s)，跳过（回填 published_at）", job.ID, existStatus)
		// 临时回填：仅更新发布时间与缺失年份，不重下图片（图片已在库）。
		if p.PublishedAt != 0 {
			if err := s.repo.BackfillRunwayPublishedAt(ctx, reuseID, p.PublishedAt, effYear, seasonCode); err != nil {
				s.failOrRetry(ctx, job.ID, "backfill published_at: "+err.Error())
				return
			}
		}
		_ = s.repo.MarkDone(ctx, job.ID)
		return
	}

	// 3) season_code 已在去重前补齐（vogue.go 历史漏填的 bug，统一在此兜底）

	// 4) 下载图片并上传到存储（结构化 Looks 优先：主图+细节图分组；否则回退 Images 全部视为主图）。
	// 内容哈希（sha1）key 保证重爬不产生孤儿文件：失败回滚删本批 key 即可。
	var cover string
	var images []model.BrandRunwayImage
	// vecs 与 images 下标一一对应（nil = 该张未产出向量）；落库拿到图片 id 后写入 image_embeddings。
	var vecs [][]float32
	var keys []string
	var imgFailed bool
	var imageCount uint16
	var timing *fetchTiming
	if len(p.Looks) > 0 {
		cover, images, vecs, keys, imgFailed, timing = s.fetchLookImages(ctx, p.Looks, "runway")
		// image_count 必须等于**实际落库的主图行数**，不能是 len(p.Looks)：
		// fetchLookImages 会跳过 look.Main == "" 的 look 与 sha1 重复的行，实际主图行可能更少。
		// 旧流程靠晋升时按实际行重算自愈，该路径已随单表发布模型删除 —— 一旦用 len(Looks)，
		// 漂移会永久留在公开卡片的「N 张」角标与 FeaturedIDs 的 SUM(image_count) 热度上。
		imageCount = countMainImages(images)
	} else {
		var fimgs []fetchedImage
		cover, fimgs, keys, imgFailed, timing = s.fetchImages(ctx, p.Images, "runway")
		imageCount = uint16(len(fimgs))
		for i, fi := range fimgs {
			images = append(images, model.BrandRunwayImage{
				Image:     fi.url,
				Name:      fmt.Sprintf("Look %d", i+1),
				SortOrder: uint32(i + 1),
				IsDetail:  0,
				ImageSize: fi.size,
			})
			vecs = append(vecs, fi.vec)
		}
	}
	log.Printf("[ingest] job=%d runway 拉图 总耗时=%v %s", job.ID, timing.total.Round(time.Millisecond), timing)
	if imgFailed {
		// 单图失败=整任务失败：先回滚本批已上传的图（S4 + 本地兜底），避免孤儿文件永远堆在存储里，
		// 然后按指数退避自动重试，达上限才置 failed 等后台手动重试。
		s.cleanupUploads(ctx, keys)
		s.failOrRetry(ctx, job.ID, "image download failed")
		return
	}

	// 5) 直写正式表（status=pending），等待后台审核通过（审核只改状态，不重建图片）
	writeStart := time.Now()
	rw := &model.BrandRunway{
		JobID:          job.ID,
		BrandID:        brandID,
		TitleEn:        p.TitleEn,
		TitleCn:        p.TitleCn,
		DescriptionEn:  p.DescriptionEn,
		DescriptionCn:  p.DescriptionCn,
		Year:           effYear,
		Season:         p.Season,
		CollectionType: p.CollectionType,
		SeasonCode:     seasonCode,
		PublishedAt:    p.PublishedAt,
		Cover:          cover,
		ImageCount:     imageCount,
		Status:         model.StatusPending,
	}

	// 入库自动补全中文标题：spider 未给 title_cn 时，用翻译服务按「记忆优先、规则兜底」推导；
	// 已给则保留（绝不覆盖爬虫提供/已有的翻译）。翻译服务未注入（nil）时跳过。
	if rw.TitleCn == "" && s.translate != nil {
		if cn, terr := s.translate.SuggestTitleCn(ctx, rw.TitleEn); terr == nil {
			rw.TitleCn = cn
		}
	}

	for i := range images {
		images[i].BrandID = brandID
	}
	var saveErr error
	if action == ingestReuseRejected {
		saveErr = s.repo.ReuseRejectedRunway(ctx, reuseID, rw, images)
	} else {
		_, saveErr = s.repo.CreateRunwayWithImages(ctx, rw, images)
	}
	if saveErr != nil {
		s.failOrRetry(ctx, job.ID, "save runway: "+saveErr.Error())
		return
	}
	// 落库成功后写入语义向量（仅后台「疑似同款」用）。GORM 批量 Create 会把自增主键回填到
	// images，因此此处 images[i].ID 即新建图片 id；写失败只记日志，绝不影响入库结果。
	embedIDs := make([]uint32, 0, len(images))
	for i := range images {
		embedIDs = append(embedIDs, images[i].ID)
	}
	s.saveEmbeddings(ctx, dto.IngestKindRunway, embedIDs, vecs)
	// 顺手把本场的「疑似同款」近邻算好落库（失败不影响入库）。
	s.precomputeSimilar(ctx, dto.IngestKindRunway, embedIDs)
	// 入库后按 sort_order 回填细节图的 parent_image_id，使新走秀自动成组（与街拍一致）。
	runwayID := rw.ID
	if runwayID == 0 {
		runwayID = reuseID
	}
	if rerr := s.repo.RepairRunwayDetailParents(ctx, runwayID); rerr != nil {
		log.Printf("[ingest] job=%d runway 回填 parent 失败: %v", job.ID, rerr)
	}
	log.Printf("[ingest] job=%d runway 写正式表 耗时=%v 图片行=%d 复用驳回=%v",
		job.ID, time.Since(writeStart).Round(time.Millisecond), len(images), action == ingestReuseRejected)
	_ = s.repo.MarkDone(ctx, job.ID)
}

// fetchLookImages 按 Looks 结构下载主图+细节图并上传到存储，返回可直接落正式图片表的
// BrandRunwayImage 行（带 parent_image_id / is_detail 分组）。任意一张下载/上传失败即把
// failed 置 true，调用方据此把整条任务判失败并回滚本批已上传的 key，符合「单图失败=整任务失败」策略。
// cover 取首个成功下载的主图；image_count 由调用方按返回行中 is_detail == 0 的数量计
//（countMainImages），而不是 len(Looks)：跳过空主图 / sha1 重复后实际行数可能更少。
//
// 实现：先把全部「主图 + 细节图」按原始顺序摊平成任务列表，用 concurrentFetch 并发完成
// 「下载 → 上传」这段 IO 密集操作；随后再按下标顺序串行做批次内去重与组装。
// 这样既拿到并发收益，又保证 cover / sort_order / 批次内去重与原串行实现一致。
func (s *IngestService) fetchLookImages(ctx context.Context, looks []dto.RunwayLook, prefix string) (string, []model.BrandRunwayImage, [][]float32, []string, bool, *fetchTiming) {
	start := time.Now()
	timing := &fetchTiming{}
	cover := ""
	rows := make([]model.BrandRunwayImage, 0)
	// vecs 与 rows 下标一一对应（nil 表示该张未产出向量），供落库后写入 image_embeddings。
	vecs := make([][]float32, 0)
	keys := make([]string, 0)
	seen := make(map[string]bool)
	failed := false

	// 摊平：顺序 = 逐个 look 先主图、再其细节图，与原串行遍历顺序完全一致。
	tasks := make([]lookImageTask, 0, len(looks))
	for li, look := range looks {
		lookIdx := li + 1
		if look.Main != "" {
			tasks = append(tasks, lookImageTask{url: look.Main, lookIdx: lookIdx})
		}
		for di, d := range look.Details {
			if d == "" {
				continue
			}
			tasks = append(tasks, lookImageTask{url: d, lookIdx: lookIdx, isDetail: true, detailNo: di + 1})
		}
	}

	// 并发段：下载 + 上传（耗时大头）。
	results := concurrentFetch(ctx, len(tasks), func(c context.Context, i int) downloadResult {
		return s.downloadAndUpload(c, tasks[i].url, prefix)
	})

	// 串行段：按原始顺序累计耗时、做批次内去重、组装行。
	order := 0
	for i, r := range results {
		task := tasks[i]
		timing.download += r.download
		timing.upload += r.upload
		timing.embed += r.embed
		if r.err != nil {
			failed = true
			continue
		}
		keys = append(keys, r.key)
		if seen[r.sha1] {
			continue
		}
		seen[r.sha1] = true
		order++
		if cover == "" && !task.isDetail {
			cover = r.url
		}
		name := fmt.Sprintf("Look %d", task.lookIdx)
		isDetail := uint8(0)
		if task.isDetail {
			name = fmt.Sprintf("Look %d — Detail %d", task.lookIdx, task.detailNo)
			isDetail = 1
		}
		rows = append(rows, model.BrandRunwayImage{
			Image:     r.url,
			Name:      name,
			SortOrder: uint32(order),
			IsDetail:  isDetail,
			ImageSize: r.size,
		})
		vecs = append(vecs, r.vec)
	}
	timing.images = len(rows)
	timing.total = time.Since(start)
	return cover, rows, vecs, keys, failed, timing
}

// countMainImages 数一批待落库图片行中的主图（is_detail = 0）数量，即 image_count 的口径。
// 与 SoftDeleteRunwayImage 审核期重算的口径一致（只计主图），保证入库值与后续重算值同源。
func countMainImages(images []model.BrandRunwayImage) uint16 {
	var n uint16
	for i := range images {
		if images[i].IsDetail == 0 {
			n++
		}
	}
	return n
}

// processStreet 街拍入库：实体键分三支（新建 / 放弃 / 复用驳回）→ 下载图 → 直写 street_snaps（无品牌）。
func (s *IngestService) processStreet(ctx context.Context, job model.IngestJob, p dto.RunwayIngest) {
	// 1) 实体键查重并决定分支（单表模型，同 runway）。
	// 实体键含 title：与下方 snap.Title 用的是同一个归一化后的 title，保证查重口径与落库内容一致，
	// 同城同年的不同专题（Day 2 / Day 3）因此不会被误判为重复。
	// 标题先经 NormalizeStreetTitle 归一化（空白折叠），消除「纯空白微调」造成的重复实体（设计规格 §12.4）。
	title := repository.NormalizeStreetTitle(p.TitleEn)
	reuseID, existStatus, exist, isDeleted, err := s.repo.StreetSnapEntityState(ctx, p.City, p.Year, title)
	if err != nil {
		// 同 runway：读失败不得当「未命中」，否则会给同一实体留两行。fail-closed + 退避重试。
		s.failOrRetry(ctx, job.ID, "entity lookup: "+err.Error())
		return
	}
	// 删除即永久：已软删的实体键不再因重爬复活（与 runway 同源）。
	if exist && isDeleted {
		log.Printf("[ingest] job=%d street 实体已删除(is_deleted=1)，跳过重爬", job.ID)
		_ = s.repo.MarkDone(ctx, job.ID)
		return
	}
	action := decideIngestAction(exist, existStatus)
	if action == ingestSkip {
		log.Printf("[ingest] job=%d street 实体已存在(status=%s)，跳过", job.ID, existStatus)
		_ = s.repo.MarkDone(ctx, job.ID)
		return
	}

	// 2) 下载图片并上传到存储（S4优先，失败兜底本地）。
	cover, fimgs, keys, imgFailed, timing := s.fetchImages(ctx, p.Images, "street")
	log.Printf("[ingest] job=%d street 拉图 总耗时=%v %s", job.ID, timing.total.Round(time.Millisecond), timing)
	if len(p.Images) > 0 && imgFailed {
		// 单图失败=整任务失败：先回滚本批已上传的图（S4 + 本地兜底），避免孤儿文件永远堆在存储里，
		// 然后按指数退避自动重试，达上限才置 failed 等后台手动重试。
		s.cleanupUploads(ctx, keys)
		s.failOrRetry(ctx, job.ID, "image download failed")
		return
	}

	// 3) 直写正式表（status=pending）
	writeStart := time.Now()

	// 街拍入库自动补全中文标题（同 runway：spider 未给才补，且翻译服务需注入）。
	titleCn := p.TitleCn
	if titleCn == "" && s.translate != nil {
		if cn, terr := s.translate.SuggestTitleCn(ctx, title); terr == nil {
			titleCn = cn
		}
	}

	snap := &model.StreetSnap{
		JobID:      job.ID,
		Title:      title, // 街拍单标题，爬虫优先填 title_en（已归一化，与查重口径一致）
		TitleCn:    titleCn,
		Year:       p.Year,
		City:       p.City,
		Cover:      cover,
		ImageCount: uint16(len(fimgs)),
		Status:     model.StatusPending,
	}
	rows := make([]model.StreetSnapImage, 0, len(fimgs))
	// vecs 与 rows 下标一一对应（nil = 该张未产出向量）；落库拿到图片 id 后写入 image_embeddings。
	vecs := make([][]float32, 0, len(fimgs))
	for i, fi := range fimgs {
		rows = append(rows, model.StreetSnapImage{
			Image:     fi.url,
			Name:      fmt.Sprintf("Look %d", i+1),
			SortOrder: uint32(i + 1),
			ImageSize: fi.size,
		})
		vecs = append(vecs, fi.vec)
	}
	var saveErr error
	if action == ingestReuseRejected {
		saveErr = s.repo.ReuseRejectedStreetSnap(ctx, reuseID, snap, rows)
	} else {
		_, saveErr = s.repo.CreateStreetSnapWithImages(ctx, snap, rows)
	}
	if saveErr != nil {
		s.failOrRetry(ctx, job.ID, "save street snap: "+saveErr.Error())
		return
	}
	// 落库成功后写入语义向量（同走秀：GORM 批量 Create 回填自增主键，故 rows[i].ID 即新建图片 id）。
	embedIDs := make([]uint32, 0, len(rows))
	for i := range rows {
		embedIDs = append(embedIDs, rows[i].ID)
	}
	s.saveEmbeddings(ctx, dto.IngestKindStreet, embedIDs, vecs)
	// 顺手把本专辑的「疑似同款」近邻算好落库（失败不影响入库）。
	s.precomputeSimilar(ctx, dto.IngestKindStreet, embedIDs)
	log.Printf("[ingest] job=%d street 写正式表 耗时=%v 图片行=%d 复用驳回=%v",
		job.ID, time.Since(writeStart).Round(time.Millisecond), len(rows), action == ingestReuseRejected)
	_ = s.repo.MarkDone(ctx, job.ID)
}

// fetchImages 下载图片并上传到存储，返回 (cover 地址, 已下载图结构, 已成功上传对象的 key 列表, 是否有任意一张失败)。
// prefix 为S4 key 前缀（runway/ 或 street/）。
// 只要任意一张下载/上传失败，failed 即置 true，调用方据此把整条任务判为失败并回滚本批已上传的对象，
// 符合「单图失败=整任务失败」策略。
//
// 实现：先用 concurrentFetch 并发完成全部图片的「下载 → 上传」，再按下标顺序串行
// 做批次内去重与组装，保证 cover / 顺序 / 批次内去重与原串行实现一致。
func (s *IngestService) fetchImages(ctx context.Context, urls []string, prefix string) (string, []fetchedImage, []string, bool, *fetchTiming) {
	start := time.Now()
	timing := &fetchTiming{}
	cover := ""
	out := make([]fetchedImage, 0, len(urls))
	keys := make([]string, 0, len(urls))
	seen := make(map[string]bool)
	failed := false

	// 并发段：下载 + 上传（耗时大头）。
	results := concurrentFetch(ctx, len(urls), func(c context.Context, i int) downloadResult {
		return s.downloadAndUpload(c, urls[i], prefix)
	})

	// 串行段：按原始顺序累计耗时、做批次内去重、组装结果。
	for _, r := range results {
		timing.download += r.download
		timing.upload += r.upload
		timing.embed += r.embed
		if r.err != nil {
			failed = true
			continue
		}
		keys = append(keys, r.key)
		if seen[r.sha1] {
			continue
		}
		seen[r.sha1] = true
		if cover == "" {
			cover = r.url
		}
		out = append(out, fetchedImage{
			url:  r.url,
			key:  r.key,
			size: r.size,
			vec:  r.vec,
		})
	}
	timing.images = len(out)
	timing.total = time.Since(start)
	return cover, out, keys, failed, timing
}

// precomputeSimilar 写完向量后顺手预计算「疑似同款」近邻并落库。
//
// 为什么放在这里：原先是在**打开图集页时**对每张图实时做向量近邻检索，一个 116 张图的
// 图集要发 116 次查询、每次约 20KB SQL；本机 PostgreSQL 跑在 WSL2 里、Windows 经 wslrelay
// 访问，该链路大 payload 吞吐实测约 0.47MB/s，合计 5 秒。挪到入库时算一次（每个图集只付一次），
// 页面就退化成一次小查询。
//
// 失败只记日志：预计算是增值数据，绝不能让「算不出同款」影响入库结果。
// 注意：这里统计的是本次新入库的图片；它们与**其他图集存量图**的配对由预计算内部双向写入，
// 因此对方图集的页面也能立刻看到提示，无需重算对方。
func (s *IngestService) precomputeSimilar(ctx context.Context, kind string, imageIDs []uint32) {
	if s.similar == nil || len(imageIDs) == 0 {
		return
	}
	start := time.Now()
	if err := s.similar.PrecomputeForGallery(ctx, kind, imageIDs); err != nil {
		log.Printf("[ingest] 「疑似同款」预计算失败 kind=%s 图=%d err=%v（不影响入库）", kind, len(imageIDs), err)
		return
	}
	log.Printf("[ingest] 「疑似同款」预计算耗时 kind=%s 图=%d 耗时=%v", kind, len(imageIDs), time.Since(start).Round(time.Millisecond))
}

// saveEmbeddings 落库成功后写入一批图片的语义向量（仅后台「疑似同款」提示用）。
//
// 向量是增值数据、不是入库的必要条件：仓储未注入（未启用视觉服务）、无向量、
// 或写入失败，都只记日志并返回，绝不影响入库结果。ids[i] 与 vecs[i] 一一对应。
func (s *IngestService) saveEmbeddings(ctx context.Context, kind string, ids []uint32, vecs [][]float32) {
	if s.embedRep == nil || len(ids) == 0 || len(vecs) == 0 {
		return
	}
	if err := s.embedRep.BulkUpsertEmbedding(ctx, kind, ids, vecs); err != nil {
		log.Printf("[ingest] 写向量失败 kind=%s 数量=%d err=%v（不影响入库）", kind, len(ids), err)
	}
}

// cleanupUploads 删除一批本批次成功上传的对象（S4 + 本地兜底），用于任务失败回滚：
// 「单图失败=整任务失败」时，前面已成功上传的图若放任不管就成了孤儿，永远堆在存储里。
// 对两种存储都尝试删除（任一不存在即按幂等成功处理）；删除失败仅告警，不阻断任务置失败。
// 注意：key 为 sha1 内容寻址，若恰与其他已晋升图集共享同一内容哈希会被一并移除，重试会重新上传补齐。
func (s *IngestService) cleanupUploads(ctx context.Context, keys []string) {
	for _, k := range keys {
		if d, ok := s.uploader.(storage.Deleter); ok {
			if err := d.Delete(ctx, k); err != nil {
				log.Printf("[warn] ingest cleanup: S4删除失败 key=%s err=%v", k, err)
			}
		}
		if s.local != nil {
			if err := s.local.Delete(ctx, k); err != nil {
				log.Printf("[warn] ingest cleanup: 本地删除失败 key=%s err=%v", k, err)
			}
		}
	}
}

// fetchedImage 是单张已下载图片的暂存结构。
type fetchedImage struct {
	url  string
	key  string
	size int64     // 原图源文件字节数，随结果带回供落库写入 image_size
	vec  []float32 // FashionSigLIP 向量（nil = 未产出），供落库后写入 image_embeddings
}

// slowImageThreshold 单张图各阶段合计超过该阈值时单独打一条告警，便于从大量图里定位异常慢图。
const slowImageThreshold = 2 * time.Second

// imageFetchConcurrency 单条入库任务内「下载 + 上传」的并发度。
//
// 默认 1（等效串行），可用环境变量 INGEST_IMAGE_CONCURRENCY（1~16）覆盖。
//
// 之所以默认 1：早期实测 theimpression.com（59 张图）显示源站对同 IP 并发有强限流——
// 并发 1 时单张 ~3.7s / 整批 3m54s；并发 5 时单张涨到 ~23s，5m24s 只完成 12/59 张
// 就因超时失败，吞吐从 ~0.27 张/秒掉到 ~0.04 张/秒。结论：并发会把任务拖垮，故取 1。
//
// 但该结论**只对 theimpression 成立，不能外推到所有源站**。2026-10-13 对 assets.vogue.com
// 的冷图（未命中 CDN 边缘）实测：串行 1582ms/张，8 并发降到 661ms/张（2.4x），并发是有收益的。
// 冷图慢是因为 CDN 要回源，属延迟受限而非带宽受限——正好是并发能掩盖的那一类。
//
// 因此把它做成可配置项：换源站时先小批量实测再调（本文件顶部注释记录了两组实测数据）。
var imageFetchConcurrency = func() int {
	v := strings.TrimSpace(os.Getenv("INGEST_IMAGE_CONCURRENCY"))
	if v == "" {
		return 1
	}
	n, err := strconv.Atoi(v)
	if err != nil || n < 1 || n > 16 {
		log.Printf("[ingest] INGEST_IMAGE_CONCURRENCY=%q 非法（应为 1~16 的整数），回退并发 1", v)
		return 1
	}
	return n
}()

// fetchTiming 汇总一次图集拉取各阶段的累计耗时，用于定位「慢在哪一步」。
// 下载与上传是网络 IO，embed 是视觉服务调用（HTTP + 模型推理），分开计时才能对症下药。
type fetchTiming struct {
	images   int           // 成功入行的图片数
	total    time.Duration // 整个拉取过程的墙钟耗时
	download time.Duration // 各张 HTTP 下载耗时之和
	upload   time.Duration // 各张上传存储耗时之和
	embed    time.Duration // 各张视觉服务编码耗时之和
}

// String 输出各阶段耗时（毫秒精度），一条日志看清瓶颈。
// 注意：下载 / 上传 / embed 是「各张图耗时之和」，并发执行下其总和会大于墙钟总耗时，
// 因此三者与「总耗时」的比值不再等于时间占比——它们用于横向对比哪个阶段最重。
func (t *fetchTiming) String() string {
	if t == nil {
		return ""
	}
	return fmt.Sprintf("图片=%d 并发=%d 下载=%v 上传=%v embed=%v（各张累计值）",
		t.images,
		imageFetchConcurrency,
		t.download.Round(time.Millisecond),
		t.upload.Round(time.Millisecond),
		t.embed.Round(time.Millisecond),
	)
}

// downloadResult 是单张图「下载 → 上传」的结果，供并发执行后按下标串行组装。
// 各阶段耗时随结果一并带回、由调用方在串行段汇总（而非直接累加到共享 timing），
// 这样并发路径上无需加锁即可保证计时准确。
type downloadResult struct {
	url      string        // 存储访问地址（失败为空）
	key      string        // 对象 key（内容寻址）
	sha1     string        // 内容 sha1，用于批次内去重
	download time.Duration // 本张下载耗时
	upload   time.Duration // 本张上传耗时
	size     int64         // 原图源文件字节数（io.ReadAll 读到的下载响应体长度）
	vec      []float32     // FashionSigLIP 向量（nil = 未产出：服务未启用/不可用/失败）
	embed    time.Duration // 本张向量编码耗时
	err      error         // 下载或上传失败原因（非空即该张失败）
}

// concurrentFetch 以 imageFetchConcurrency 路并发执行 fn(ctx, i)（i ∈ [0,n)），返回长度 n、
// 下标与入参一一对齐的结果切片。
//
// 下标对齐是刻意的：调用方随后按原始顺序串行组装，使 cover / sort_order / 批次内去重结果
// 与串行实现完全一致，只把「下载+上传」这段 IO 并行化。
// 每个下标只由一个 goroutine 写入，故无需加锁。
func concurrentFetch(ctx context.Context, n int, fn func(context.Context, int) downloadResult) []downloadResult {
	results := make([]downloadResult, n)
	if n == 0 {
		return results
	}
	sem := make(chan struct{}, imageFetchConcurrency)
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			sem <- struct{}{}        // 占坑：超过并发度即在此排队
			defer func() { <-sem }() // 释放坑位
			results[i] = fn(ctx, i)
		}(i)
	}
	wg.Wait()
	return results
}

// lookImageTask 是 fetchLookImages 摊平后的一张待下载任务：记录它属于哪个 look、
// 是主图还是第几张细节图，供并发下载完成后重建 Name / 主副图归属 / IsDetail。
type lookImageTask struct {
	url      string
	lookIdx  int  // 1-based look 序号
	isDetail bool // true=细节图
	detailNo int  // 细节图序号（1-based），主图为 0
}

// downloadAndUpload 把单张远程图下载后上传到存储（S4优先，失败兜底本地）。
// 内容 sha1 用于生成内容寻址的存储 key（重爬幂等）与批次内去重；访问地址可直接写入数据库
// （S4为完整 https URL，本地为相对 /uploads 路径）。
//
// 本函数是并发调用点：每次调用各自返回一份独立的 downloadResult，不共享可变状态；
// 耗时汇总与去重留痕由调用方在串行段统一完成。
func (s *IngestService) downloadAndUpload(ctx context.Context, u, prefix string) downloadResult {
	res := downloadResult{}

	// ① 下载
	dlStart := time.Now()
	resp, err := s.httpClient.Get(u)
	if err != nil {
		res.err = err
		return res
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		res.err = fmt.Errorf("status %d", resp.StatusCode)
		return res
	}
	data, err := io.ReadAll(resp.Body)
	if err != nil {
		res.err = err
		return res
	}
	res.download = time.Since(dlStart)
	res.size = int64(len(data))

	ext := path.Ext(u)
	if ext == "" || len(ext) > 5 {
		ext = ".jpg"
	}
	// 内容哈希作 key：同一张图（无论来自哪篇文章/job）永远得到相同 key，
	// S4覆盖写即天然幂等，不会因重复采集、崩溃重试、reject 重爬而累积孤儿文件。
	h := sha1.Sum(data)
	res.sha1 = hex.EncodeToString(h[:])
	res.key = fmt.Sprintf("%s/%s%s", prefix, res.sha1, ext)
	ct := resp.Header.Get("Content-Type")

	// ② 视觉服务编码（可选）：产出后台「疑似同款」用的语义向量。
	// 未启用 / 不可用 / 超时 / 维度不符一律留空 vec —— 绝不阻断入库，也不在此重试
	// （重试会拖慢整批；熔断由客户端统一处理）。
	if s.vision != nil && s.vision.Enabled() {
		embedStart := time.Now()
		vecs, _, eerr := s.vision.Embed(ctx, [][]byte{data})
		res.embed = time.Since(embedStart)
		if eerr == nil && len(vecs) > 0 {
			res.vec = vecs[0]
		} else if eerr != nil {
			log.Printf("[ingest] embed 失败 url=%s err=%v（不写向量，继续入库）", u, eerr)
		}
	}

	// ③ 上传（主存储 S4，失败兜底本地）
	upStart := time.Now()
	if url, uerr := s.uploader.Upload(ctx, res.key, data, ct); uerr == nil {
		res.url = url
	} else if s.local != nil {
		// 兜底本地，避免图片完全丢失
		if lurl, lerr := s.local.Upload(ctx, res.key, data, ct); lerr == nil {
			res.url = lurl
		} else {
			res.err = uerr
		}
	} else {
		res.err = uerr
	}
	res.upload = time.Since(upStart)

	// 单图异常慢（合计超阈值）单独告警：便于从几十上百张图里一眼锁定是哪张、卡在哪一段。
	if total := res.download + res.embed + res.upload; total > slowImageThreshold {
		log.Printf("[ingest] 慢图 url=%s 合计=%v 下载=%v embed=%v 上传=%v",
			u, total.Round(time.Millisecond),
			res.download.Round(time.Millisecond), res.embed.Round(time.Millisecond),
			res.upload.Round(time.Millisecond))
	}
	return res
}

// ingestActionKind 入库分支决策结果（单表发布模型）。
type ingestActionKind int

const (
	ingestCreate        ingestActionKind = iota // 实体键未命中 → 新建记录
	ingestSkip                                  // 命中 pending/published → 放弃本次入库
	ingestReuseRejected                         // 命中 rejected → 复用该行重审
)

// decideIngestAction 按实体键命中情况与既有行状态决定入库分支。
// 抽成纯函数，是为了让「驳回不是永久黑名单」这条规则有独立的、不需要数据库的测试。
func decideIngestAction(found bool, status string) ingestActionKind {
	switch {
	case !found:
		return ingestCreate
	case status == model.StatusRejected:
		return ingestReuseRejected
	default:
		return ingestSkip
	}
}