<commit_before>package org.tribbloid.spookystuff.entity

import java.io._
import java.util.{Date, UUID}

import de.l3s.boilerpipe.extractors.ArticleExtractor
import org.apache.commons.io.IOUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.http.entity.ContentType
import org.jsoup.Jsoup
import org.jsoup.nodes.{Document, Element}
import org.tribbloid.spookystuff.entity.client.Action
import org.tribbloid.spookystuff.{Const, SpookyContext, Utils}

import scala.collection.JavaConversions._

object Page {

  def load(fullPath: Path)(hConf: Configuration): Array[Byte] = {

    val fs = fullPath.getFileSystem(hConf)

    if (fs.exists(fullPath)) {

      val fis = fs.open(fullPath)

      try {
        IOUtils.toByteArray(fis)
      }
      finally {
        fis.close()
      }
    }
    else null
  }

  //unlike save, this will store all information in an unreadable, serialized, probably compressed file
  def cache(
             pages: Seq[Page],
             path: String,
             overwrite: Boolean = false
             )(hConf: Configuration): Unit = {

    var fullPath = new Path(path)

    val fs = fullPath.getFileSystem(hConf)

    if (!overwrite && fs.exists(fullPath)) fullPath = new Path(path +"-"+ UUID.randomUUID())

    val fos = fs.create(fullPath, overwrite)
    val objectOS = new ObjectOutputStream(fos)

    try {
      objectOS.writeObject(pages)
    }
    finally {
      objectOS.close()
    }
  }

  def autoCache(
                 pages: Seq[Page],
                 uid: PageUID,
                 spooky: SpookyContext
                 ): Unit = {
    val pathStr = Utils.urlConcat(
      spooky.autoCacheRoot,
      spooky.autoCacheLookup(uid).toString,
      UUID.randomUUID().toString
    )

    Page.cache(pages, pathStr)(spooky.hConf)
  }

  def restore(fullPath: Path)(hConf: Configuration): Seq[Page] = {

    val fs = fullPath.getFileSystem(hConf)

    if (fs.exists(fullPath)) {
      val fis = fs.open(fullPath)
      val objectIS = new ObjectInputStream(fis)

      try {
        objectIS.readObject().asInstanceOf[Seq[Page]]
      }
      finally {
        objectIS.close()
      }
    }
    else null
  }

  //  class PrefixFilter(val prefix: String) extends PathFilter {
  //
  //    override def accept(path: Path): Boolean = path.getName.startsWith(prefix)
  //  }
  //
  //  def getDirsByPrefix(dirPath: Path, prefix: String)(hConf: Configuration): Seq[Path] = {
  //
  //    val fs = dirPath.getFileSystem(hConf)
  //
  //    if (fs.getFileStatus(dirPath).isDir) {
  //      val status = fs.listStatus(dirPath, new PrefixFilter(prefix))
  //
  //      status.map(_.getPath)
  //    }
  //    else Seq()
  //  }

  //restore latest in a directory
  //returns: Seq() => has backtrace dir but contains no page
  //returns null => no backtrace dir
  def restoreLatest(
                     dirPath: Path,
                     earliestModificationTime: Long = 0
                     )(hConf: Configuration): Seq[Page] = {

    val fs = dirPath.getFileSystem(hConf)

    if (fs.exists(dirPath) && fs.getFileStatus(dirPath).isDir) {
      //      val results = new ArrayBuffer[Page]()

      val statuses = fs.listStatus(dirPath)

      val latestStatus = statuses.filter(status => !status.isDir && status.getModificationTime >= earliestModificationTime).sortBy(_.getModificationTime).lastOption

      latestStatus match {
        case Some(status) => restore(status.getPath)(hConf)
        case _ => null
      }
    }
    else null
  }

