Fix issues and add ffmpeg

README.md

@@ -12,6 +12,7 @@ This bot can download music and videos from youtube.
 BOT_TOKEN=<YOUR_TELEGRAM_BOT_TOKEN>
 ADMIN_LIST=<Admin ID list, separate ", "> (example: 123,321, 678, 911 , 228)
 WHITELIST=<User list, separate ", " (Important: Admin list will add to whitelist automatly)> (example: 123,321, 678, 911 , 228)
+SHOW_ALL_FORMATS=<Show all formats from video (1, 0 - Only '.mp4', 'm4a')>
 ```
 6) Build docker container:
 `docker build -t ytdlp_node:release .`

src/bot.js

@@ -1,6 +1,6 @@
 const TelegramBot = require("node-telegram-bot-api");
 const fs = require("fs");
-const DownloadVideo = require("./download");
+const { DownloadVideo, YtdlUpdater } = require("./download");
 const https = require("https");
 
 async function getImage(url) {
@@ -23,6 +23,7 @@ async function getImage(url) {
 const config = {
 	token: process.env.BOT_TOKEN,
 	admins: process.env.ADMIN_LIST?.split(/\s*,\s*/g).map((x) => +x) ?? [],
+	allFormats: process.env.SHOW_ALL_FORMATS === "1"
 };
 config.whitelist = [].concat(
 	process.env.WHITELIST?.split(/\s*,\s*/g).map((x) => +x) ?? [],
@@ -50,6 +51,8 @@ class TgBotSender {
 		this.bot = bot;
 		this.msg = msg;
 		
+		this.curText = null;
+
 		this.memory = memory;
 	}
 
@@ -62,11 +65,13 @@ class TgBotSender {
 	}
 
 	async text(text, reply_markup) {
+		if (text === this.curText) return;
 		await this.bot.editMessageText(text, {
 			chat_id: this.msg.chat.id,
 			message_id: this.msg.message_id,
 			reply_markup,
 		});
+		this.curText = text;
 	}
 
 	async download(video, quality = DownloadVideo.qualities.mp4_best) {
@@ -77,14 +82,17 @@ class TgBotSender {
 		await video.setupQuality(quality, false);
 		let cooldown = new Date();
 		video
-			.download(({ percent }) => {
+			.download(({ percent, downloaded, fine, msg }) => {
+				//console.debug({ percent, fine, downloaded });
 				const now = new Date();
-				if (now - cooldown <= 1500) return;
+				if (!fine && now - cooldown <= 3500) return;
 				cooldown = new Date();
-				sent.text(`Downloaded (${percent}%)`);
+				if (!fine)
-			})
+					sent.text(`Downloaded (${percent}%), ${downloaded} bytes`);
+				else
+					sent.text(!msg ? "Download complete" : msg);
+			}, true)
 			.then((downloaded) => {
-				sent.text("Download complete");
 				if (downloaded.format.type === "video") {
 					this.bot.sendVideo(
 						msg.chat.id,
@@ -114,7 +122,44 @@ class TgBotSender {
 	}
 }
 
-const actions = new Object();
+class VideosGarbageCollector {
+	constructor () {
+		this.actions = new Object();
+		this.pinnedVideos = new Object();
+		this.removeTable = new Object();
+		
+		setInterval(() => {
+			const tasks = [];
+			
+			for (let registerDate in this.removeTable) {
+				const key = this.removeTable[registerDate];
+				const savedTo = new Date(registerDate);
+				savedTo.setHours(savedTo.getHours() + 3);
+				
+				if ((new Date()) >= savedTo) {
+					const video = this.pinnedVideos[key];
+					tasks.push(() => {
+						// Call destructor
+						video.destructor();
+						// Delete..
+						delete this.actions[key];
+						delete this.pinnedVideos[key];
+						delete this.removeTable[registerDate];
+					});
+				}
+			}
+			
+			tasks.forEach(task => task());
+			console.info("GC Clean up", tasks.length, "items(s)");
+		}, 60000);
+	}
+	register (key, act, pinnedVideo) {
+		this.actions[key] = act;
+		this.pinnedVideos[key] = pinnedVideo;
+		this.removeTable[new Date()] = key;
+	}
+}
+
 class TgBot {
 	constructor() {
 		this.bot = new TelegramBot(config.token, {
@@ -122,12 +167,14 @@ class TgBot {
 		});
 		this.bot.on("text", (m) => this.newMessageHandler(m));
 		this.bot.on("callback_query", (m) => this.cbHandler(m));
+		this.videosGC = new VideosGarbageCollector();
+		this.updater = new YtdlUpdater();
 	}
 
 	async cbHandler(query) {
 		const key = query.from.id + ":" + query.data;
-		//console.debug(query, actions[key], key);
+		//console.debug(query, this.videosGC.actions[key], key);
-		actions[key]?.();
+		this.videosGC.actions[key]?.();
 	}
 
 	async send(msg, txt) {
@@ -141,16 +188,18 @@ class TgBot {
 				if (ytLinkRegexp.test(msg.text)) {
 					const sent = await this.send(msg, "Wait..");
 					const errHandler = (err) => {
-						//console.error(err.stack);
+						console.error(err.stack);
 						sent.text("Error. Invalid/Hidden video or Forbidden for download");
 					};
 
 					try {
 						const video = new DownloadVideo(msg.text);
+						video.setupParent(this);
 						const info = await video.getVideoInfo();
+						sent.text("Info getted, getting formats...");
 						//console.debug(info);
 						// Download only video with best quality
-						const qualities = await video.qualities();
+						const qualities = await video.qualities(true);
 						//console.debug(qualities);
 						let keyboard = [];
 						let keyboardM = [];
@@ -160,6 +209,7 @@ class TgBot {
 						for (let quality in qualities) {
 							const qualityItem = qualities[quality];
 							if (qualityItem.type !== null) {
+								if (config.allFormats || ({ mp4: true, m4a: true })[qualityItem.exc]) {
 									if (qualityItem.type === "audio")
 										kbChunkM.push({
 											text: `🎧 ${qualityItem.exc}/${qualityItem.resolution}`,
@@ -180,6 +230,7 @@ class TgBot {
 									}
 								}
 							}
+						}
 						if (kbChunk.length > 0) keyboard.push(kbChunk);
 						if (kbChunkM.length > 0) keyboardM.push(kbChunkM);
 						keyboard = keyboard.concat(keyboardM);
@@ -191,12 +242,19 @@ class TgBot {
 								const qualityItem = qualities[quality];
 								const actKey =
 									msg.from.id + ":" + quality + "_" + msg.message_id;
-								actions[actKey] = () => {
+								this.videosGC.register(actKey, () => {
+									//console.debug("ACT btn:", actKey);
 									sent
 										.download(video, qualityItem)
 										.then(() => {})
 										.catch(errHandler);
-								};
+								}, video);
+								/*this.videosGC.actions[actKey] = () => {
+									sent
+										.download(video, qualityItem)
+										.then(() => {})
+										.catch(errHandler);
+								};)*/
 							}
 							return Promise.resolve(null);
 						};

src/download.js

@@ -1,32 +1,114 @@
 const { spawn } = require("child_process");
+const { Readable } = require("stream");
+const EventEmitter = require("events");
 const mime = require("mime");
 const path = require("path");
+const FFMPEG = require("./ffmpeg");
 
 const formatsCache = new Map();
 const infoCache = new Map();
+
+const d = r => console.debug("DEBUG:", r) || r;
+
+class YtdlUpdater extends EventEmitter {
+	constructor () {
+		super();
+		this.updateRunned = false;
+		const updInterval = 3600000 * 24;  // production
+		//const updInterval = 30000;  // test
+		this.update();
+		setInterval(() => this.update(), updInterval);
+		this.locks = 0;
+	}
+	
+	lock (lockedBy) {
+		if (this.updateRunned) {
+			return false;  // Denied
+		}
+		this.locks++;
+		lockedBy.once("finish", () => {
+			if (--this.locks === 0) {
+				this.emit("resume");
+			}
+		});
+		return true;
+	}
+	
+	async update () {
+		if (this.updateRunned) return;
+		this.updateRunned = true;
+		this.emit("init");
+		if (this.locks > 0) await new Promise(rs => {
+			this.once("resume", rs);
+		});
+		this.emit("run");
+		
+		console.info("RUN UPDATE", new Date());
+		const code = await new Promise((resolve, reject) => {
+			const child = spawn("yt-dlp", ["-U"]);
+			child.on("close", (code) => {
+				resolve(code);
+			});
+		});
+		if (code !== 0) console.warn("yt-dlp update failed with error: " + code);
+		else console.info("Update finished");
+		
+		this.emit("finish");
+		this.updateRunned = false;
+	}
+}
+
 class DownloadVideo {
 	static get qualities () {
 		const q = {
 			webm_best: { exc: "webm", flag: "best", mime: "video/webm" },
-			mp4_best: { exc: "mp4", flag: "best[ext=mp4]", mime: "video/mp4" },
+			//mp4_best: { exc: "mp4", flag: "best[ext=mp4]", mime: "video/mp4" },
+			mp4_best: { exc: "mp4", flag: [
+				"bestvideo[height<=1080]+bestaudio/best", 
+				"--merge-output-format", "mp4"
+			], mime: "video/mp4" },
 		};
 		q.default = q.webm_best;
 		
 		return q;
 	}
 	
+	destructor () {
+		CLASS
+			.instances
+			.pop(CLASS
+				.instances
+				.indexOf(this));
+	}
+	
 	constructor(url, quality) {
 		const CLASS = this.__proto__.constructor;
+		if (!CLASS.instances) CLASS.instances = new Array();
+		CLASS.instances.push(this);
 
 		this.url = url;
 		this.quality = quality;
 		if (!this.quality) this.quality = CLASS.qualities.default;
 	}
 	
+	setupParent (parent) {
+		this.parent = parent;
+	}
+	
 	async getVideoInfo() {
 		const fromCache = infoCache.get(this.url);
 		if (fromCache) return fromCache;
 		
+		const signal = new EventEmitter();
+		if (!this.parent.updater.lock(signal)) {
+			await new Promise((rs => {
+				this.parent.updater.once("finish", () => {
+					this.parent.updater.lock(this);
+					rs();
+				});
+			}));
+		}
+		
 		const result = await new Promise((resolve, reject) => {
 			const child = spawn("yt-dlp", ["--cookies", "cookies.txt", "-j", this.url]);
 
@@ -63,6 +145,7 @@ class DownloadVideo {
 			});
 		});
 		infoCache.set(this.url, result);
+		signal.emit("finish");
 		return result;
 	}
 
@@ -70,6 +153,16 @@ class DownloadVideo {
 		const fromCache = formatsCache.get(this.url);
 		if (fromCache) return fromCache;
 		
+		const signal = new EventEmitter();
+		if (!this.parent.updater.lock(signal)) {
+			await new Promise((rs => {
+				this.parent.updater.once("finish", () => {
+					this.parent.updater.lock(this);
+					rs();
+				});
+			}));
+		}
+		
 		const result = await new Promise((resolve, reject) => {
 			const child = spawn("yt-dlp", ["--cookies", "cookies.txt", "-j", this.url]);
 			let data = "";
@@ -82,10 +175,12 @@ class DownloadVideo {
 			});
 		});
 		formatsCache.set(this.url, result);
+		signal.emit("finish");
 		return result;
 	}
 	
-	async qualities() {
+	async qualities(withDefaults = false) {
+		const CLASS = this.__proto__.constructor;
 		const formats = await this.getFormats();
 		const q = {};
 
@@ -98,7 +193,19 @@ class DownloadVideo {
 			
 			let flag = `bestaudio[ext=${f.ext}]`;
 			if (f.height) {
-				flag = `best[height<=${f.height}][ext=${f.ext}]`;
+				//flag = `best[height<=${f.height}][ext=${f.ext}]`;
+				flag = [
+					/*`bestvideo[height<=${
+						f.height
+					}]+bestaudio[ext=m4a]/best`,*/
+					`bestvideo[height<=${
+						f.height
+					}][vcodec^=avc1]+bestaudio[ext=m4a]/best[ext=${f.ext}]`,
+					/*`bestvideo[height<=${
+						f.height
+					}][vcodec^=mov]+bestaudio[ext=m4a]/best`,*/
+					"--merge-output-format", f.ext
+				];
 			}
 			//const flag = f.format_id;
 			
@@ -115,6 +222,12 @@ class DownloadVideo {
 				mime: mimeType, type: !mimeType ? null : type
 			};
 		});
+		if (withDefaults) {
+			for (let quality in CLASS.qualities) {
+				if (!q[quality])
+					q[quality] = CLASS.qualities[quality];
+			}
+		}
 
 		return q;
 	}
@@ -134,33 +247,90 @@ class DownloadVideo {
 		return false;
 	}
 	
-	async download(cb) {
+	async download(cb, isStream = false) {
-		const data = await new Promise((resolve, reject) => {
+		const signal = new EventEmitter();
+		if (!this.parent.updater.lock(signal)) {
+			await new Promise((rs => {
+				this.parent.updater.once("finish", () => {
+					this.parent.updater.lock(this);
+					rs();
+				});
+			}));
+		}
+		
+		const data = await new Promise(async (resolve, reject) => {
+			const flags = !Array.isArray(this.quality.flag) ? 
+				[this.quality.flag] : this.quality.flag;
 			const params = [
 				"--cookies", "cookies.txt",
-				"-f", this.quality.flag,
+				"-f", ...flags, "%%",
 				"-o", "-", this.url,
 			];
+			d(params);
 			const child = spawn("yt-dlp", params);
-
-			const chunks = [];
 			let downloaded = 0;
-
+			let lastPercent = 0;
-			child.stdout.on("data", (chunk) => {
+			let sizeTg = 0;
-				chunks.push(chunk);
+			let fine = false;
-				downloaded += chunk.length;
+			const getDwnld = () => `Real size: ${downloaded}, total: ${sizeTg}`;
-				//if (cb) cb({ downloaded });
-			});
 
 			child.stderr.on("data", (data) => {
 				const str = data.toString();
+				//console.debug(str);
 				const match = str.match(/(\d+\.\d+)%/);
 				if (match && cb) {
-					cb({ percent: parseFloat(match[1]), downloaded });
+					const percent = parseFloat(match[1]);
+					if (percent > lastPercent)
+						lastPercent = percent;
+					cb({ percent: lastPercent, downloaded: getDwnld(), fine });
 				}
 			});
 			
+			if (isStream) {
+				const stream = new Readable({ read() {} });
+				const ffmpeg = new FFMPEG(stream);
+
+				child.stdout.on("data", (chunk) => {
+					stream.push(chunk);
+					downloaded += chunk.length;
+					cb({ percent: lastPercent, downloaded: getDwnld(), fine });
+				});
+				
 				child.on("close", (code) => {
+					stream.push(null);
+					signal.emit("finish");
+					cb({ percent: 100, downloaded, fine: true });
+					fine = true;
+					if (code !== 0) reject(new Error(`yt-dlp exited with code ${code}`));
+				});
+
+				child.on("error", reject);
+				//return resolve(stream);
+				const resultedStream = await ffmpeg.convert();
+				//const resultedStream = stream;
+				
+				resultedStream.on("data", chunk => {
+					sizeTg += chunk.length;
+					if (sizeTg > 50 * 1024 * 1024) {
+						fine = true;
+						cb({ percent: lastPercent, downloaded, fine, msg: "Download failed: 50MB Telegram limit reached" });
+						resultedStream.push(null);
+					}
+				});
+				return resolve(resultedStream);
+			}
+
+			const chunks = [];
+			child.stdout.on("data", (chunk) => {
+				chunks.push(chunk);
+				downloaded += chunk.length;
+				cb({ percent: lastPercent, downloaded, fine: false });
+			});
+
+			child.on("close", (code) => {
+				fine = true;
+				signal.emit("finish");
+				cb({ percent: 100, downloaded, fine });
 				if (code === 0) {
 					resolve(Buffer.concat(chunks));
 				} else {
@@ -177,4 +347,4 @@ class DownloadVideo {
 	}
 }
 
-module.exports = DownloadVideo;
+module.exports = { DownloadVideo, YtdlUpdater };

src/ffmpeg.js

@@ -0,0 +1,59 @@
+const { Readable, PassThrough } = require("stream");
+const { spawn } = require("child_process");
+
+class FFMPEG {
+	constructor(data) {
+		this.data = data; // Can be Readable or Buffer
+		this.isBuffer = data instanceof Buffer;
+	}
+
+	convert() {
+		return new Promise((resolve, reject) => {
+			const ffmpegArgs = [
+				"-i",
+				"pipe:0",
+				"-c:v",
+				"libx264",
+				"-profile:v",
+				"high",
+				"-level",
+				"4.1",
+				"-c:a",
+				"aac",
+				"-crf",
+				"26",
+				"-preset",
+				"medium",
+				"-movflags",
+				"+frag_keyframe+empty_moov+default_base_moof",
+				"-f",
+				"mp4",
+				"pipe:1",
+			];
+
+			const ff = spawn("ffmpeg", ffmpegArgs);
+
+			const inputStream = this.isBuffer ? Readable.from(this.data) : this.data;
+
+			inputStream.pipe(ff.stdin);
+
+			const output = new PassThrough();
+			ff.stdout.pipe(output);
+
+			let errData = "";
+			ff.stderr.on("data", (chunk) => (errData += chunk.toString()));
+
+			ff.on("close", (code) => {
+				if (code !== 0) {
+					reject(new Error(`FFMPEG exited with code ${code}\n${errData}`));
+				}
+			});
+
+			ff.on("error", reject);
+
+			resolve(output);
+		});
+	}
+}
+
+module.exports = FFMPEG;