fixes

Author: Chetan11-dev

js/botasaurus-js/src/task.ts

@@ -20,6 +20,8 @@ import { FormatType } from './formats'
 export type TaskRunOptions<I=any> = {
     data: I
     metadata: any
+    taskId: number
+    parentTaskId: number | null
     isAborted: () => boolean
     pushData: (data: Record<string, any> | Record<string, any>[]) => Promise<void> 
 }

js/botasaurus-server-js/src/routes-db-logic.ts

@@ -261,7 +261,11 @@ async function performPatchTask(action: string, taskId: number): Promise<void> {
   const task = await new Promise<any>((resolve, reject) => {
     db.findOne(
       { id: taskId },
-      { id:1, is_all_task: 1, parent_task_id: 1, scraper_name: 1, status: 1 },
+      { 
+        // id:1,
+         is_all_task: 1, parent_task_id: 1, scraper_name: 1,
+        //  status: 1,
+         },
       (err, task) => {
         if (err) {
           reject(err);
@@ -276,14 +280,14 @@ async function performPatchTask(action: string, taskId: number): Promise<void> {
 
   if (task) {
     const removeDuplicatesBy = Server.getRemoveDuplicatesBy(task.scraper_name);
-    const { is_all_task, parent_task_id, status } = task;
+    const { is_all_task, parent_task_id } = task;
 
     if (action === 'delete') {
       await deleteTask(taskId, is_all_task, parent_task_id, removeDuplicatesBy);
     } else if (action === 'abort') {
-      await abortTask(taskId, is_all_task, parent_task_id, removeDuplicatesBy, status);
+      await abortTask(taskId, is_all_task, parent_task_id, removeDuplicatesBy);
     } else if (action === 'retry') {
-      await retryTask(taskId, is_all_task, parent_task_id, status);
+      await retryTask(taskId, is_all_task, parent_task_id);
     }
   }
 }
@@ -1071,13 +1075,12 @@ function convertUnicodeDictToAsciiDictInPlace(inputList: any[]): any[] {
     is_all_task: boolean,
     parentId: number | null,
     removeDuplicatesBy: any,
-    status: string
   ) {
     // Only abort tasks that are in PENDING or IN_PROGRESS status
-    const abortableStatuses: string[] = [TaskStatus.PENDING, TaskStatus.IN_PROGRESS];
-    if (!abortableStatuses.includes(status)) {
-      return;
-    }
+    // const abortableStatuses: string[] = [TaskStatus.PENDING, TaskStatus.IN_PROGRESS];
+    // if (!abortableStatuses.includes(status)) {
+    //   return;
+    // }
 
     let fn: (() => Promise<void>) | null = null;
 
@@ -1142,12 +1145,7 @@ function convertUnicodeDictToAsciiDictInPlace(inputList: any[]): any[] {
     taskId: number,
     is_all_task: boolean,
     parent_task_id: number | null,
-    status: string
   ): Promise<void> {
-    // Only allow retry for failed tasks
-    if (status !== TaskStatus.FAILED) {
-      return;
-    }
 
     if (is_all_task) {
         // Retry all failed children by setting them to pending

js/botasaurus-server-js/src/server.ts

@@ -103,6 +103,14 @@ type EmailSupportOptions = {
   body: string; // Default email body
 };
 
+export type OnCompleteInfo = {
+  taskId: number;
+  scraperName: string;
+  parentTaskId: number | null;
+  taskData: any;
+  result: any[];
+};
+
 type Scraper = {
   name: string;
   function: Function;
@@ -118,6 +126,7 @@ type Scraper = {
   default_sort: string;
   remove_duplicates_by: string | null;
   isGoogleChromeRequired: boolean;
+  onComplete?: (info: OnCompleteInfo) => void | Promise<void>;
 };
 class _Server {
   scrapers: Record<string, Scraper> = {};
@@ -244,7 +253,8 @@ class _Server {
       sorts = [],
       views = [],
       removeDuplicatesBy = null,
-      isGoogleChromeRequired
+      isGoogleChromeRequired,
+      onComplete
   }: {
     productName?: string | null;
     getTaskName?: Function;
@@ -255,6 +265,7 @@ class _Server {
     views?: View[];
     removeDuplicatesBy?: string | null;
     isGoogleChromeRequired?: boolean;
+    onComplete?: (info: OnCompleteInfo) => void | Promise<void>;
   } = {}
   ): void {
     // @ts-ignore
@@ -355,7 +366,8 @@ class _Server {
       views,
       default_sort: defaultSort as any,
       remove_duplicates_by: removeDuplicatesBy,
-      isGoogleChromeRequired: isGoogleChromeRequired || (scraper_type === ScraperType.BROWSER)
+      isGoogleChromeRequired: isGoogleChromeRequired || (scraper_type === ScraperType.BROWSER),
+      onComplete
     };
   }
 
@@ -413,6 +425,10 @@ class _Server {
     return this.scrapers[scraperName].remove_duplicates_by;
   }
 
+  getOnComplete(scraperName: string): ((info: OnCompleteInfo) => void | Promise<void>) | undefined {
+    return this.scrapers[scraperName]?.onComplete;
+  }
+
   getScrapersNames(): string[] {
     return Object.keys(this.scrapers);
   }

js/botasaurus-server-js/src/task-executor.ts

@@ -9,6 +9,7 @@ import { formatExc } from 'botasaurus/utils'
 import { cleanDataInPlace } from 'botasaurus/output'
 import { getScraperErrorMessage, Server } from './server'
 import { isLargeFile, isNotEmptyObject, omitKeys } from './utils'
+import { readNdJsonCallback } from './ndjson'
 
 /**
  * Task priority levels for queue ordering
@@ -511,6 +512,8 @@ class TaskExecutor {
         try {
             const result = await fn(taskData, {
                 ...metadata,
+                taskId,
+                parentTaskId: parent_task_id,
                 isAborted,
                 pushData,
                 parallel: null,
@@ -565,6 +568,25 @@ class TaskExecutor {
                 }
                 await this.reportTaskSuccess(taskId, processedResult, isResultDontCached, scraperName, taskData, parent_task_id, key)
             }
+
+            const onComplete = Server.getOnComplete(scraperName)
+            if (onComplete) {
+                try {
+                    let onCompleteResult: any[]
+                    if (pushDataWriter.wasUsed()) {
+                        onCompleteResult = []
+                        await readNdJsonCallback(pushDataWriter.getFilePath(), async (item) => {
+                            onCompleteResult.push(item)
+                        })
+                    } else {
+                        onCompleteResult = Array.isArray(processedResult) ? processedResult : []
+                    }
+                    await onComplete({ taskId, scraperName, parentTaskId: parent_task_id, taskData: task.data, result: onCompleteResult })
+                } catch (error) {
+                    console.error(`[onComplete] Error in callback for ${scraperName}:`, error)
+                }
+            }
+
         } catch (error) {
             // Release task resources FIRST to prevent capacity leak
             releaseTask()