<commit_before>#pragma once

#include <boost/config.hpp>

#include "blackhole/attribute.hpp"
#include "blackhole/repository/factory/traits.hpp"
#include "blackhole/sink/files/backend.hpp"
#include "blackhole/sink/files/config.hpp"
#include "blackhole/sink/files/flusher.hpp"
#include "blackhole/sink/files/rotation.hpp"
#include "blackhole/sink/files/writer.hpp"
#include "blackhole/utils/unique.hpp"

namespace blackhole {

namespace sink {

template<class Backend, class Rotator, class = void>
class file_handler_t;

template<class Backend>
class file_handler_t<Backend, NoRotation, void> {
    Backend m_backend;
    files::writer_t<Backend> writer;
    files::flusher_t<Backend> flusher;
public:
    file_handler_t(const std::string& path, const files::config_t<NoRotation>& config) :
        m_backend(path),
        writer(m_backend),
        flusher(config.autoflush, m_backend)
    {}

    void handle(const std::string& message) {
        writer.write(message);
        flusher.flush();
    }

    const Backend& backend() {
        return m_backend;
    }
};

template<class Backend, class Rotator>
class file_handler_t<
    Backend,
    Rotator,
    typename std::enable_if<
        !std::is_same<Rotator, NoRotation>::value
    >::type>
{
    Backend m_backend;
    files::writer_t<Backend> writer;
    files::flusher_t<Backend> flusher;
    Rotator rotator;
public:
    file_handler_t(const std::string& path, const files::config_t<Rotator>& config) :
        m_backend(path),
        writer(m_backend),
        flusher(config.autoflush, m_backend),
        rotator(config.rotation, m_backend)
    {}

    void handle(const std::string& message) {
        writer.write(message);
        flusher.flush();
        if (rotator.necessary(message)) {
            rotator.rotate();
        }
    }

    const Backend& backend() {
        return m_backend;
    }
};

template<class Backend = files::boost_backend_t, class Rotator = NoRotation>
class files_t {
    typedef file_handler_t<Backend, Rotator> handler_type;
    typedef std::unordered_map<std::string, std::shared_ptr<handler_type>> handlers_type;

    files::config_t<Rotator> config;
    handlers_type m_handlers;
public:
    typedef files::config_t<Rotator> config_type;

    static const char* name() {
        return "files";
    }

    files_t(const config_type& config) :
        config(config)
    {}

    void consume(const std::string& message, const log::attributes_t& attributes = log::attributes_t()) {
        auto filename = make_filename(attributes);
        auto it = m_handlers.find(filename);
        if (it == m_handlers.end()) {
            it = m_handlers.insert(it, std::make_pair(filename, std::make_shared<handler_type>(filename, config)));
        }
        it->second->handle(message);
    }

    const handlers_type& handlers() {
        return m_handlers;
    }

    std::string make_filename(const log::attributes_t& attributes) const {
        auto it = config.path.begin();
        auto end = config.path.end();
        std::string filename;
        while (it != end) {
            if (*it == '%' && it + 1 != end && *(it + 1) == '(') {
                it += 2;

                std::string key;
                bool found = false;
                while (it != end) {
                    if (*it == ')' && it + 1 != end && *(it + 1) == 's') {
                        found = true;
                        it += 2;
                        break;
                    }
                    key.push_back(*it);
                    it++;
                }

                if (found) {
                    auto ait = attributes.find(key);
                    if (ait != attributes.end()) {
                        std::ostringstream stream;
                        stream << ait->second.value;
                        filename.append(stream.str());
                    } else {
                        filename.append(key);
                    }
                } else {
                    filename.append(key);
                }
            }

            if (it == end) {
                break;
            }
            filename.push_back(*it);
            it++;
        }
//        std::cout << "filename=" << filename << std::endl;
        return filename;
    }
};

} // namespace sink

template<class Backend, class Watcher>
struct unique_id_traits<sink::files_t<Backend, sink::rotator_t<Backend, Watcher>>> {
    typedef sink::rotator_t<Backend, Watcher> rotator_type;
    typedef sink::files_t<Backend, rotator_type> sink_type;

    typedef std::map<std::string, boost::any> variant_map_t;

