<commit_before>import java.util.ArrayList;

import java.net.URI;

import org.apache.hadoop._;
import org.apache.hadoop.fs._;
import org.apache.hadoop.io._;
import org.apache.hadoop.mapreduce._;
import org.apache.hadoop.mapreduce.lib.input._;

import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;

import org.apache.pig._;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data._;

import org.archive.io.{ArchiveReader,ArchiveReaderFactory,ArchiveRecord};
import org.cdlib.was.ngIndexer._;

class ArchiveListLoader extends LoadFunc {
  val tupleFactory = TupleFactory.getInstance();
  val bagFactory = BagFactory.getInstance();
  val client = new SimpleHttpClient;
  val parser = new MyParser;

  var in : RecordReader[_,_] = null;
  var response : HttpResponse = null;
  var reader : ArchiveReader = null;
  var it : java.util.Iterator[ArchiveRecord] = null;
  var arcName : String = _;

  def setupNextArchiveReader : Boolean = {
    if (!in.nextKeyValue()) {
      return false;
    } else {
      if (this.response != null) {
        EntityUtils.consume(response.getEntity);
      }
      this.reader = null;
      this.it = null;
      val value = this.in.getCurrentValue.asInstanceOf[Text]
      val uri = new URI(value.toString);
      val matcher = Utility.ARC_RE.pattern.matcher(uri.getPath);
      if (!matcher.matches) {
        return false;
      } else {
        arcName = matcher.group(1);
        client.getUriResponse(uri) match {
          case None => return setupNextArchiveReader;
          case Some(response) => {
            this.response = response;
            reader = ArchiveReaderFactory.get(arcName, response.getEntity.getContent, true);
            it = reader.iterator;
            return true;
          }
        }
      }
    }
  }      
  
  def getNextArchiveRecord : Option[ArchiveRecordWrapper] = {
    if (it == null || !it.hasNext) {
      /* try to get a new ArchiveReader, otherwise return null */
      if (!setupNextArchiveReader) {
        return None;
      }
    }
    val next = it.next;
    val rec = new ArchiveRecordWrapper(next, arcName);
    if (!rec.isHttpResponse) {
      /* try again */
      return getNextArchiveRecord;
    } else {
      return Some(rec);
    }
  }

  override def getNext : Tuple = {
    try {
      getNextArchiveRecord match {
        case None => return null;
        case Some(rec) => {
          var tuple = tupleFactory.newTupleNoCopy(new java.util.ArrayList[java.lang.Object]());
          val result = parser.parse(rec, rec.getMediaTypeStr, rec.getUrl, rec.getDate)
          val outlinks = bagFactory.newDistinctBag;
          for (link <- result.outlinks) { outlinks.add(tupleFactory.newTuple(link)); }
          tuple.append(rec.getUrl)
          tuple.append(result.content);
          tuple.append(result.mediaType);
          tuple.append(result.title);
          tuple.append(outlinks);
          return tuple;
        }
      }
    } catch {<commit_msg>add digest to generated tuple<commit_after>package org.cdlib.was.ngIndexer.pig;

import java.util.ArrayList;

import java.net.URI;

import org.apache.hadoop._;
import org.apache.hadoop.fs._;
import org.apache.hadoop.io._;
import org.apache.hadoop.mapreduce._;
import org.apache.hadoop.mapreduce.lib.input._;

import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;

import org.apache.pig._;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data._;

import org.archive.io.{ArchiveReader,ArchiveReaderFactory,ArchiveRecord};
import org.cdlib.was.ngIndexer._;

class ArchiveListLoader extends LoadFunc {
  val tupleFactory = TupleFactory.getInstance();
  val bagFactory = BagFactory.getInstance();
  val client = new SimpleHttpClient;
  val parser = new MyParser;

  var in : RecordReader[_,_] = null;
  var response : HttpResponse = null;
  var reader : ArchiveReader = null;
  var it : java.util.Iterator[ArchiveRecord] = null;
  var arcName : String = _;

  def setupNextArchiveReader : Boolean = {
    if (!in.nextKeyValue()) {
      return false;
    } else {
      if (this.response != null) {
        EntityUtils.consume(response.getEntity);
      }
      this.reader = null;
      this.it = null;
      val value = this.in.getCurrentValue.asInstanceOf[Text]
      val uri = new URI(value.toString);
      val matcher = Utility.ARC_RE.pattern.matcher(uri.getPath);
      if (!matcher.matches) {
        return false;
      } else {
        arcName = matcher.group(1);
        client.getUriResponse(uri) match {
          case None => return setupNextArchiveReader;
          case Some(response) => {
            this.response = response;
            reader = ArchiveReaderFactory.get(arcName, response.getEntity.getContent, true);
            it = reader.iterator;
            return true;
          }
        }
      }
    }
  }      
  
  def getNextArchiveRecord : Option[ArchiveRecordWrapper] = {
    if (it == null || !it.hasNext) {
      /* try to get a new ArchiveReader, otherwise return null */
      if (!setupNextArchiveReader) {
        return None;
      }
    }
    val next = it.next;
    val rec = new ArchiveRecordWrapper(next, arcName);
    if (!rec.isHttpResponse) {
      /* try again */
      return getNextArchiveRecord;
    } else {
      return Some(rec);
    }
  }

  override def getNext : Tuple = {
    try {
      getNextArchiveRecord match {
        case None => return null;
        case Some(rec) => {
          var tuple = tupleFactory.newTupleNoCopy(new java.util.ArrayList[java.lang.Object]());
          val result = parser.parse(rec, rec.getMediaTypeStr, rec.getUrl, rec.getDate)
          val outlinks = bagFactory.newDistinctBag;
          for (link <- result.outlinks) { outlinks.add(tupleFactory.newTuple(link)); }
          tuple.append(rec.getUrl)
          tuple.append(result.content);
          tuple.append(result.mediaType);
          tuple.append(result.title);
          tuple.append(rec.getDigestStr.getOrElse("-"));
          tuple.append(outlinks);
          return tuple;
        }
      }
    } catch {
{'fraction_non_alphanumeric': 0.11247411183686475, 'fraction_numerical': 0.0003186235462800701, 'mean_word_length': 2.6436448055716775, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '5217608', 'n_tokens_mistral': 1907, 'n_tokens_neox': 1862, 'n_words': 393}