<commit_before>package org.textfiledatatools.core;

import org.textfiledatatools.core.consumer.LoggerConsumer;
import org.textfiledatatools.core.consumer.RecordConsumer;
import org.textfiledatatools.core.consumer.UncheckedConsumerException;
import org.textfiledatatools.core.filter.RecordFilter;
import org.textfiledatatools.core.logger.RecordLogger;
import org.textfiledatatools.core.logger.SystemOutLogger;
import org.textfiledatatools.core.message.CompareMessageBuilder;
import org.textfiledatatools.core.message.RecordMessage;
import org.textfiledatatools.core.producer.RecordProducer;
import org.textfiledatatools.core.producer.UncheckedProducerException;

import java.util.*;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * This class consists of {@code static} utility methods for operating on record streams.
 *
 * @author <NAME>
 * @since 0.1
 */
public final class RecordStreams {

    private RecordStreams() {
    }

    public static Stream<Record> empty() {
        return Stream.empty();
    }

    public static <T extends Record> Stream<T> of(T record) {
        return Stream.of(record);
    }

    @SafeVarargs
    public static <T extends Record> Stream<T> of(T... records) {
        return Stream.of(records);
    }

    public static <T extends Record> Stream<T> produceStream(RecordProducer<T> recordProducer) throws UncheckedProducerException {
        return recordProducer.produceStream();
    }

    public static <T extends Record> Stream<T> generate(Supplier<T> recordSupplier) {
        return Stream.generate(recordSupplier);
    }

    public static <T extends Record> Stream<T> concat(Stream<T> firstRecordStream, Stream<T> secondRecordStream) {
        return Stream.concat(firstRecordStream, secondRecordStream);
    }

    @SafeVarargs
    public static <T extends Record> Stream<T> concat(Stream<T>... recordStreams) {
        // TODO Validate: concatWithFlatMap or concatWithReduce
        return concatWithFlatMap(recordStreams);
    }

    @SafeVarargs
    static <T extends Record> Stream<T> concatWithFlatMap(Stream<T>... recordStreams) {
        return Stream.of(recordStreams).flatMap(Function.identity());
    }

    @SafeVarargs
    static <T extends Record> Stream<T> concatWithReduce(Stream<T>... recordStreams) {
        return Stream.of(recordStreams).reduce(Stream.empty(), Stream::concat);
    }

    public static <R extends Record, T extends R> RecordConsumer<R> consume(Stream<T> recordStream,
                                                                            RecordConsumer<R> recordConsumer) throws UncheckedConsumerException {
        recordStream.forEachOrdered(recordConsumer::consume);
        return recordConsumer;
    }

    public static <T extends Record> List<T> collect(Stream<T> recordStream) {
        return recordStream.collect(Collectors.toList());
    }

    public static <T extends Record> List<String> collectMessages(Stream<T> recordStream,
                                                                  RecordMessage<? super T> recordMessage) {
        Objects.requireNonNull(recordMessage);
        return recordStream.map(recordMessage::createMessage).collect(Collectors.toList());
    }

    public static <T extends Record> Map<String, List<String>> collectMessages(Stream<T> recordStream,
                                                                               RecordMessage<? super T> keyMessage,
                                                                               RecordMessage<? super T> valueMessage) {
        Objects.requireNonNull(keyMessage);
        Objects.requireNonNull(valueMessage);
        return recordStream.collect(Collectors.groupingBy(keyMessage::createMessage, HashMap::new,
                Collectors.mapping(valueMessage::createMessage, Collectors.toList())));
    }

    public static <T extends Record> String joinMessages(Stream<T> recordStream,
                                                         RecordMessage<? super T> recordMessage,
                                                         String delimiter) {
        Objects.requireNonNull(recordMessage);
        Objects.requireNonNull(delimiter);
        return recordStream.map(recordMessage::createMessage).collect(Collectors.joining(delimiter));
    }

    public static void printLines(Stream<Record> recordStream) {
        consume(recordStream, new LoggerConsumer<>(new SystemOutLogger()));
    }