  //TODO: cannot handle infinite duration, avoid using it!
  def autoRestoreLatest(
                         uid: PageUID,
                         spooky: SpookyContext
                         ): Seq[Page] = {
    val pathStr = Utils.urlConcat(
      spooky.autoCacheRoot,
      spooky.autoCacheLookup(uid).toString
    )

    restoreLatest(new Path(pathStr), System.currentTimeMillis() - spooky.pageExpireAfter.toMillis)(spooky.hConf)
  }
}

/**
 * Created by peng on 04/06/14.
 */
//use to genterate a lookup key for each page so
case class PageUID(
                    backtrace: Seq[Action],
                    blockKey: Int = -1 //-1 is no sub key
                    )

//immutable! we don't want to lose old pages
//keep small, will be passed around by Spark
//@SerialVersionUID(1925602137496052L)
case class Page(
                 uid: PageUID,

                 resolvedUrl: String,
                 contentType: String,
                 content: Array[Byte],

                 name: String = null,
                 //                 cookie: Seq[SerializableCookie] = Seq(),
                 timestamp: Date = new Date,
                 saved: String = null
                 )
  extends Serializable {

  private final val serialVersionUID: Long = 1925602137496052L

  @transient lazy val parsedContentType: ContentType = {
    var result = ContentType.parse(this.contentType)
    if (result.getCharset == null) result = result.withCharset(Const.defaultCharset)
    result
  }
  @transient lazy val contentStr: String = new String(this.content,this.parsedContentType.getCharset)

  @transient lazy val doc: Option[Any] = if (parsedContentType.getMimeType.contains("html")){
    Some(Jsoup.parse(this.contentStr, resolvedUrl)) //not serialize, parsing is faster
  }
  else{
    None
  }

  def backtrace = this.uid.backtrace
  def blockKey = this.uid.blockKey

  //this will lose information as charset encoding will be different
  def save(<commit_msg>cache delegated to SparkEnv.get.Serializer<commit_after>package org.tribbloid.spookystuff.entity

import java.io._
import java.util.{Date, UUID}

import de.l3s.boilerpipe.extractors.ArticleExtractor
import org.apache.commons.io.IOUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.http.entity.ContentType
import org.apache.spark.{SparkEnv, SparkContext, SparkConf}
import org.apache.spark.serializer.JavaSerializer
import org.jsoup.Jsoup
import org.jsoup.nodes.{Document, Element}
import org.tribbloid.spookystuff.entity.client.Action
import org.tribbloid.spookystuff.{Const, SpookyContext, Utils}

import scala.collection.JavaConversions._

object Page {

  def load(fullPath: Path)(hConf: Configuration): Array[Byte] = {

    val fs = fullPath.getFileSystem(hConf)

    if (fs.exists(fullPath)) {

      val fis = fs.open(fullPath)

      try {
        IOUtils.toByteArray(fis)
      }
      finally {
        fis.close()
      }
    }
    else null
  }

  //unlike save, this will store all information in an unreadable, serialized, probably compressed file
  def cache(
             pages: Seq[Page],
             path: String,
             overwrite: Boolean = false
             )(hConf: Configuration): Unit = {

    var fullPath = new Path(path)

    val fs = fullPath.getFileSystem(hConf)

    if (!overwrite && fs.exists(fullPath)) fullPath = new Path(path +"-"+ UUID.randomUUID())

    val ser = SparkEnv.get.serializer.newInstance()
    val copy = ser.serialize(pages)

    val fos = fs.create(fullPath, overwrite)
    try {
      fos.write(copy.array())
    }
    finally {
      fos.close()
    }
  }

  def autoCache(
                 pages: Seq[Page],
                 uid: PageUID,
                 spooky: SpookyContext
                 ): Unit = {
    val pathStr = Utils.urlConcat(
      spooky.autoCacheRoot,
      spooky.autoCacheLookup(uid).toString,
      UUID.randomUUID().toString
    )

    Page.cache(pages, pathStr)(spooky.hConf)
  }

  def restore(fullPath: Path)(hConf: Configuration): Seq[Page] = {

    val fs = fullPath.getFileSystem(hConf)

    if (fs.exists(fullPath)) {
      val fis = fs.open(fullPath)

      val ser = SparkEnv.get.serializer.newInstance()

      val serIn = ser.deserializeStream(fis)
      val obj = serIn.readObject[Seq[Page]]()
      serIn.close()
      obj
    }
    else null
  }

  //  class PrefixFilter(val prefix: String) extends PathFilter {
  //
  //    override def accept(path: Path): Boolean = path.getName.startsWith(prefix)
  //  }
  //
  //  def getDirsByPrefix(dirPath: Path, prefix: String)(hConf: Configuration): Seq[Path] = {
  //
  //    val fs = dirPath.getFileSystem(hConf)
  //
  //    if (fs.getFileStatus(dirPath).isDir) {
  //      val status = fs.listStatus(dirPath, new PrefixFilter(prefix))
  //
  //      status.map(_.getPath)
  //    }
  //    else Seq()
  //  }

  //restore latest in a directory
  //returns: Seq() => has backtrace dir but contains no page
  //returns null => no backtrace dir
  def restoreLatest(
                     dirPath: Path,
                     earliestModificationTime: Long = 0
                     )(hConf: Configuration): Seq[Page] = {

    val fs = dirPath.getFileSystem(hConf)

    if (fs.exists(dirPath) && fs.getFileStatus(dirPath).isDir) {
      //      val results = new ArrayBuffer[Page]()

      val statuses = fs.listStatus(dirPath)

      val latestStatus = statuses.filter(status => !status.isDir && status.getModificationTime >= earliestModificationTime).sortBy(_.getModificationTime).lastOption

      latestStatus match {
        case Some(status) => restore(status.getPath)(hConf)
        case _ => null
      }
    }
    else null
  }

  //TODO: cannot handle infinite duration, avoid using it!
  def autoRestoreLatest(
                         uid: PageUID,
                         spooky: SpookyContext
                         ): Seq[Page] = {
    val pathStr = Utils.urlConcat(
      spooky.autoCacheRoot,
      spooky.autoCacheLookup(uid).toString
    )

    restoreLatest(new Path(pathStr), System.currentTimeMillis() - spooky.pageExpireAfter.toMillis)(spooky.hConf)
  }
}

/**
 * Created by peng on 04/06/14.
 */
//use to genterate a lookup key for each page so
case class PageUID(
                    backtrace: Seq[Action],
                    blockKey: Int = -1 //-1 is no sub key
                    )

//immutable! we don't want to lose old pages
//keep small, will be passed around by Spark
@SerialVersionUID(1925602137496052L)
case class Page(
                 uid: PageUID,

                 resolvedUrl: String,
                 contentType: String,
                 content: Array[Byte],

                 name: String = null,
                 //                 cookie: Seq[SerializableCookie] = Seq(),
                 timestamp: Date = new Date,
                 saved: String = null
                 )
  extends Serializable {

//  private final val serialVersionUID: Long = 1925602137496052L

  @transient lazy val parsedContentType: ContentType = {
    var result = ContentType.parse(this.contentType)
    if (result.getCharset == null) result = result.withCharset(Const.defaultCharset)
    result
  }
  @transient lazy val contentStr: String = new String(this.content,this.parsedContentType.getCharset)

  @transient lazy val doc: Option[Any] = if (parsedContentType.getMimeType.contains("html")){
    Some(Jsoup.parse(this.contentStr, resolvedUrl)) //not serialize, parsing is faster
  }
  else{
    None
  }

  def backtrace = this.uid.backtrace
  def blockKey = this.uid.blockKey

  //this will lose information as charset encoding will be different
  def save(
{'fraction_non_alphanumeric': 0.10697674418604651, 'fraction_numerical': 0.0075134168157423974, 'mean_word_length': 2.3436004784688995, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7589028', 'n_tokens_mistral': 3527, 'n_tokens_neox': 3399, 'n_words': 791}