From 65f3cb6561f7bde99e6fa39489452570d810656f Mon Sep 17 00:00:00 2001
From: Christian Weilbach <ch_weil@topiq.es>
Date: Fri, 12 Apr 2024 06:58:32 +0000
Subject: [PATCH] Factor text-extraction runtime.

---
 .../simm/runtimes/relational_assistance.clj   |  60 +---------
 src/ie/simm/runtimes/text_extractor.clj       | 103 ++++++++++++++++++
 src/ie/simm/towers.clj                        |   7 +-
 3 files changed, 111 insertions(+), 59 deletions(-)
 create mode 100644 src/ie/simm/runtimes/text_extractor.clj

diff --git a/src/ie/simm/runtimes/relational_assistance.clj b/src/ie/simm/runtimes/relational_assistance.clj
index c14ae03..0ce5754 100644
--- a/src/ie/simm/runtimes/relational_assistance.clj
+++ b/src/ie/simm/runtimes/relational_assistance.clj
@@ -34,7 +34,7 @@
                 conn)
               (catch Exception _
                 (d/connect cfg)))]
-        (d/transact conn default-schema)
+        #_(d/transact conn default-schema)
         (swap! peer assoc-in [:conn chat-id] conn)
         conn)))
 
@@ -101,26 +101,6 @@
         (when (seq tags)
           {:message/tag tags}))]))))
 
