<commit_before>/* Copyright (c) 2011 The Regents of the University of California */

package org.cdlib.was.ngIndexer;

import java.io.InputStream;

import org.apache.http.{Header,HeaderElement,NameValuePair};
import org.apache.http.ParseException;
import org.apache.http.impl.DefaultHttpResponseFactory;
import org.apache.http.impl.io.AbstractSessionInputBuffer;
import org.apache.http.impl.io.{AbstractMessageParser,HttpResponseParser};
import org.apache.http.message.{BasicHeaderValueParser,HeaderValueParser,ParserCursor};
import org.apache.http.params.BasicHttpParams;
import org.apache.http.util.CharArrayBuffer;

import org.apache.tika.metadata.HttpHeaders;

import org.archive.io.ArchiveRecord;
import org.archive.io.arc.ARCRecord;
import org.archive.io.warc.{WARCConstants,WARCRecord};
import org.archive.util.ArchiveUtils;

import org.apache.http.message.BasicLineParser;

import org.cdlib.was.ngIndexer.Utility.string2date;

import scala.util.matching.Regex;

/**
 * A wrapper for ArchiveRecord objects to provide a more consistent
 * interface.
 */
class ArchiveRecordWrapper (rec : ArchiveRecord, filename : String) 
  extends InputStream with WASArchiveRecord with Logger {

  private var statusCode : Option[Int] = None;
  private var ready : Boolean = false;
  private var contentTypeStr : Option[String] = None;
  private var httpResponse : Boolean = false;
  private var contentType : Option[ContentType] = None;
  private var closed : Boolean = false;

  /**
   * Parse the headers from a WARCRecord.
   */
  def parseWarcHttpHeaders {
    val lineParser = new BasicLineParser;
    val firstLine = ArchiveRecordWrapper.readLine(rec);
    catchAndLogExceptions {
      val statusLine = lineParser.parseStatusLine(firstLine, new ParserCursor(0, firstLine.length - 1));
      val headers = ArchiveRecordWrapper.readHeaderLines(rec).map(lineParser.parseHeader(_));
        val headerMap = headers.map {(h)=> h.getName.toLowerCase->h }.toMap;
      statusCode = Some(statusLine.getStatusCode);
      contentTypeStr = 
        headerMap.get(HttpHeaders.CONTENT_TYPE.toLowerCase).map(_.getValue);
    }
  }
  
  /**
   * Cue up a record to its start, fill in httpResponse & contentType.
   */
  private def cueUp {
    if (((rec.getClass == classOf[WARCRecord]) &&
         (rec.getHeader.getMimetype == "application/http; msgtype=response")) ||
        ((rec.getClass == classOf[ARCRecord]) &&
         rec.asInstanceOf[ARCRecord].getMetaData.getUrl.startsWith("http"))) {
           httpResponse = true;
           parseWarcHttpHeaders;
         }
    contentType = contentTypeStr.flatMap(ContentType.parse(_));
    ready = true;
  }

  /**
   * Return the response code. Returns -1 if this is not an HTTP response
   * record, or there was a failure in parsing.
   */
  def getStatusCode : Int = {
    if (!ready) cueUp;
    if (statusCode.isEmpty) {
      return -1;
    } else {
      return statusCode.get;
    }
  }

  lazy val getContentType : ContentType = {
    if (!ready) cueUp;
    contentType.getOrElse(ContentType.DEFAULT);
  }

  /**
   * Return true if this is an HTTP response record.
   */
  lazy val isHttpResponse : Boolean = {
    if (!ready) cueUp;
    httpResponse;
  }

  lazy val getUrl = rec.getHeader.getUrl;

  lazy val getLength = rec.getHeader.getLength;

  lazy val getDate : java.util.Date = rec.getHeader.getDate;

  val SHA1_RE = new Regex("""^sha1:([a-zA-Z0-9]+)$""");

  lazy val getDigestStr : Option[String] = {
    if (!ready) cueUp;
    rec match {
      case arc : ARCRecord => {
        if (!this.closed) {
            /* must wait until finished */
          None;
        }
        Some(rec.getDigestStr);
      }
      case warc : WARCRecord => {
        warc.getHeader.getHeaderValue(WARCConstants.HEADER_KEY_PAYLOAD_DIGEST) match {
          case SHA1_RE(s) => Some(s);
          case _          => None;
        }
      }
    }
  }

  lazy val getFilename = filename;

  /* InputStream wrapper */
  override def available = { 
    if (!ready) cueUp;
    rec.available; 
  }
  
  override def close = {
    if (!this.closed) rec.close;
    this.closed = true;
  }

  override def mark (readlimit : Int) = {
    if (!ready) cueUp; 
    rec.mark(readlimit);
  }
  
  override def markSupported = rec.markSupported;

  override def read = {
    if (!ready) cueUp; 
    rec.read;
  }

  override def read (buff : Array[Byte]) = {
    if (!ready) cueUp; 
    rec.read(buff);
  }

  override def read (buff : Array[Byte], off : Int, len : Int) = {
    if (!ready) cueUp; 
    rec.read(buff, off, len);
  }

  override def reset = {
    if (!ready) cueUp;
    rec.reset;
  }

  override def skip (n : Long) = {
    if (!ready) cueUp; 
    rec.skip(n);
  }
}

