<commit_before>package vtpassim

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

import org.apache.hadoop.fs.{FileSystem,Path}

import org.apache.spark.sql.catalyst.util.DateTimeUtils
import java.sql.Date

import collection.JavaConversions._
import scala.collection.mutable.ListBuffer
import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.Queue

object MetsAlto {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("MetsAlto Application")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive", "true")

    sc.binaryFiles(args(0))
      .filter(_._1.endsWith(".zip"))
      .flatMap( x => {
        try {
          val fname = new java.io.File(new java.net.URL(x._1).toURI)
          val issue = fname.getName.replaceAll(".zip$", "")

          val zfile = new java.util.zip.ZipFile(fname)
          import scala.collection.JavaConversions._
          val (series, date, title, lang) =
            try {
              val mfile = zfile.entries.filter(_.getName.endsWith("mets.xml")).toSeq.head
              val t = scala.xml.XML.load(zfile.getInputStream(mfile))
              ((t \\ "identifier").head.text,
                (t \\ "dateIssued").head.text
                  .replaceAll("""^(\d\d)\.(\d\d)\.(\d\d\d\d)$""", """$3-$2-$1"""),
                (t \\ "title").head.text,
                (t \\ "languageTerm").head.text)
            } catch {
              case e: Exception => ("", "", "", "")
            }

          zfile.entries.filter(f => f.getName.endsWith(".xml") && !f.getName.endsWith("mets.xml"))
            .map(f => {
              val t = scala.xml.XML.load(zfile.getInputStream(f))
              val res = (t \\ "TextLine").map { line =>
                (line \ "_").map({ e =>
                  if ( e.label == "String" ) {
                    (e \ "@CONTENT").text.replaceAll("<", "&lt;")
                  } else if ( e.label == "SP" ) {
                    " "
                  } else if ( e.label == "HYP" ) {
                    "\u00ad"
                  }
                })
                  .mkString("")
            }
              (f.getName.replaceAll(".xml$", ""), issue, series, date,
                title, lang, res.mkString("\n"))
            })
        } catch {
          case e: Exception => {
            println(x._1 + e.toString)
            None
          }
        }
      }
    )
      .toDF("id", "issue", "series", "date", "title", "lang", "text")
      .write.save(args(1))
  }
}<commit_msg>Include title in records.<commit_after>package vtpassim

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

import org.apache.hadoop.fs.{FileSystem,Path}

import org.apache.spark.sql.catalyst.util.DateTimeUtils
import java.sql.Date

import collection.JavaConversions._
import scala.collection.mutable.ListBuffer
import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.Queue

object MetsAlto {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("MetsAlto Application")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive", "true")

    sc.binaryFiles(args(0), sc.defaultParallelism)
      .filter(_._1.endsWith(".zip"))
      .flatMap( x => {
        try {
          val fname = new java.io.File(new java.net.URL(x._1).toURI)
          val issue = fname.getName.replaceAll(".zip$", "")

          val zfile = new java.util.zip.ZipFile(fname)
          import scala.collection.JavaConversions._
          val (series, date, lang) =
            try {
              val mfile = zfile.entries.filter(_.getName.endsWith("mets.xml")).toSeq.head
              val t = scala.xml.XML.load(zfile.getInputStream(mfile))
              ((t \\ "identifier").head.text,
                (t \\ "dateIssued").head.text
                  .replaceAll("""^(\d\d)\.(\d\d)\.(\d\d\d\d)$""", """$3-$2-$1"""),
                (t \\ "languageTerm").head.text)
            } catch {
              case e: Exception => ("", "", "")
            }

          zfile.entries.filter(f => f.getName.endsWith(".xml") && !f.getName.endsWith("mets.xml"))
            .map(f => {
              val t = scala.xml.XML.load(zfile.getInputStream(f))
              val res = (t \\ "TextLine").map { line =>
                (line \ "_").map({ e =>
                  if ( e.label == "String" ) {
                    (e \ "@CONTENT").text.replaceAll("<", "&lt;")
                  } else if ( e.label == "SP" ) {
                    " "
                  } else if ( e.label == "HYP" ) {
                    "\u00ad"
                  }
                })
                  .mkString("")
            }
              (f.getName.replaceAll(".xml$", ""), issue, series, date, lang, res.mkString("\n"))
            })
        } catch {
          case e: Exception => {
            println(x._1 + e.toString)
            None
          }
        }
      }
    )
      .toDF("id", "issue", "series", "date", "lang", "text")
      .write.save(args(1))
  }
}
{'fraction_non_alphanumeric': 0.15960013788348845, 'fraction_numerical': 0.003447087211306446, 'mean_word_length': 1.881330685203575, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1052227', 'n_tokens_mistral': 1848, 'n_tokens_neox': 1781, 'n_words': 307}