File size: 10,749 Bytes
426e8cd
 
 
b5bcffb
 
ebf7adb
426e8cd
 
 
 
684d20b
426e8cd
ebf7adb
 
684d20b
426e8cd
 
 
 
 
 
 
 
 
 
684d20b
426e8cd
 
 
 
 
b5bcffb
7dc09d2
 
 
 
 
eb7c18d
7dc09d2
 
 
 
 
eb7c18d
7dc09d2
 
 
 
 
 
 
 
eb7c18d
7dc09d2
 
b5bcffb
3aa3b77
 
 
 
b5bcffb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3aa3b77
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7dc09d2
 
3aa3b77
7dc09d2
 
 
 
b5bcffb
3aa3b77
7dc09d2
684d20b
 
 
 
426e8cd
 
684d20b
 
 
 
 
 
 
3aa3b77
684d20b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3aa3b77
684d20b
 
 
 
 
 
 
 
 
426e8cd
684d20b
 
3aa3b77
 
426e8cd
684d20b
426e8cd
3aa3b77
426e8cd
684d20b
426e8cd
 
 
684d20b
426e8cd
3aa3b77
426e8cd
3aa3b77
684d20b
3aa3b77
684d20b
3aa3b77
426e8cd
3aa3b77
426e8cd
 
 
684d20b
 
 
3aa3b77
 
684d20b
7dc09d2
684d20b
 
 
 
 
 
 
 
 
3aa3b77
 
684d20b
 
 
426e8cd
 
3aa3b77
426e8cd
7dc09d2
 
3aa3b77
7dc09d2
 
 
426e8cd
 
3aa3b77
426e8cd
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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(); });