From acb9a34520b85c61861d86afc42992935ea1917c Mon Sep 17 00:00:00 2001
From: root <fish93558@gmail.com>
Date: Thu, 29 Oct 2026 20:18:36 +0800
Subject: [PATCH] =?UTF-8?q?[AI]=20fix:=20=E4=BF=AE=E5=A4=8D=E7=9F=AD?=
 =?UTF-8?q?=E5=89=A7=E5=89=A7=E9=9B=86=E9=87=8D=E5=A4=8D=E6=98=BE=E7=A4=BA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

两处原因，分别处理：

1. contentsync 只按 _id 水位线插新增，源库事后打上的 isDelete 传不过来，
   源站删掉的重复集在目标库一直留着。每轮同步后补扫一次 media /
   media_content / video_info，把源库的软删除标记同步过去，并按
   web/service/mediacontentser 的同一口径重算父级 currentEpisode /
   freeEpisode。只做「源库删了→目标也删」一个方向，不与本地下架保护冲突。

2. 各站 skd 的媒资库同步会为同一集(mediaId+hashId 相同)另插一条自己的 _id，
   与 contentsync 带过来的源库那条并存。媒资库那条在 SaleCoin=0 时
   listenPermission=2，本该收费的集可以白看。插入时若发现本地已有同一集，
   把本地那条 isActive:false 下架；并新增 skd 配置开关
   switch.mediaSyncDisabled，非源站可直接关掉这条同步链。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 scripts/contentsync/README.md | 134 ++++++++
 scripts/contentsync/main.go   | 567 ++++++++++++++++++++++++++++++++++
 scripts/contentsync/spec.go   |  77 +++++
 skd/job/job.go                |  20 +-
 skd/skdg/skdg.go              |   9 +
 5 files changed, 799 insertions(+), 8 deletions(-)
 create mode 100644 scripts/contentsync/README.md
 create mode 100644 scripts/contentsync/main.go
 create mode 100644 scripts/contentsync/spec.go

