gongxiangp2/scripts/migrate-s3-materials.ts

513 lines
14 KiB
TypeScript
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/**
* S3素材无感迁移脚本
*
* 目标:将老素材的S3文件迁移到新位置(统一使用路径前缀)
* 原则:
* 1. 所有素材、人员信息、元数据都不能动
* 2. 无感迁移,用户无感知
* 3. 支持断点续传和错误重试
* 4. 迁移后验证所有功能正常
*/
import { PrismaClient } from '@prisma/client'
import { S3Client, GetObjectCommand, PutObjectCommand, HeadObjectCommand } from '@aws-sdk/client-s3'
import { Readable } from 'stream'
import { writeFile, readFile, mkdir } from 'fs/promises'
import { existsSync } from 'fs'
import path from 'path'
const prisma = new PrismaClient()
// S3客户端配置
const s3Client = new S3Client({
endpoint: process.env.S3_ENDPOINT || '',
region: process.env.S3_REGION || 'us-east-1',
credentials: {
accessKeyId: process.env.S3_ACCESS_KEY_ID!,
secretAccessKey: process.env.S3_SECRET_ACCESS_KEY!,
},
forcePathStyle: true,
requestHandler: {
requestTimeout: 300000, // 5分钟超时
connectionTimeout: 30000, // 30秒连接超时
},
maxAttempts: 10,
retryMode: 'adaptive',
})
const S3_BUCKET = process.env.S3_BUCKET || ''
const S3_PUBLIC_DOMAIN = process.env.S3_PUBLIC_DOMAIN || ''
const S3_FILE_PATH = process.env.S3_FILE_PATH || '由壹点AI/'
// 迁移日志文件
const MIGRATION_LOG_DIR = path.join(process.cwd(), 'logs', 'migration')
const MIGRATION_LOG_FILE = path.join(MIGRATION_LOG_DIR, `migration-${Date.now()}.json`)
interface MigrationLog {
materialId: string
field: string
oldUrl: string
newUrl: string
status: 'success' | 'failed' | 'skipped'
error?: string
timestamp: string
}
interface MigrationStats {
total: number
success: number
failed: number
skipped: number
errors: Array<{ materialId: string; field: string; error: string }>
}
/**
* 判断URL是否为老素材(不包含路径前缀)
*/
function isOldMaterialUrl(url: string): boolean {
if (!url || !url.includes('s3.1080p.hk') || !url.includes('/jsxiangche/')) {
return false
}
// 老素材URL格式:https://s3.1080p.hk/jsxiangche/preview-xxx.webp
// 新素材URL格式:https://s3.1080p.hk/jsxiangche/由壹点AI/video-thumb-xxx.jpg
return !url.includes(S3_FILE_PATH)
}
/**
* 从URL提取S3 Key
*/
function extractS3Key(url: string): string {
try {
const urlObj = new URL(url)
// 移除开头的斜杠和bucket名称
let key = urlObj.pathname.slice(1) // 移除开头的斜杠
if (key.startsWith('jsxiangche/')) {
key = key.replace('jsxiangche/', '')
}
return key
} catch (error) {
throw new Error(`Invalid URL format: ${url}`)
}
}
/**
* 生成新URL(带路径前缀)
*/
function generateNewUrl(oldKey: string): string {
const newKey = `${S3_FILE_PATH}${oldKey}`
return `${S3_PUBLIC_DOMAIN}/${newKey}`
}
/**
* 从S3下载文件
*/
async function downloadFromS3(key: string): Promise<Buffer> {
const command = new GetObjectCommand({
Bucket: S3_BUCKET,
Key: key,
})
const response = await s3Client.send(command)
const stream = response.Body as Readable
const chunks: Buffer[] = []
for await (const chunk of stream) {
chunks.push(Buffer.from(chunk))
}
return Buffer.concat(chunks)
}
/**
* 上传文件到S3
*/
async function uploadToS3(buffer: Buffer, key: string, contentType?: string): Promise<void> {
const command = new PutObjectCommand({
Bucket: S3_BUCKET,
Key: key,
Body: buffer,
ContentType: contentType || 'application/octet-stream',
ACL: 'public-read',
})
await s3Client.send(command)
}
/**
* 检查S3文件是否存在
*/
async function checkS3FileExists(key: string): Promise<boolean> {
try {
const command = new HeadObjectCommand({
Bucket: S3_BUCKET,
Key: key,
})
await s3Client.send(command)
return true
} catch (error: any) {
if (error.name === 'NotFound' || error.$metadata?.httpStatusCode === 404) {
return false
}
throw error
}
}
/**
* 迁移单个文件
*/
async function migrateFile(
materialId: string,
field: string,
oldUrl: string
): Promise<{ newUrl: string; skipped: boolean }> {
try {
// 提取旧Key
const oldKey = extractS3Key(oldUrl)
// 生成新Key和URL
const newKey = `${S3_FILE_PATH}${oldKey}`
const newUrl = generateNewUrl(oldKey)
// 检查新文件是否已存在
const exists = await checkS3FileExists(newKey)
if (exists) {
console.log(` ✅ 文件已存在,跳过: ${newKey}`)
return { newUrl, skipped: true }
}
// 下载文件
console.log(` 📥 下载: ${oldKey}`)
const buffer = await downloadFromS3(oldKey)
// 确定ContentType
const contentType = getContentTypeFromUrl(oldUrl)
// 上传到新位置
console.log(` 📤 上传: ${newKey}`)
await uploadToS3(buffer, newKey, contentType)
// 验证新文件可访问
const newExists = await checkS3FileExists(newKey)
if (!newExists) {
throw new Error('上传后验证失败:文件不存在')
}
console.log(` ✅ 迁移成功: ${newUrl}`)
return { newUrl, skipped: false }
} catch (error: any) {
console.error(` ❌ 迁移失败: ${error.message}`)
throw error
}
}
/**
* 从URL获取ContentType
*/
function getContentTypeFromUrl(url: string): string {
const ext = path.extname(url).toLowerCase()
const contentTypeMap: Record<string, string> = {
'.webp': 'image/webp',
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.png': 'image/png',
'.gif': 'image/gif',
'.mp4': 'video/mp4',
'.mp3': 'audio/mpeg',
'.wav': 'audio/wav',
'.flac': 'audio/flac',
}
return contentTypeMap[ext] || 'application/octet-stream'
}
/**
* 迁移单个素材的所有S3字段
*/
async function migrateMaterial(material: any): Promise<MigrationLog[]> {
const logs: MigrationLog[] = []
const updates: Record<string, any> = {}
// 需要迁移的字段列表
const s3Fields = [
'previewUrl',
'zipFileName', // 仅当包含S3 URL时
'videoSourceUrl',
'videoPreview540Url',
'videoPreviewClipUrl',
'videoCoverUrl',
]
// 迁移各个S3字段
for (const field of s3Fields) {
const url = material[field]
if (!url || typeof url !== 'string' || !isOldMaterialUrl(url)) {
continue
}
try {
console.log(`[${material.id}] 迁移字段 ${field}: ${url}`)
const { newUrl, skipped } = await migrateFile(material.id, field, url)
updates[field] = newUrl
logs.push({
materialId: material.id,
field,
oldUrl: url,
newUrl,
status: skipped ? 'skipped' : 'success',
timestamp: new Date().toISOString(),
})
} catch (error: any) {
logs.push({
materialId: material.id,
field,
oldUrl: url,
newUrl: '',
status: 'failed',
error: error.message,
timestamp: new Date().toISOString(),
})
}
}
// 迁移displayImages数组
if (Array.isArray(material.displayImages) && material.displayImages.length > 0) {
const newDisplayImages: string[] = []
let hasChanges = false
for (let i = 0; i < material.displayImages.length; i++) {
const imageUrl = material.displayImages[i]
if (typeof imageUrl === 'string' && isOldMaterialUrl(imageUrl)) {
try {
console.log(`[${material.id}] 迁移 displayImages[${i}]: ${imageUrl}`)
const { newUrl, skipped } = await migrateFile(material.id, `displayImages[${i}]`, imageUrl)
newDisplayImages.push(newUrl)
hasChanges = true
logs.push({
materialId: material.id,
field: `displayImages[${i}]`,
oldUrl: imageUrl,
newUrl,
status: skipped ? 'skipped' : 'success',
timestamp: new Date().toISOString(),
})
} catch (error: any) {
newDisplayImages.push(imageUrl) // 失败时保留原URL
logs.push({
materialId: material.id,
field: `displayImages[${i}]`,
oldUrl: imageUrl,
newUrl: imageUrl,
status: 'failed',
error: error.message,
timestamp: new Date().toISOString(),
})
}
} else {
newDisplayImages.push(imageUrl)
}
}
if (hasChanges) {
updates.displayImages = newDisplayImages
}
}
// 更新数据库
if (Object.keys(updates).length > 0) {
try {
await prisma.material.update({
where: { id: material.id },
data: updates,
})
console.log(`[${material.id}] ✅ 数据库更新成功`)
} catch (error: any) {
console.error(`[${material.id}] ❌ 数据库更新失败: ${error.message}`)
// 标记所有日志为失败
logs.forEach(log => {
if (log.status === 'success') {
log.status = 'failed'
log.error = `数据库更新失败: ${error.message}`
}
})
}
}
return logs
}
/**
* 保存迁移日志
*/
async function saveMigrationLog(logs: MigrationLog[]): Promise<void> {
if (!existsSync(MIGRATION_LOG_DIR)) {
await mkdir(MIGRATION_LOG_DIR, { recursive: true })
}
await writeFile(MIGRATION_LOG_FILE, JSON.stringify(logs, null, 2), 'utf-8')
console.log(`\n📝 迁移日志已保存: ${MIGRATION_LOG_FILE}`)
}
/**
* 加载迁移日志(用于断点续传)
*/
async function loadMigrationLog(): Promise<Set<string>> {
// 查找最新的日志文件
// 这里简化处理,实际可以扫描目录找到最新的日志文件
return new Set()
}
/**
* 主迁移函数
*/
async function main() {
console.log('🚀 开始S3素材无感迁移...\n')
// 检查环境变量
if (!S3_BUCKET || !S3_PUBLIC_DOMAIN) {
console.error('❌ 错误:缺少必要的环境变量')
console.error(' 需要: S3_BUCKET, S3_PUBLIC_DOMAIN')
process.exit(1)
}
console.log('📋 配置信息:')
console.log(` S3_BUCKET: ${S3_BUCKET}`)
console.log(` S3_PUBLIC_DOMAIN: ${S3_PUBLIC_DOMAIN}`)
console.log(` S3_FILE_PATH: ${S3_FILE_PATH}`)
console.log('')
// 查询需要迁移的素材
console.log('🔍 查询需要迁移的素材...')
const materials = await prisma.material.findMany({
where: {
deletedAt: null, // 只迁移未删除的素材
OR: [
{ previewUrl: { contains: 's3.1080p.hk' } },
{ zipFileName: { contains: 's3.1080p.hk' } },
{ videoSourceUrl: { contains: 's3.1080p.hk' } },
{ videoPreview540Url: { contains: 's3.1080p.hk' } },
{ videoPreviewClipUrl: { contains: 's3.1080p.hk' } },
{ videoCoverUrl: { contains: 's3.1080p.hk' } },
],
},
select: {
id: true,
previewUrl: true,
zipFileName: true,
displayImages: true,
videoSourceUrl: true,
videoPreview540Url: true,
videoPreviewClipUrl: true,
videoCoverUrl: true,
},
})
// 过滤出真正需要迁移的素材(包含老URL的)
const materialsToMigrate = materials.filter(material => {
const s3Fields = [
material.previewUrl,
material.zipFileName,
material.videoSourceUrl,
material.videoPreview540Url,
material.videoPreviewClipUrl,
material.videoCoverUrl,
]
// 检查S3字段
if (s3Fields.some(url => url && isOldMaterialUrl(url))) {
return true
}
// 检查displayImages
if (Array.isArray(material.displayImages)) {
if (material.displayImages.some((url: string) => url && isOldMaterialUrl(url))) {
return true
}
}
return false
})
console.log(` 找到 ${materialsToMigrate.length} 个素材需要迁移\n`)
if (materialsToMigrate.length === 0) {
console.log('✅ 没有需要迁移的素材')
return
}
// 执行迁移
const allLogs: MigrationLog[] = []
const stats: MigrationStats = {
total: 0,
success: 0,
failed: 0,
skipped: 0,
errors: [],
}
for (let i = 0; i < materialsToMigrate.length; i++) {
const material = materialsToMigrate[i]
console.log(`\n[${i + 1}/${materialsToMigrate.length}] 迁移素材: ${material.id}`)
const logs = await migrateMaterial(material)
allLogs.push(...logs)
// 更新统计
logs.forEach(log => {
stats.total++
if (log.status === 'success') {
stats.success++
} else if (log.status === 'failed') {
stats.failed++
stats.errors.push({
materialId: log.materialId,
field: log.field,
error: log.error || '未知错误',
})
} else if (log.status === 'skipped') {
stats.skipped++
}
})
// 每10个素材保存一次日志(防止中断丢失)
if ((i + 1) % 10 === 0) {
await saveMigrationLog(allLogs)
}
}
// 保存最终日志
await saveMigrationLog(allLogs)
// 输出统计
console.log('\n' + '='.repeat(60))
console.log('📊 迁移统计:')
console.log(` 总计: ${stats.total}`)
console.log(` 成功: ${stats.success}`)
console.log(` 跳过: ${stats.skipped}`)
console.log(` 失败: ${stats.failed}`)
console.log('='.repeat(60))
if (stats.errors.length > 0) {
console.log('\n❌ 失败详情:')
stats.errors.forEach((error, index) => {
console.log(` ${index + 1}. [${error.materialId}] ${error.field}: ${error.error}`)
})
}
if (stats.failed === 0) {
console.log('\n✅ 迁移完成!所有素材已成功迁移')
} else {
console.log(`\n⚠️ 迁移完成,但有 ${stats.failed} 个文件迁移失败,请检查日志`)
}
}
// 执行迁移
main()
.catch((error) => {
console.error('❌ 迁移过程出错:', error)
process.exit(1)
})
.finally(async () => {
await prisma.$disconnect()
})