[AI] chore: 纳入 dbseed 一次性建库播种工具
hgmj 从 91dj_video 建库时写的工具,user 表按 uid < 30000 过滤, 排除交易/分析/IM/AI/UGC 等大表。之前一直没入库。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,77 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import "go.mongodb.org/mongo-driver/bson"
|
||||||
|
|
||||||
|
// SeedSpec 一个要播种的集合。Filter 为 nil 表示整表搬。
|
||||||
|
type SeedSpec struct {
|
||||||
|
Name string
|
||||||
|
Filter bson.M
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedSpecs 是 91dj_video -> hgmj 的播种清单。
|
||||||
|
//
|
||||||
|
// 只搬「内容」与「配置」,外加用户表的一个子集。刻意排除的几类:
|
||||||
|
// - 交易与资产:transaction / wallet / recharge_order / product_history /
|
||||||
|
// media_buy_record / pay_video_log / advance_order / backpack / video_coupon /
|
||||||
|
// fund_transfer_log / withdraw_order
|
||||||
|
// - 埋点与事件:vip_card_analytics_event(496万) / event(270万) / user_act /
|
||||||
|
// operation_log / ip_block / user_watch_record(474万)
|
||||||
|
// - IM 与 AI 生成内容:im_user(82万) / im_group / chat_message / ai_plaza(15.7万) /
|
||||||
|
// ai_un_dress / ai_change_face / ai_change_face_img / ai_text_to_* / ai_image_to_video
|
||||||
|
// - 用户行为与 UGC:comment(33.7万) / like(17.6万) / collect(9.8万) /
|
||||||
|
// media_bookshelf(8万) / user_follow / user_checkin / invitation / user_feedback
|
||||||
|
// - 运行时任务:task / export_task
|
||||||
|
//
|
||||||
|
// 排除它们的原因:用户表只搬 205 个 uid,这些按 uid 关联的表搬过去会产生大量
|
||||||
|
// 指向不存在用户的孤儿记录;埋点和事件是源站历史,对新站报表无意义。
|
||||||
|
var seedSpecs = []SeedSpec{
|
||||||
|
// ---- 内容 ----
|
||||||
|
{Name: "video_info"}, // 帖子与视频(newsType 区分 SP/COVER/SHORT/PIC/SEED_LINK)
|
||||||
|
{Name: "section_video"}, // 模块下的视频编排
|
||||||
|
{Name: "section"}, // 模块/专题
|
||||||
|
{Name: "media"}, // 短剧主表
|
||||||
|
{Name: "media_content"}, // 剧集内容
|
||||||
|
{Name: "media_tag"}, // 短剧标签
|
||||||
|
{Name: "media_tag_dimension"}, // 标签维度
|
||||||
|
{Name: "tag"}, // 视频标签
|
||||||
|
{Name: "ai_change_face_video"}, // AI 换脸视频模版(模版属于内容,不是用户生成结果)
|
||||||
|
|
||||||
|
// ---- 配置 ----
|
||||||
|
{Name: "sys_conf"}, // 系统通用配置
|
||||||
|
{Name: "module_conf"}, // 模块配置
|
||||||
|
{Name: "product"}, // 商品(会员卡)
|
||||||
|
{Name: "product_benefit"}, // 商品权益
|
||||||
|
{Name: "product_position"}, // 商品位
|
||||||
|
{Name: "vip_card_experiment"}, // 会员卡 A/B 实验配置(源库有 5 条,会在新站生效)
|
||||||
|
{Name: "currency"}, // 币种
|
||||||
|
{Name: "jingang"}, // 金刚区
|
||||||
|
{Name: "banner_jump"}, // Banner 跳转
|
||||||
|
{Name: "official_config"}, // 官方配置
|
||||||
|
{Name: "daily_task"}, // 每日任务
|
||||||
|
{Name: "once_task"}, // 一次性任务
|
||||||
|
{Name: "task_config"}, // 任务配置
|
||||||
|
{Name: "sign_task_config"}, // 签到任务配置
|
||||||
|
{Name: "checkin_config"}, // 签到配置
|
||||||
|
{Name: "checkin_prize"}, // 签到奖品
|
||||||
|
{Name: "integral_config"}, // 积分配置
|
||||||
|
{Name: "video_gold_coin_ser"}, // 金币视频配置
|
||||||
|
{Name: "activity_prize"}, // 活动奖品
|
||||||
|
{Name: "transaction_act"}, // 交易活动配置
|
||||||
|
{Name: "withdraw_channel"}, // 提现渠道
|
||||||
|
{Name: "withdraw_tariff"}, // 提现费率
|
||||||
|
{Name: "source"}, // 资源管理
|
||||||
|
{Name: "filter"}, // 敏感词
|
||||||
|
{Name: "reject_template"}, // 驳回模板
|
||||||
|
{Name: "location"}, // 地区
|
||||||
|
{Name: "announce"}, // 公告
|
||||||
|
{Name: "quick_search"}, // 快捷搜索
|
||||||
|
{Name: "ip_white"}, // IP 白名单(ip_block 是运行时封禁,不搬)
|
||||||
|
{Name: "admin"}, // 后台账号
|
||||||
|
{Name: "authority"}, // 权限
|
||||||
|
{Name: "auto_id"}, // 自增序列:必须搬,否则新站发号可能与已搬内容的 ID 冲撞
|
||||||
|
|
||||||
|
// ---- 用户子集 ----
|
||||||
|
// uid 在源库只连续分配到 205,之后直接跳到 100 万以上,
|
||||||
|
// 所以 <30000 实际就是这 205 个早期账号。
|
||||||
|
{Name: "user", Filter: bson.M{"uid": bson.M{"$lt": 30000}}},
|
||||||
|
}
|
||||||
@@ -0,0 +1,165 @@
|
|||||||
|
// dbseed 把源库的一批集合播种到目标库,用于开新站。
|
||||||
|
//
|
||||||
|
// 与 scripts/contentsync 的区别:contentsync 是每天跑的增量同步,只认 _id 水位线;
|
||||||
|
// dbseed 是一次性播种,可带过滤条件,按 _id 升序分批 upsert,中断后重跑幂等。
|
||||||
|
//
|
||||||
|
// 集合清单与过滤条件写在 collections.go,改范围改那里。
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"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"
|
||||||
|
)
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
var (
|
||||||
|
srcURI = flag.String("src", "", "源 URI")
|
||||||
|
srcDB = flag.String("src-db", "", "源库名")
|
||||||
|
dstURI = flag.String("dst", "", "目标 URI")
|
||||||
|
dstDB = flag.String("dst-db", "", "目标库名")
|
||||||
|
only = flag.String("only", "", "只跑指定集合,逗号分隔")
|
||||||
|
dryRun = flag.Bool("dry-run", false, "只统计不写入")
|
||||||
|
batchSize = flag.Int("batch", 500, "每批条数")
|
||||||
|
timeout = flag.Duration("timeout", 6*time.Hour, "整体超时")
|
||||||
|
)
|
||||||
|
flag.Parse()
|
||||||
|
if *srcURI == "" || *srcDB == "" || *dstURI == "" || *dstDB == "" {
|
||||||
|
fmt.Fprintln(os.Stderr, "必须指定 --src --src-db --dst --dst-db")
|
||||||
|
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())
|
||||||
|
dstCli, err := mongo.Connect(ctx, options.Client().ApplyURI(*dstURI))
|
||||||
|
if err != nil {
|
||||||
|
fatal("连接目标库失败: %v", err)
|
||||||
|
}
|
||||||
|
defer dstCli.Disconnect(context.Background())
|
||||||
|
|
||||||
|
src := srcCli.Database(*srcDB)
|
||||||
|
dst := dstCli.Database(*dstDB)
|
||||||
|
|
||||||
|
wanted := map[string]bool{}
|
||||||
|
for _, n := range strings.Split(*only, ",") {
|
||||||
|
if n = strings.TrimSpace(n); n != "" {
|
||||||
|
wanted[n] = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
started := time.Now()
|
||||||
|
var total int
|
||||||
|
failed := false
|
||||||
|
for _, spec := range seedSpecs {
|
||||||
|
if len(wanted) > 0 && !wanted[spec.Name] {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
n, err := seedOne(ctx, src, dst, spec, *dryRun, *batchSize)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Printf("[FAIL] %-26s %v\n", spec.Name, err)
|
||||||
|
failed = true
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
total += n
|
||||||
|
note := ""
|
||||||
|
if spec.Filter != nil {
|
||||||
|
note = " (带过滤)"
|
||||||
|
}
|
||||||
|
fmt.Printf("[ok] %-26s 写入 %7d%s\n", spec.Name, n, note)
|
||||||
|
}
|
||||||
|
fmt.Printf("\n合计写入 %d 条,耗时 %s\n", total, time.Since(started).Round(time.Second))
|
||||||
|
if failed {
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func seedOne(ctx context.Context, src, dst *mongo.Database, spec SeedSpec,
|
||||||
|
dryRun bool, batchSize int) (written int, err error) {
|
||||||
|
|
||||||
|
filter := bson.M{}
|
||||||
|
if spec.Filter != nil {
|
||||||
|
filter = spec.Filter
|
||||||
|
}
|
||||||
|
cur, err := src.Collection(spec.Name).Find(ctx, filter,
|
||||||
|
options.Find().SetSort(bson.D{{Key: "_id", Value: 1}}).
|
||||||
|
SetBatchSize(int32(batchSize)).SetNoCursorTimeout(true))
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
defer cur.Close(ctx)
|
||||||
|
|
||||||
|
target := dst.Collection(spec.Name)
|
||||||
|
buf := make([]mongo.WriteModel, 0, batchSize)
|
||||||
|
scanned := 0
|
||||||
|
lastLog := time.Now()
|
||||||
|
|
||||||
|
flush := func() error {
|
||||||
|
if len(buf) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
models := buf
|
||||||
|
buf = make([]mongo.WriteModel, 0, batchSize)
|
||||||
|
if dryRun {
|
||||||
|
written += len(models)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
res, e := target.BulkWrite(ctx, models, options.BulkWrite().SetOrdered(false))
|
||||||
|
if res != nil {
|
||||||
|
written += int(res.UpsertedCount) + int(res.ModifiedCount)
|
||||||
|
}
|
||||||
|
return e
|
||||||
|
}
|
||||||
|
|
||||||
|
for cur.Next(ctx) {
|
||||||
|
var doc bson.M
|
||||||
|
if e := cur.Decode(&doc); e != nil {
|
||||||
|
return written, e
|
||||||
|
}
|
||||||
|
scanned++
|
||||||
|
if time.Since(lastLog) >= 30*time.Second {
|
||||||
|
fmt.Printf(" %-26s 已扫描 %d 条…\n", spec.Name, scanned)
|
||||||
|
lastLog = time.Now()
|
||||||
|
}
|
||||||
|
id := doc["_id"]
|
||||||
|
delete(doc, "_id")
|
||||||
|
if len(doc) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// 整条替换:目标是新库,没有需要保留的本地字段。
|
||||||
|
// upsert 保证中断重跑幂等。
|
||||||
|
buf = append(buf, mongo.NewUpdateOneModel().
|
||||||
|
SetFilter(bson.M{"_id": id}).
|
||||||
|
SetUpdate(bson.M{"$set": doc}).
|
||||||
|
SetUpsert(true))
|
||||||
|
if len(buf) >= batchSize {
|
||||||
|
if e := flush(); e != nil {
|
||||||
|
return written, e
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if e := cur.Err(); e != nil {
|
||||||
|
return written, e
|
||||||
|
}
|
||||||
|
return written, flush()
|
||||||
|
}
|
||||||
|
|
||||||
|
var _ = primitive.NilObjectID
|
||||||
|
|
||||||
|
func fatal(f string, a ...any) {
|
||||||
|
fmt.Fprintf(os.Stderr, f+"\n", a...)
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user