-
-;; TODO factor into youtube middleware
-;; require libpython
-
-(require '[libpython-clj2.require :refer [require-python]] 
-         '[libpython-clj2.python :refer [py. py.. py.-] :as py])
-
-(require-python '[youtube_transcript_api :refer [YouTubeTranscriptApi]])
-
-(defn youtube-transcript [video-id]
-  ;; " ".join([t['text'] for t in transcript])
-  (let [transcript (py. YouTubeTranscriptApi get_transcript video-id)]
-    (str/join " " (map :text transcript))))
-
-(comment
-  (youtube-transcript "20TAkcy3aBY")
-
-  )
-
-
 (def window-size 10)
 
 (defn summarize [S conn conv chat]
@@ -195,33 +175,6 @@
     (.close zip-out)
     zip-file))
 
-(defn extract-url [S text chat]
-  (go-try S
-          (if-let [;; if text matches http or https web URL extrect URL with regex
-                   url (re-find #"https?://\S+" text)]
-            (if-let [;; extract youtube video id from URL
-                     youtube-id (second (or (re-find #"youtube.com/watch\?v=([^&]+)" url)
-                                            (re-find #"youtu.be/([^\?]+)" url)))]
-              (try
-                (debug "summarizing youtube transcript" youtube-id)
-                (let [transcript (youtube-transcript youtube-id)
-                      summary (<? S (cheap-llm (format pr/summarization transcript)))
-                      summary (str "Youtube transcript summary:\n" summary "\n" url)]
-                  (<? S (send-text! (:id chat) summary))
-                  summary)
-                (catch Exception e
-                  (warn "Could not extract transcript from youtube video" youtube-id e)
-                  text))
-              (try
-                (let [body (<? S (extract-body url))
-                      summary (<? S (cheap-llm (format pr/summarization body)))]
-                  (<? S (send-text! (:id chat) summary))
-                  (str "Website summary:\n" summary "\n" url))
-                (catch Exception e
-                  (warn "Could not extract body from URL" url e)
-                  text)))
-            text)))
-
 (defn relational-assistance
   "This interpreter can derive facts and effects through a relational database."
   [[S peer [in out]]]
@@ -250,20 +203,11 @@
                  (when m
                    (binding [lb/*chans* [next-in pi out po]]
                      (let [{:keys [msg]
-                            {:keys [text chat voice-path from]} :msg} m]
+                            {:keys [chat from]} :msg} m]
                        (try
                          (let [_ (debug "received message" m)
                                firstname (:first_name from)
                                start-time-ms (System/currentTimeMillis)
-                               text (if-not voice-path text
-                                            (let [transcript (<? S (stt-basic voice-path))
-                                                  transcript (str "Voice transcript " (:username from) ":\n" transcript)]
-                                              (when text (warn "Ignoring text in favor of voice message"))
-                                              (debug "created transcript" transcript)
-                                              (<? S (send-text! (:id chat) transcript))
-                                              transcript))
-                               text (<? S (extract-url S text chat))
-                               msg (assoc msg :text text)
 
                                conn (ensure-conn peer (:id chat))
                                _ (debug "conn" conn)
diff --git a/src/ie/simm/runtimes/text_extractor.clj b/src/ie/simm/runtimes/text_extractor.clj
new file mode 100644
index 0000000..43018a8
--- /dev/null
+++ b/src/ie/simm/runtimes/text_extractor.clj
@@ -0,0 +1,103 @@
+(ns ie.simm.runtimes.text-extractor
+  "Extract texts from incoming message and augment the text.
+   
+   Properties: stateless"
+  (:require [ie.simm.languages.bindings :as lb]
+            [ie.simm.languages.gen-ai :refer [cheap-llm stt-basic]]
+            [ie.simm.languages.chat :refer [send-text!]]
+            [ie.simm.languages.browser :refer [extract-body]]
+            [ie.simm.prompts :as pr]
+            [superv.async :refer [<?? go-try S go-loop-try <? >? put? go-for] :as sasync]
+            [clojure.core.async :refer [chan pub sub mult tap timeout] :as async]
+            [taoensso.timbre :refer [debug info warn error]]
+            [hasch.core :refer [uuid]]
+            [clojure.string :as str]
+            [libpython-clj2.require :refer [require-python]]
+            [libpython-clj2.python :refer [py. py.. py.-] :as py]))
+
+(require-python '[youtube_transcript_api :refer [YouTubeTranscriptApi]])
+
+(defn youtube-transcript [video-id]
+  ;; " ".join([t['text'] for t in transcript])
+  (let [transcript (py. YouTubeTranscriptApi get_transcript video-id)]
+    (str/join " " (map :text transcript))))
+
+(comment
+  (youtube-transcript "20TAkcy3aBY")
+
+  )
+
+(defn extract-url [S text chat]
+  (go-try S
+          (if-let [;; if text matches http or https web URL extrect URL with regex
+                   url (re-find #"https?://\S+" text)]
+            (if-let [;; extract youtube video id from URL
+                     youtube-id (second (or (re-find #"youtube.com/watch\?v=([^&]+)" url)
+                                            (re-find #"youtu.be/([^\?]+)" url)))]
+              (try
+                (debug "summarizing youtube transcript" youtube-id)
+                (let [transcript (youtube-transcript youtube-id)
+                      summary (<? S (cheap-llm (format pr/summarization transcript)))
+                      summary (str "Youtube transcript summary:\n" summary "\n" url)]
+                  (<? S (send-text! (:id chat) summary))
+                  summary)
+                (catch Exception e
+                  (warn "Could not extract transcript from youtube video" youtube-id e)
+                  text))
+              (try
+                (let [body (<? S (extract-body url))
+                      summary (<? S (cheap-llm (format pr/summarization body)))]
+                  (<? S (send-text! (:id chat) summary))
+                  (str "Website summary:\n" summary "\n" url))
+                (catch Exception e
+                  (warn "Could not extract body from URL" url e)
+                  text)))
+            text)))
+
+(defn text-extractor
+  [[S peer [in out]]]
+  ;; pass everything from in to next-in for the next middleware
+  ;; and create publication channel for runtime context
+  (let [pi (pub in :type)
+        next-in (chan 1000)
+
+        ;; subscriptions for this runtime context
+        p (pub in (fn [{:keys [type]}]
+                    (or ({:ie.simm.runtimes.telegram/message ::message} type)
+                        :unrelated)))
+        msg-ch (chan 1000)
+        _ (sub p ::message msg-ch)
+        _ (sub p :unrelated next-in)
+
+       ;; do the same in reverse for outputs from below
+        prev-out (chan)
+        mo (mult prev-out)
+        _ (tap mo out)
+        pub-out (chan)
+        _ (tap mo pub-out)
+        po (pub pub-out :type)]
+    (go-loop-try S [m (<? S msg-ch)]
+                 (when m
+                   (binding [lb/*chans* [next-in pi out po]]
+                     (let [{:keys [msg]
+                            {:keys [text chat voice-path from]} :msg} m]
+                       (try
+                         (let [_ (debug "received message" m)
+                               text (if-not voice-path text
+                                            (let [transcript (<? S (stt-basic voice-path))
+                                                  transcript (str "Voice transcript " (:username from) ":\n" transcript)]
+                                              (when text (warn "Ignoring text in favor of voice message"))
+                                              (debug "created transcript" transcript)
+                                              (<? S (send-text! (:id chat) transcript))
+                                              transcript))
+                               text (<? S (extract-url S text chat))
+                               msg (assoc msg :text text)
+                               m (assoc m :msg msg)]
+                           (prn "NEW M" m)
+                           (>? S next-in m))
+                         (catch Exception e
+                           (let [error-id (uuid)]
+                             (error "Could not process message(" error-id "): " m e)
+                             (<? S (send-text! (:id chat) (str "Sorry, I could not process your message. Error: " error-id))))))))
+                   (recur (<? S msg-ch))))
+    [S peer [next-in prev-out]]))
diff --git a/src/ie/simm/towers.clj b/src/ie/simm/towers.clj
index 9d684ed..6c22850 100644
--- a/src/ie/simm/towers.clj
+++ b/src/ie/simm/towers.clj
@@ -4,6 +4,7 @@
             [ie.simm.runtimes.report :refer [report]]
             [ie.simm.runtimes.etaoin :refer [etaoin]]
             [ie.simm.runtimes.telegram :refer [telegram long-polling]]
+            [ie.simm.runtimes.text-extractor :refer [text-extractor]]
             [ie.simm.runtimes.relational-assistance :refer [relational-assistance]]
             [kabel.peer :refer [drain]]
             [clojure.core.async :refer [close!]]
@@ -17,7 +18,7 @@
   [S peer [in out]])
 
 (defn default []
-  (comp drain brave etaoin openai relational-assistance telegram codrain))
+  (comp drain brave etaoin openai relational-assistance text-extractor telegram codrain))
 
 (defn debug [] 
   (comp drain 
@@ -30,6 +31,8 @@
         (partial report #(println "openai: " (:type %) (:request-id %)))
         relational-assistance
         (partial report #(println "relational-assistance: " (:type %) (:request-id %)))
+        text-extractor
+        (partial report #(println "text-extractor: " (:type %) (:request-id %)))
         telegram
         (partial report #(println "telegram: " (:type %) (:request-id %)))
         codrain))
@@ -45,6 +48,8 @@
          (partial report #(println "openai: " (:type %) (:request-id %)))
          relational-assistance
          (partial report #(println "relational-assistance: " (:type %) (:request-id %)))
+         text-extractor
+         (partial report #(println "text-extractor: " (:type %) (:request-id %)))
          (partial telegram long-polling)
          (partial report #(println "telegram: " (:type %) (:request-id %)))
          codrain))
\ No newline at end of file