<commit_before>package viscel.crawl

import java.time.Instant

import akka.http.scaladsl.HttpExt
import akka.http.scaladsl.coding.{Deflate, Gzip}
import akka.http.scaladsl.model.headers.{HttpEncodings, Location, Referer, `Accept-Encoding`, `Last-Modified`, ETag}
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.Materializer
import viscel.shared.Log
import viscel.store.Vurl
import cats.implicits.catsSyntaxEitherId

import scala.concurrent.duration.{FiniteDuration, SECONDS}
import scala.concurrent.{ExecutionContext, Future}


class AkkaHttpRequester(ioHttp: HttpExt)
                       (implicit val ec: ExecutionContext, materializer: Materializer)
  extends WebRequestInterface {

  val timeout = FiniteDuration(300, SECONDS)

  private def decompress(r: HttpResponse): Future[HttpResponse] =
    Deflate.decodeMessage(Gzip.decodeMessage(r)).toStrict(timeout)
  private def requestDecompressed(request: HttpRequest): Future[HttpResponse] =
    ioHttp.singleRequest(request.withHeaders(`Accept-Encoding`(HttpEncodings.deflate, HttpEncodings.gzip)))
    .flatMap(decompress)

  def extractResponseLocation(base: Vurl, httpResponse: HttpResponse): Vurl =
    httpResponse.header[Location].fold(base)(l => Vurl.fromUri(l.uri.resolvedAgainst(base.uri)))
  def extractLastModified(httpResponse: HttpResponse): Option[Instant] =
    httpResponse.header[`Last-Modified`].map(h => Instant.ofEpochMilli(h.date.clicks))
  def extractEtag(httpResponse: HttpResponse): Option[String] =
    httpResponse.header[ETag].map(_.toString())


  private def requestWithRedirects(request: HttpRequest, redirects: Int = 10): Future[HttpResponse] = {
    Log.Crawl.info(s"request ${request.uri}" + request.header[Referer].fold("")(r => s" ($r)"))

    requestDecompressed(request).flatMap { response =>
      if (response.status.isRedirection() && response.header[Location].isDefined) {
        val loc = response.header[Location].get.uri.resolvedAgainst(request.uri)
        requestWithRedirects(request.withUri(loc), redirects = redirects - 1)
      }
      else if (response.status.isSuccess()) {
        // if the response has no location header, we insert the url from the request as a location,
        // this allows all other systems to use the most accurate location available
        Future.successful(
          response.addHeader(Location.apply(
            extractResponseLocation(Vurl.fromUri(request.uri), response).uri)))
      }
      else Future.failed(RequestException(request.uri.toString(), response.status.toString()))
    }
  }


  private def requestInternal[R](request: VRequest): Future[VResponse[HttpResponse]] = {
    val req = HttpRequest(
      method = HttpMethods.GET,
      uri = request.href.uri,
      headers = request.origin.map(x => Referer.apply(x.uri)).toList)
    requestWithRedirects(req).map{ resp =>
      VResponse(resp,
                location = extractResponseLocation(request.href, resp),
                mime = resp.entity.contentType.mediaType.toString(),
                lastModified = extractLastModified(resp),
                etag = extractEtag(resp))
    }
  }


  override def get(request: VRequest): Future[VResponse[Either[Array[Byte], String]]] = {
    for {
      resp <- requestInternal(request)
      content <- if (resp.content.entity.contentType.mediaType.isText)
                   Unmarshal(resp.content).to[String].map(_.asRight)
                 else resp.content.entity.toStrict(timeout).map(_.data.toArray[Byte].asLeft)
    } yield resp.copy(content = content)
  }

}<commit_msg>fix binary non binary distinction<commit_after>package viscel.crawl

import java.time.Instant

import akka.http.scaladsl.HttpExt
import akka.http.scaladsl.coding.{Deflate, Gzip}
import akka.http.scaladsl.model.headers.{HttpEncodings, Location, Referer, `Accept-Encoding`, `Last-Modified`, ETag}
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.Materializer
import viscel.shared.Log
import viscel.store.Vurl
import cats.implicits.catsSyntaxEitherId

import scala.concurrent.duration.{FiniteDuration, SECONDS}
import scala.concurrent.{ExecutionContext, Future}


class AkkaHttpRequester(ioHttp: HttpExt)
                       (implicit val ec: ExecutionContext, materializer: Materializer)
  extends WebRequestInterface {

  val timeout = FiniteDuration(300, SECONDS)

  private def decompress(r: HttpResponse): Future[HttpResponse] =
    Deflate.decodeMessage(Gzip.decodeMessage(r)).toStrict(timeout)
  private def requestDecompressed(request: HttpRequest): Future[HttpResponse] =
    ioHttp.singleRequest(request.withHeaders(`Accept-Encoding`(HttpEncodings.deflate, HttpEncodings.gzip)))
    .flatMap(decompress)

  def extractResponseLocation(base: Vurl, httpResponse: HttpResponse): Vurl =
    httpResponse.header[Location].fold(base)(l => Vurl.fromUri(l.uri.resolvedAgainst(base.uri)))
  def extractLastModified(httpResponse: HttpResponse): Option[Instant] =
    httpResponse.header[`Last-Modified`].map(h => Instant.ofEpochMilli(h.date.clicks))
  def extractEtag(httpResponse: HttpResponse): Option[String] =
    httpResponse.header[ETag].map(_.toString())


  private def requestWithRedirects(request: HttpRequest, redirects: Int = 10): Future[HttpResponse] = {
    Log.Crawl.info(s"request ${request.uri}" + request.header[Referer].fold("")(r => s" ($r)"))

    requestDecompressed(request).flatMap { response =>
      if (response.status.isRedirection() && response.header[Location].isDefined) {
        val loc = response.header[Location].get.uri.resolvedAgainst(request.uri)
        requestWithRedirects(request.withUri(loc), redirects = redirects - 1)
      }
      else if (response.status.isSuccess()) {
        // if the response has no location header, we insert the url from the request as a location,
        // this allows all other systems to use the most accurate location available
        Future.successful(
          response.addHeader(Location.apply(
            extractResponseLocation(Vurl.fromUri(request.uri), response).uri)))
      }
      else Future.failed(RequestException(request.uri.toString(), response.status.toString()))
    }
  }


  private def requestInternal[R](request: VRequest): Future[VResponse[HttpResponse]] = {
    val req = HttpRequest(
      method = HttpMethods.GET,
      uri = request.href.uri,
      headers = request.origin.map(x => Referer.apply(x.uri)).toList)
    requestWithRedirects(req).map{ resp =>
      VResponse(resp,
                location = extractResponseLocation(request.href, resp),
                mime = resp.entity.contentType.mediaType.toString(),
                lastModified = extractLastModified(resp),
                etag = extractEtag(resp))
    }
  }


  override def get(request: VRequest): Future[VResponse[Either[Array[Byte], String]]] = {
    for {
      resp <- requestInternal(request)
      content <- if (!resp.content.entity.contentType.binary)
                   Unmarshal(resp.content).to[String].map(_.asRight)
                 else resp.content.entity.toStrict(timeout).map(_.data.toArray[Byte].asLeft)
    } yield resp.copy(content = content)
  }

}
{'fraction_non_alphanumeric': 0.11861841199944957, 'fraction_numerical': 0.001651300399064263, 'mean_word_length': 4.0262793914246195, 'pattern_counts': {'":': 0, '<': 7, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2489995', 'n_tokens_mistral': 2175, 'n_tokens_neox': 2111, 'n_words': 390}