diff --git a/scripts/contentsync/README.md b/scripts/contentsync/README.md
new file mode 100644
index 0000000..deef2fd
--- /dev/null
+++ b/scripts/contentsync/README.md
@@ -0,0 +1,134 @@
+# contentsync
+
+把源库 `91porn_video` 的**帖子 / 视频 / 短剧**增量同步到 `91fans`、`huangguo`、`tangguo` 三个库。
+
+## 为什么不能直接整文档覆盖
+
+对源库与目标库做过逐字段抽样比对（video_info 400 条 / 其余各 300 条），
+74 个字段里只有 9 个分叉，且全部是**站点本地产生**的数据：
+
+| 集合 | 分叉字段 | 分叉比例 |
+|---|---|---|
+| `video_info` | pageViewCount / playCount / fakePlayCount / likeCount / fakeLikeCount / collectCount / forwardCount / purchaseCount / updatedAt | 最高 88% |
+| `media` | countBrowse / countCollect / countLike / countPurchases / hot / updateTime | 最高 39% |
+| `tag` | tPlayCount / vPlayCount / playRating / videoCount / updatedAt | 最高 42% |
+| `media_content` / `media_tag` / `media_tag_dimension` | 无 | 0% |
+
+内容字段（title、cover、tags、status、newsType、coins…）**完全一致**。
+
+所以同步必须是字段级的。本地字段分三类，都不能被源库覆盖：
+
+1. **运行时计数器** —— 各站点自己累积的播放 / 点赞 / 收藏 / 购买数
+2. **本地定时任务写入** —— `skd/job/hotvalue/video.go` 会重算并写回 `hot`
+3. **后台运营编排** —— `web/service/vidser/vid.go` 写 `happinessPlazaTop`、`liaoBaTopSort`
+
+清单见 `spec.go`，新增字段前先确认它属于「源库内容」还是「站点本地」。
+
+## 写入语义
+
+一条 upsert 同时处理新增和更新：
+
+```
+UpdateOne({_id}, {
+    $set:         <内容字段>,    // 每次同步都以源库为准
+    $setOnInsert: <本地字段>,    // 只在首次插入时给初值，之后永不再碰
+}, upsert: true)
+```
+
+只更新不删除。源库下架通过 `status` 字段自然同步过去。
+
+## 本地下架保护
+
+每批写入前先查目标库这批 `_id` 里哪些是 `status: 0`，**整条踢出**——连内容字段
+也不写。站点运营下架之后，源库无权再把它改回上架。
+
+水位线仍会越过被跳过的文档，否则它会永久卡住水位线导致每次重扫。
+
+没有 `status` 字段的集合（media_tag / media_tag_dimension / tag）自然查不出东西，
+等于不生效。
+
+## 软删除补扫
+
+水位线只管新增，所以源库**事后**打上的 `isDelete: true` 永远传不过来：源站把重复
+剧集软删掉了，目标库却还留着，App 剧集列表就出现「1 1 1 2 2 2」这种重复集。
+2026-10-29 排查 `琼明神女录` 显示 6 集（实际 2 集）就是这个原因。
+
+每轮同步结束后对 `media` / `media_content` / `video_info` 补扫一次：把源库
+`isDelete: true` 的文档在目标库也标成已删除。`--sweep-deleted` 默认开，三个库 6 秒。
+
+- **只做「源库删了 → 目标也删」这一个方向**，绝不把目标库已删的恢复回来，所以和
+  上面的本地下架保护不冲突
+- 只写 `isDelete`，不碰 `updateTime`（那是站点本地字段）
+- 顺带按 `web/service/mediacontentser` 的同一口径重算父级 `currentEpisode` /
+  `freeEpisode`——这两个字段原本只在后台编辑子集时才重算，不补的话集数会停在
+  含重复集的旧值
+
+一次性修复历史欠账时实测：三个库各 261 条 `media_content` + 18 条 `media`，
+涉及 28 部短剧。`freeEpisode` 只减不增（减少量正好等于被去掉的重复免费集数）。
+
+## 媒资库重复集去重
+
+各站点的 skd 自己也在跑媒资库同步（`common/media/syncMedia.go`，每 30 分钟），
+它按 `(mediaId, hashId)` 去重、用**自己的 `_id`** 为同一集插一条；本同步再把源库
+那条插进来，同一集就成了两条。两个源库的 `(mediaId, hashId)` 重复组都是 0，
+只有目标站有——这是本同步引入的。
+
+更要紧的是媒资库那条在 `SaleCoin = 0` 时 `listenPermission = 2`（免费），
+**本该收费的集可以白看**。2026-10-29 的存量：
+
+| 站点 | 重复组 | 其中免费副本 |
+|---|---|---|
+| tangguo / huangguo / 91fans | 各 268 | 各 241 |
+| hgmj | 97 | 84 |
+| 51dj | 102 | 97 |
+
+购买记录全部挂在源库那条上（本地副本购买恒为 0，因为它免费、没人需要买），
+所以**以源库那条为准**：插入时若发现本地已有同一集，把本地那条 `isActive: false`
+下架（不删除，便于回查）。
+
+根治办法是关掉各站 skd 的媒资库同步：`skd.json` 里 `switch.mediaSyncDisabled: true`
+（见 `skd/job/job.go`）。**源站必须保持开启**，否则新剧集无从进库。
+关掉之后新剧集改由本同步每天 04:00 从源站带过来，延迟从 30 分钟变成最多 24 小时。
+
+## 增量水位线
+
+每个「目标库 × 集合」各记一份水位线，存在 `state.json`：
+
+常规同步**只取新增**：`_id > lastID`，按 `_id` 升序。ObjectId 单调递增且天然有索引，
+所以每次只读上次之后产生的文档，耗时与集合总量无关（实测 1~3 秒）。
+
+**刻意不做存量修改比对。** 源库改了已存在的文档——改标题、下架、调价、换封面——
+不会同步过来。需要补齐时手动跑 `--full` 重扫全部（三个库约 17 分钟起）。
+
+这是明确的取舍。参考量级：一次 `--full` 里 `video_info` 更新了约 7300 条
+（占 273K 的 2.7%）、`media` 更新了约 5900 条（占 20K 的 29%）。
+
+## 用法
+
+```bash
+# 增量：只取新增。cron 每天北京时间 04:00 跑的就是这个，耗时数秒
+/server/sync/run.sh
+
+# 全量：忽略水位线重扫，补齐存量修改。约 17 分钟起，手动跑
+/server/sync/run.sh --full
+
+# 只跑某几个集合 / 干跑
+/server/sync/contentsync --src "$SRC_URI" --dst "tangguo=$DST_TANGGUO" \
+    --only video_info,media --dry-run --state /server/sync/state.json
+```
+
+## 部署
+
+服务器上无 Go / node / mongosh，所以本地交叉编译后 scp 静态二进制：
+
+```bash
+GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o contentsync ./scripts/contentsync/
+scp contentsync root@172.104.39.32:/server/sync/
+```
+
+`/server/sync/env.sh`（600，含连接串）由 `run.sh` source。不走 Jenkins。
+
+cron：`0 20 * * * /server/sync/run.sh`
+
+注意机器系统时区是 **UTC**，且 Debian 系的 `cron 3.0pl1` **不支持 `CRON_TZ`**
+（那是 cronie 的特性），所以只能直接写 UTC 时刻：20:00 UTC = 北京时间 04:00。
diff --git a/scripts/contentsync/main.go b/scripts/contentsync/main.go
new file mode 100644
index 0000000..39d0a84
--- /dev/null
+++ b/scripts/contentsync/main.go
@@ -0,0 +1,567 @@
+// 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) 相同但 _id 不同。
+	//
+	// 各站点自己也在跑媒资库同步(common/media/syncMedia.go)，它按
+	// (mediaId, hashId) 去重、用自己的 _id 为同一集插一条；本同步再把源库那条
+	// 插进来，同一集就变成两条，前端剧集列表显示「1 1 2 2」。
+	// 更要紧的是媒资库那条在 SaleCoin=0 时 listenPermission=2（免费），
+	// 等于用户可以白看本该收费的集。
+	//
+	// 以源库那条为准（购买记录都挂在它上面），把本地那条下架 isActive:false，
+	// 不删除，便于事后回查。
+	localTwins := func(docs []bson.M) (map[any]any, error) {
+		hashes := make([]any, 0, len(docs))
+		want := map[string]any{} // mediaId|hashId -> 源库 _id
+		for _, d := range docs {
+			h, ok := d["hashId"].(string)
+			if !ok || h == "" {
+				continue
+			}
+			mid, ok := d["mediaId"].(primitive.ObjectID)
+			if !ok {
+				continue
+			}
+			hashes = append(hashes, h)
+			want[mid.Hex()+"|"+h] = d["_id"]
+		}
+		if len(hashes) == 0 {
+			return nil, nil
+		}
+		// hashId 上有索引，按它捞再在内存里配 mediaId
+		c, e := dst.Find(ctx, bson.M{"hashId": bson.M{"$in": hashes}, "isActive": true},
+			options.Find().SetProjection(bson.M{"_id": 1, "mediaId": 1, "hashId": 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"`
+			}
+			if e := c.Decode(&row); e != nil {
+				return nil, e
+			}
+			srcID, ok := want[row.MediaID.Hex()+"|"+row.HashID]
+			if !ok {
+				continue
+			}
+			if sid, ok := srcID.(primitive.ObjectID); ok && sid == row.ID {
+				continue // 就是源库那条自己，不是重复
+			}
+			out[srcID] = row.ID
+		}
+		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)
+}
diff --git a/scripts/contentsync/spec.go b/scripts/contentsync/spec.go
new file mode 100644
index 0000000..7b71079
--- /dev/null
+++ b/scripts/contentsync/spec.go
@@ -0,0 +1,77 @@
+package main
+
+// CollSpec 描述一个集合怎么同步。
+//
+// localFields 是「站点本地字段」：目标库自己产生、源库无权覆盖的字段。
+// 判定依据有三类，改动前请先确认属于哪一类：
+//  1. 运行时计数器 —— 各站点自己累积的播放/点赞/收藏/购买数
+//  2. 本地定时任务写入 —— 如 skd/job/hotvalue 会重算并写回 hot
+//  3. 后台运营编排 —— 如 web/service/vidser 写的 happinessPlazaTop、liaoBaTopSort
+//
+// 这些字段走 $setOnInsert：新文档插入时给初值，之后永不再覆盖。
+// 其余字段走 $set。
+//
+// 注意：常规同步只按 _id 水位线取新增，不比对存量修改，所以 $set 实际只在
+// 新文档插入、或手动 --full 重扫时才起作用。
+type CollSpec struct {
+	Name string
+	// LocalFields 走 $setOnInsert，只在首次插入时写入，之后永不覆盖。
+	LocalFields []string
+}
+
+// 各集合的本地字段清单，来自对源库与目标库 400/300 条抽样的逐字段比对，
+// 并交叉核对了 web/ 与 skd/ 中的写入点。
+var specs = []CollSpec{
+	{
+		Name: "video_info",
+		LocalFields: []string{
+			// 运行时计数器
+			"playCount", "fakePlayCount", "effectivePlayCount", "pageViewCount",
+			"likeCount", "fakeLikeCount", "collectCount", "forwardCount",
+			"shareCount", "fakeShareCount", "commentCount", "fakeCommentCount",
+			"validCommentCount", "purchaseCount", "rewarded", "fakeRewarded",
+			// 本地 skd 重算：skd/job/hotvalue/video.go
+			"hot", "hotCommentExposureUntil", "lastHotTriggerAt",
+			// 后台运营编排：web/service/vidser/vid.go
+			"happinessPlazaTop", "liaoBaTop", "liaoBaTopSort",
+			"isTopping", "isRecommend", "chosen", "chosenDate",
+			"sortCode", "tagSort", "recoWeight", "upTag", "shareSort", "worksSort",
+			// 本地修改时间戳
+			"updatedAt",
+		},
+	},
+	{
+		Name: "media",
+		LocalFields: []string{
+			"countBrowse", "countCollect", "countComment", "countContentPurchases",
+			"countDisLike", "countLike", "countPurchases", "countView",
+			"hot",
+			// 短剧在各站点的编排位
+			"choice", "choiceSort", "general", "generalSort", "sectionSort",
+			"updateTime", "updatedAt", "updatedAct", "updatedDesc",
+		},
+	},
+	{
+		Name: "media_content",
+		LocalFields: []string{
+			"countBrowse", "countLike", "countPurchases", "fakeCountBrowse",
+			"updateTime", "updatedAt", "updatedAct",
+		},
+	},
+	{
+		Name:        "media_tag",
+		LocalFields: []string{"updatedAt"},
+	},
+	{
+		Name:        "media_tag_dimension",
+		LocalFields: []string{"updatedAt"},
+	},
+	{
+		Name: "tag",
+		LocalFields: []string{
+			"collCount", "fakeCollCount", "fakePlayCount", "likeCount",
+			"playRating", "tPlayCount", "vPlayCount", "videoCount", "hotMark",
+			"updatedAt",
+		},
+	},
+}
diff --git a/skd/job/job.go b/skd/job/job.go
index 9f9580a..84231e0 100644
--- a/skd/job/job.go
+++ b/skd/job/job.go
@@ -93,14 +93,18 @@ func start() {
 		RunPendingSyncIMUsers()
 	})
 	// 同步acg订阅的动漫
