File size: 6,904 Bytes
ee41abd
 
 
 
 
 
 
 
 
 
 
50264f7
ee41abd
 
 
 
50264f7
 
 
2f6c6cd
 
ec948f4
2f6c6cd
50264f7
 
2f6c6cd
 
 
 
ee41abd
ec948f4
50264f7
 
 
 
 
 
 
 
 
ee41abd
c6e17b0
ec948f4
ee41abd
 
c6e17b0
50264f7
 
c6e17b0
 
ee41abd
 
 
 
 
 
 
 
aaee7b0
50264f7
ee41abd
 
ec948f4
ee41abd
50264f7
 
ee41abd
 
 
c6e17b0
ec948f4
a4b1221
 
 
ec948f4
a4b1221
c6e17b0
 
 
50264f7
ee41abd
 
 
aaee7b0
 
 
 
 
 
 
50264f7
d9c2090
ee41abd
aaee7b0
ee41abd
 
aaee7b0
a4b1221
 
 
 
 
 
 
 
efcac7e
 
a4b1221
 
efcac7e
 
 
 
 
a4b1221
 
 
 
aaee7b0
a4b1221
 
ec948f4
aaee7b0
 
 
 
 
a4b1221
 
 
aaee7b0
ec948f4
a4b1221
aaee7b0
 
 
a4b1221
 
 
ec948f4
 
a4b1221
 
ec948f4
a4b1221
 
 
 
 
 
 
 
 
 
50264f7
 
ec948f4
ee41abd
c6e17b0
ec948f4
ee41abd
c6e17b0
ec948f4
c6e17b0
 
ec948f4
50264f7
c6e17b0
 
d9c2090
50264f7
 
ec948f4
ee41abd
 
 
78a8ad2
f695c1d
78a8ad2
f695c1d
 
ee41abd
 
 
d9c2090
7291ebb
c6e17b0
ec948f4
c6e17b0
d9c2090
 
 
 
 
 
7291ebb
 
2f6c6cd
 
 
ec948f4
 
2f6c6cd
 
 
 
 
 
 
 
50264f7
2f6c6cd
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
const express = require("express");
const fs = require("fs");
const path = require("path");
const ffmpeg = require("fluent-ffmpeg");
const cors = require("cors");

const app = express();
const PORT = 7860;

app.use(cors());
app.use(express.json());
app.use(express.static("public"));

const hlsFolder = path.join(__dirname, "hls");
if (!fs.existsSync(hlsFolder)) fs.mkdirSync(hlsFolder);

const streams = {}; // Active FFmpeg processes
const queues = loadQueues(); // Load persisted queues
const timestamps = loadTimestamps(); // Load timestamps from file
const activeChannels = new Set(Object.keys(queues)); // Restore active channels

// Load and save queues
function loadQueues() {
    const file = path.join(__dirname, "queues.json");
    return fs.existsSync(file) ? JSON.parse(fs.readFileSync(file)) : {};
}
function saveQueues() {
    fs.writeFileSync(path.join(__dirname, "queues.json"), JSON.stringify(queues, null, 2));
}

// Load and save timestamps
function loadTimestamps() {
    const file = path.join(__dirname, "timestamps.json");
    return fs.existsSync(file) ? JSON.parse(fs.readFileSync(file)) : {};
}
function saveTimestamps() {
    fs.writeFileSync(path.join(__dirname, "timestamps.json"), JSON.stringify(timestamps, null, 2));
}

// Add video to queue
app.get("/add", (req, res) => {
    const { channel, video } = req.query;
    if (!channel || !video) return res.status(400).json({ error: "Channel and video URL required" });

    if (!queues[channel]) queues[channel] = [];
    queues[channel].push(video);
    activeChannels.add(channel);
    saveQueues();

    console.log(`Added video to channel ${channel}: ${video}`);

    if (!streams[channel]) {
        startStream(channel);
    }

    res.json({ message: "Video added to queue", queue: queues[channel] });
});

