Merge branch 'main' into update-formula-aarch64-apple-darwin-0e24c5ca…

…a49839487e3c18fcd69931d1a61feccb

Formula/screenpipe.rb

@@ -10,7 +10,7 @@ class Screenpipe < Formula
       sha256 "924b079f24939be270dc74c2040e117ff7229ed285fb7f246ed52c6dbf6d160d" # arm64
     else
       url "https://github.com/mediar-ai/screenpipe/releases/download/v#{version}/screenpipe-#{version}-x86_64-apple-darwin.tar.gz"
-      sha256 "8acabe46b935ac8ed6607a43a5d47e81b5647ee2c52193bcb3b646628a4ced9b" # x86_64
+      sha256 "98e91de2eb3f6823ca5974ab48fb1d28b09ae5c382893f510838f9213dfe618a" # x86_64
     end
   end
 

examples/typescript/pipe-sync-meetings-to-notion/README.md

@@ -0,0 +1,87 @@
+
+
+in screenpipe we have a plugin system called "pipe store" or "pipes"
+
+think of it like this:
+
+screenpipe data -> your pipe like "AI annotate" or "send to salesforce"
+
+a more dev-friendly explanation:
+
+screenpipe | AI tag | notion update
+
+or 
+
+screenpipe | AI tag | slack send report
+
+or 
+
+screenpipe | fill salesforce
+
+or 
+
+screenpipe | logs daily
+
+basically it would read, process, annotate, analyse, summarize, send, your data customisable to your desire, effortlessly
+
+### pipe-sync-meetings-to-notion
+
+this is an experimental, but official pipe, that will sync your meetings to notion
+
+this is how you run it through the app:
+
+
+## Setup
+
+1. create a notion integration:
+   - go to https://www.notion.so/my-integrations
+   - click "new integration"
+   - give it a name (e.g., "screenpipe meeting sync")
+   - select the workspace where you want to sync your meetings
+   - click "submit" to create the integration
+
+2. get your notion api key:
+   - in the integration page, find the "internal integration token"
+   - copy this token, you'll need it later
+
+3. create a database in notion:
+   - create a new page in notion
+   - add a database to this page
+   - add columns: title, date, transcription + optionally notion ai columns if you want
+   - share this page with your integration (click three dots, connections, your integration)
+
+4. get your notion database id:
+   - open your database in notion
+   - look at the url, it should look like: https://www.notion.so/yourworkspace/83c75a51b3bd4a)
+   - the part after the last slash and before the ? is your database id
+
+now, your meeting transcriptions will automatically sync to your notion database!
+
+
+if you're in dev mode you can run the cli like this:
+
+```bash
+export SCREENPIPE_NOTION_API_KEY=secret_abcd
+export SCREENPIPE_NOTION_DATABASE_ID=1234567890
+screenpipe --pipe ./examples/typescript/pipe-sync-meetings-to-notion/main.js
+```
+
+please look at the code, it's 99% normal JS but there are limitations currently:
+- you cannot use dependencies (yet)
+- untested with typescript (but will make pipes TS first soon)
+
+i recommend you copy paste the current main.js file into AI and ask some changes for whatever you want to do, make sure to run an infinite loop also
+
+get featured in the pipe store:
+
+<img width="1312" alt="Screenshot 2024-08-27 at 17 06 45" src="https://github.com/user-attachments/assets/b6856bf4-2cfd-4888-be11-ee7baae6b84b">
+
+just ask @louis030195
+
+### what's next for pipes
+
+- use dependencies (like vercel/ai so cool)
+- TS
+- access to screenpipe desktop api (e.g. trigger notifications, customise what cursor-like @ are in the chat, etc.)
+- easier to publish your pipes (like obsidian store)
+- everything frictionless, effortless, and maximize the value you get out of screenpipe

examples/typescript/pipe-sync-meetings-to-notion/main.js