object ArchiveRecordWrapper {
  /**
   * Method to read one line of input from an InputStream. A line<commit_msg>fix digest creation for arc files<commit_after>/* Copyright (c) 2011 The Regents of the University of California */

package org.cdlib.was.ngIndexer;

import java.io.InputStream;
import java.security.MessageDigest;

import org.apache.http.{Header,HeaderElement,NameValuePair};
import org.apache.http.ParseException;
import org.apache.http.impl.DefaultHttpResponseFactory;
import org.apache.http.impl.io.AbstractSessionInputBuffer;
import org.apache.http.impl.io.{AbstractMessageParser,HttpResponseParser};
import org.apache.http.message.{BasicHeaderValueParser,HeaderValueParser,ParserCursor};
import org.apache.http.params.BasicHttpParams;
import org.apache.http.util.CharArrayBuffer;

import org.apache.tika.metadata.HttpHeaders;

import org.archive.io.ArchiveRecord;
import org.archive.io.arc.ARCRecord;
import org.archive.io.warc.{WARCConstants,WARCRecord};
import org.archive.util.{ArchiveUtils,Base32};

import org.apache.http.message.BasicLineParser;

import org.cdlib.was.ngIndexer.Utility.{dumpStream,string2date};

import scala.util.matching.Regex;

/**
 * A wrapper for ArchiveRecord objects to provide a more consistent
 * interface.
 */
class ArchiveRecordWrapper (rec : ArchiveRecord, filename : String) 
  extends InputStream with WASArchiveRecord with Logger {

  private var statusCode : Option[Int] = None;
  private var ready : Boolean = false;
  private var contentTypeStr : Option[String] = None;
  private var httpResponse : Boolean = false;
  private var contentType : Option[ContentType] = None;
  private var closed : Boolean = false;
  private var digest : Option[MessageDigest] = 
    rec match {
      case arc : ARCRecord =>
        Some(MessageDigest.getInstance("SHA-1"));
      case _ => None;
    }

  /**
   * Parse the headers from a WARCRecord.
   */
  def parseWarcHttpHeaders {
    val lineParser = new BasicLineParser;
    val firstLine = ArchiveRecordWrapper.readLine(rec);
    catchAndLogExceptions {
      val statusLine = lineParser.parseStatusLine(firstLine, new ParserCursor(0, firstLine.length - 1));
      val headers = ArchiveRecordWrapper.readHeaderLines(rec).map(lineParser.parseHeader(_));
        val headerMap = headers.map {(h)=> h.getName.toLowerCase->h }.toMap;
      statusCode = Some(statusLine.getStatusCode);
      contentTypeStr = 
        headerMap.get(HttpHeaders.CONTENT_TYPE.toLowerCase).map(_.getValue);
    }
  }
  
  /**
   * Cue up a record to its start, fill in httpResponse & contentType.
   */
  private def cueUp {
    if (((rec.getClass == classOf[WARCRecord]) &&
         (rec.getHeader.getMimetype == "application/http; msgtype=response")) ||
        ((rec.getClass == classOf[ARCRecord]) &&
         rec.asInstanceOf[ARCRecord].getMetaData.getUrl.startsWith("http"))) {
           httpResponse = true;
           parseWarcHttpHeaders;
         }
    contentType = contentTypeStr.flatMap(ContentType.parse(_));
    ready = true;
  }

  /**
   * Return the response code. Returns -1 if this is not an HTTP response
   * record, or there was a failure in parsing.
   */
  def getStatusCode : Int = {
    if (!ready) cueUp;
    if (statusCode.isEmpty) {
      return -1;
    } else {
      return statusCode.get;
    }
  }

  lazy val getContentType : ContentType = {
    if (!ready) cueUp;
    contentType.getOrElse(ContentType.DEFAULT);
  }

  /**
   * Return true if this is an HTTP response record.
   */
  lazy val isHttpResponse : Boolean = {
    if (!ready) cueUp;
    httpResponse;
  }

  lazy val getUrl = rec.getHeader.getUrl;

  lazy val getLength = rec.getHeader.getLength;

  lazy val getDate : java.util.Date = rec.getHeader.getDate;

  val SHA1_RE = new Regex("""^sha1:([a-zA-Z0-9]+)$""");

  lazy val getDigestStr : Option[String] = {
    if (!ready) cueUp;
    rec match {
      case arc : ARCRecord => {
        if (!this.closed) {
            /* must wait until finished */
          throw new Exception("You must close the ARC record before getting the digest string!");
        }
        this.digest.map(dig=>Base32.encode(dig.digest));
      }
      case warc : WARCRecord => {
        warc.getHeader.getHeaderValue(WARCConstants.HEADER_KEY_PAYLOAD_DIGEST) match {
          case SHA1_RE(s) => Some(s);
          case _          => None;
        }
      }
    }
  }

  lazy val getFilename = filename;

  /* InputStream wrapper */
  override def available = { 
    if (!ready) cueUp;
    rec.available; 
  }
  
  override def close = {
    dumpStream(this);
    if (!this.closed) rec.close;
    this.closed = true;
  }

  override def mark (readlimit : Int) = {
    if (!ready) cueUp; 
    rec.mark(readlimit);
  }
  
  override def markSupported = rec.markSupported;

  override def read = {
    if (!ready) cueUp; 
    val retval = rec.read;
    digest.map(_.update(retval.asInstanceOf[Byte]));
    retval;
  }

  override def read (buff : Array[Byte]) = {
    if (!ready) cueUp; 
    val retval = rec.read(buff);
    if (retval > 0) digest.map(_.update(buff, 0, retval));
    retval;
  }

  override def read (buff : Array[Byte], off : Int, len : Int) = {
    if (!ready) cueUp; 
    val retval = rec.read(buff, off, len);
    if (retval > 0) digest.map(_.update(buff, off, retval));
    retval;
  }

  override def reset = {
    if (!ready) cueUp;
    rec.reset;
  }

  override def skip (n : Long) = {
    if (!ready) cueUp; 
    rec.skip(n);
  }
}

object ArchiveRecordWrapper {
  /**
   * Method to read one line of input from an InputStream. A line
{'fraction_non_alphanumeric': 0.11708029197080291, 'fraction_numerical': 0.0035036496350364962, 'mean_word_length': 3.063266113088177, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 20, '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': '1218515', 'n_tokens_mistral': 3293, 'n_tokens_neox': 3232, 'n_words': 787}