    static std::string generate(const boost::any& config) {
        const variant_map_t& cfg = boost::any_cast<variant_map_t>(config);

        auto rotation_it = cfg.find("rotation");
        if (rotation_it != cfg.end()) {
            const variant_map_t& rotation = boost::any_cast<variant_map_t>(rotation_it->second);

            const bool has_size_watcher = rotation.find("size") != rotation.end();
            const bool has_datetime_watcher = rotation.find("period") != rotation.end();
<commit_msg>[Aux] Files sink now uses generic string formatter for substituting attributes into generic filename pattern.<commit_after>#pragma once

#include <boost/config.hpp>

#include "blackhole/attribute.hpp"
#include "blackhole/detail/string/formatting/formatter.hpp"
#include "blackhole/repository/factory/traits.hpp"
#include "blackhole/sink/files/backend.hpp"
#include "blackhole/sink/files/config.hpp"
#include "blackhole/sink/files/flusher.hpp"
#include "blackhole/sink/files/rotation.hpp"
#include "blackhole/sink/files/writer.hpp"
#include "blackhole/utils/unique.hpp"

namespace blackhole {

namespace sink {

template<class Backend, class Rotator, class = void>
class file_handler_t;

template<class Backend>
class file_handler_t<Backend, NoRotation, void> {
    Backend m_backend;
    files::writer_t<Backend> writer;
    files::flusher_t<Backend> flusher;
public:
    file_handler_t(const std::string& path, const files::config_t<NoRotation>& config) :
        m_backend(path),
        writer(m_backend),
        flusher(config.autoflush, m_backend)
    {}

    void handle(const std::string& message) {
        writer.write(message);
        flusher.flush();
    }

    const Backend& backend() {
        return m_backend;
    }
};

template<class Backend, class Rotator>
class file_handler_t<
    Backend,
    Rotator,
    typename std::enable_if<
        !std::is_same<Rotator, NoRotation>::value
    >::type>
{
    Backend m_backend;
    files::writer_t<Backend> writer;
    files::flusher_t<Backend> flusher;
    Rotator rotator;
public:
    file_handler_t(const std::string& path, const files::config_t<Rotator>& config) :
        m_backend(path),
        writer(m_backend),
        flusher(config.autoflush, m_backend),
        rotator(config.rotation, m_backend)
    {}

    void handle(const std::string& message) {
        writer.write(message);
        flusher.flush();
        if (rotator.necessary(message)) {
            rotator.rotate();
        }
    }

    const Backend& backend() {
        return m_backend;
    }
};

struct substitute_attribute_t {
    const log::attributes_t& attributes;

    void operator()(aux::attachable_ostringstream& stream, const std::string& placeholder) const {
        auto it = attributes.find(placeholder);
        if (it == attributes.end()) {
            stream.rdbuf()->storage()->append(placeholder);
        } else {
            stream << it->second.value;
        }
    }
};

template<class Backend = files::boost_backend_t, class Rotator = NoRotation>
class files_t {
    typedef file_handler_t<Backend, Rotator> handler_type;
    typedef std::unordered_map<std::string, std::shared_ptr<handler_type>> handlers_type;

    files::config_t<Rotator> config;
    handlers_type m_handlers;
//    aux::formatter_t formatter;
public:
    typedef files::config_t<Rotator> config_type;

    static const char* name() {
        return "files";
    }

    files_t(const config_type& config) :
        config(config)
//        formatter(config.path)
    {}

    void consume(const std::string& message, const log::attributes_t& attributes = log::attributes_t()) {
        auto filename = make_filename(attributes);
        auto it = m_handlers.find(filename);
        if (it == m_handlers.end()) {
            it = m_handlers.insert(it, std::make_pair(filename, std::make_shared<handler_type>(filename, config)));
        }
        it->second->handle(message);
    }

    const handlers_type& handlers() {
        return m_handlers;
    }

    std::string make_filename(const log::attributes_t& attributes) {
        aux::formatter_t formatter(config.path);
        formatter.on_placeholder(substitute_attribute_t { attributes });
        return formatter.execute();
    }
};

} // namespace sink

template<class Backend, class Watcher>
struct unique_id_traits<sink::files_t<Backend, sink::rotator_t<Backend, Watcher>>> {
    typedef sink::rotator_t<Backend, Watcher> rotator_type;
    typedef sink::files_t<Backend, rotator_type> sink_type;

    typedef std::map<std::string, boost::any> variant_map_t;

    static std::string generate(const boost::any& config) {
        const variant_map_t& cfg = boost::any_cast<variant_map_t>(config);

        auto rotation_it = cfg.find("rotation");
        if (rotation_it != cfg.end()) {
            const variant_map_t& rotation = boost::any_cast<variant_map_t>(rotation_it->second);

            const bool has_size_watcher = rotation.find("size") != rotation.end();
            const bool has_datetime_watcher = rotation.find("period") != rotation.end();

{'fraction_non_alphanumeric': 0.1252590136759221, 'fraction_numerical': 0.0006216328222130128, 'mean_word_length': 2.7618862042088854, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 73, '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': '2858170', 'n_tokens_mistral': 2953, 'n_tokens_neox': 2901, 'n_words': 582}