corp-transfer/scripts/replace-corpid.js

596 lines
21 KiB
JavaScript
Raw Permalink Normal View History

#!/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);
});