    public static <T extends Record> Stream<T> filterAndLog(Stream<T> recordStream,
                                                            RecordFilter<T> recordFilter,
                                                            RecordLogger<T> validLogger,
                                                            RecordLogger<T> invalidLogger) {
        return recordStream.peek(r -> {
            if (recordFilter.isValid(r)) {
                validLogger.log(r);
            } else {<commit_msg>Add log and filter methods<commit_after>package org.textfiledatatools.core;

import org.textfiledatatools.core.consumer.LoggerConsumer;
import org.textfiledatatools.core.consumer.RecordConsumer;
import org.textfiledatatools.core.consumer.UncheckedConsumerException;
import org.textfiledatatools.core.filter.RecordFilter;
import org.textfiledatatools.core.logger.RecordLogger;
import org.textfiledatatools.core.logger.SystemOutLogger;
import org.textfiledatatools.core.message.CompareMessageBuilder;
import org.textfiledatatools.core.message.RecordMessage;
import org.textfiledatatools.core.producer.RecordProducer;
import org.textfiledatatools.core.producer.UncheckedProducerException;

import java.util.*;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * This class consists of {@code static} utility methods for operating on record streams.
 *
 * @author <NAME>
 * @since 0.1
 */
public final class RecordStreams {

    private RecordStreams() {
    }

    public static Stream<Record> empty() {
        return Stream.empty();
    }

    public static <T extends Record> Stream<T> of(T record) {
        return Stream.of(record);
    }

    @SafeVarargs
    public static <T extends Record> Stream<T> of(T... records) {
        return Stream.of(records);
    }

    public static <T extends Record> Stream<T> produceStream(RecordProducer<T> recordProducer) throws UncheckedProducerException {
        return recordProducer.produceStream();
    }

    public static <T extends Record> Stream<T> generate(Supplier<T> recordSupplier) {
        return Stream.generate(recordSupplier);
    }

    public static <T extends Record> Stream<T> concat(Stream<T> firstRecordStream, Stream<T> secondRecordStream) {
        return Stream.concat(firstRecordStream, secondRecordStream);
    }

    @SafeVarargs
    public static <T extends Record> Stream<T> concat(Stream<T>... recordStreams) {
        // TODO Validate: concatWithFlatMap or concatWithReduce
        return concatWithFlatMap(recordStreams);
    }

    @SafeVarargs
    static <T extends Record> Stream<T> concatWithFlatMap(Stream<T>... recordStreams) {
        return Stream.of(recordStreams).flatMap(Function.identity());
    }

    @SafeVarargs
    static <T extends Record> Stream<T> concatWithReduce(Stream<T>... recordStreams) {
        return Stream.of(recordStreams).reduce(Stream.empty(), Stream::concat);
    }

    public static <R extends Record, T extends R> RecordConsumer<R> consume(Stream<T> recordStream,
                                                                            RecordConsumer<R> recordConsumer) throws UncheckedConsumerException {
        recordStream.forEachOrdered(recordConsumer::consume);
        return recordConsumer;
    }

    public static <T extends Record> List<T> collect(Stream<T> recordStream) {
        return recordStream.collect(Collectors.toList());
    }

    public static <T extends Record> List<String> collectMessages(Stream<T> recordStream,
                                                                  RecordMessage<? super T> recordMessage) {
        Objects.requireNonNull(recordMessage);
        return recordStream.map(recordMessage::createMessage).collect(Collectors.toList());
    }

    public static <T extends Record> Map<String, List<String>> collectMessages(Stream<T> recordStream,
                                                                               RecordMessage<? super T> keyMessage,
                                                                               RecordMessage<? super T> valueMessage) {
        Objects.requireNonNull(keyMessage);
        Objects.requireNonNull(valueMessage);
        return recordStream.collect(Collectors.groupingBy(keyMessage::createMessage, HashMap::new,
                Collectors.mapping(valueMessage::createMessage, Collectors.toList())));
    }

    public static <T extends Record> String joinMessages(Stream<T> recordStream,
                                                         RecordMessage<? super T> recordMessage,
                                                         String delimiter) {
        Objects.requireNonNull(recordMessage);
        Objects.requireNonNull(delimiter);
        return recordStream.map(recordMessage::createMessage).collect(Collectors.joining(delimiter));
    }

    public static void printLines(Stream<Record> recordStream) {
        consume(recordStream, new LoggerConsumer<>(new SystemOutLogger()));
    }

    public static <T extends Record> Stream<T> log(Stream<T> recordStream, RecordLogger<? super T> recordLogger) {
        return recordStream.peek(recordLogger::log);
    }

    public static <T extends Record> Stream<T> filter(Stream<T> recordStream, RecordFilter<? super T> recordFilter) {
        return recordStream.filter(recordFilter::isValid);
    }

    public static <T extends Record> Stream<T> filterAndLog(Stream<T> recordStream,
                                                            RecordFilter<? super T> recordFilter,
                                                            RecordLogger<? super T> validLogger,
                                                            RecordLogger<? super T> invalidLogger) {
        return recordStream.peek(r -> {
            if (recordFilter.isValid(r)) {
                validLogger.log(r);
            } else {
{'fraction_non_alphanumeric': 0.09178975637828506, 'fraction_numerical': 0.0003836562440053712, 'mean_word_length': 2.552640545144804, 'pattern_counts': {'":': 0, '<': 117, '<?xml version=': 0, '>': 119, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 42, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2012399', 'n_tokens_mistral': 2615, 'n_tokens_neox': 2553, 'n_words': 574}