save changes

src/generateImage.ts

@@ -24,8 +24,11 @@ const servers = [
     },
 ];
 
-async function processServerQueue(server: any, videos: any[]) {
+const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
-    for (const video of videos) {
+
+async function processVideo(server: any, video: any) {
+    const MAX_RETRIES = 3;
+    for (let i = 0; i < MAX_RETRIES; i++) {
         try {
             const fileName = `${video.id}_${video.genre}_${video.sub_genre}.png`.replace(/\s/g, '_');
             const imagePath = await generateImage(
@@ -39,34 +42,47 @@ async function processServerQueue(server: any, videos: any[]) {
 
             await query('UPDATE video SET image_path = ? WHERE id = ?', [imagePath, video.id]);
             logger.info(`Generated and saved image for video ${video.id} at ${imagePath}`);
+            return; // Success
         } catch (error) {
-            logger.error(`Failed to generate image for video ${video.id}:`, error);
+            logger.error(`Failed to generate image for video ${video.id} on server ${server.baseUrl}:`, error);
+            if (i < MAX_RETRIES - 1) {
+                logger.info(`Retrying in 1 minute... (${i + 1}/${MAX_RETRIES})`);
+                await sleep(60000);
+            } else {
+                logger.error(`All retries failed for video ${video.id} on server ${server.baseUrl}.`);
+            }
         }
     }
 }
 
+async function worker(server: any) {
+    while (true) {
+        const videosToProcess = (await query(
+            "SELECT * FROM video WHERE image_prompt IS NOT NULL AND (image_path IS NULL OR image_path = '') LIMIT 1"
+        )) as any[];
+
+        if (videosToProcess.length === 0) {
+            logger.info('No more images to generate.');
+            await sleep(10000); // Wait for 10 seconds before checking again
+            continue;
+        }
+
+        const video = videosToProcess[0];
+        // Mark the video as being processed to avoid other workers picking it up
+        await query("UPDATE video SET image_path = 'processing' WHERE id = ?", [video.id]);
+
+        await processVideo(server, video);
+    }
+}
+
 async function main() {
     try {
-        const videosToProcess = await query(
+        const promises = servers.map(server => {
-            "SELECT * FROM video WHERE image_prompt IS NOT NULL AND (image_path IS NULL OR image_path = '')"
-        ) as any[];
-
-        if (videosToProcess.length === 0) {
-            logger.info('No images to generate.');
-            return;
-        }
-
-        const queues: VideoRecord[][] = servers.map(() => []);
-        videosToProcess.forEach((video, index) => {
-            queues[index % servers.length].push(video);
-        });
-
-        const promises = servers.map((server, index) => {
             if (!server.baseUrl || !server.outputDir) {
-                logger.warn(`Server ${index + 1} is not configured. Skipping.`);
+                logger.warn(`Server is not configured. Skipping.`);
                 return Promise.resolve();
             }
-            return processServerQueue(server, queues[index]);
+            return worker(server);
         });
 
         await Promise.all(promises);

src/generateVideo.ts

@@ -26,12 +26,26 @@ const servers = [
     },
 ];
 
-async function processServerQueue(server: any, videos: VideoRecord[]) {
+const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
-    for (const video of videos) {
+
+async function processVideo(server: any, video: VideoRecord) {
+    const MAX_RETRIES = 3;
+    for (let i = 0; i < MAX_RETRIES; i++) {
         try {
             const imageFileName = path.basename(video.image_path);
-            const serverImagePath = path.join(server.outputDir.replace('output','input'), imageFileName);
+            const generatedPath = path.resolve(__dirname, '..', 'generated');
-            await fs.copyFile(video.image_path, serverImagePath);
+            const localImagePath = path.join(generatedPath, imageFileName);
+
+            try {
+                await fs.access(localImagePath);
+            } catch (error) {
+                logger.error(`Image file not found for video ${video.id} at ${localImagePath}. Skipping.`);
+                await query("UPDATE video SET video_path = 'image not found' WHERE id = ?", [video.id]);
+                return;
+            }
+
+            const serverImagePath = path.join(server.outputDir.replace('output', 'input'), imageFileName);
+            await fs.copyFile(localImagePath, serverImagePath);
 
             const videoFileName = `${video.id}_${video.genre}_${video.sub_genre}.mp4`.replace(/\s/g, '_');
             const videoPath = await generateVideo(
@@ -46,34 +60,47 @@ async function processServerQueue(server: any, videos: VideoRecord[]) {
             const absolutePath = path.resolve(videoPath);
             await query('UPDATE video SET video_path = ? WHERE id = ?', [absolutePath, video.id]);
             logger.info(`Generated and saved video for video ${video.id} at ${absolutePath}`);
+            return; // Success
         } catch (error) {
-            logger.error(`Failed to generate video for video ${video.id}:`, error);
+            logger.error(`Failed to generate video for video ${video.id} on server ${server.baseUrl}:`, error);
+            if (i < MAX_RETRIES - 1) {
+                logger.info(`Retrying in 1 minute... (${i + 1}/${MAX_RETRIES})`);
+                await sleep(60000);
+            } else {
+                logger.error(`All retries failed for video ${video.id} on server ${server.baseUrl}.`);
+            }
         }
     }
 }
 
+async function worker(server: any) {
+    while (true) {
+        const videosToProcess = (await query(
+            "SELECT * FROM video WHERE video_prompt IS NOT NULL AND image_path IS NOT NULL AND image_path != 'processing' AND (video_path IS NULL OR video_path = '') LIMIT 1"
+        )) as VideoRecord[];
+
+        if (videosToProcess.length === 0) {
+            logger.info('No more videos to generate.');
+            await sleep(10000); // Wait for 10 seconds before checking again
+            continue;
+        }
+
+        const video = videosToProcess[0];
+        // Mark the video as being processed to avoid other workers picking it up
+        await query("UPDATE video SET video_path = 'processing' WHERE id = ?", [video.id]);
+
+        await processVideo(server, video);
+    }
+}
+
 async function main() {
     try {
-        const videosToProcess = await query(
+        const promises = servers.map(server => {
-            "SELECT * FROM video WHERE video_prompt IS NOT NULL AND image_path IS NOT NULL AND (video_path IS NULL OR video_path = '')"
-        ) as VideoRecord[];
-
-        if (videosToProcess.length === 0) {
-            logger.info('No videos to generate.');
-            return;
-        }
-
-        const queues: VideoRecord[][] = servers.map(() => []);
-        videosToProcess.forEach((video, index) => {
-            queues[index % servers.length].push(video);
-        });
-
-        const promises = servers.map((server, index) => {
             if (!server.baseUrl || !server.outputDir) {
-                logger.warn(`Server ${index + 1} is not configured. Skipping.`);
+                logger.warn(`Server is not configured. Skipping.`);
                 return Promise.resolve();
             }
-            return processServerQueue(server, queues[index]);
+            return worker(server);
         });
 
         await Promise.all(promises);