<commit_before>#include "product_price_trend.h"

#include <common/Utilities.h>
#include <log-manager/PriceHistory.h>
#include <document-manager/DocumentManager.h>

#include <am/range/AmIterator.h>
#include <libcassandra/util_functions.h>
#include <glog/logging.h>

#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/filesystem.hpp>
#include <algorithm>

using namespace std;
using namespace libcassandra;
using izenelib::util::UString;

namespace sf1r
{

ProductPriceTrend::ProductPriceTrend(
        const boost::shared_ptr<DocumentManager>& document_manager,
        const CassandraStorageConfig& cassandraConfig,
        const string& data_dir,
        const vector<string>& group_prop_vec,
        const vector<uint32_t>& time_int_vec)
    : document_manager_(document_manager)
    , cassandraConfig_(cassandraConfig)
    , data_dir_(data_dir)
    , group_prop_vec_(group_prop_vec)
    , time_int_vec_(time_int_vec)
    , buffer_size_(0)
    , enable_tpc_(!group_prop_vec_.empty() && !time_int_vec_.empty())
{
}

ProductPriceTrend::~ProductPriceTrend()
{
    for (TPCStorage::iterator it = tpc_storage_.begin();
            it != tpc_storage_.end(); ++it)
    {
        for (uint32_t i = 0; i < time_int_vec_.size(); i++)
            delete it->second[i];
    }
}

bool ProductPriceTrend::Init()
{
    if (!cassandraConfig_.enable)
        return false;

    price_history_.reset(new PriceHistory(cassandraConfig_.keyspace));
    price_history_->createColumnFamily();

    if (!price_history_->is_enabled)
        return false;

    for (vector<string>::const_iterator it = group_prop_vec_.begin();
            it != group_prop_vec_.end(); ++it)
    {
        vector<TPCBTree *>& prop_tpc = tpc_storage_[*it];
        for (uint32_t i = 0; i < time_int_vec_.size(); i++)
        {
            string db_path = data_dir_ + "/" + *it + "." + boost::lexical_cast<string>(time_int_vec_[i]) + ".tpc";
            prop_tpc.push_back(new TPCBTree(db_path));
            if (!prop_tpc.back()->open())
            {
                boost::filesystem::remove_all(db_path);
                prop_tpc.back()->open();
            }
        }
    }
    price_history_buffer_.resize(2000);

    return true;
}

bool ProductPriceTrend::Insert(
        const string& docid,
        const ProductPrice& price,
        time_t timestamp)
{
    price_history_buffer_[buffer_size_].resetKey(docid);
    price_history_buffer_[buffer_size_].resetHistory(0, timestamp, price);
    ++buffer_size_;

    if (IsBufferFull_())
    {
        return Flush();
    }

    return true;
}

bool ProductPriceTrend::Update(
        const string& docid,
        const ProductPrice& price,
        time_t timestamp,
        map<string, string>& group_prop_map)
{
    price_history_buffer_[buffer_size_].resetKey(docid);
    price_history_buffer_[buffer_size_].resetHistory(0, timestamp, price);
    ++buffer_size_;

    if (enable_tpc_ && !group_prop_map.empty() && price.value.first > 0)
    {
        PropItemType& prop_item = prop_map_[s_docid];
        prop_item.first = price.value.first;
        prop_item.second.swap(group_prop_map);
    }

    if (IsBufferFull_())
    {
        return Flush();
    }

    return true;
}

bool ProductPriceTrend::IsBufferFull_() const
{
    return buffer_size_ >= 2000;
}

bool ProductPriceTrend::Flush()
{
    bool ret = true;
    time_t now = Utilities::createTimeStamp();

    if (!prop_map_.empty())
    {
        for (uint32_t i = 0; i < time_int_vec_.size(); i++)<commit_msg>tiny fix<commit_after>#include "product_price_trend.h"

#include <common/Utilities.h>
#include <log-manager/PriceHistory.h>
#include <document-manager/DocumentManager.h>

#include <am/range/AmIterator.h>
#include <libcassandra/util_functions.h>
#include <glog/logging.h>

#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/filesystem.hpp>
#include <algorithm>

using namespace std;
using namespace libcassandra;
using izenelib::util::UString;

namespace sf1r
{

ProductPriceTrend::ProductPriceTrend(
        const boost::shared_ptr<DocumentManager>& document_manager,
        const CassandraStorageConfig& cassandraConfig,
        const string& data_dir,
        const vector<string>& group_prop_vec,
        const vector<uint32_t>& time_int_vec)
    : document_manager_(document_manager)
    , cassandraConfig_(cassandraConfig)
    , data_dir_(data_dir)
    , group_prop_vec_(group_prop_vec)
    , time_int_vec_(time_int_vec)
    , buffer_size_(0)
    , enable_tpc_(!group_prop_vec_.empty() && !time_int_vec_.empty())
{
}

ProductPriceTrend::~ProductPriceTrend()
{
    for (TPCStorage::iterator it = tpc_storage_.begin();
            it != tpc_storage_.end(); ++it)
    {
        for (uint32_t i = 0; i < time_int_vec_.size(); i++)
            delete it->second[i];
    }
}

bool ProductPriceTrend::Init()
{
    if (!cassandraConfig_.enable)
        return false;

    price_history_.reset(new PriceHistory(cassandraConfig_.keyspace));
    price_history_->createColumnFamily();

    if (!price_history_->is_enabled)
        return false;

    for (vector<string>::const_iterator it = group_prop_vec_.begin();
            it != group_prop_vec_.end(); ++it)
    {
        vector<TPCBTree *>& prop_tpc = tpc_storage_[*it];
        for (uint32_t i = 0; i < time_int_vec_.size(); i++)
        {
            string db_path = data_dir_ + "/" + *it + "." + boost::lexical_cast<string>(time_int_vec_[i]) + ".tpc";
            prop_tpc.push_back(new TPCBTree(db_path));
            if (!prop_tpc.back()->open())
            {
                boost::filesystem::remove_all(db_path);
                prop_tpc.back()->open();
            }
        }
    }
    price_history_buffer_.resize(2000);

    return true;
}

bool ProductPriceTrend::Insert(
        const string& docid,
        const ProductPrice& price,
        time_t timestamp)
{
    price_history_buffer_[buffer_size_].resetKey(docid);
    price_history_buffer_[buffer_size_].resetHistory(0, timestamp, price);
    ++buffer_size_;

    if (IsBufferFull_())
    {
        return Flush();
    }

    return true;
}

bool ProductPriceTrend::Update(
        const string& docid,
        const ProductPrice& price,
        time_t timestamp,
        map<string, string>& group_prop_map)
{
    price_history_buffer_[buffer_size_].resetKey(docid);
    price_history_buffer_[buffer_size_].resetHistory(0, timestamp, price);
    ++buffer_size_;

    if (enable_tpc_ && !group_prop_map.empty() && price.value.first > 0)
    {
        PropItemType& prop_item = prop_map_[docid];
        prop_item.first = price.value.first;
        prop_item.second.swap(group_prop_map);
    }

    if (IsBufferFull_())
    {
        return Flush();
    }

    return true;
}

bool ProductPriceTrend::IsBufferFull_() const
{
    return buffer_size_ >= 2000;
}

bool ProductPriceTrend::Flush()
{
    bool ret = true;
    time_t now = Utilities::createTimeStamp();

    if (!prop_map_.empty())
    {
        for (uint32_t i = 0; i < time_int_vec_.size(); i++)
{'fraction_non_alphanumeric': 0.12658951278754108, 'fraction_numerical': 0.006858122588941277, 'mean_word_length': 3.2016806722689077, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 49, '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': '7053273', 'n_tokens_mistral': 2372, 'n_tokens_neox': 2311, 'n_words': 390}