move cold processing files
This commit is contained in:
localhost
parent
da0c9c23f5
commit
86e2dc41f3
|
|
@ -1,200 +0,0 @@
|
|||
import { readdir } from 'node:fs/promises'
|
||||
|
||||
type FfprobeStream = {
|
||||
codec_name?: string
|
||||
codec_type?: string
|
||||
duration?: string
|
||||
width?: number
|
||||
height?: number
|
||||
avg_frame_rate?: string
|
||||
r_frame_rate?: string
|
||||
}
|
||||
|
||||
type FfprobeFormat = {
|
||||
duration?: string
|
||||
}
|
||||
|
||||
type FfprobeOutput = {
|
||||
streams?: FfprobeStream[]
|
||||
format?: FfprobeFormat
|
||||
}
|
||||
|
||||
function printUsage() {
|
||||
console.error('Usage: bun run generateColdRow.ts --input-folder=/mnt/cold --prefix=cold/ --suffix=.webm')
|
||||
console.error('Example: bun run generateColdRow.ts --input-folder=./cold --prefix=cold/ --suffix=.webm')
|
||||
}
|
||||
|
||||
function formatResolution(width?: number, height?: number): string {
|
||||
if (!width || !height) return 'unknown'
|
||||
return `${width}x${height}`
|
||||
}
|
||||
|
||||
function parseDuration(durationText?: string): number {
|
||||
if (!durationText) return 0
|
||||
const duration = Number(durationText)
|
||||
return Number.isFinite(duration) ? duration : 0
|
||||
}
|
||||
|
||||
function parseFps(rate?: string): number {
|
||||
if (!rate) return 0
|
||||
const [numeratorText, denominatorText] = rate.split('/')
|
||||
const numerator = Number(numeratorText)
|
||||
const denominator = denominatorText ? Number(denominatorText) : 1
|
||||
|
||||
if (!Number.isFinite(numerator) || !Number.isFinite(denominator) || denominator === 0) {
|
||||
return 0
|
||||
}
|
||||
|
||||
return Number((numerator / denominator).toFixed(3))
|
||||
}
|
||||
|
||||
async function hashFile(path: string): Promise<string> {
|
||||
const proc = Bun.spawn(['sha256sum', '--', path], {
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
})
|
||||
|
||||
const [stdout, stderr, exitCode] = await Promise.all([
|
||||
new Response(proc.stdout).text(),
|
||||
new Response(proc.stderr).text(),
|
||||
proc.exited,
|
||||
])
|
||||
|
||||
if (exitCode !== 0) {
|
||||
throw new Error(stderr.trim() || 'sha256sum failed')
|
||||
}
|
||||
|
||||
const hash = stdout.trim().split(/\s+/)[0]
|
||||
if (!hash) throw new Error('sha256sum did not return a hash')
|
||||
return hash
|
||||
}
|
||||
|
||||
async function getVideoMetadata(path: string): Promise<FfprobeOutput> {
|
||||
const proc = Bun.spawn([
|
||||
'ffprobe',
|
||||
'-v',
|
||||
'error',
|
||||
'-print_format',
|
||||
'json',
|
||||
'-show_format',
|
||||
'-show_streams',
|
||||
path,
|
||||
], {
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
})
|
||||
|
||||
const [stdout, stderr, exitCode] = await Promise.all([
|
||||
new Response(proc.stdout).text(),
|
||||
new Response(proc.stderr).text(),
|
||||
proc.exited,
|
||||
])
|
||||
|
||||
if (exitCode !== 0) {
|
||||
throw new Error(stderr.trim() || 'ffprobe failed')
|
||||
}
|
||||
|
||||
return JSON.parse(stdout) as FfprobeOutput
|
||||
}
|
||||
|
||||
function normalizeDirectory(path: string): string {
|
||||
return path.endsWith('/') ? path : `${path}/`
|
||||
}
|
||||
|
||||
function escapeRegExp(text: string): string {
|
||||
return text.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')
|
||||
}
|
||||
|
||||
function readFlagValue(arg: string, longName: string, shortName: string): string | undefined {
|
||||
if (arg.startsWith(`--${longName}=`)) {
|
||||
return arg.slice(longName.length + 3)
|
||||
}
|
||||
|
||||
if (arg.startsWith(`-${shortName}=`)) {
|
||||
return arg.slice(shortName.length + 2)
|
||||
}
|
||||
}
|
||||
|
||||
function parseArgs(args: string[]): { inputFolder: string, suffix: string } {
|
||||
let inputFolder: string | undefined
|
||||
let prefix: string | undefined
|
||||
let suffix: string | undefined
|
||||
|
||||
for (const arg of args) {
|
||||
inputFolder ??= readFlagValue(arg, 'input-folder', 'input-folder')
|
||||
inputFolder ??= readFlagValue(arg, 'input', 'input')
|
||||
prefix ??= readFlagValue(arg, 'prefix', 'prefix')
|
||||
suffix ??= readFlagValue(arg, 'suffix', 'suffix')
|
||||
}
|
||||
|
||||
if (!prefix || !suffix) {
|
||||
printUsage()
|
||||
process.exit(1)
|
||||
}
|
||||
|
||||
return {
|
||||
inputFolder: normalizeDirectory(inputFolder ?? prefix),
|
||||
suffix,
|
||||
}
|
||||
}
|
||||
|
||||
async function readInputIds(inputFolder: string, suffix: string): Promise<string[]> {
|
||||
const entries = await readdir(inputFolder, { withFileTypes: true })
|
||||
const suffixPattern = new RegExp(`${escapeRegExp(suffix)}$`)
|
||||
|
||||
return entries
|
||||
.filter(entry => entry.isFile() && entry.name.endsWith(suffix))
|
||||
.map(entry => entry.name.replace(suffixPattern, ''))
|
||||
.sort()
|
||||
}
|
||||
|
||||
async function generateRow(videoId: string, inputFolder: string, suffix: string) {
|
||||
const filePath = `${inputFolder}${videoId}${suffix}`
|
||||
const file = Bun.file(filePath)
|
||||
|
||||
if (!(await file.exists())) {
|
||||
throw new Error(`File not found for ${videoId}: ${filePath}`)
|
||||
}
|
||||
|
||||
const filename = filePath.split('/').pop()
|
||||
if (!filename) {
|
||||
throw new Error(`Invalid file path for ${videoId}: ${filePath}`)
|
||||
}
|
||||
|
||||
const [hash, metadata] = await Promise.all([
|
||||
hashFile(filePath),
|
||||
getVideoMetadata(filePath),
|
||||
])
|
||||
|
||||
const videoStream = metadata.streams?.find(stream => stream.codec_type === 'video')
|
||||
const audioStream = metadata.streams?.find(stream => stream.codec_type === 'audio')
|
||||
|
||||
return {
|
||||
videoId,
|
||||
filename,
|
||||
hash,
|
||||
hash_algorithm: 'sha256',
|
||||
size_bytes: file.size,
|
||||
duration_seconds: parseDuration(videoStream?.duration ?? metadata.format?.duration),
|
||||
video_codec: videoStream?.codec_name ?? 'unknown',
|
||||
audio_codec: audioStream?.codec_name ?? 'unknown',
|
||||
resolution: formatResolution(videoStream?.width, videoStream?.height),
|
||||
fps: parseFps(videoStream?.avg_frame_rate ?? videoStream?.r_frame_rate),
|
||||
}
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const { inputFolder, suffix } = parseArgs(Bun.argv.slice(2))
|
||||
const ids = await readInputIds(inputFolder, suffix)
|
||||
if (ids.length === 0) {
|
||||
console.error(`No files ending in ${suffix} found in ${inputFolder}`)
|
||||
printUsage()
|
||||
process.exit(1)
|
||||
}
|
||||
|
||||
for (const videoId of ids) {
|
||||
console.log(await generateRow(videoId, inputFolder, suffix))
|
||||
}
|
||||
}
|
||||
|
||||
await main()
|
||||
179
insertColdRow.ts
179
insertColdRow.ts
|
|
@ -1,179 +0,0 @@
|
|||
import { db } from '@/utils/database'
|
||||
import type { NewFile } from '@/types'
|
||||
|
||||
const DEFAULT_INPUT_PATH = './coldRows.txt'
|
||||
|
||||
function chunkArray<T>(array: T[], size: number): T[][] {
|
||||
const result: T[][] = []
|
||||
for (let i = 0; i < array.length; i += size) {
|
||||
result.push(array.slice(i, i + size))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
function getLineAndColumn(text: string, index: number): { line: number; column: number; context: string } {
|
||||
const lines = text.slice(0, index).split('\n')
|
||||
const line = lines.length
|
||||
const column = lines[lines.length - 1].length + 1
|
||||
|
||||
const allLines = text.split('\n')
|
||||
const startLineIdx = Math.max(0, line - 3)
|
||||
const endLineIdx = Math.min(allLines.length - 1, line + 2)
|
||||
|
||||
const contextLines = allLines.slice(startLineIdx, endLineIdx + 1).map((l, i) => {
|
||||
const currentLineNum = startLineIdx + i + 1
|
||||
const prefix = currentLineNum === line ? '-> ' : ' '
|
||||
return `${prefix}${String(currentLineNum).padStart(4, ' ')} | ${l}`
|
||||
})
|
||||
|
||||
return {
|
||||
line,
|
||||
column,
|
||||
context: contextLines.join('\n'),
|
||||
}
|
||||
}
|
||||
|
||||
function extractObjectBlocks(text: string): string[] {
|
||||
const blocks: string[] = []
|
||||
let depth = 0
|
||||
let startIndex = -1
|
||||
const openBraceIndices: number[] = []
|
||||
|
||||
for (let index = 0; index < text.length; index++) {
|
||||
const char = text[index]
|
||||
|
||||
if (char === '{') {
|
||||
if (depth === 0) {
|
||||
startIndex = index
|
||||
}
|
||||
depth += 1
|
||||
openBraceIndices.push(index)
|
||||
continue
|
||||
}
|
||||
|
||||
if (char !== '}') continue
|
||||
|
||||
depth -= 1
|
||||
openBraceIndices.pop()
|
||||
|
||||
if (depth < 0) {
|
||||
const { line, column, context } = getLineAndColumn(text, index)
|
||||
console.error(`Error: Unexpected closing brace at line ${line}, column ${column}:\n${context}`)
|
||||
throw new Error(`Unexpected closing brace at position ${index}`)
|
||||
}
|
||||
|
||||
if (depth === 0 && startIndex !== -1) {
|
||||
blocks.push(text.slice(startIndex, index + 1))
|
||||
startIndex = -1
|
||||
}
|
||||
}
|
||||
|
||||
if (depth !== 0) {
|
||||
console.error(`Error: Input contains an unterminated object literal (depth: ${depth}).`)
|
||||
console.error(`There are ${openBraceIndices.length} unclosed '{' brace(s).`)
|
||||
|
||||
const maxToPrint = Math.min(openBraceIndices.length, 3)
|
||||
for (let i = 0; i < maxToPrint; i++) {
|
||||
const braceIndex = openBraceIndices[i]
|
||||
const { line, column, context } = getLineAndColumn(text, braceIndex)
|
||||
console.error(`\nUnclosed '{' #${i + 1} at line ${line}, column ${column}:`)
|
||||
console.error(context)
|
||||
}
|
||||
|
||||
if (openBraceIndices.length > maxToPrint) {
|
||||
console.error(`\n... and ${openBraceIndices.length - maxToPrint} more unclosed brace(s).`)
|
||||
}
|
||||
|
||||
throw new Error('Input contains an unterminated object literal')
|
||||
}
|
||||
|
||||
return blocks
|
||||
}
|
||||
|
||||
function parseColdRow(block: string): NewFile {
|
||||
const parsed = Function(`"use strict"; return (${block})`)()
|
||||
|
||||
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
|
||||
throw new Error('Parsed row is not an object')
|
||||
}
|
||||
|
||||
return parsed as NewFile
|
||||
}
|
||||
|
||||
async function loadRows(inputPath: string): Promise<NewFile[]> {
|
||||
const text = await Bun.file(inputPath).text()
|
||||
const blocks = extractObjectBlocks(text)
|
||||
|
||||
if (blocks.length === 0) {
|
||||
throw new Error(`No row objects found in ${inputPath}`)
|
||||
}
|
||||
|
||||
return blocks.map(parseColdRow)
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const args = Bun.argv.slice(2)
|
||||
const dryRun = args.includes('--dry-run')
|
||||
const inputPath = args.find(arg => arg !== '--dry-run') ?? DEFAULT_INPUT_PATH
|
||||
const rows = await loadRows(inputPath)
|
||||
const videoIds = [...new Set(rows.map(row => row.videoId))]
|
||||
|
||||
const uniqueHashes = [...new Set(rows.map(r => r.hash))]
|
||||
const existingFilesChunks = []
|
||||
for (const chunk of chunkArray(uniqueHashes, 1000)) {
|
||||
const filesChunk = await db.selectFrom('files')
|
||||
.where('hash', 'in', chunk)
|
||||
.selectAll()
|
||||
.execute()
|
||||
existingFilesChunks.push(filesChunk)
|
||||
}
|
||||
const existingFiles = existingFilesChunks.flat()
|
||||
|
||||
const rowsToInsert: NewFile[] = []
|
||||
for (const row of rows) {
|
||||
const matches = existingFiles.filter(f => f.hash === row.hash)
|
||||
const isDuplicate = matches.some(match => {
|
||||
return Object.keys(row).every(key => {
|
||||
const k = key as keyof NewFile
|
||||
return row[k] == (match as any)[k]
|
||||
})
|
||||
})
|
||||
|
||||
if (!isDuplicate) {
|
||||
rowsToInsert.push(row)
|
||||
}
|
||||
}
|
||||
|
||||
if (dryRun) {
|
||||
console.log(`Parsed ${rows.length} row(s) from ${inputPath}`)
|
||||
console.log(`Would insert ${rowsToInsert.length} new row(s) and skip ${rows.length - rowsToInsert.length} duplicate(s)`)
|
||||
console.log(JSON.stringify(rowsToInsert, null, 2))
|
||||
console.log(`Would set deletion_stage to cold_storage for ${videoIds.length} video(s)`)
|
||||
return
|
||||
}
|
||||
|
||||
if (rowsToInsert.length > 0) {
|
||||
for (const chunk of chunkArray(rowsToInsert, 100)) {
|
||||
await db.insertInto('files')
|
||||
.values(chunk)
|
||||
.execute()
|
||||
}
|
||||
}
|
||||
|
||||
for (const chunk of chunkArray(videoIds, 1000)) {
|
||||
await db.updateTable('videos')
|
||||
.where('id', 'in', chunk)
|
||||
.set({ deletion_stage: 'cold_storage' })
|
||||
.execute()
|
||||
}
|
||||
|
||||
console.log(`Inserted ${rowsToInsert.length} row(s) from ${inputPath}`)
|
||||
console.log(`Skipped ${rows.length - rowsToInsert.length} duplicate row(s)`)
|
||||
console.log(`Set deletion_stage to cold_storage for ${videoIds.length} video(s)`)
|
||||
}
|
||||
|
||||
try {
|
||||
await main()
|
||||
} finally {
|
||||
await db.destroy()
|
||||
}
|
||||
Loading…
Reference in New Issue