yeyinly/MoonTVPlus
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

数据迁移并发提升性能

Browse Source
This commit is contained in:
mtvpls
2026-02-24 21:03:18 +08:00
parent ef5656cde5
commit ac68bca649
5 changed files with 557 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/api/admin/data_migration/import/route.ts
+246 -157
View File
@@ -8,6 +8,7 @@ import { getAuthInfoFromCookie } from '@/lib/auth';
import { configSelfCheck, setCachedConfig } from '@/lib/config';
import { SimpleCrypto } from '@/lib/crypto';
import { db } from '@/lib/db';
import { updateProgress, clearProgress } from '../progress/route';
export const runtime = 'nodejs';
@@ -78,6 +79,7 @@ export async function POST(req: NextRequest) {
}
// 开始导入数据 - 先清空现有数据
updateProgress(authInfo.username, 'import', 'clearing', 0, 1, '正在清空现有数据...');
await db.clearAllData();
// 额外清除所有V2用户（clearAllData可能只清除旧版用户）
@@ -109,185 +111,267 @@ export async function POST(req: NextRequest) {
const userCount = Object.keys(userData).length;
console.log(`准备导入 ${userCount} 个用户的数据`);
updateProgress(authInfo.username, 'import', 'importing', 0, userCount, '开始导入用户数据...');
// 分块处理用户，每批处理数量可通过环境变量配置
const CHUNK_SIZE = parseInt(process.env.DATA_MIGRATION_CHUNK_SIZE || '10', 10);
const usernames = Object.keys(userData);
let importedCount = 0;
for (const username in userData) {
const user = userData[username];
// 为所有有passwordV2的用户创建user:info
if (user.passwordV2) {
const userV2 = usersV2Map.get(username) as any;
for (let i = 0; i < usernames.length; i += CHUNK_SIZE) {
const chunk = usernames.slice(i, i + CHUNK_SIZE);
console.log(`处理第 ${Math.floor(i / CHUNK_SIZE) + 1} 批用户 (${chunk.length} 个)`);
updateProgress(
authInfo.username,
'import',
'importing',
importedCount,
userCount,
`正在导入用户数据 (${importedCount}/${userCount})...`
);
// 确定角色：站长为owner，其他用户从usersV2获取或默认为user
let role: 'owner' | 'admin' | 'user' = 'user';
if (username === process.env.USERNAME) {
role = 'owner';
} else if (userV2) {
role = userV2.role === 'owner' ? 'user' : userV2.role;
}
// 并行导入当前批次的用户
const importPromises = chunk.map(async (username) => {
try {
const user = userData[username];
// 数据批处理大小（用于播放记录、收藏夹等）
const DATA_BATCH_SIZE = parseInt(process.env.DATA_MIGRATION_CHUNK_SIZE || '10', 10);
const createdAt = userV2?.created_at || Date.now();
// 为所有有passwordV2的用户创建user:info
if (user.passwordV2) {
const userV2 = usersV2Map.get(username) as any;
// 根据存储类型使用不同的导入方法
if (storageType === 'd1') {
// D1 存储：使用 createUserWithHashedPassword 方法
try {
if (typeof storage.createUserWithHashedPassword === 'function') {
await storage.createUserWithHashedPassword(
username,
user.passwordV2, // 已经是hash过的密码
role,
createdAt,
userV2?.tags,
userV2?.oidcSub,
userV2?.enabledApis,
userV2?.banned
);
importedCount++;
console.log(`用户 ${username} 导入成功 (D1)`);
// 确定角色：站长为owner，其他用户从usersV2获取或默认为user
let role: 'owner' | 'admin' | 'user' = 'user';
if (username === process.env.USERNAME) {
role = 'owner';
} else if (userV2) {
role = userV2.role === 'owner' ? 'user' : userV2.role;
}
const createdAt = userV2?.created_at || Date.now();
// 根据存储类型使用不同的导入方法
if (storageType === 'd1') {
// D1 存储：使用 createUserWithHashedPassword 方法
if (typeof storage.createUserWithHashedPassword === 'function') {
await storage.createUserWithHashedPassword(
username,
user.passwordV2,
role,
createdAt,
userV2?.tags,
userV2?.oidcSub,
userV2?.enabledApis,
userV2?.banned
);
console.log(`用户 ${username} 导入成功 (D1)`);
} else {
console.error(`D1 storage 缺少 createUserWithHashedPassword 方法`);
return false;
}
} else if (storageType === 'postgres') {
// Postgres 存储：使用 createUserWithHashedPassword 方法
if (typeof storage.createUserWithHashedPassword === 'function') {
await storage.createUserWithHashedPassword(
username,
user.passwordV2,
role,
createdAt,
userV2?.tags,
userV2?.oidcSub,
userV2?.enabledApis,
userV2?.banned
);
console.log(`用户 ${username} 导入成功 (Postgres)`);
} else {
console.error(`Postgres storage 缺少 createUserWithHashedPassword 方法`);
return false;
}
} else {
console.error(`D1 storage 缺少 createUserWithHashedPassword 方法`);
}
} catch (err) {
console.error(`导入用户 ${username} 失败:`, err);
}
} else if (storageType === 'postgres') {
// Postgres 存储：使用 createUserWithHashedPassword 方法
try {
if (typeof storage.createUserWithHashedPassword === 'function') {
await storage.createUserWithHashedPassword(
username,
user.passwordV2, // 已经是hash过的密码
// Redis 存储：直接设置用户信息
const userInfoKey = `user:${username}:info`;
const userInfo: Record<string, string> = {
role,
createdAt,
userV2?.tags,
userV2?.oidcSub,
userV2?.enabledApis,
userV2?.banned
);
importedCount++;
console.log(`用户 ${username} 导入成功 (Postgres)`);
} else {
console.error(`Postgres storage 缺少 createUserWithHashedPassword 方法`);
banned: String(userV2?.banned || false),
password: user.passwordV2,
created_at: createdAt.toString(),
};
if (userV2?.tags && userV2.tags.length > 0) {
userInfo.tags = JSON.stringify(userV2.tags);
}
if (userV2?.oidcSub) {
userInfo.oidcSub = userV2.oidcSub;
}
if (userV2?.enabledApis && userV2.enabledApis.length > 0) {
userInfo.enabledApis = JSON.stringify(userV2.enabledApis);
}
await storage.withRetry(() => storage.client.hSet(userInfoKey, userInfo));
await storage.withRetry(() => storage.client.zAdd('user:list', {
score: createdAt,
value: username,
}));
if (userV2?.oidcSub) {
const oidcSubKey = `oidc:sub:${userV2.oidcSub}`;
await storage.withRetry(() => storage.client.set(oidcSubKey, username));
}
console.log(`用户 ${username} 导入成功 (Redis)`);
}
} catch (err) {
console.error(`导入用户 ${username} 失败:`, err);
}
} else {
// Redis 存储：直接设置用户信息
const userInfoKey = `user:${username}:info`;
const userInfo: Record<string, string> = {
role,
banned: String(userV2?.banned || false),
password: user.passwordV2, // 已经是hash过的密码，直接使用
created_at: createdAt.toString(),
};
if (userV2?.tags && userV2.tags.length > 0) {
userInfo.tags = JSON.stringify(userV2.tags);
} else {
console.log(`跳过用户 ${username}：没有passwordV2`);
return false;
}
if (userV2?.oidcSub) {
userInfo.oidcSub = userV2.oidcSub;
}
// 并行导入用户的各类数据
await Promise.all([
// 导入播放记录（批量）
(async () => {
if (user.playRecords) {
const entries = Object.entries(user.playRecords);
// 使用配置的批处理大小
for (let j = 0; j < entries.length; j += DATA_BATCH_SIZE) {
const batch = entries.slice(j, j + DATA_BATCH_SIZE);
await Promise.all(
batch.map(([key, record]) =>
(db as any).storage.setPlayRecord(username, key, record)
)
);
}
}
})(),
if (userV2?.enabledApis && userV2.enabledApis.length > 0) {
userInfo.enabledApis = JSON.stringify(userV2.enabledApis);
}
// 导入收藏夹（批量）
(async () => {
if (user.favorites) {
const entries = Object.entries(user.favorites);
for (let j = 0; j < entries.length; j += DATA_BATCH_SIZE) {
const batch = entries.slice(j, j + DATA_BATCH_SIZE);
await Promise.all(
batch.map(([key, favorite]) =>
(db as any).storage.setFavorite(username, key, favorite)
)
);
}
}
})(),
// 使用storage.withRetry直接设置用户信息
await storage.withRetry(() => storage.client.hSet(userInfoKey, userInfo));
// 导入搜索历史（批量）
(async () => {
if (user.searchHistory && Array.isArray(user.searchHistory)) {
const reversed = user.searchHistory.reverse();
for (let j = 0; j < reversed.length; j += DATA_BATCH_SIZE) {
const batch = reversed.slice(j, j + DATA_BATCH_SIZE);
await Promise.all(
batch.map(keyword => db.addSearchHistory(username, keyword))
);
}
}
})(),
// 添加到用户列表
await storage.withRetry(() => storage.client.zAdd('user:list', {
score: createdAt,
value: username,
}));
// 导入跳过片头片尾配置（批量）
(async () => {
if (user.skipConfigs) {
const entries = Object.entries(user.skipConfigs);
for (let j = 0; j < entries.length; j += DATA_BATCH_SIZE) {
const batch = entries.slice(j, j + DATA_BATCH_SIZE);
await Promise.all(
batch.map(([key, skipConfig]) => {
const [source, id] = key.split('+');
if (source && id) {
return db.setSkipConfig(username, source, id, skipConfig as any);
}
return Promise.resolve();
})
);
}
}
})(),
// 如果有oidcSub，创建映射
if (userV2?.oidcSub) {
const oidcSubKey = `oidc:sub:${userV2.oidcSub}`;
await storage.withRetry(() => storage.client.set(oidcSubKey, username));
}
// 导入音乐播放记录（批量）
(async () => {
if (user.musicPlayRecords) {
const entries = Object.entries(user.musicPlayRecords);
for (let j = 0; j < entries.length; j += DATA_BATCH_SIZE) {
const batch = entries.slice(j, j + DATA_BATCH_SIZE);
await Promise.all(
batch.map(([key, record]) => {
const [platform, id] = key.split('+');
if (platform && id) {
return db.saveMusicPlayRecord(username, platform, id, record as any);
}
return Promise.resolve();
})
);
}
}
})(),
importedCount++;
console.log(`用户 ${username} 导入成功 (Redis)`);
// 导入音乐歌单
(async () => {
if (user.musicPlaylists && Array.isArray(user.musicPlaylists)) {
for (const playlist of user.musicPlaylists) {
await db.createMusicPlaylist(username, {
id: playlist.id,
name: playlist.name,
description: playlist.description,
cover: playlist.cover,
});
// 批量导入歌单中的歌曲
if (playlist.songs && Array.isArray(playlist.songs)) {
for (let j = 0; j < playlist.songs.length; j += DATA_BATCH_SIZE) {
const batch = playlist.songs.slice(j, j + DATA_BATCH_SIZE);
await Promise.all(
batch.map(song =>
db.addSongToPlaylist(playlist.id, {
platform: song.platform,
id: song.id,
name: song.name,
artist: song.artist,
album: song.album,
pic: song.pic,
duration: song.duration || 0,
})
)
);
}
}
}
}
})()
]);
return true;
} catch (error) {
console.error(`导入用户 ${username} 失败:`, error);
return false;
}
} else {
console.log(`跳过用户 ${username}：没有passwordV2`);
}
});
// 导入播放记录
if (user.playRecords) {
for (const [key, record] of Object.entries(user.playRecords)) {
await (db as any).storage.setPlayRecord(username, key, record);
}
}
// 等待当前批次完成
const results = await Promise.all(importPromises);
importedCount += results.filter(r => r).length;
// 导入收藏夹
if (user.favorites) {
for (const [key, favorite] of Object.entries(user.favorites)) {
await (db as any).storage.setFavorite(username, key, favorite);
}
}
// 导入搜索历史
if (user.searchHistory && Array.isArray(user.searchHistory)) {
for (const keyword of user.searchHistory.reverse()) { // 反转以保持顺序
await db.addSearchHistory(username, keyword);
}
}
// 导入跳过片头片尾配置
if (user.skipConfigs) {
for (const [key, skipConfig] of Object.entries(user.skipConfigs)) {
const [source, id] = key.split('+');
if (source && id) {
await db.setSkipConfig(username, source, id, skipConfig as any);
}
}
}
// 导入音乐播放记录
if (user.musicPlayRecords) {
for (const [key, record] of Object.entries(user.musicPlayRecords)) {
const [platform, id] = key.split('+');
if (platform && id) {
await db.saveMusicPlayRecord(username, platform, id, record as any);
}
}
}
// 导入音乐歌单
if (user.musicPlaylists && Array.isArray(user.musicPlaylists)) {
for (const playlist of user.musicPlaylists) {
// 创建歌单
await db.createMusicPlaylist(username, {
id: playlist.id,
name: playlist.name,
description: playlist.description,
cover: playlist.cover,
});
// 导入歌单中的歌曲
if (playlist.songs && Array.isArray(playlist.songs)) {
for (const song of playlist.songs) {
await db.addSongToPlaylist(playlist.id, {
platform: song.platform,
id: song.id,
name: song.name,
artist: song.artist,
album: song.album,
pic: song.pic,
duration: song.duration || 0,
});
}
}
}
}
console.log(`已完成 ${importedCount}/${userCount} 个用户`);
updateProgress(
authInfo.username,
'import',
'importing',
importedCount,
userCount,
`已导入 ${importedCount}/${userCount} 个用户`
);
}
console.log(`成功导入 ${importedCount} 个用户的user:info`);
updateProgress(authInfo.username, 'import', 'completed', importedCount, userCount, '导入完成！');
setTimeout(() => clearProgress(authInfo.username, 'import'), 3000);
return NextResponse.json({
message: '数据导入成功',
@@ -299,6 +383,11 @@ export async function POST(req: NextRequest) {
} catch (error) {
console.error('数据导入失败:', error);
// 清除进度信息
const authInfo = getAuthInfoFromCookie(req);
if (authInfo?.username) {
clearProgress(authInfo.username, 'import');
}
return NextResponse.json(
{ error: error instanceof Error ? error.message : '导入失败' },
{ status: 500 }