Files
rootandClaude Opus 5 ed51664a09 [AI] fix: 剧集去重键补上 videoUrl
媒资库会给同一集重新发 hashId(「JK学姐不为人知的秘密」源库那条
hashId 6aa23652…、本地那条 6acaf74c…,videoUrl 一模一样),
只比 hashId 会漏掉这一类重复,前端仍显示「1 1 2 2」。
改成 mediaId 下 hashId 或 videoUrl 任一相同即认定是同一集。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-10-29 20:24:38 +08:00

586 lines
17 KiB
Go
Raw Permalink 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.
// contentsync 把源库(91porn_video)的帖子/视频/短剧增量同步到若干目标库。
//
// 增量策略(每个集合、每个目标库各自记水位线):
// - 新增:按 _id 升序游标,_id > lastID。ObjectId 单调递增且天然有索引。
// - 修改:按 tsField > lastTS。仅对有该字段且建了索引的集合启用。
// 没有可靠时间字段的集合(media_content/media_tag/tag)靠定期 --full 兜底。
//
// 写入语义:$set 内容字段 + $setOnInsert 站点本地字段(见 spec.go)。
// 只更新不删除;源库下架通过 status 字段自然同步。
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
"strings"
"time"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
)
// 源库名由 --src-db 指定。两个源库 91porn_video 与 91dj_video 老数据同源但已分叉
// (最新 500 条零重叠),所以每个目标库只能配一个源,不能混。
const defaultSrcDBName = "91porn_video"
type mark struct {
LastID string `json:"lastId"`
LastTS time.Time `json:"lastTs"`
}
// state: target -> collection -> mark
type state map[string]map[string]*mark
func loadState(p string) state {
s := state{}
b, err := os.ReadFile(p)
if err != nil {
return s
}
_ = json.Unmarshal(b, &s)
return s
}
func saveState(p string, s state) error {
b, err := json.MarshalIndent(s, "", " ")
if err != nil {
return err
}
tmp := p + ".tmp"
if err := os.WriteFile(tmp, b, 0o600); err != nil {
return err
}
return os.Rename(tmp, p) // 原子替换,避免崩溃时留下半截状态
}
func (s state) get(target, coll string) *mark {
if s[target] == nil {
s[target] = map[string]*mark{}
}
if s[target][coll] == nil {
s[target][coll] = &mark{}
}
return s[target][coll]
}
// dstList 让 --dst 可以重复出现。MongoDB URI 自身含逗号(副本集主机列表),
// 所以不能用逗号分隔多个目标。
type dstList []string
func (d *dstList) String() string { return strings.Join(*d, " ") }
func (d *dstList) Set(v string) error {
*d = append(*d, v)
return nil
}
type target struct {
name string
db *mongo.Database
}
func main() {
var (
srcURI = flag.String("src", "", "源库 URI")
dsts dstList
srcDBName = flag.String("src-db", defaultSrcDBName, "源库名")
statePath = flag.String("state", "/server/sync/state.json", "水位线状态文件")
full = flag.Bool("full", false, "忽略水位线做全量扫描(用于兜底修复)")
dryRun = flag.Bool("dry-run", false, "只统计不写入")
only = flag.String("only", "", "只同步指定集合,逗号分隔")
batchSize = flag.Int("batch", 500, "每批写入条数")
sweep = flag.Bool("sweep-deleted", true, "回传源库软删除标记(清理重复剧集)")
timeout = flag.Duration("timeout", 30*time.Minute, "整体超时")
)
flag.Var(&dsts, "dst", "目标库,格式 name=uri;可重复指定多个")
flag.Parse()
if *srcURI == "" || len(dsts) == 0 {
fmt.Fprintln(os.Stderr, "必须指定 --src 和 --dst")
os.Exit(2)
}
ctx, cancel := context.WithTimeout(context.Background(), *timeout)
defer cancel()
srcCli, err := mongo.Connect(ctx, options.Client().ApplyURI(*srcURI))
if err != nil {
fatal("连接源库失败: %v", err)
}
defer srcCli.Disconnect(context.Background())
srcDB := srcCli.Database(*srcDBName)
var targets []target
for _, part := range dsts {
part = strings.TrimSpace(part)
if part == "" {
continue
}
name, uri, ok := strings.Cut(part, "=")
if !ok {
fatal("--dst 格式错误,应为 name=uri: %s", part)
}
cli, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
if err != nil {
fatal("连接目标库 %s 失败: %v", name, err)
}
defer cli.Disconnect(context.Background())
targets = append(targets, target{name: name, db: cli.Database(name)})
}
wanted := map[string]bool{}
for _, n := range strings.Split(*only, ",") {
if n = strings.TrimSpace(n); n != "" {
wanted[n] = true
}
}
st := loadState(*statePath)
started := time.Now()
var totalUp, totalIns int
failed := false
for _, spec := range specs {
if len(wanted) > 0 && !wanted[spec.Name] {
continue
}
for _, t := range targets {
ins, up, err := syncOne(ctx, srcDB, t, spec, st, *full, *dryRun, *batchSize)
if err != nil {
fmt.Printf("[FAIL] %-20s -> %-10s %v\n", spec.Name, t.name, err)
failed = true
continue
}
totalIns += ins
totalUp += up
fmt.Printf("[ok] %-20s -> %-10s 新增 %6d 更新 %6d\n", spec.Name, t.name, ins, up)
if !*dryRun {
if err := saveState(*statePath, st); err != nil {
fmt.Printf("[WARN] 写状态文件失败: %v\n", err)
}
}
}
}
if *sweep {
for _, coll := range tombstoneColls {
if len(wanted) > 0 && !wanted[coll] {
continue
}
for _, t := range targets {
n, fixed, err := sweepDeleted(ctx, srcDB, t, coll, *dryRun)
if err != nil {
fmt.Printf("[FAIL] 补扫软删除 %-14s -> %-10s %v\n", coll, t.name, err)
failed = true
continue
}
fmt.Printf("[ok] 补扫软删除 %-14s -> %-10s 标记 %5d 重算父级 %4d\n", coll, t.name, n, fixed)
}
}
}
fmt.Printf("\n合计 新增 %d 更新 %d,耗时 %s\n", totalIns, totalUp, time.Since(started).Round(time.Second))
if failed {
os.Exit(1)
}
}
func syncOne(ctx context.Context, srcDB *mongo.Database, t target, spec CollSpec,
st state, full, dryRun bool, batchSize int) (ins, up int, err error) {
m := st.get(t.name, spec.Name)
local := map[string]bool{}
for _, f := range spec.LocalFields {
local[f] = true
}
// 只按 _id 水位线取新增。ObjectId 单调递增且天然有索引,
// 所以每次只读上次同步之后产生的文档,跟集合总量无关。
//
// 刻意不做存量修改比对:源库改了已存在的文档不会同步过来。
// 需要补齐时用 --full 手动重扫一次。
filter := bson.M{}
if !full && m.LastID != "" {
oid, e := primitive.ObjectIDFromHex(m.LastID)
if e == nil {
filter["_id"] = bson.M{"$gt": oid}
}
}
ins, up, maxID, err := runPass(ctx, srcDB, t, spec, local, filter,
bson.D{{Key: "_id", Value: 1}}, dryRun, batchSize)
if err != nil {
return ins, up, err
}
// 水位线只在真跑时推进,干跑不能改状态
if !dryRun && maxID != primitive.NilObjectID {
m.LastID = maxID.Hex()
}
return ins, up, nil
}
func runPass(ctx context.Context, srcDB *mongo.Database, t target, spec CollSpec,
local map[string]bool, filter bson.M, sortBy bson.D, dryRun bool, batchSize int,
) (ins, up int, maxID primitive.ObjectID, err error) {
srcCur, err := srcDB.Collection(spec.Name).Find(ctx, filter,
options.Find().SetSort(sortBy).SetBatchSize(int32(batchSize)).SetNoCursorTimeout(true))
if err != nil {
return 0, 0, maxID, err
}
defer srcCur.Close(ctx)
dst := t.db.Collection(spec.Name)
pending := make([]bson.M, 0, batchSize)
scanned, skipped, deduped := 0, 0, 0
lastLog := time.Now()
// blockedIDs 查出这批里哪些在目标库已被本地下架(status:0)。
// 目标库没有 status 字段的集合自然查不出东西,等于不生效。
blockedIDs := func(docs []bson.M) (map[any]bool, error) {
out := map[any]bool{}
ids := make([]any, 0, len(docs))
for _, d := range docs {
ids = append(ids, d["_id"])
}
c, e := dst.Find(ctx, bson.M{"_id": bson.M{"$in": ids}, "status": 0},
options.Find().SetProjection(bson.M{"_id": 1}))
if e != nil {
return nil, e
}
defer c.Close(ctx)
for c.Next(ctx) {
var row bson.M
if e := c.Decode(&row); e == nil {
out[row["_id"]] = true
}
}
return out, c.Err()
}
// localTwins 查出这批源文档在目标库的「同一集另一条」:同属一个 mediaId,
// 且 hashId 或 videoUrl 相同,但 _id 不同。
//
// 各站点自己也在跑媒资库同步(common/media/syncMedia.go),它按
// (mediaId, hashId) 去重、用自己的 _id 为同一集插一条;本同步再把源库那条
// 插进来,同一集就变成两条,前端剧集列表显示「1 1 2 2」。
// 更要紧的是媒资库那条在 SaleCoin=0 时 listenPermission=2(免费),
// 等于用户可以白看本该收费的集。
//
// 必须同时认 videoUrl:媒资库会给同一集重新发 hashId(实测
// 「JK学姐不为人知的秘密」源库那条 hashId 6aa23652…、本地那条
// 6acaf74c…,videoUrl 却一模一样),只比 hashId 会漏掉这一类。
//
// 以源库那条为准(购买记录都挂在它上面),把本地那条下架 isActive:false,
// 不删除,便于事后回查。
localTwins := func(docs []bson.M) (map[any]any, error) {
keys := make([]any, 0, len(docs)*2) // 待查的 hashId 与 videoUrl
want := map[string]any{} // mediaId|key -> 源库 _id
add := func(mid primitive.ObjectID, v string, id any) {
if v == "" {
return
}
keys = append(keys, v)
want[mid.Hex()+"|"+v] = id
}
for _, d := range docs {
mid, ok := d["mediaId"].(primitive.ObjectID)
if !ok {
continue
}
h, _ := d["hashId"].(string)
u, _ := d["videoUrl"].(string)
add(mid, h, d["_id"])
add(mid, u, d["_id"])
}
if len(keys) == 0 {
return nil, nil
}
// hashId 上有索引;videoUrl 没有,所以放在同一个 $or 里一次查完
c, e := dst.Find(ctx, bson.M{"isActive": true, "$or": []bson.M{
{"hashId": bson.M{"$in": keys}},
{"videoUrl": bson.M{"$in": keys}},
}}, options.Find().SetProjection(bson.M{"_id": 1, "mediaId": 1, "hashId": 1, "videoUrl": 1}))
if e != nil {
return nil, e
}
defer c.Close(ctx)
out := map[any]any{}
for c.Next(ctx) {
var row struct {
ID primitive.ObjectID `bson:"_id"`
MediaID primitive.ObjectID `bson:"mediaId"`
HashID string `bson:"hashId"`
VideoURL string `bson:"videoUrl"`
}
if e := c.Decode(&row); e != nil {
return nil, e
}
for _, k := range []string{row.HashID, row.VideoURL} {
if k == "" {
continue
}
srcID, ok := want[row.MediaID.Hex()+"|"+k]
if !ok {
continue
}
if sid, ok := srcID.(primitive.ObjectID); ok && sid == row.ID {
continue // 就是源库那条自己,不是重复
}
out[srcID] = row.ID
break
}
}
return out, c.Err()
}
flush := func() error {
if len(pending) == 0 {
return nil
}
docs := pending
pending = make([]bson.M, 0, batchSize)
blocked, e := blockedIDs(docs)
if e != nil {
return e
}
twins, e := localTwins(docs)
if e != nil {
return e
}
models := make([]mongo.WriteModel, 0, len(docs))
for _, doc := range docs {
id := doc["_id"]
// 站点运营下架之后,源库无权再改它——整条跳过,内容字段也不写。
if blocked[id] {
skipped++
continue
}
set := bson.M{}
onInsert := bson.M{}
for k, v := range doc {
if k == "_id" {
continue
}
if local[k] {
onInsert[k] = v
} else {
set[k] = v
}
}
update := bson.M{}
if len(set) > 0 {
update["$set"] = set
}
if len(onInsert) > 0 {
update["$setOnInsert"] = onInsert
}
if len(update) == 0 {
continue
}
models = append(models, mongo.NewUpdateOneModel().
SetFilter(bson.M{"_id": id}).SetUpdate(update).SetUpsert(true))
// 同一集本地已有另一条(媒资库同步插的),把它下架
if twinID, ok := twins[id]; ok {
models = append(models, mongo.NewUpdateOneModel().
SetFilter(bson.M{"_id": twinID}).
SetUpdate(bson.M{"$set": bson.M{"isActive": false}}))
deduped++
}
}
if len(models) == 0 {
return nil
}
if dryRun {
up += len(models)
return nil
}
res, e := dst.BulkWrite(ctx, models, options.BulkWrite().SetOrdered(false))
if e != nil {
if res != nil {
ins += int(res.UpsertedCount)
up += int(res.ModifiedCount)
}
return e
}
ins += int(res.UpsertedCount)
up += int(res.ModifiedCount)
return nil
}
for srcCur.Next(ctx) {
var doc bson.M
if e := srcCur.Decode(&doc); e != nil {
return ins, up, maxID, e
}
scanned++
// 大集合动辄几十万条,cron 里长时间无输出无法判断死活
if time.Since(lastLog) >= 30*time.Second {
fmt.Printf(" %-20s -> %-10s 已扫描 %d 条…\n", spec.Name, t.name, scanned)
lastLog = time.Now()
}
// 水位线推进到扫过的最大 _id。即使这条因本地下架被跳过也要推进,
// 否则它会永久卡住水位线,导致每次都从它开始重扫。
if id, ok := doc["_id"].(primitive.ObjectID); ok && !id.IsZero() {
if maxID.IsZero() || id.Hex() > maxID.Hex() {
maxID = id
}
}
pending = append(pending, doc)
if len(pending) >= batchSize {
if e := flush(); e != nil {
return ins, up, maxID, e
}
}
}
if e := srcCur.Err(); e != nil {
return ins, up, maxID, e
}
if e := flush(); e != nil {
return ins, up, maxID, e
}
if skipped > 0 {
fmt.Printf(" %-20s -> %-10s 跳过 %d 条(本地已下架)\n", spec.Name, t.name, skipped)
}
if deduped > 0 {
fmt.Printf(" %-20s -> %-10s 下架 %d 条本地重复集(媒资库同步插的同一集)\n", spec.Name, t.name, deduped)
}
return ins, up, maxID, nil
}
// tombstoneColls 需要回传源库软删除标记的集合。
//
// 常规同步只按 _id 水位线插新增,源库事后打上的 isDelete 永远传不过来:
// 源站把重复剧集软删掉了,目标库却还留着,App 的剧集列表就会出现
// 「1 1 1 2 2 2」这种重复集。每轮补扫一次把标记补齐。
//
// 只做「源库删了 → 目标库也删」这一个方向,绝不把目标库已删的恢复回来,
// 所以和「本地下架后不再同步」不冲突。
var tombstoneColls = []string{"media", "media_content", "video_info"}
// sweepDeleted 把源库 isDelete:true 的文档在目标库也标成已删除,
// 并按 web/service/mediacontentser 的同一口径重算父级媒体的集数。
func sweepDeleted(ctx context.Context, srcDB *mongo.Database, t target, coll string, dryRun bool,
) (marked, fixed int, err error) {
cur, err := srcDB.Collection(coll).Find(ctx, bson.M{"isDelete": true},
options.Find().SetProjection(bson.M{"_id": 1}).SetNoCursorTimeout(true))
if err != nil {
return 0, 0, err
}
defer cur.Close(ctx)
dst := t.db.Collection(coll)
parents := map[primitive.ObjectID]bool{}
batch := make([]primitive.ObjectID, 0, 1000)
flush := func() error {
if len(batch) == 0 {
return nil
}
ids := batch
batch = make([]primitive.ObjectID, 0, 1000)
// 先把本轮真正要改的那几条捞出来,顺带记下父级,用于之后重算集数
hit := make([]primitive.ObjectID, 0, len(ids))
c, e := dst.Find(ctx, bson.M{"_id": bson.M{"$in": ids}, "isDelete": false},
options.Find().SetProjection(bson.M{"_id": 1, "mediaId": 1}))
if e != nil {
return e
}
defer c.Close(ctx)
for c.Next(ctx) {
var row struct {
ID primitive.ObjectID `bson:"_id"`
MediaID primitive.ObjectID `bson:"mediaId"`
}
if e := c.Decode(&row); e != nil {
return e
}
hit = append(hit, row.ID)
if coll == "media_content" && !row.MediaID.IsZero() {
parents[row.MediaID] = true
} else if coll == "media" {
parents[row.ID] = true
}
}
if e := c.Err(); e != nil {
return e
}
if len(hit) == 0 {
return nil
}
if dryRun {
marked += len(hit)
return nil
}
// 只写 isDelete,不碰 updateTime —— 那是站点本地字段
res, e := dst.UpdateMany(ctx, bson.M{"_id": bson.M{"$in": hit}},
bson.M{"$set": bson.M{"isDelete": true}})
if e != nil {
return e
}
marked += int(res.ModifiedCount)
return nil
}
for cur.Next(ctx) {
var row struct {
ID primitive.ObjectID `bson:"_id"`
}
if e := cur.Decode(&row); e != nil {
return marked, fixed, e
}
batch = append(batch, row.ID)
if len(batch) >= 1000 {
if e := flush(); e != nil {
return marked, fixed, e
}
}
}
if e := cur.Err(); e != nil {
return marked, fixed, e
}
if e := flush(); e != nil {
return marked, fixed, e
}
if dryRun || len(parents) == 0 {
return marked, len(parents), nil
}
// 父级的 currentEpisode/freeEpisode 原本只在后台编辑子集时重算,
// 这里补上,否则集数会停留在含重复集的旧值(如 2 集的剧显示 6 集)。
mc := t.db.Collection("media_content")
media := t.db.Collection("media")
for mid := range parents {
base := bson.M{"mediaId": mid, "isActive": true, "isDelete": false}
cnt, e := mc.CountDocuments(ctx, base)
if e != nil {
return marked, fixed, e
}
free, e := mc.CountDocuments(ctx, bson.M{"mediaId": mid, "listenPermission": 2,
"isActive": true, "isDelete": false})
if e != nil {
return marked, fixed, e
}
res, e := media.UpdateOne(ctx, bson.M{"_id": mid},
bson.M{"$set": bson.M{"currentEpisode": cnt, "freeEpisode": free}})
if e != nil {
return marked, fixed, e
}
fixed += int(res.ModifiedCount)
}
return marked, fixed, nil
}
func fatal(f string, a ...any) {
fmt.Fprintf(os.Stderr, f+"\n", a...)
os.Exit(1)
}