@@ -0,0 +1,94 @@
+const INTERVAL = 30 * 1000; // 30 seconds in milliseconds
+const NOTION_API_URL = 'https://api.notion.com/v1/pages';
+const NOTION_DATABASE_ID = process.env.SCREENPIPE_NOTION_DATABASE_ID;
+const NOTION_API_KEY = process.env.SCREENPIPE_NOTION_API_KEY;
+
+
+
+async function queryScreenpipe(startTime, endTime) {
+    try {
+        const queryParams = `start_time=${startTime}&end_time=${endTime}&limit=50&content_type=audio`;
+        const response = await fetch(`http://localhost:3030/search?${queryParams}`);
+        if (!response.ok) {
+            throw new Error(`HTTP error! status: ${response.status}`);
+        }
+        return await response.json();
+    } catch (error) {
+        console.error("Error querying screenpipe:", error.toString());
+        return [];
+    }
+}
+
+async function syncAudioToNotion(audioData) {
+    try {
+        const title = `Audio - ${audioData.content.timestamp}`;
+        const date = audioData.content.timestamp;
+        const transcription = audioData.content.transcription;
+
+        // Split transcription into chunks of 2000 characters
+        const chunks = splitTranscription(transcription);
+
+        for (let i = 0; i < chunks.length; i++) {
+            const response = await fetch(NOTION_API_URL, {
+                method: 'POST',
+                headers: {
+                    'Authorization': `Bearer ${NOTION_API_KEY}`,
+                    'Notion-Version': '2022-06-28',
+                    'Content-Type': 'application/json',
+                },
+                body: JSON.stringify({
+                    parent: { database_id: NOTION_DATABASE_ID },
+                    properties: {
+                        Title: { title: [{ text: { content: `${title} (Part ${i + 1}/${chunks.length})` } }] },
+                        Date: { date: { start: date } },
+                        Transcription: { rich_text: [{ text: { content: chunks[i] } }] },
+                    },
+                }),
+            });
+
+            if (!response.ok) {
+                const errorBody = await response.text();
+                throw new Error(`HTTP error! status: ${response.status}, body: ${errorBody}`);
+            }
+        }
+
+        console.log("Audio synced to Notion successfully");
+    } catch (error) {
+        console.error("Error syncing audio to Notion:", error);
+    }
+}
+
+function splitTranscription(transcription, chunkSize = 2000) {
+    const chunks = [];
+    for (let i = 0; i < transcription.length; i += chunkSize) {
+        chunks.push(transcription.slice(i, i + chunkSize));
+    }
+    return chunks;
+}
+
+async function streamAudioToNotion() {
+    console.log("Starting Audio Stream to Notion");
+
+    while (true) {
+        try {
+            const now = new Date();
+            const thirtySecondsAgo = new Date(now.getTime() - INTERVAL);
+
+            const audioData = await queryScreenpipe(thirtySecondsAgo.toISOString(), now.toISOString());
+
+            console.log("Audio data:", audioData);
+            for (const audio of audioData.data) {
+                await syncAudioToNotion(audio);
+            }
+        } catch (error) {
+            console.error("Error syncing audio to Notion:", {
+                message: error.message,
+                stack: error.stack,
+                audioData: JSON.stringify(audioData)
+            });
+        }
+        await new Promise((resolve) => setTimeout(resolve, INTERVAL));
+    }
+}
+
+streamAudioToNotion();

screenpipe-core/src/deno/runtime.js

@@ -32,10 +32,37 @@ const pipe = {
         const response = await ops.op_fetch_post(url, body);
         return JSON.parse(response);
     },
