summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorTrygve Laugstøl <trygvis@inamo.no>2012-08-14 01:02:48 +0200
committerTrygve Laugstøl <trygvis@inamo.no>2012-08-14 01:02:48 +0200
commitd5a8420694bc9f5b269f46097829d3178e116923 (patch)
tree9d21cf08ef3cd73f43fb79128af5afa4dd122328
downloadimap-to-db-d5a8420694bc9f5b269f46097829d3178e116923.tar.gz
imap-to-db-d5a8420694bc9f5b269f46097829d3178e116923.tar.bz2
imap-to-db-d5a8420694bc9f5b269f46097829d3178e116923.tar.xz
imap-to-db-d5a8420694bc9f5b269f46097829d3178e116923.zip
o Initial import.
-rw-r--r--.gitignore1
-rw-r--r--build.sbt21
-rw-r--r--project/plugins.sbt14
-rw-r--r--src/main/scala/main.scala350
-rw-r--r--version.sbt1
5 files changed, 387 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..eb5a316
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+target
diff --git a/build.sbt b/build.sbt
new file mode 100644
index 0000000..d51170c
--- /dev/null
+++ b/build.sbt
@@ -0,0 +1,21 @@
+import AssemblyKeys._
+
+assemblySettings
+
+organization := "io.trygvis.imap-to-db"
+
+name := "imap-to-db"
+
+scalaVersion := "2.9.1"
+
+libraryDependencies ++= Seq(
+ "org.constretto" % "constretto-core" % "2.0.1",
+ "ch.qos.logback" % "logback-core" % "1.0.6",
+ "org.slf4j" % "slf4j-log4j12" % "1.6.1",
+ "javax.mail" % "mail" % "1.4.5",
+ "postgresql" % "postgresql" % "9.1-901-1.jdbc4",
+ "commons-io" % "commons-io" % "2.2",
+ "commons-net" % "commons-net" % "3.1"
+)
+
+mainClass in assembly := Some("ImapToDb")
diff --git a/project/plugins.sbt b/project/plugins.sbt
new file mode 100644
index 0000000..c8f0531
--- /dev/null
+++ b/project/plugins.sbt
@@ -0,0 +1,14 @@
+resolvers += Resolver.url(
+ "sbt-plugin-releases",
+ new URL("http://scalasbt.artifactoryonline.com/scalasbt/sbt-plugin-releases/")
+)(Resolver.ivyStylePatterns)
+
+// IDEA plugin
+resolvers += "sbt-idea-repo" at "http://mpeltonen.github.com/maven/"
+
+addSbtPlugin("com.github.mpeltonen" % "sbt-idea" % "1.0.0")
+
+// Release plugin
+// addSbtPlugin("com.github.gseitz" % "sbt-release" % "0.5")
+
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.8.3")
diff --git a/src/main/scala/main.scala b/src/main/scala/main.scala
new file mode 100644
index 0000000..e8e1d6d
--- /dev/null
+++ b/src/main/scala/main.scala
@@ -0,0 +1,350 @@
+import com.sun.mail.imap._
+import com.sun.mail.util.MailSSLSocketFactory
+import java.io._
+import java.sql.{Array => SqlArray, Date => SqlDate, _}
+import java.util._
+import java.util.concurrent._
+import javax.mail._
+import javax.mail.event._
+import javax.mail.internet._
+import org.apache.commons.io.IOUtils
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import scala.collection.JavaConverters._
+
+object MyConnectionListener extends ConnectionListener {
+
+ def closed(e: ConnectionEvent) {
+ println("MyConnectionListener: closed");
+ }
+
+ def disconnected(e: ConnectionEvent) {
+ println("MyConnectionListener: disconnected");
+ }
+
+ def opened(e: ConnectionEvent) {
+ println("MyConnectionListener: opened");
+ }
+}
+
+class MyMessageCountListener(f: IMAPFolder, deque: Deque[Long]) extends MessageCountListener {
+ def messagesAdded(e: MessageCountEvent) {
+ println("messagesAdded: added count=" + e.getMessages.length)
+ e.getMessages.foreach { message =>
+ val uid = f.getUID(message)
+ println("uid=" + uid)
+ deque.push(uid)
+ }
+ }
+
+ def messagesRemoved(e: MessageCountEvent) {
+ println("messagesRemoved: removed count=" + e.getMessages.length + ", isRemoved=" + e.isRemoved)
+ e.getMessages.foreach { message =>
+ println("message: " + message.getFrom)
+ }
+ }
+}
+
+/*
+object MyMessageChangedListener extends MessageChangedListener {
+ def messageChanged(e: MessageChangedEvent) {
+ val t = if(e.getMessageChangeType() == MessageChangedEvent.FLAGS_CHANGED) "FLAGS_CHANGED" else "ENVELOPE_CHANGED"
+ println("messageChanged: " + t)
+ println("message: " + e.getMessage())
+ }
+}
+*/
+
+object MyFolderListener extends FolderListener {
+ def folderCreated(e: FolderEvent) {
+ println("folder created")
+ }
+
+ def folderDeleted(e: FolderEvent) {
+ println("folder deleted")
+ }
+
+ def folderRenamed(e: FolderEvent) {
+ println("folder renamed")
+ }
+}
+
+object MyStoreListener extends StoreListener {
+ def notification(e: StoreEvent) {
+ val t = if(e.getMessageType() == StoreEvent.NOTICE) "NOTICE" else "ALERT"
+ println("store event: " + t)
+ println("message: " + e.getMessage())
+ }
+}
+
+class Dao(c: Connection) extends Closeable {
+ def close() {
+ c.close()
+ }
+
+ val s = c.prepareStatement("BEGIN")
+ def begin() { s.executeUpdate() }
+ def commit() { c.commit() }
+ def rollback() { c.rollback() }
+
+ val countMail = c.prepareStatement("SELECT count(uid) FROM mail WHERE uid=?")
+ def countMail(uid: Long): Long = {
+ countMail.setLong(1, uid)
+ val rs = countMail.executeQuery()
+ rs.next()
+ val count = rs.getLong(1)
+ rs.close()
+ count
+ }
+
+ val insertMail = c.prepareStatement(
+ "INSERT INTO mail(uid, header_keys, header_values) VALUES(?, ?, ?)")
+ def insertMail(uid: Long, keys: Array[String], values: Array[String]) {
+ insertMail.setLong(1, uid)
+ insertMail.setArray(2, c.createArrayOf("varchar", keys.asInstanceOf[Array[Object]]))
+ insertMail.setArray(3, c.createArrayOf("varchar", values.asInstanceOf[Array[Object]]))
+ insertMail.executeUpdate()
+ }
+
+ val insertMailPart = c.prepareStatement(
+ "INSERT INTO mail_part(mail_uid, index, header_keys, header_values, data) VALUES(?, ?, ?, ?, ?)")
+ def insertMailPart(uid: Long, index: Int, keys: Array[String], values: Array[String], is: InputStream, size: Int) {
+ insertMailPart.setLong(1, uid)
+ insertMailPart.setInt(2, index)
+ insertMailPart.setArray(3, c.createArrayOf("varchar", keys.asInstanceOf[Array[Object]]))
+ insertMailPart.setArray(4, c.createArrayOf("varchar", values.asInstanceOf[Array[Object]]))
+ insertMailPart.setBinaryStream(5, is, size)
+ insertMailPart.executeUpdate()
+ }
+}
+
+object Service {
+ def headers(part: Part): (Array[String], Array[String]) = {
+ var headers = part.getAllHeaders.asInstanceOf[java.util.Enumeration[Header]].asScala.toList
+ var (names, values) = headers.unzip({
+ header => println(header.getName + ": " + header.getValue)
+ (header.getName, header.getValue)
+ })
+ return (names.toArray, values.toArray)
+ }
+ def processMail(dao: Dao, uid: Long, message: Message) {
+ println("Importing message, uid=" + uid)
+
+ var (names, values) = headers(message)
+ dao.insertMail(uid, names, values)
+
+ /*
+ message.getFrom.foreach { address =>
+ if(address.isInstanceOf[InternetAddress]) {
+ val a = address.asInstanceOf[InternetAddress]
+ println("From: " + a.getPersonal + " <" + a.getAddress + ">")
+ }
+ else
+ println("From: " + address)
+ }
+ message.getAllRecipients.foreach { address =>
+ println("To: " + address)
+ }
+ println("Subject: " + message.getSubject)
+ */
+
+ if(!message.isInstanceOf[MimeMessage])
+ error("Unprocessable: not a mime message")
+ else {
+ val m = message.asInstanceOf[MimeMessage]
+ /*
+ println("Mime mail")
+ println("Content-Type: " + m.getContentType)
+ */
+ val content = m.getContent
+// println("Is multipart? " + m.getContent.isInstanceOf[Multipart])
+ if(m.getContent.isInstanceOf[Multipart]) {
+ val multipart = m.getContent.asInstanceOf[Multipart]
+ 0.until(multipart.getCount).foreach { i =>
+ val body = multipart.getBodyPart(i)
+ var (names, values) = headers(body)
+ /*
+ println("Part #" + i)
+ println("Content-Type: " + body.getContentType)
+ println("Description: " + body.getDescription)
+ println("Content-Disposition: " + body.getDisposition)
+ println("FileName: " + body.getFileName)
+ println("Downloading part #" + i)
+ */
+ val start = System.currentTimeMillis
+ var bytes = IOUtils.toByteArray(m.getInputStream)
+ val end = System.currentTimeMillis
+ println("Downloaded " + bytes.length + " bytes in " + (end - start) + "ms")
+ dao.insertMailPart(uid, i, names, values, new ByteArrayInputStream(bytes), bytes.length)
+ }
+ }
+ else {
+ // TODO: what to do?
+ }
+ }
+ }
+}
+
+class ConnectionHandler(f: IMAPFolder, dao: Dao) {
+ def tx[T](f: Dao => T): Option[T] = {
+ dao.begin()
+ try {
+ val t = f(dao)
+ dao.commit()
+ Some(t)
+ } catch {
+ case e =>
+ e.printStackTrace
+ dao.rollback()
+ None
+ }
+ }
+
+ def process() {
+ val deque = new LinkedBlockingDeque[Long]()
+
+ println("Folder: " + f.getFullName())
+ f.open(Folder.READ_WRITE)
+ println("Folder has " + f.getMessageCount() + " messages, " + f.getNewMessageCount + " new and " + f.getDeletedMessageCount + " deleted.")
+ // f.addMessageChangedListener(MyMessageChangedListener)
+ f.addMessageCountListener(new MyMessageCountListener(f, deque))
+
+ f.getMessages().foreach(processMessage)
+
+ do {
+ println("IDLE")
+ f.idle(true);
+ println("Waiting for message..")
+ /*
+ val message = deque.poll
+ if(message != null)
+ processMessage(message)
+ */
+ } while(true)
+ }
+
+ def processMessage(message: Message) {
+ val uid = f.getUID(message)
+ var ok = tx { dao =>
+ println(new Date)
+ if(dao.countMail(uid) > 0) {
+ println("Message already imported: uid=" + uid)
+ }
+ else {
+ Service.processMail(dao, uid, message)
+ }
+ }
+ if(ok.isDefined) {
+ message.setFlag(Flags.Flag.DELETED, true)
+ }
+ }
+}
+
+class ImapToDb(f: File) {
+ def work() {
+ val p = new Properties()
+ var r: InputStream = null
+ try {
+ r = new FileInputStream(f)
+ p.load(r)
+ } catch {
+ case _ =>
+ System.err.println("Could not read configuraion file: " + f)
+ return
+ } finally {
+ IOUtils.closeQuietly(r)
+ }
+
+ def remove(key: String) = {
+ val value = p.getProperty(key)
+ println(key + ":" + value)
+ if(value != null) {
+ p.remove(key)
+ Some(value)
+ }
+ else
+ None
+ }
+ val params = for {
+ dbUrl: String <- remove("db.url")
+ dbUsername: String <- remove("db.username")
+ dbPassword: String <- remove("db.password")
+ imapHost: String <- remove("imap.host")
+ imapUsername: String <- remove("imap.username")
+ imapPassword: String <- remove("imap.password")
+ } yield (dbUrl, dbUsername, dbPassword, imapHost, imapUsername, imapPassword)
+
+ params match {
+ case None =>
+ System.err.println("Required properties:")
+ System.err.println(" db.url, db.username db.password")
+ System.err.println(" imap.host, imap.username imap.password")
+ case Some((dbUrl, dbUsername, dbPassword, imapHost, imapUsername, imapPassword)) =>
+ for(klass: String <- remove("db.driver") if klass.length > 0)
+ Class.forName(klass)
+
+ val dbProps = new Properties()
+ println("Db properties: ")
+ p.asScala.foreach { case (key, value) if key.startsWith("db.") =>
+ dbProps.put(key.substring(3), value)
+ println(key + ": " + value)
+ }
+
+ val imapProps = new Properties()
+ println("Imap properties: ")
+ p.asScala.foreach { case (key, value) if key.startsWith("imap.") =>
+ imapProps.put(key.substring(5), value)
+ println(key + ": " + value)
+ }
+
+ val sf = new MailSSLSocketFactory()
+ sf.setTrustAllHosts(true)
+ imapProps.put("mail.imaps.ssl.socketFactory", sf)
+ val session = Session.getDefaultInstance(imapProps, null)
+ def connectToFolder() = {
+ val store = session.getStore("imaps")
+ store.connect(imapHost, imapUsername, imapPassword)
+ store.addConnectionListener(MyConnectionListener)
+ store.addStoreListener(MyStoreListener)
+ // store.addFolderListener(MyFolderListener)
+ val folder = store.getFolder("INBOX")
+ // folder.addConnectionListener(MyConnectionListener)
+ // folder.addFolderListener(MyFolderListener)
+ // folder.addMessageCountListener(MyMessageCountListener)
+ // folder.addMessageChangedListener(MyMessageChangedListener)
+ if(!folder.isInstanceOf[IMAPFolder]) {
+ error("folder is not an IMAPFolder")
+ }
+ folder.asInstanceOf[IMAPFolder]
+ }
+
+ def connectToDb() = {
+ val c = DriverManager.getConnection(dbUrl, dbProps)
+ c.setAutoCommit(false)
+ new Dao(c)
+ }
+
+ while(true) {
+ var folder: IMAPFolder = null
+ var dao: Dao = null
+ try {
+ folder = connectToFolder()
+ dao = connectToDb()
+ new ConnectionHandler(folder, dao).process()
+ } catch {
+ case e =>
+ System.out.println("Uncaught exception: " + e.getMessage)
+ } finally {
+ try { folder.getStore.close() } catch { case _ => }
+ IOUtils.closeQuietly(dao)
+ Thread.sleep(1000)
+ }
+ }
+ }
+ }
+
+}
+
+object ImapToDb extends App {
+ new ImapToDb(new File("imap-to-db.properties")).work()
+}
diff --git a/version.sbt b/version.sbt
new file mode 100644
index 0000000..6243c9f
--- /dev/null
+++ b/version.sbt
@@ -0,0 +1 @@
+version in ThisBuild := "1.0-SNAPSHOT"