Spaces:
Sleeping
Sleeping
import express from 'express'; | |
import multer from 'multer'; | |
import { spawn } from 'child_process'; | |
import { writeFile, unlink, readFile } from 'fs/promises'; | |
import { createReadStream } from 'fs'; | |
import path from 'path'; | |
import { v4 as uuidv4 } from 'uuid'; | |
import fetch from 'node-fetch'; | |
import { startCleanupJob } from './cleanup.js'; | |
const TEMP_DIR = '/tmp'; | |
const app = express(); | |
const PORT = process.env.PORT || 7860; | |
export const tasks = {}; | |
app.use(express.json()); | |
app.use(express.urlencoded({ extended: true })); | |
const storage = multer.memoryStorage(); | |
const upload = multer({ storage: storage }); | |
const downloadFile = async (url) => { | |
const response = await fetch(url); | |
if (!response.ok) { | |
throw new Error(`Failed to download file: ${response.statusText}`); | |
} | |
const arrayBuffer = await response.arrayBuffer(); | |
return Buffer.from(arrayBuffer); | |
}; | |
// --- "УМНЫЙ" ЭНДПОИНТ ДЛЯ ПОТОКОВОЙ ОБРАБОТКИ --- | |
app.post('/api/run/stream', upload.single('file'), async (req, res) => { | |
try { | |
const { command, args: argsJson, file_url } = req.body; | |
const file = req.file; | |
if (!command) return res.status(400).json({ error: 'Parameter "command" is required.' }); | |
let args; | |
try { | |
args = argsJson ? JSON.parse(argsJson) : []; | |
} catch(e) { | |
return res.status(400).json({ error: 'Parameter "args" must be a valid JSON array.' }); | |
} | |
let inputBuffer; | |
if (file) { | |
inputBuffer = file.buffer; | |
} else if (file_url) { | |
inputBuffer = await downloadFile(file_url); | |
} else { | |
return res.status(400).json({ error: 'A file must be provided via "file" or "file_url".' }); | |
} | |
// --- СПЕЦИАЛЬНАЯ ЛОГИКА ДЛЯ FFMPEG --- | |
if (command === 'ffmpeg') { | |
const inputFilePath = path.join(TEMP_DIR, `${uuidv4()}-input`); | |
const outputFilePath = path.join(TEMP_DIR, `${uuidv4()}-output`); | |
try { | |
await writeFile(inputFilePath, inputBuffer); | |
const processedArgs = args.map(arg => | |
arg.replace('{INPUT_FILE}', inputFilePath) | |
.replace('{OUTPUT_FILE}', outputFilePath) | |
); | |
const process = spawn(command, processedArgs); | |
let stderrChunks = []; | |
process.stderr.on('data', (data) => stderrChunks.push(data)); | |
process.on('close', async (code) => { | |
if (code === 0) { | |
try { | |
const outputBuffer = await readFile(outputFilePath); | |
res.setHeader('Content-Type', 'application/octet-stream'); | |
res.send(outputBuffer); | |
} catch (readError) { | |
res.status(500).json({ error: "Command succeeded, but failed to read output file.", message: readError.message }); | |
} | |
} else { | |
const stderr = Buffer.concat(stderrChunks).toString('utf8'); | |
res.status(500).json({ error: 'Command execution failed.', code: code, stderr: stderr }); | |
} | |
// Очистка | |
await unlink(inputFilePath).catch(()=>{}); | |
await unlink(outputFilePath).catch(()=>{}); | |
}); | |
} catch (execError) { | |
res.status(500).json({ error: "Failed during ffmpeg file operations.", message: execError.message }); | |
await unlink(inputFilePath).catch(()=>{}); | |
await unlink(outputFilePath).catch(()=>{}); | |
} | |
// --- ОБЫЧНАЯ ЛОГИКА ДЛЯ ДРУГИХ КОМАНД --- | |
} else { | |
const process = spawn(command, args); | |
let stdoutChunks = []; | |
let stderrChunks = []; | |
process.stdout.on('data', (data) => stdoutChunks.push(data)); | |
process.stderr.on('data', (data) => stderrChunks.push(data)); | |
process.on('close', (code) => { | |
if (code === 0) { | |
res.setHeader('Content-Type', 'application/octet-stream'); | |
res.send(Buffer.concat(stdoutChunks)); | |
} else { | |
const stderr = Buffer.concat(stderrChunks).toString('utf8'); | |
res.status(500).json({ error: 'Command execution failed.', code: code, stderr: stderr }); | |
} | |
}); | |
process.stdin.write(inputBuffer); | |
process.stdin.end(); | |
} | |
} catch (error) { | |
res.status(500).json({ error: 'Server error during stream processing.', message: error.message }); | |
} | |
}); | |
// --- СИСТЕМА АСИНХРОННЫХ ЗАДАЧ (ОСТАЕТСЯ КАК АЛЬТЕРНАТИВА) --- | |
// ... (остальной код для /api/task/* и /api/download/* без изменений) ... | |
const executeTask = async (taskId) => { | |
const task = tasks[taskId]; | |
const { command, args, inputBuffer, outputFilename } = task.payload; | |
const tempFiles = []; | |
try { | |
tasks[taskId].status = 'processing'; | |
tasks[taskId].startTime = Date.now(); | |
const originalName = task.payload.originalName?.replace(/[^a-zA-Z0-9._-]/g, '') || 'input'; | |
const inputFilePath = path.join(TEMP_DIR, `${uuidv4()}-${originalName}`); | |
const outputFilePath = path.join(TEMP_DIR, `${uuidv4()}-${outputFilename || 'output'}`); | |
tempFiles.push(inputFilePath, outputFilePath); | |
task.outputFilePath = outputFilePath; | |
const processedArgs = args.map(arg => arg.replace('{INPUT_FILE}', inputFilePath).replace('{OUTPUT_FILE}', outputFilePath)); | |
await writeFile(inputFilePath, inputBuffer); | |
const process = spawn(command, processedArgs); | |
let stderrOutput = ''; | |
let totalDuration = 0; | |
process.stderr.on('data', (data) => { | |
const stderrLine = data.toString(); | |
stderrOutput += stderrLine; | |
if (!totalDuration) { | |
const durationMatch = stderrLine.match(/Duration: (\d{2}):(\d{2}):(\d{2})\.\d{2}/); | |
if (durationMatch) { | |
totalDuration = parseInt(durationMatch[1]) * 3600 + parseInt(durationMatch[2]) * 60 + parseInt(durationMatch[3]); | |
task.estimatedTotalTime = totalDuration; | |
} | |
} | |
const timeMatch = stderrLine.match(/time=(\d{2}):(\d{2}):(\d{2})\.\d{2}/); | |
if (timeMatch && totalDuration) { | |
const currentTime = parseInt(timeMatch[1]) * 3600 + parseInt(timeMatch[2]) * 60 + parseInt(timeMatch[3]); | |
task.progress = Math.min(100, Math.round((currentTime / totalDuration) * 100)); | |
} | |
}); | |
await new Promise((resolve, reject) => { | |
process.on('close', (code) => { | |
task.endTime = Date.now(); | |
task.stderr = stderrOutput; | |
if (code === 0) { | |
task.status = 'completed'; | |
task.result = { download_url: `/api/download/${path.basename(outputFilePath)}` }; | |
resolve(); | |
} else { | |
const error = new Error(`Process exited with code ${code}`); | |
error.code = code; | |
reject(error); | |
} | |
}); | |
process.on('error', reject); | |
}); | |
} catch (error) { | |
tasks[taskId].status = 'failed'; | |
tasks[taskId].endTime = Date.now(); | |
tasks[taskId].error = { message: error.message, code: error.code }; | |
for (const filePath of tempFiles) { unlink(filePath).catch(() => {}); } | |
} | |
}; | |
app.get('/', (req, res) => { res.send('Task-based remote execution server is ready.'); }); | |
app.post('/api/task/create', upload.single('file'), async (req, res) => { | |
try { | |
const { command, args: argsJson, file_url, output_filename } = req.body; | |
const file = req.file; | |
if (!command) return res.status(400).json({ error: 'Parameter "command" is required.' }); | |
let args; | |
try { args = argsJson ? JSON.parse(argsJson) : []; } catch(e) { return res.status(400).json({ error: 'Parameter "args" must be a valid JSON array.' }); } | |
let inputBuffer; | |
if (file) { inputBuffer = file.buffer; } else if (file_url) { inputBuffer = await downloadFile(file_url); } else { return res.status(400).json({ error: 'A file must be provided via "file" or "file_url".' }); } | |
const taskId = uuidv4(); | |
tasks[taskId] = { id: taskId, status: 'queued', progress: 0, submittedAt: Date.now(), payload: { command, args, outputFilename: output_filename, originalName: file?.originalname, inputBuffer, } }; | |
executeTask(taskId); | |
res.status(202).json({ message: "Task accepted.", taskId: taskId, status_url: `/api/task/status/${taskId}` }); | |
} catch (error) { | |
res.status(500).json({ error: 'Failed to create task.', message: error.message }); | |
} | |
}); | |
app.get('/api/task/status/:taskId', (req, res) => { | |
const { taskId } = req.params; | |
const task = tasks[taskId]; | |
if (!task) { return res.status(404).json({ error: 'Task not found.' }); } | |
const response = { id: task.id, status: task.status }; | |
if (task.startTime) { | |
const endTime = task.endTime || Date.now(); | |
response.elapsedTimeSeconds = Math.round((endTime - task.startTime) / 1000); | |
} | |
if (task.status === 'processing') { | |
response.progress = task.progress; | |
if (task.estimatedTotalTime && response.elapsedTimeSeconds) { | |
const remaining = task.estimatedTotalTime - response.elapsedTimeSeconds; | |
response.estimatedTimeLeftSeconds = Math.max(0, remaining); | |
} | |
} | |
if (task.status === 'completed') { response.result = task.result; } | |
if (task.status === 'failed') { response.error = task.error; } | |
res.status(200).json(response); | |
}); | |
app.get('/api/download/:fileId', (req, res) => { | |
const { fileId } = req.params; | |
if (fileId.includes('..')) { return res.status(400).send('Invalid file ID.'); } | |
const filePath = path.join(TEMP_DIR, fileId); | |
const stream = createReadStream(filePath); | |
stream.on('error', (err) => { | |
if (err.code === 'ENOENT') { res.status(404).send('File not found or has been cleaned up.'); } else { res.status(500).send('Server error.'); } | |
}); | |
res.setHeader('Content-Type', 'application/octet-stream'); | |
stream.pipe(res); | |
}); | |
app.listen(PORT, () => { startCleanupJob(); }); | |