feat: add Gumloop flow trigger pipe

examples/typescript/pipe-gumloop-flow-executor/pipe.json

@@ -0,0 +1,70 @@
+{
+  "fields": [
+    {
+      "name": "interval",
+      "type": "number",
+      "default": 60,
+      "description": "Interval in seconds to trigger the Gumloop flow."
+    },
+    {
+      "name": "Frequency",
+      "type": "string",
+      "default": "daily",
+      "description": "Frequency of summary execution: 'daily' for once a day at flowTime, or 'hourly:X' for every X hours (e.g., 'hourly:4')."
+    },
+    {
+      "name": "flowTime",
+      "type": "time",
+      "default": "11:00",
+      "description": "Time to trigger the Gumloop flow daily (only if Frequency is 'daily')."
+    },
+    {
+      "name": "gumloopApiToken",
+      "type": "string",
+      "default": "<your_gumloop_api_token>",
+      "description": "API token for Gumloop authentication."
+    },
+    {
+      "name": "userId",
+      "type": "string",
+      "default": "<your_user_id>",
+      "description": "User ID for the Gumloop API request."
+    },
+    {
+      "name": "flowId",
+      "type": "string",
+      "default": "<your_flow_id>",
+      "description": "The saved item ID for the Gumloop flow to be triggered."
+    },
+    {
+      "name": "pageSize",
+      "type": "number",
+      "default": 100,
+      "description": "Number of records to retrieve from Screenpipe per page for structured extraction. Increase this value if using audio."
+    },
+    {
+      "name": "customPrompt",
+      "type": "string",
+      "default": "You are an AI assistant tasked with extracting structured information from screen data (OCR). Analyze the following screen data and extract relevant information about my daily activity.",
+      "description": "Custom prompt for the AI assistant that will be used to extract information from the screen data."
+    },
+    {
+      "name": "windowName",
+      "type": "string",
+      "default": "",
+      "description": "Filter by specific window name in screen data (e.g., 'gmail', 'slack', etc.)."
+    },
+    {
+      "name": "appName",
+      "type": "string",
+      "default": "",
+      "description": "Filter by specific application name in screen data (e.g., 'Chrome', 'Visual Studio Code', etc.)."
+    },
+    {
+      "name": "contentType",
+      "type": "string",
+      "default": "ocr",
+      "description": "Type of content to analyze: 'ocr', 'audio', or 'all'."
+    }
+  ]
+}

examples/typescript/pipe-gumloop-flow-executor/pipe.ts

@@ -0,0 +1,112 @@
+import * as fs from "node:fs";
+import { queryScreenpipe, loadPipeConfig, ContentItem, pipe } from "@screenpipe/js";
+import fetch from "node-fetch";
+import process from "node:process";
+
+async function triggerGumloopFlow(
+  gumloopApiToken: string,
+  userId: string,
+  flowId: string,
+  prompt: string,
+  screenData: ContentItem[]
+): Promise<void> {
+  const url = 'https://api.gumloop.com/api/v1/start_pipeline';
+  const headers = {
+    'Content-Type': 'application/json',
+    Authorization: `Bearer ${gumloopApiToken}`,
+  };
+
+  const data = {
+    user_id: userId,
+    saved_item_id: flowId,
+    pipeline_inputs: [
+      { input_name: 'prompt', value: prompt },
+      { input_name: 'screen_data', value: JSON.stringify(screenData) }
+    ]
+  };
+
+  const response = await fetch(url, {
+    method: 'POST',
+    headers: headers,
+    body: JSON.stringify(data),
+  });
+
+  const result = await response.json();
+  console.log(`Gumloop flow triggered with response: ${JSON.stringify(result)}`);
+}
+
+async function dailyGumloopPipeline(): Promise<void> {
+  console.log("Starting Gumloop flow execution");
+
+  const config = await loadPipeConfig();
+  const interval = config.interval * 1000;
+  const customPrompt = config.customPrompt;
+  const gumloopApiToken = config.gumloopApiToken;
+  const userId = config.userId;
+  const flowId = config.flowId;
+  const windowName = config.windowName || "";
+  const appName = config.appName || "";
+  const pageSize = config.pageSize;
+  const contentType = config.contentType || "ocr";
+
+  pipe.scheduler
+    .task("triggerGumloopFlow")
+    .every(interval)
+    .do(async () => {
+      const now = new Date();
+      const oneMinuteAgo = new Date(now.getTime() - interval);
+
+      const screenData = await queryScreenpipe({
+        startTime: oneMinuteAgo.toISOString(),
+        endTime: now.toISOString(),
+        windowName: windowName,
+        appName: appName,
+        limit: pageSize,
+        contentType: contentType,
+      });
+
+      if (screenData && screenData.data && screenData.data.length > 0) {
+        await triggerGumloopFlow(gumloopApiToken, userId, flowId, customPrompt, screenData.data);
+      }
+    });
+
+  if (config.Frequency === "daily") {
+    const [flowHour, flowMinute] = config.flowTime.split(":").map(Number);
+    pipe.scheduler
+      .task("triggerDailyGumloopFlow")
+      .every("1 day")
+      .at(`${flowHour}:${flowMinute}`)
+      .do(async () => {
+        const todayLogs = getTodayLogs();
+        if (todayLogs.length > 0) {
+          await triggerGumloopFlow(gumloopApiToken, userId, flowId, customPrompt, todayLogs);
+        }
+      });
+  } else if (config.Frequency.startsWith("hourly:")) {
+    const hours = parseInt(config.Frequency.split(":")[1], 10);
+    pipe.scheduler
+      .task("triggerHourlyGumloopFlow")
+      .every(`${hours} hours`)
+      .do(async () => {
+        const todayLogs = getTodayLogs();
+        if (todayLogs.length > 0) {
+          await triggerGumloopFlow(gumloopApiToken, userId, flowId, customPrompt, todayLogs);
+        }
+      });
+  }
+}
+
+function getTodayLogs(): ContentItem[] {
+  const logsDir = `${process.env.PIPE_DIR}/logs`;
+  const today = new Date().toISOString().replace(/:/g, "-").split("T")[0];
+  const files = fs.readdirSync(logsDir);
+  const todayFiles = files.filter(file => file.startsWith(today));
+  const logs: ContentItem[] = [];
+  for (const file of todayFiles) {
+    const content = fs.readFileSync(`${logsDir}/${file}`, "utf8");
+    logs.push(JSON.parse(content));
+  }
+  return logs;
+}
+
+dailyGumloopPipeline();