<commit_before>package com.spotify.scio.io

import java.io.{File, FileInputStream, InputStream, SequenceInputStream}
import java.net.URI
import java.util.Collections

import com.fasterxml.jackson.databind.{ObjectMapper, SerializationFeature}
import com.google.api.client.googleapis.auth.oauth2.GoogleCredential
import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
import com.google.api.client.json.jackson2.JacksonFactory
import com.google.api.client.util.Charsets
import com.google.api.services.storage.Storage
import com.spotify.scio.bigquery.TableRow
import org.apache.avro.file.DataFileStream
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.specific.SpecificDatumReader
import org.apache.commons.io.{FileUtils, IOUtils}

import scala.collection.JavaConverters._
import scala.reflect.ClassTag

private object FileStorage {
  def apply(path: String): FileStorage =
    if (new URI(path).getScheme == "gs") new GcsStorage(path) else new LocalStorage(path)
}

private trait FileStorage {

  protected val path: String

  def genericAvroFile: Iterator[GenericRecord] = {
    val reader = new GenericDatumReader[GenericRecord]()
    new DataFileStream[GenericRecord](getDirectoryInputStream(path), reader).iterator().asScala
  }

  def specificAvroFile[T: ClassTag]: Iterator[T] = {
    val cls = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]
    val reader = new SpecificDatumReader[T](cls)
    new DataFileStream[T](getDirectoryInputStream(path), reader).iterator().asScala
  }

  def textFile: Iterator[String] =
    IOUtils.lineIterator(getDirectoryInputStream(path), Charsets.UTF_8).asScala

  def tableRowJsonFile: Iterator[TableRow] = {
    val mapper = new ObjectMapper().disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)
    textFile.map(mapper.readValue(_, classOf[TableRow]))
  }

  def delete(): Unit

  private def getDirectoryInputStream(path: String): InputStream = {
    val inputs = list.map(getObjectInputStream).asJava
    new SequenceInputStream(Collections.enumeration(inputs))
  }

  protected def list: Seq[String]

  protected def getObjectInputStream(path: String): InputStream

}

private class GcsStorage(protected val path: String) extends FileStorage {

  lazy private val storage = new Storage.Builder(
    GoogleNetHttpTransport.newTrustedTransport(),
    JacksonFactory.getDefaultInstance,
    GoogleCredential.getApplicationDefault).build()

  override def delete(): Unit = {
    val uri = new URI(path)
    require(uri.getScheme == "gs", s"Not a GCS path: $path")
    val bucket = uri.getHost
    val prefix = uri.getPath.replaceAll("^/", "") + (if (uri.getPath.endsWith("/")) "" else "/")
    val objects = storage.objects()
      .list(bucket)
      .setPrefix(prefix)
      .execute()
      .getItems.asScala
    objects.foreach { o =>
      storage.objects().delete(bucket, o.getName).execute()
    }
  }

  override protected def list: Seq[String] = {
    val uri = new URI(path)
    require(uri.getScheme == "gs", s"Not a GCS path: $path")
    val bucket = uri.getHost
    val prefix = uri.getPath.replaceAll("^/", "") + (if (uri.getPath.endsWith("/")) "" else "/")
    val pDelimiters = prefix.count(isDelimiter)
    storage.objects()
      .list(bucket)
      .setPrefix(prefix)
      .execute()
      .getItems.asScala
      .filter{ o =>
        val n = o.getName
        val nDelimiters: Int = n.count(isDelimiter)<commit_msg>minor style fix in FileStorage<commit_after>package com.spotify.scio.io

import java.io.{File, FileInputStream, InputStream, SequenceInputStream}
import java.net.URI
import java.util.Collections

import com.fasterxml.jackson.databind.{ObjectMapper, SerializationFeature}
import com.google.api.client.googleapis.auth.oauth2.GoogleCredential
import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
import com.google.api.client.json.jackson2.JacksonFactory
import com.google.api.client.util.Charsets
import com.google.api.services.storage.Storage
import com.spotify.scio.bigquery.TableRow
import org.apache.avro.file.DataFileStream
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.specific.SpecificDatumReader
import org.apache.commons.io.{FileUtils, IOUtils}

import scala.collection.JavaConverters._
import scala.reflect.ClassTag

private object FileStorage {
  def apply(path: String): FileStorage =
    if (new URI(path).getScheme == "gs") new GcsStorage(path) else new LocalStorage(path)
}

private trait FileStorage {

  protected val path: String

  def genericAvroFile: Iterator[GenericRecord] = {
    val reader = new GenericDatumReader[GenericRecord]()
    new DataFileStream[GenericRecord](getDirectoryInputStream(path), reader).iterator().asScala
  }

  def specificAvroFile[T: ClassTag]: Iterator[T] = {
    val cls = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]
    val reader = new SpecificDatumReader[T](cls)
    new DataFileStream[T](getDirectoryInputStream(path), reader).iterator().asScala
  }

  def textFile: Iterator[String] =
    IOUtils.lineIterator(getDirectoryInputStream(path), Charsets.UTF_8).asScala

  def tableRowJsonFile: Iterator[TableRow] = {
    val mapper = new ObjectMapper().disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)
    textFile.map(mapper.readValue(_, classOf[TableRow]))
  }

  def delete(): Unit

  private def getDirectoryInputStream(path: String): InputStream = {
    val inputs = list.map(getObjectInputStream).asJava
    new SequenceInputStream(Collections.enumeration(inputs))
  }

  protected def list: Seq[String]

  protected def getObjectInputStream(path: String): InputStream

}

private class GcsStorage(protected val path: String) extends FileStorage {

  lazy private val storage = new Storage.Builder(
    GoogleNetHttpTransport.newTrustedTransport(),
    JacksonFactory.getDefaultInstance,
    GoogleCredential.getApplicationDefault).build()

  override def delete(): Unit = {
    val uri = new URI(path)
    require(uri.getScheme == "gs", s"Not a GCS path: $path")
    val bucket = uri.getHost
    val prefix = uri.getPath.replaceAll("^/", "") + (if (uri.getPath.endsWith("/")) "" else "/")
    val objects = storage.objects()
      .list(bucket)
      .setPrefix(prefix)
      .execute()
      .getItems.asScala
    objects.foreach { o =>
      storage.objects().delete(bucket, o.getName).execute()
    }
  }

  override protected def list: Seq[String] = {
    val uri = new URI(path)
    require(uri.getScheme == "gs", s"Not a GCS path: $path")
    val bucket = uri.getHost
    val prefix = uri.getPath.replaceAll("^/", "") + (if (uri.getPath.endsWith("/")) "" else "/")
    val pDelimiters = prefix.count(isDelimiter)
    storage.objects()
      .list(bucket)
      .setPrefix(prefix)
      .execute()
      .getItems.asScala
      .filter { o =>
        val n = o.getName
        val nDelimiters: Int = n.count(isDelimiter)
{'fraction_non_alphanumeric': 0.1179026651216686, 'fraction_numerical': 0.0008690614136732329, 'mean_word_length': 4.386115444617785, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '3831412', 'n_tokens_mistral': 2125, 'n_tokens_neox': 2088, 'n_words': 437}