replying to emails

src/clj/auto_ap/background/mail.clj

@@ -1,18 +1,61 @@
 (ns auto-ap.background.mail
   (:require [amazonica.aws.sqs :as sqs]
             [amazonica.aws.s3 :as s3]
-            [clojure.data.json :as json])
-  )
+            [clojure.data.json :as json]
+            [clojure-mail.message :as message]
+            [clojure.string :as str]
+            [clojure.java.io :as io]
+            [auto-ap.parse :as parse]
+            [auto-ap.db.invoices :as invoices]
+            [auto-ap.db.companies :as companies]
+            )
+  (:import [java.util Properties]
+           [java.util UUID]
+           [javax.mail.search FlagTerm]
+           [java.io FileInputStream File]
+           [javax.mail.internet MimeMessage]
+           [javax.mail Session
+            Folder
+            Flags
+            Flags$Flag AuthenticationFailedException]
+           (com.sun.mail.imap IMAPStore)))
+
+(def  queue-url "https://sqs.us-east-1.amazonaws.com/679918342773/integreat-mail-prod")
 
 (defn process-sqs []
-  (doseq [message (:messages (sqs/receive-message {:queue-url "https://sqs.us-east-1.amazonaws.com/679918342773/integreat-mail-prod"
-                                                   :wait-time-seconds 5
-                                                   :max-number-of-messages 10
-                                                   #_#_:attribute-names ["All"]}))]
-    (let [message-body (json/read-str (:body message)
-                                      :key-fn keyword)]
-      (doseq [r (:Records message-body)] 
-        (println (-> (s3/get-object {:key (-> r :s3 :object :key)
-                                     :bucket-name (-> r :s3 :bucket :name)})
-                     :input-stream
-                     slurp))))))
+  (println "Fetching messages from sqs...")
+  (let [companies (companies/get-all)]
+    (doseq [message (:messages (sqs/receive-message {:queue-url queue-url 
+                                                     :wait-time-seconds 5
+                                                     :max-number-of-messages 10
+                                                     #_#_:attribute-names ["All"]}))]
+      (let [message-body (json/read-str (:body message)
+                                        :key-fn keyword)]
+        (doseq [r (:Records message-body)] 
+          (println "Processing record " r)
+          (let [props (Session/getDefaultInstance (Properties.))
+                message-stream (-> (s3/get-object {:key (-> r :s3 :object :key)
+                                                   :bucket-name (-> r :s3 :bucket :name)})
+                                   :input-stream)
+                mail (message/read-message (MimeMessage. props message-stream))]
+            (doseq [pdf-stream (->> (-> mail :body)
+                                    (filter :content-type)
+                                    (filter #(re-find #"application/pdf" (:content-type %)) ))]
+              (let [filename (str "/tmp/" (UUID/randomUUID) ".pdf")]
+                (io/copy (:body pdf-stream) (io/file filename))
+                (invoices/insert-multi!
+                 (for [{:keys [total date invoice-number customer-identifier vendor] :as row}
+                       (parse/parse-file filename filename)]
+                   (do
+                     (println row)
+                     (assoc row
+                            :company (:name (parse/best-match companies customer-identifier))
+                            :imported false
+                            :potential-duplicate false))))
+                
+                (io/delete-file filename))))))
+      (sqs/delete-message (assoc message :queue-url queue-url )))))
+
+(defn always-process-sqs []
+  (while (not (Thread/interrupted))
+    (process-sqs)))