Files
backend_v2/internal/pkg/storage/s4.go
toom1996 151e09ab30 feat: 抓取任务队列 + 媒体机直传 + 向量回填
crawl(品牌抓取任务队列)
- /admin/internal/crawl/tasks 领取 / :id/result 回报,spider 可常驻拉任务
- 后台品牌列表支持「加入抓取队列」,队列页可看状态与重试
- 租约超时的 running 自动回收为 queued,避免 worker 崩溃后任务卡死

media(媒体机直传)
- ingest 载荷新增 main_key/detail_keys/main_size/detail_sizes:
  有 key 就直接落库,跳过下载与上传;与 URL 形态可逐图混用
- 细节图按 Details/DetailKeys 两者较长者遍历(只给 key 时不再丢图)
- 媒体机上传的 key 不进失败回滚清单(内容寻址、可能与他人共享)
- /admin/internal/upload-url 签发 S4 PUT 预签名(spider 不持有 AK/SK)

embeddings(向量回填)
- GET /admin/internal/embeddings/pending 派发缺向量的图(含可直接下载的 url)
- POST /admin/internal/embeddings 回写向量并重算近邻;失败项计入
  image_embedding_failures,连续失败到上限后退出队列(坏图不占位)
- 近邻重算改为异步串行:同步会让 100 张的批次 POST 超时(实测 >60s)
- 新增 key 分支单测:混用 key/url、细节图只给 key、字节数、回滚清单、向量占位
2026-10-25 00:37:13 +08:00

156 lines
5.4 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package storage
import (
"bytes"
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/aws/aws-sdk-go-v2/aws"
awsconfig "github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/s3"
)
// s4DefaultEndpoint 缤纷云 S4 的 S3 兼容接入点(官方文档确认 https://s3.bitiful.net)。
// 采用虚拟主机风格寻址,请求落在 https://<bucket>.s3.bitiful.net/<key>。
const s4DefaultEndpoint = "https://s3.bitiful.net"
// S4Uploader 缤纷云 S4 对象存储上传器(兼容 S3 协议,基于 aws-sdk-go-v2)。
//
// 图片经 CoreIX 公开样式(如 key!style:high)对外提供,VIP 与免费用户同质量、匿名可访问。
// 库里只存对象 key(如 runway/abc.jpg),完整 URL 由 imgurl.Composer 拼装。
// 本类型同时实现 storage.Uploader 与 storage.Deleter。
type S4Uploader struct {
client *s3.Client
bucket string
baseURL string
enabled bool
}
// NewS4Uploader 构造缤纷云 S4 上传器。
// - ak/sk:缤纷云子账户 AccessKey/SecretKey(须有 PutObject/DeleteObject/GetObject 权限)。
// - bucket:存储桶名(须为「公开桶」,图片经 CoreIX 公开样式对外提供)。
// - endpoint:S3 接入点,默认 https://s3.bitiful.net;虚拟主机风格寻址。
// - baseURL:对外访问域名(GET 图片用)。为空时自动推导为 https://<bucket>.s3.bitiful.net,
// 须与该 S4 接入点的虚拟主机域名一致,保证展示图 URL 同源。
// - region:任意值(如 cn-east-1),缤纷云不校验。
func NewS4Uploader(ak, sk, bucket, endpoint, baseURL, region string) *S4Uploader {
if ak == "" || sk == "" || bucket == "" {
return &S4Uploader{enabled: false}
}
if endpoint == "" {
endpoint = s4DefaultEndpoint
}
if region == "" {
region = "cn-east-1"
}
cfg, err := awsconfig.LoadDefaultConfig(context.Background(),
awsconfig.WithRegion(region),
awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(ak, sk, "")),
)
if err != nil {
return &S4Uploader{enabled: false}
}
client := s3.NewFromConfig(cfg, func(o *s3.Options) {
o.BaseEndpoint = aws.String(endpoint)
o.UsePathStyle = false // 虚拟主机风格:https://<bucket>.s3.bitiful.net/<key>
})
if baseURL == "" {
baseURL = deriveS4BaseURL(endpoint, bucket)
}
return &S4Uploader{
client: client,
bucket: bucket,
baseURL: strings.TrimRight(baseURL, "/"),
enabled: true,
}
}
// PresignPut 为指定 key 生成有时效的 PUT 预签名 URL。
//
// 用途:让**不持有 S4 密钥**的客户端(例如另部署的 spider 媒体机)直传对象。
// 密钥只留在后端,客户端拿到 URL 直接 HTTP PUT 即可 —— 不必在爬虫里引入 S3 SDK,
// 也不用把 AK/SK 分发到更多机器。
//
// contentType 必须与 PUT 请求头一致,否则 S4 会因签名不匹配拒绝(403)。
func (u *S4Uploader) PresignPut(ctx context.Context, key, contentType string, ttl time.Duration) (string, error) {
if !u.enabled || u.client == nil {
return "", errors.New("S4 未启用(缺少 AK/SK 或 bucket)")
}
if key == "" {
return "", errors.New("key 不能为空")
}
if contentType == "" {
contentType = "application/octet-stream"
}
if ttl <= 0 {
ttl = 15 * time.Minute
}
pc := s3.NewPresignClient(u.client)
out, err := pc.PresignPutObject(ctx, &s3.PutObjectInput{
Bucket: aws.String(u.bucket),
Key: aws.String(key),
ContentType: aws.String(contentType),
}, s3.WithPresignExpires(ttl))
if err != nil {
return "", err
}
return out.URL, nil
}
// deriveS4BaseURL 由 endpoint + bucket 推导虚拟主机风格访问域名:
// endpoint=https://s3.bitiful.net, bucket=toom → https://toom.s3.bitiful.net
func deriveS4BaseURL(endpoint, bucket string) string {
endpoint = strings.TrimRight(endpoint, "/")
scheme := "https://"
host := endpoint
if strings.HasPrefix(host, "https://") {
host = host[len("https://"):]
} else if strings.HasPrefix(host, "http://") {
scheme = "http://"
host = host[len("http://"):]
}
return scheme + bucket + "." + host
}
// PublicBaseURL 返回对外访问域名(可能由 NewS4Uploader 自动推导)。
func (s *S4Uploader) PublicBaseURL() string { return s.baseURL }
// Enabled 凭证齐全才视为启用。
func (s *S4Uploader) Enabled() bool { return s.enabled }
// Upload 把 data 直传到缤纷云 bucket 的 key,返回 key(库里只存 key,baseURL 由 Composer 拼)。
func (s *S4Uploader) Upload(ctx context.Context, key string, data []byte, contentType string) (string, error) {
if !s.enabled {
return "", fmt.Errorf("s4 uploader not enabled")
}
input := &s3.PutObjectInput{
Bucket: aws.String(s.bucket),
Key: aws.String(key),
Body: bytes.NewReader(data),
ContentType: aws.String(contentType),
}
if _, err := s.client.PutObject(ctx, input); err != nil {
return "", fmt.Errorf("s4 put object failed: %w", err)
}
return key, nil
}
// Delete 删除缤纷云 bucket 中的对象(按 key)。key 可能带 /uploads/ 等前缀,先剥离。
func (s *S4Uploader) Delete(ctx context.Context, key string) error {
if !s.enabled {
return nil
}
key = strings.TrimPrefix(key, "/")
if _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{
Bucket: aws.String(s.bucket),
Key: aws.String(key),
}); err != nil {
return fmt.Errorf("s4 delete object failed: %w", err)
}
return nil
}