-	_, _ = statCron.AddFunc("0 0/30 * * * ?", func() { //每30分钟执行一次
-		ms := mediaService.NewMediaSync(skdg.Redis, mediaCenter.NewMediaCenterService(mediaCenter.Config{
-			Url:    skdg.Conf.MediaResourcesCfg.UploadMediaUrl,
-			ProId:  skdg.Conf.MediaResourcesCfg.ProId,
-			Secret: skdg.Conf.MediaResourcesCfg.AppSecret,
-		}))
-		ms.SyncAll("系统")
-	})
+	if skdg.Conf.Switch.MediaSyncDisabled {
+		log.Info("【ACG同步任务】已按配置关闭(switch.mediaSyncDisabled)，内容改由 contentsync 从源站同步")
+	} else {
+		_, _ = statCron.AddFunc("0 0/30 * * * ?", func() { //每30分钟执行一次
+			ms := mediaService.NewMediaSync(skdg.Redis, mediaCenter.NewMediaCenterService(mediaCenter.Config{
+				Url:    skdg.Conf.MediaResourcesCfg.UploadMediaUrl,
+				ProId:  skdg.Conf.MediaResourcesCfg.ProId,
+				Secret: skdg.Conf.MediaResourcesCfg.AppSecret,
+			}))
+			ms.SyncAll("系统")
+		})
+	}
 	common.Go(func() {
 		err := aiService.NewAiService(
 			aiService.AppId(int(commod.KFK_APPID)),
diff --git a/skd/skdg/skdg.go b/skd/skdg/skdg.go
index 8713427..e111493 100644
--- a/skd/skdg/skdg.go
+++ b/skd/skdg/skdg.go
@@ -79,6 +79,15 @@ type GlobalConfig struct {
 	} `json:"url"`
 	Switch struct {
 		NotifyExpire bool `json:"notifyExpire"`
+		// MediaSyncDisabled 关掉 skd 每 30 分钟的媒资库同步。
+		//
+		// 本站内容由 scripts/contentsync 从源站整体同步过来，源站自己会跑媒资库
+		// 同步；本站再跑一遍会为同一集(mediaId+hashId 相同)另插一条自己的 _id，
+		// 剧集列表就出现「1 1 2 2」，而且媒资库那条在 SaleCoin=0 时
+		// listenPermission=2（免费），等于本该收费的集可以白看。
+		//
+		// 留空/false 表示照旧开启——源站必须保持开启，否则新剧集无从进库。
+		MediaSyncDisabled bool `json:"mediaSyncDisabled"`
 	} `json:"switch"`
 	StartDate         string `json:"startDate"`
 	MediaResourcesCfg struct {