// Start streaming next video in queue
// Start streaming next video in queue
function startStream(channel) {
    if (!queues[channel] || queues[channel].length === 0) {
        console.log(`No more videos for channel ${channel}`);
        delete streams[channel];
        activeChannels.delete(channel);
        saveQueues();
        return;
    }

    if (streams[channel]) {
        console.log(`Stopping existing stream for ${channel}`);
        try {
            streams[channel].kill("SIGKILL");
        } catch (error) {
            console.error(`Error stopping stream for ${channel}:`, error);
        }
        delete streams[channel];
    }

    const videoUrl = queues[channel].shift();
    const outputFolder = path.join(hlsFolder, channel);
    if (!fs.existsSync(outputFolder)) fs.mkdirSync(outputFolder, { recursive: true });

    // Ensure timestamps[channel] is set up correctly and only updates if needed
    if (!timestamps[channel]) {
        timestamps[channel] = { start: Date.now(), elapsed: 0 };
    } else {
        timestamps[channel].elapsed = timestamps[channel].elapsed || 0;  // Retain the elapsed time if available
    }

    saveQueues();
    saveTimestamps();

    console.log(`Streaming ${videoUrl} on channel ${channel} from ${timestamps[channel].elapsed}s`);

    streams[channel] = ffmpeg(videoUrl)
        .inputOptions(timestamps[channel].elapsed > 0 ? [`-ss ${timestamps[channel].elapsed}`] : []) 
        .output(`${outputFolder}/index.m3u8`)
        .addOptions([
            "-c:v libx264",
            "-preset veryfast",
            "-crf 23",
            "-b:v 1000k",
            "-c:a aac",
            "-b:a 128k",
            "-hls_time 4",
            "-hls_list_size 30",
            "-hls_flags append_list+independent_segments",
            "-hls_segment_type mpegts",
            "-hls_delete_threshold 50",
            "-g 24",
            "-r 24",
            "-bufsize 4000k",
            "-maxrate 1200k",
            "-strict -2",
        ])
        .on("start", () => {
            console.log(`FFmpeg started for ${channel}`);
            // We don't reset start time in the start event, so it stays consistent
        })
        .on("end", () => {
            console.log(`Finished streaming for ${channel}`);
            
            // Calculate and store the final elapsed time before stopping
            const now = Date.now();
            timestamps[channel].elapsed += Math.floor((now - timestamps[channel].start) / 1000);
            
            delete streams[channel];

            if (queues[channel] && queues[channel].length > 0) {
                timestamps[channel].start = now; // Reset start time for next video
                saveTimestamps();
                startStream(channel);
            } else {
                delete timestamps[channel]; // Remove if no more videos
                saveTimestamps();
            }
        })
        .on("error", (err) => {
            console.error(`FFmpeg error for ${channel}:`, err);
            saveTimestamps();

            if (err.message.includes("Invalid data found") || err.message.includes("No such file")) {
                console.log(`Skipping corrupted video: ${videoUrl}`);
                delete timestamps[channel];
            }

            delete streams[channel];

            if (queues[channel] && queues[channel].length > 0) {
                startStream(channel);
            }
        })
        .run();
}

// Skip to next video
app.get("/next", (req, res) => {
    const { channel } = req.query;
    if (!channel || !streams[channel]) return res.status(400).json({ error: "No active stream" });

    if (!queues[channel] || queues[channel].length === 0) {
        return res.json({ message: "No more videos." });
    }

    console.log(`Skipping video for channel ${channel}`);
    streams[channel].kill("SIGKILL");
    delete streams[channel];

    timestamps[channel] = { start: Date.now(), elapsed: 0 };
    saveTimestamps();
    startStream(channel);

    res.json({ message: "Skipped to next video" });
});

// Get active channels
app.get("/channels", (req, res) => {
    res.json({ channels: Array.from(activeChannels) });
});

// Serve HLS files
app.use("/hls", express.static(hlsFolder));

// Get real-time timestamp
app.get("/timestamp", (req, res) => {
    const { channel } = req.query;
    if (!channel) return res.status(400).json({ error: "Channel is required" });

    if (!timestamps[channel] || !timestamps[channel].start) {
        return res.json({ elapsed: 0 });
    }

    const elapsed = Math.floor((Date.now() - timestamps[channel].start) / 1000) + timestamps[channel].elapsed;
    res.json({ elapsed });
});

// Recover streams after restart
function recoverStreams() {
    for (const channel of Object.keys(queues)) {
        if (queues[channel].length > 0 && !streams[channel]) {
            console.log(`Recovering stream for ${channel}`);
            startStream(channel);
        }
    }
}

// Start the server
app.listen(PORT, () => {
    console.log(`Server running on http://localhost:${PORT}`);
    recoverStreams();
});