+    fetch: async (url, options) => {
+        try {
+            const responseString = await ops.op_fetch(url, options);
+            const response = JSON.parse(responseString);
+            return {
+                ok: response.status >= 200 && response.status < 300,
+                status: response.status,
+                statusText: response.statusText,
+                headers: response.headers, // Use the headers directly without wrapping in Headers object
+                text: async () => response.text,
+                json: async () => {
+                    try {
+                        return JSON.parse(response.text);
+                    } catch (error) {
+                        console.error("Error parsing JSON:", error);
+                        return response.text;
+                    }
+                },
+            };
+        } catch (error) {
+            console.error("Fetch error:", error);
+            throw error;
+        }
+    },
 };
 
 globalThis.setTimeout = (callback, delay) => {
     ops.op_set_timeout(delay).then(callback);
 };
 globalThis.console = console;
-globalThis.pipe = pipe;
+globalThis.pipe = pipe;
+globalThis.fetch = pipe.fetch;
+
+

screenpipe-core/src/pipes.rs

@@ -12,9 +12,77 @@ mod pipes {
     use reqwest::header::HeaderMap;
     use reqwest::header::HeaderValue;
     use reqwest::header::CONTENT_TYPE;
+    use std::collections::HashMap;
     use std::env;
     use std::rc::Rc;
 
+    use reqwest::Client;
+    use serde_json::Value;
+
+    #[op2]
+    #[string]
+    fn op_get_env(#[string] key: String) -> Option<String> {
+        env::var(&key).ok()
+    }
+
+    #[op2(async)]
+    #[string]
+    async fn op_fetch(
+        #[string] url: String,
+        #[serde] options: Option<Value>,
+    ) -> anyhow::Result<String> {
+        let client = Client::new();
+        let mut request = client.get(&url);
+
+        if let Some(opts) = options {
+            if let Some(method) = opts.get("method").and_then(|m| m.as_str()) {
+                request = match method.to_uppercase().as_str() {
+                    "GET" => client.get(&url),
+                    "POST" => client.post(&url),
+                    "PUT" => client.put(&url),
+                    "DELETE" => client.delete(&url),
+                    // Add other methods as needed
+                    _ => return Err(anyhow::anyhow!("Unsupported HTTP method")),
+                };
+            }
+
+            if let Some(headers) = opts.get("headers").and_then(|h| h.as_object()) {
+                for (key, value) in headers {
+                    if let Some(value_str) = value.as_str() {
+                        request = request.header(key, value_str);
+                    }
+                }
+            }
+
+            if let Some(body) = opts.get("body").and_then(|b| b.as_str()) {
+                request = request.body(body.to_string());
+            }
+        }
+
+        let response = match request.send().await {
+            Ok(resp) => resp,
+            Err(e) => return Err(anyhow::anyhow!(e)),
+        };
+
+        let status = response.status();
+        let headers = response.headers().clone();
+        let text = match response.text().await {
+            Ok(t) => t,
+            Err(e) => return Err(anyhow::anyhow!(e)),
+        };
+
+        let result = serde_json::json!({
+            "status": status.as_u16(),
+            "statusText": status.to_string(),
+            "headers": headers.iter()
+                .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
+                .collect::<HashMap<String, String>>(),
+            "text": text,
+        });
+
+        Ok(result.to_string())
+    }
+
     #[op2(async)]
     #[string]
     async fn op_read_file(#[string] path: String) -> anyhow::Result<String> {
@@ -177,6 +245,8 @@ mod pipes {
             op_fetch_get,
             op_fetch_post,
             op_set_timeout,
+            op_fetch,
+            op_get_env,
         ]
     }
 
@@ -189,6 +259,18 @@ mod pipes {
             ..Default::default()
         });
 
+        // add all env var starting with SCREENPIPE_ to the global scope in process.env
+
+        // first init the process.env object
+        js_runtime.execute_script("main", "globalThis.process = { env: {} }")?;
+
+        for (key, value) in env::vars() {
+            if key.starts_with("SCREENPIPE_") {
+                js_runtime
+                    .execute_script("main", format!("process.env['{}'] = '{}'", key, value))?;
+            }
+        }
+
         let mod_id = js_runtime.load_main_es_module(&main_module).await?;
         let result = js_runtime.mod_evaluate(mod_id);
         js_runtime.run_event_loop(Default::default()).await?;