596 lines
21 KiB
JavaScript
596 lines
21 KiB
JavaScript
#!/usr/bin/env node
|
||
/**
|
||
* replace-corpid
|
||
*
|
||
* 目的：
|
||
* - 默认扫描所有数据库下的 collection（默认排除 `config` 和 `local`，包含 `admin`）
|
||
* - 把字段名命中 `corpId/corpid/corp_id` 且值等于旧 corpId 的字段，替换为新 corpId
|
||
* - 对 `corp` collection 中命中的文档，补充/更新 `internal_agentid`
|
||
*
|
||
* 安全：
|
||
* - 默认 dry-run（不落库）
|
||
* - `--apply --yes` 才会写入
|
||
*/
|
||
|
||
const { MongoClient } = require("mongodb");
|
||
const dotenv = require("dotenv");
|
||
const fs = require("fs");
|
||
const path = require("path");
|
||
|
||
function isNonEmptyString(v) {
|
||
return typeof v === "string" && v.trim().length > 0;
|
||
}
|
||
|
||
function safeNumber(n, fallback) {
|
||
if (!Number.isFinite(n) || n <= 0) return fallback;
|
||
return n;
|
||
}
|
||
|
||
function loadDotEnv() {
|
||
const nodeEnv = process.env.NODE_ENV || "development";
|
||
const envFileName = `.env.${nodeEnv}`;
|
||
const plainEnvFileName = `env.${nodeEnv}`;
|
||
const explicitEnvFile = process.env.ENV_FILE || process.env.DOTENV_CONFIG_PATH || "";
|
||
const candidates = [];
|
||
|
||
if (isNonEmptyString(explicitEnvFile)) {
|
||
candidates.push(
|
||
path.isAbsolute(explicitEnvFile) ? explicitEnvFile : path.resolve(process.cwd(), explicitEnvFile),
|
||
path.isAbsolute(explicitEnvFile) ? explicitEnvFile : path.resolve(__dirname, explicitEnvFile)
|
||
);
|
||
}
|
||
|
||
candidates.push(
|
||
path.resolve(process.cwd(), envFileName),
|
||
path.resolve(__dirname, envFileName),
|
||
path.resolve(process.cwd(), plainEnvFileName),
|
||
path.resolve(__dirname, plainEnvFileName),
|
||
path.resolve(process.cwd(), ".env"),
|
||
path.resolve(__dirname, ".env")
|
||
);
|
||
|
||
const picked = candidates.find((p) => {
|
||
try {
|
||
return fs.existsSync(p);
|
||
} catch {
|
||
return false;
|
||
}
|
||
});
|
||
|
||
if (picked) {
|
||
dotenv.config({ path: picked });
|
||
console.log(`[env] loaded: ${picked}`);
|
||
return;
|
||
}
|
||
dotenv.config();
|
||
}
|
||
|
||
loadDotEnv();
|
||
|
||
function createDefaultOptions() {
|
||
return {
|
||
dbName: "",
|
||
corpIdFields: ["corpId", "corpid", "corp_id"],
|
||
collections: [],
|
||
batchDocs: 200,
|
||
skipDocs: 0,
|
||
limitDocs: 0,
|
||
sortById: true,
|
||
dryRun: true,
|
||
yes: false,
|
||
help: false,
|
||
reportZh: false,
|
||
logFile: "",
|
||
mongoUri: "",
|
||
mongoUriEnv: "",
|
||
mongoHost: "",
|
||
mongoPort: "",
|
||
mongoUser: "",
|
||
mongoPass: "",
|
||
mongoAuthDb: "",
|
||
sourceCorpId: "",
|
||
newCorpId: "",
|
||
internalAgentId: 1000092,
|
||
};
|
||
}
|
||
|
||
function parseArgs(argv) {
|
||
const args = argv.slice(2);
|
||
const opts = createDefaultOptions();
|
||
|
||
function takeValue(i) {
|
||
if (i + 1 >= args.length) return "";
|
||
return args[i + 1];
|
||
}
|
||
|
||
function setKeyValue(keyRaw, valueRaw) {
|
||
const key = String(keyRaw || "").replace(/^--/, "").trim().toLowerCase();
|
||
const value = valueRaw;
|
||
if (!key) return;
|
||
if (key === "dbname" || key === "db") opts.dbName = value;
|
||
else if (key === "collections")
|
||
opts.collections = String(value || "").split(",").map((s) => s.trim()).filter(Boolean);
|
||
else if (key === "corpidfields")
|
||
opts.corpIdFields = String(value || "").split(",").map((s) => s.trim()).filter(Boolean);
|
||
else if (key === "mongouri" || key === "mongourl") opts.mongoUri = value;
|
||
else if (key === "mongourienv") opts.mongoUriEnv = value;
|
||
else if (key === "mongohost") opts.mongoHost = value;
|
||
else if (key === "mongoport") opts.mongoPort = value;
|
||
else if (key === "mongouser") opts.mongoUser = value;
|
||
else if (key === "mongopass") opts.mongoPass = value;
|
||
else if (key === "mongoauthdb") opts.mongoAuthDb = value;
|
||
else if (key === "sourcecorpid" || key === "oldcorpid") opts.sourceCorpId = value;
|
||
else if (key === "newcorpid" || key === "targetcorpid") opts.newCorpId = value;
|
||
else if (key === "internalagentid" || key === "internal_agentid") opts.internalAgentId = Number(value);
|
||
else if (key === "batchdocs") opts.batchDocs = Number(value);
|
||
else if (key === "skipdocs" || key === "skip-docs") opts.skipDocs = Number(value);
|
||
else if (key === "limitdocs" || key === "limit-docs") opts.limitDocs = Number(value);
|
||
else if (key === "logfile" || key === "log_file") opts.logFile = value;
|
||
}
|
||
|
||
for (let i = 0; i < args.length; i++) {
|
||
const a = args[i];
|
||
if (!a) continue;
|
||
if (a === "--") continue;
|
||
|
||
if (typeof a === "string" && a.includes("=")) {
|
||
const [k, ...rest] = a.split("=");
|
||
setKeyValue(k, rest.join("="));
|
||
continue;
|
||
}
|
||
|
||
if (a === "--help" || a === "-h") opts.help = true;
|
||
else if (a === "--apply") opts.dryRun = false;
|
||
else if (a === "--yes") opts.yes = true;
|
||
else if (a === "--reportZh" || a === "--zh") opts.reportZh = true;
|
||
else if (a === "--dbName" || a === "--db") {
|
||
opts.dbName = takeValue(i);
|
||
i++;
|
||
} else if (a === "--collections") {
|
||
opts.collections = String(takeValue(i) || "").split(",").map((s) => s.trim()).filter(Boolean);
|
||
i++;
|
||
} else if (a === "--corpIdFields") {
|
||
opts.corpIdFields = String(takeValue(i) || "").split(",").map((s) => s.trim()).filter(Boolean);
|
||
i++;
|
||
} else if (a === "--mongoUri" || a === "--mongoUrl" || a === "--mongoURL") {
|
||
opts.mongoUri = takeValue(i);
|
||
i++;
|
||
} else if (a === "--mongoUriEnv") {
|
||
opts.mongoUriEnv = takeValue(i);
|
||
i++;
|
||
} else if (a === "--mongoHost") {
|
||
opts.mongoHost = takeValue(i);
|
||
i++;
|
||
} else if (a === "--mongoPort") {
|
||
opts.mongoPort = takeValue(i);
|
||
i++;
|
||
} else if (a === "--mongoUser") {
|
||
opts.mongoUser = takeValue(i);
|
||
i++;
|
||
} else if (a === "--mongoPass") {
|
||
opts.mongoPass = takeValue(i);
|
||
i++;
|
||
} else if (a === "--mongoAuthDb") {
|
||
opts.mongoAuthDb = takeValue(i);
|
||
i++;
|
||
} else if (a === "--sourceCorpId" || a === "--oldCorpId") {
|
||
opts.sourceCorpId = takeValue(i);
|
||
i++;
|
||
} else if (a === "--newCorpId" || a === "--targetCorpId") {
|
||
opts.newCorpId = takeValue(i);
|
||
i++;
|
||
} else if (a === "--internalAgentId") {
|
||
opts.internalAgentId = Number(takeValue(i));
|
||
i++;
|
||
} else if (a === "--batchDocs") {
|
||
opts.batchDocs = Number(takeValue(i));
|
||
i++;
|
||
} else if (a === "--skipDocs" || a === "--skip-docs") {
|
||
opts.skipDocs = Number(takeValue(i));
|
||
i++;
|
||
} else if (a === "--limitDocs" || a === "--limit-docs") {
|
||
opts.limitDocs = Number(takeValue(i));
|
||
i++;
|
||
} else if (a === "--logFile" || a === "--log-file") {
|
||
opts.logFile = takeValue(i);
|
||
i++;
|
||
}
|
||
}
|
||
|
||
return opts;
|
||
}
|
||
|
||
function applyEnvFallbacks(opts) {
|
||
if (!opts.mongoUri) opts.mongoUri = process.env.MONGO_URI || process.env.GK_MONGO_URI || "";
|
||
if (!opts.sourceCorpId && process.env.TARGET_CORP_ID) opts.sourceCorpId = process.env.TARGET_CORP_ID;
|
||
if (!opts.newCorpId && process.env.TOKEN_CORP_ID) opts.newCorpId = process.env.TOKEN_CORP_ID;
|
||
if ((!opts.internalAgentId || !Number.isFinite(opts.internalAgentId)) && process.env.INTERNAL_AGENT_ID) {
|
||
opts.internalAgentId = Number(process.env.INTERNAL_AGENT_ID);
|
||
}
|
||
}
|
||
|
||
function die(msg) {
|
||
console.error(msg);
|
||
process.exit(1);
|
||
}
|
||
|
||
function formatTimestampForFile(d = new Date()) {
|
||
const pad = (n) => String(n).padStart(2, "0");
|
||
return [
|
||
d.getFullYear(),
|
||
pad(d.getMonth() + 1),
|
||
pad(d.getDate()),
|
||
"-",
|
||
pad(d.getHours()),
|
||
pad(d.getMinutes()),
|
||
pad(d.getSeconds()),
|
||
].join("");
|
||
}
|
||
|
||
function maskSensitiveValue(key, value) {
|
||
const k = String(key || "").toLowerCase();
|
||
if (value == null) return value;
|
||
if (k.includes("pass")) return "***";
|
||
return value;
|
||
}
|
||
|
||
function sanitizeOptionsForLog(opts) {
|
||
const out = {};
|
||
for (const [k, v] of Object.entries(opts || {})) {
|
||
out[k] = maskSensitiveValue(k, v);
|
||
}
|
||
return out;
|
||
}
|
||
|
||
function initFileLogger(opts) {
|
||
const scriptDir = path.dirname(path.resolve(process.argv[1] || __filename));
|
||
const logsDir = path.resolve(scriptDir, "logs");
|
||
fs.mkdirSync(logsDir, { recursive: true });
|
||
const filePath = isNonEmptyString(opts.logFile)
|
||
? path.resolve(process.cwd(), opts.logFile)
|
||
: path.join(logsDir, `corpid-${formatTimestampForFile()}.log`);
|
||
|
||
fs.mkdirSync(path.dirname(filePath), { recursive: true });
|
||
fs.writeFileSync(filePath, "", "utf8");
|
||
|
||
const rawStdout = process.stdout.write.bind(process.stdout);
|
||
const rawStderr = process.stderr.write.bind(process.stderr);
|
||
const stream = fs.createWriteStream(filePath, { flags: "a", encoding: "utf8" });
|
||
|
||
const writeLine = (chunk) => {
|
||
try {
|
||
stream.write(typeof chunk === "string" ? chunk : chunk.toString("utf8"));
|
||
} catch {
|
||
// ignore
|
||
}
|
||
};
|
||
|
||
process.stdout.write = function patchedStdout(chunk, encoding, cb) {
|
||
writeLine(chunk);
|
||
return rawStdout(chunk, encoding, cb);
|
||
};
|
||
process.stderr.write = function patchedStderr(chunk, encoding, cb) {
|
||
writeLine(chunk);
|
||
return rawStderr(chunk, encoding, cb);
|
||
};
|
||
|
||
stream.write(`[log] started at ${new Date().toISOString()}\n`);
|
||
stream.write(`[log] cwd=${process.cwd()}\n`);
|
||
stream.write(`[log] scriptDir=${scriptDir}\n`);
|
||
stream.write(`[log] argv=${JSON.stringify(process.argv)}\n`);
|
||
stream.write(`[log] options=${JSON.stringify(sanitizeOptionsForLog(opts))}\n`);
|
||
console.log(`[log] writing to ${filePath}`);
|
||
return filePath;
|
||
}
|
||
|
||
function normalizeKeyName(key) {
|
||
return String(key || "")
|
||
.trim()
|
||
.toLowerCase()
|
||
.replace(/[_\-\s]/g, "");
|
||
}
|
||
|
||
function isCorpFieldKey(key, corpIdFields) {
|
||
const normalized = normalizeKeyName(key);
|
||
if (!normalized) return false;
|
||
return (corpIdFields || []).some((field) => normalizeKeyName(field) === normalized);
|
||
}
|
||
|
||
function isPlainObject(v) {
|
||
if (!v || typeof v !== "object") return false;
|
||
if (Array.isArray(v)) return false;
|
||
return Object.prototype.toString.call(v) === "[object Object]";
|
||
}
|
||
|
||
function docContainsCorpId(value, { corpIdFields, corpIds }) {
|
||
const corpIdSet = new Set((corpIds || []).filter(Boolean));
|
||
let found = false;
|
||
|
||
function walk(node, keyForContext) {
|
||
if (found || node == null) return;
|
||
if (typeof node === "string") {
|
||
if (isCorpFieldKey(keyForContext, corpIdFields) && corpIdSet.has(node)) found = true;
|
||
return;
|
||
}
|
||
if (Array.isArray(node)) {
|
||
for (const item of node) walk(item, keyForContext);
|
||
return;
|
||
}
|
||
if (isPlainObject(node)) {
|
||
for (const [k, v] of Object.entries(node)) walk(v, k);
|
||
}
|
||
}
|
||
|
||
walk(value, "");
|
||
return found;
|
||
}
|
||
|
||
function migrateDocInPlace(doc, { corpIdFields, sourceCorpId, newCorpId, collectionName, internalAgentId }) {
|
||
let corpReplaced = 0;
|
||
let internalAgentUpdated = 0;
|
||
|
||
function walk(node, keyForContext, parent, parentKey) {
|
||
if (node == null) return;
|
||
if (typeof node === "string") {
|
||
if (isCorpFieldKey(keyForContext, corpIdFields) && node === sourceCorpId) {
|
||
parent[parentKey] = newCorpId;
|
||
corpReplaced++;
|
||
}
|
||
return;
|
||
}
|
||
if (Array.isArray(node)) {
|
||
for (let i = 0; i < node.length; i++) {
|
||
const item = node[i];
|
||
if (typeof item === "string") {
|
||
if (isCorpFieldKey(keyForContext, corpIdFields) && item === sourceCorpId) {
|
||
node[i] = newCorpId;
|
||
corpReplaced++;
|
||
}
|
||
} else {
|
||
walk(item, keyForContext, node, i);
|
||
}
|
||
}
|
||
return;
|
||
}
|
||
if (isPlainObject(node)) {
|
||
for (const [k, v] of Object.entries(node)) walk(v, k, node, k);
|
||
}
|
||
}
|
||
|
||
if (isPlainObject(doc)) {
|
||
for (const [k, v] of Object.entries(doc)) walk(v, k, doc, k);
|
||
}
|
||
|
||
if (collectionName === "corp") {
|
||
const belongsToCorp = corpReplaced > 0 || docContainsCorpId(doc, {
|
||
corpIdFields,
|
||
corpIds: [sourceCorpId, newCorpId],
|
||
});
|
||
if (belongsToCorp && doc.internal_agentid !== internalAgentId) {
|
||
doc.internal_agentid = internalAgentId;
|
||
internalAgentUpdated++;
|
||
}
|
||
}
|
||
|
||
return {
|
||
corpReplaced,
|
||
internalAgentUpdated,
|
||
touched: corpReplaced > 0 || internalAgentUpdated > 0,
|
||
};
|
||
}
|
||
|
||
async function listCollections(db) {
|
||
const cols = await db.listCollections().toArray();
|
||
return cols.map((c) => c.name);
|
||
}
|
||
|
||
async function listDatabaseNames(client) {
|
||
const adminDb = client.db().admin();
|
||
const res = await adminDb.listDatabases();
|
||
const dbs = Array.isArray(res?.databases) ? res.databases : [];
|
||
const excluded = new Set(["config", "local"]);
|
||
return dbs
|
||
.map((db) => db?.name)
|
||
.filter((name) => isNonEmptyString(name) && !excluded.has(name));
|
||
}
|
||
|
||
async function buildMongoUrl(opts) {
|
||
let mongoUrl = opts.mongoUri;
|
||
if (!isNonEmptyString(mongoUrl) && isNonEmptyString(opts.mongoUriEnv) && isNonEmptyString(process.env[opts.mongoUriEnv])) {
|
||
mongoUrl = process.env[opts.mongoUriEnv];
|
||
}
|
||
if (isNonEmptyString(mongoUrl)) return mongoUrl;
|
||
|
||
const host = opts.mongoHost || process.env.MONGO_HOST || process.env.CONFIG_DB_HOST;
|
||
const port = opts.mongoPort || process.env.MONGO_PORT || process.env.CONFIG_DB_PORT || "27017";
|
||
const username = opts.mongoUser || process.env.MONGO_USER || process.env.CONFIG_DB_USERNAME;
|
||
const password = opts.mongoPass || process.env.MONGO_PASS || process.env.CONFIG_DB_PASSWORD;
|
||
const authDb = opts.mongoAuthDb || process.env.MONGO_AUTH_DB || "admin";
|
||
|
||
if (!host || !username || !password) {
|
||
die("缺少 MongoDB 连接信息：请传 --mongoUri，或传 --mongoHost/--mongoUser/--mongoPass。");
|
||
}
|
||
|
||
return `mongodb://${encodeURIComponent(username)}:${encodeURIComponent(password)}@${host}:${port}/${authDb}`;
|
||
}
|
||
|
||
async function main() {
|
||
const opts = parseArgs(process.argv);
|
||
applyEnvFallbacks(opts);
|
||
initFileLogger(opts);
|
||
|
||
if (opts.help) {
|
||
console.log(`replace-corpid
|
||
|
||
用法：
|
||
node corpid.js --apply --yes
|
||
|
||
常用参数：
|
||
--dbName <db> 仅处理单个数据库；不传则扫描全部库（默认排除 config/local，包含 admin）
|
||
--collections a,b,c 仅处理指定 collection；不传则扫描命中数据库下全部 collection
|
||
--sourceCorpId <corpId> 旧 corpId（默认取 TARGET_CORP_ID）
|
||
--newCorpId <corpId> 新 corpId（默认取 TOKEN_CORP_ID）
|
||
--internalAgentId <id> corp 集合补 internal_agentid（默认 1000092）
|
||
--apply --yes 真正写库；默认 dry-run
|
||
--logFile <path> 指定日志文件
|
||
--limitDocs <n> 每个 collection 最多处理 n 条
|
||
--skipDocs <n> 每个 collection 跳过前 n 条
|
||
`);
|
||
return;
|
||
}
|
||
|
||
opts.batchDocs = safeNumber(opts.batchDocs, 200);
|
||
opts.skipDocs = Number.isFinite(opts.skipDocs) && opts.skipDocs > 0 ? Math.floor(opts.skipDocs) : 0;
|
||
opts.limitDocs = Number.isFinite(opts.limitDocs) && opts.limitDocs > 0 ? Math.floor(opts.limitDocs) : 0;
|
||
opts.internalAgentId = Number.isFinite(opts.internalAgentId) ? Math.floor(opts.internalAgentId) : 1000092;
|
||
|
||
if (!isNonEmptyString(opts.sourceCorpId)) die("缺少旧 corpId：请传 --sourceCorpId，或在环境里提供 TARGET_CORP_ID");
|
||
if (!isNonEmptyString(opts.newCorpId)) die("缺少新 corpId：请传 --newCorpId，或在环境里提供 TOKEN_CORP_ID");
|
||
if (opts.sourceCorpId === opts.newCorpId) die("旧 corpId 和新 corpId 相同，无需执行。");
|
||
|
||
if (!opts.dryRun && !opts.yes) {
|
||
die("该脚本会直接回写原 collection。请确认后加上 --yes 再执行。");
|
||
}
|
||
|
||
const mongoUrl = await buildMongoUrl(opts);
|
||
const client = new MongoClient(mongoUrl, {
|
||
serverSelectionTimeoutMS: 30000,
|
||
connectTimeoutMS: 30000,
|
||
});
|
||
|
||
console.log("[step] 正在连接 MongoDB...");
|
||
await client.connect();
|
||
console.log("[step] MongoDB 已连接");
|
||
|
||
try {
|
||
const dbNames = isNonEmptyString(opts.dbName) ? [opts.dbName] : await listDatabaseNames(client);
|
||
console.log(`[step] 数据库列表（共 ${dbNames.length} 个）: ${dbNames.join(", ") || "（空）"}`);
|
||
console.log(
|
||
`[step] 本次执行范围: ${isNonEmptyString(opts.dbName) ? `仅指定数据库 -> ${opts.dbName}` : "未指定 --dbName，将扫描全部库（默认排除 config/local，包含 admin）"}`
|
||
);
|
||
console.log(
|
||
`[step] collection 范围: ${opts.collections?.length ? `仅指定 collection -> ${opts.collections.join(", ")}` : "未指定 --collections，将扫描命中数据库下全部 collection"}`
|
||
);
|
||
console.log(
|
||
`[step] oldCorpId=${opts.sourceCorpId} newCorpId=${opts.newCorpId} internal_agentid=${opts.internalAgentId} mode=${opts.dryRun ? "dry-run" : "apply"}`
|
||
);
|
||
|
||
let scannedDatabases = 0;
|
||
let scannedCollections = 0;
|
||
let processedTotal = 0;
|
||
let touchedDocsTotal = 0;
|
||
let wroteDocsTotal = 0;
|
||
let corpReplacedTotal = 0;
|
||
let internalAgentUpdatedTotal = 0;
|
||
for (const dbName of dbNames) {
|
||
scannedDatabases++;
|
||
const db = client.db(dbName);
|
||
const names = await listCollections(db);
|
||
const selected = opts.collections?.length ? opts.collections : names;
|
||
console.log(`[step] [db=${dbName}] collection 列表（共 ${names.length} 个）: ${names.join(", ") || "（空）"}`);
|
||
|
||
for (const colName of selected) {
|
||
scannedCollections++;
|
||
if (opts.collections?.length && !names.includes(colName)) {
|
||
console.log(`[${dbName}.${colName}] collection 不存在（或无权限看到），将跳过。`);
|
||
continue;
|
||
}
|
||
|
||
const col = db.collection(colName);
|
||
let cursor = col.find({}, { batchSize: opts.batchDocs });
|
||
if (opts.sortById) cursor = cursor.sort({ _id: 1 });
|
||
if (opts.skipDocs) cursor = cursor.skip(opts.skipDocs);
|
||
if (opts.limitDocs) cursor = cursor.limit(opts.limitDocs);
|
||
|
||
let processed = 0;
|
||
let touchedDocs = 0;
|
||
let wroteDocs = 0;
|
||
let corpReplaced = 0;
|
||
let internalAgentUpdated = 0;
|
||
let batchIndex = 0;
|
||
|
||
console.log(`[step] [${dbName}.${colName}] 开始遍历文档（batchDocs=${opts.batchDocs} limitDocs=${opts.limitDocs || "无限制"} skipDocs=${opts.skipDocs}）...`);
|
||
while (await cursor.hasNext()) {
|
||
const batch = [];
|
||
while (batch.length < opts.batchDocs && (await cursor.hasNext())) {
|
||
batch.push(await cursor.next());
|
||
}
|
||
if (batch.length === 0) break;
|
||
batchIndex++;
|
||
console.log(`[step] [${dbName}.${colName}] batch#${batchIndex} 读取文档 ${batch.length} 条（已处理 ${processed} 条）`);
|
||
|
||
const changedDocs = [];
|
||
for (const doc of batch) {
|
||
const result = migrateDocInPlace(doc, {
|
||
corpIdFields: opts.corpIdFields,
|
||
sourceCorpId: opts.sourceCorpId,
|
||
newCorpId: opts.newCorpId,
|
||
collectionName: colName,
|
||
internalAgentId: opts.internalAgentId,
|
||
});
|
||
corpReplaced += result.corpReplaced;
|
||
internalAgentUpdated += result.internalAgentUpdated;
|
||
if (result.touched) {
|
||
touchedDocs++;
|
||
changedDocs.push(doc);
|
||
}
|
||
}
|
||
|
||
if (!opts.dryRun && changedDocs.length > 0) {
|
||
console.log(`[step] [${dbName}.${colName}] batch#${batchIndex} 正在写入 ${changedDocs.length} 条变更...`);
|
||
const ops = changedDocs.map((doc) => ({
|
||
replaceOne: {
|
||
filter: { _id: doc._id },
|
||
replacement: doc,
|
||
upsert: true,
|
||
},
|
||
}));
|
||
await col.bulkWrite(ops, { ordered: false });
|
||
wroteDocs += changedDocs.length;
|
||
console.log(`[step] [${dbName}.${colName}] batch#${batchIndex} 写入完成`);
|
||
}
|
||
|
||
processed += batch.length;
|
||
}
|
||
|
||
console.log(
|
||
`[${dbName}.${colName}] done. processed=${processed} touchedDocs=${touchedDocs} wrote=${wroteDocs} corpReplaced=${corpReplaced} internalAgentUpdated=${internalAgentUpdated}`
|
||
);
|
||
|
||
processedTotal += processed;
|
||
touchedDocsTotal += touchedDocs;
|
||
wroteDocsTotal += wroteDocs;
|
||
corpReplacedTotal += corpReplaced;
|
||
internalAgentUpdatedTotal += internalAgentUpdated;
|
||
}
|
||
}
|
||
|
||
console.log(
|
||
`[mongo] summary: scannedDatabases=${scannedDatabases} scannedCollections=${scannedCollections} processed=${processedTotal} touchedDocs=${touchedDocsTotal} wrote=${wroteDocsTotal} corpReplaced=${corpReplacedTotal} internalAgentUpdated=${internalAgentUpdatedTotal}`
|
||
);
|
||
|
||
if (opts.reportZh) {
|
||
console.log("\n中文统计摘要：");
|
||
console.log(`- 数据库范围：${isNonEmptyString(opts.dbName) ? opts.dbName : "全部库（默认排除 config/local，包含 admin）"}`);
|
||
console.log(`- 旧 corpId：${opts.sourceCorpId}`);
|
||
console.log(`- 新 corpId：${opts.newCorpId}`);
|
||
console.log(`- corp 集合 internal_agentid：${opts.internalAgentId}`);
|
||
console.log(`- 运行模式：${opts.dryRun ? "dry-run（不落库）" : "apply（回写原 collection）"}`);
|
||
console.log(`- 扫描数据库数：${scannedDatabases}`);
|
||
console.log(`- 扫描 collection 数：${scannedCollections}`);
|
||
console.log(`- 处理文档数：${processedTotal}`);
|
||
console.log(`- 命中文档数：${touchedDocsTotal}`);
|
||
console.log(`- 实际写入文档数：${wroteDocsTotal}`);
|
||
console.log(`- corpId/corpid/corp_id 替换次数：${corpReplacedTotal}`);
|
||
console.log(`- internal_agentid 更新次数：${internalAgentUpdatedTotal}`);
|
||
}
|
||
} finally {
|
||
await client.close().catch(() => {});
|
||
}
|
||
}
|
||
|
||
main().catch((e) => {
|
||
console.error(e?.stack || e?.message || String(e));
|
||
process.exit(1);
|
||
});
|