<commit_before>#include "product_manager.h"
#include "product_data_source.h"
#include "operation_processor.h"
#include "uuid_generator.h"
#include "product_editor.h"
#include "product_clustering.h"
#include "product_price_trend.h"

#include <log-manager/LogServerRequest.h>
#include <log-manager/LogServerConnection.h>

#include <common/ScdWriter.h>
#include <common/Utilities.h>

#include <boost/unordered_set.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/dynamic_bitset.hpp>
#include <glog/logging.h>

using namespace sf1r;
using izenelib::util::UString;

//#define PM_PROFILER

ProductManager::ProductManager(
        const std::string& work_dir,
        ProductDataSource* data_source,
        OperationProcessor* op_processor,
        ProductPriceTrend* price_trend,
        const PMConfig& config)
    : work_dir_(work_dir)
    , config_(config)
    , data_source_(data_source)
    , op_processor_(op_processor)
    , price_trend_(price_trend)
    , clustering_(NULL)
    , editor_(new ProductEditor(data_source, op_processor, config))
    , util_(config, data_source)
    , has_price_trend_(false)
    , inhook_(false)
{
    if (price_trend_)
    {
        if (price_trend_->Init())
            has_price_trend_ = true;
        else
            std::cerr << "Error: Price trend has not been properly initialized" << std::endl;
    }

//     if (!config_.backup_path.empty())
//     {
//         backup_ = new ProductBackup(config_.backup_path);
//     }
}

ProductManager::~ProductManager()
{
    delete editor_;
}

bool ProductManager::Recover()
{
//     if (backup_==NULL)
//     {
//         error_ = "Backup not initialzed.";
//         return false;
//     }
//     if (!backup_->Recover(this))
//     {
// //         error_ = "Backup recover failed.";
//         return false;
//     }
//     if (!GenOperations_())
//     {
//         return false;
//     }
//     return true;
    return false;
}

ProductClustering* ProductManager::GetClustering_()
{
    if (!clustering_)
    {
        if (config_.enable_clustering_algo)
        {
            clustering_ = new ProductClustering(work_dir_+"/clustering", config_);
            if (!clustering_->Open())
            {
                std::cout<<"ProductClustering open failed"<<std::endl;
                delete clustering_;
                clustering_ = NULL;
            }
        }
    }
    return clustering_;
}

bool ProductManager::HookInsert(PMDocumentType& doc, izenelib::ir::indexmanager::IndexerDocument& index_document, time_t timestamp)
{
    inhook_ = true;
    boost::mutex::scoped_lock lock(human_mutex_);

    if (has_price_trend_)
    {
        ProductPrice price;
        if (util_.GetPrice(doc, price))
        {
            UString docid;
            if (doc.getProperty(config_.docid_property_name, docid))
            {
                std::string docid_str;
                docid.convertString(docid_str, UString::UTF_8);
                if (timestamp == -1) GetTimestamp_(doc, timestamp);
                std::map<std::string, std::string> group_prop_map;
//              task_type task = boost::bind(&ProductPriceTrend::Insert, price_trend_, docid_str, price, timestamp);
                GetGroupProperties_(doc, group_prop_map);
                task_type task = boost::bind(&ProductPriceTrend::Update, price_trend_, docid_str, price, timestamp, group_prop_map);
                jobScheduler_.addTask(task);
            }
        }
    }
    ProductClustering* clustering = GetClustering_();
    if (clustering)
    {
        UString uuid;
        doc.property(config_.uuid_property_name) = uuid;
        if (!data_source_->SetUuid(index_document, uuid)) return false; // set a empty uuid for rtype update later
        clustering->Insert(doc);
    }
    return true;
}

bool ProductManager::HookUpdate(PMDocumentType& to, izenelib::ir::indexmanager::IndexerDocument& index_document, time_t timestamp, bool r_type)
{
    inhook_ = true;
    boost::mutex::scoped_lock lock(human_mutex_);

    uint32_t fromid = index_document.getId(); //oldid
    PMDocumentType from;
    if (!data_source_->GetDocument(fromid, from)) return false;
    UString from_uuid;
    if (!from.getProperty(config_.uuid_property_name, from_uuid) ) return false;
    ProductPrice from_price;
    ProductPrice to_price;
    util_.GetPrice(fromid, from_price);
    util_.GetPrice(to, to_price);

    if (has_price_trend_ && to_price.Valid() && from_price != to_price)
    {
        UString docid;
        if (to.getProperty(config_.docid_property_name, docid ) )
        {
            std::string docid_str;
            docid.convertString(docid_str, UString::UTF_8);
            if (timestamp == -1) GetTimestamp_(to, timestamp);
            std::map<std::string, std::string> group_prop_map;
            GetGroupProperties_(to, group_prop_map);
            task_type task = boost::bind(&ProductPriceTrend::Update, price_trend_, docid_str, to_price, timestamp, group_prop_map);
            jobScheduler_.addTask(task);
        }
    }

    if (config_.enable_clustering_algo)
    {
        std::vector<uint32_t> docid_list;
        data_source_->GetDocIdList(from_uuid, docid_list, fromid); // except from.docid
        if (docid_list.empty()) // the from doc is unique, so delete it and insert 'to'
        {
            if (!data_source_->SetUuid(index_document, from_uuid)) return false;
            PMDocumentType new_doc(to);
            to.property(config_.uuid_property_name) = from_uuid;
            new_doc.property(config_.docid_property_name) = from_uuid;
            util_.SetItemCount(new_doc, 1);
            op_processor_->Append(2, new_doc);// if r_type, only numeric properties in 'to'
        }
        else
        {
            //need not to update(insert) to.uuid,
            if (!data_source_->SetUuid(index_document, from_uuid)) return false;<commit_msg>tiny fix<commit_after>#include "product_manager.h"
#include "product_data_source.h"
#include "operation_processor.h"
#include "uuid_generator.h"
#include "product_editor.h"
#include "product_clustering.h"
#include "product_price_trend.h"

#include <log-manager/LogServerRequest.h>
#include <log-manager/LogServerConnection.h>

#include <common/ScdWriter.h>
#include <common/Utilities.h>

#include <boost/unordered_set.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/dynamic_bitset.hpp>
#include <glog/logging.h>

using namespace sf1r;
using izenelib::util::UString;

//#define PM_PROFILER

ProductManager::ProductManager(
        const std::string& work_dir,
        ProductDataSource* data_source,
        OperationProcessor* op_processor,
        ProductPriceTrend* price_trend,
        const PMConfig& config)
    : work_dir_(work_dir)
    , config_(config)
    , data_source_(data_source)
    , op_processor_(op_processor)
    , price_trend_(price_trend)
    , clustering_(NULL)
    , editor_(new ProductEditor(data_source, op_processor, config))
    , util_(config, data_source)
    , has_price_trend_(false)
    , inhook_(false)
{
    if (price_trend_)
    {
        if (price_trend_->Init())
            has_price_trend_ = true;
        else
            std::cerr << "Error: Price trend has not been properly initialized" << std::endl;
    }

//     if (!config_.backup_path.empty())
//     {
//         backup_ = new ProductBackup(config_.backup_path);
//     }
}

ProductManager::~ProductManager()
{
    delete editor_;
}

bool ProductManager::Recover()
{
//     if (backup_==NULL)
//     {
//         error_ = "Backup not initialzed.";
//         return false;
//     }
//     if (!backup_->Recover(this))
//     {
// //         error_ = "Backup recover failed.";
//         return false;
//     }
//     if (!GenOperations_())
//     {
//         return false;
//     }
//     return true;
    return false;
}

ProductClustering* ProductManager::GetClustering_()
{
    if (!clustering_)
    {
        if (config_.enable_clustering_algo)
        {
            clustering_ = new ProductClustering(work_dir_+"/clustering", config_);
            if (!clustering_->Open())
            {
                std::cout<<"ProductClustering open failed"<<std::endl;
                delete clustering_;
                clustering_ = NULL;
            }
        }
    }
    return clustering_;
}

bool ProductManager::HookInsert(PMDocumentType& doc, izenelib::ir::indexmanager::IndexerDocument& index_document, time_t timestamp)
{
    inhook_ = true;
    boost::mutex::scoped_lock lock(human_mutex_);

    if (has_price_trend_)
    {
        ProductPrice price;
        if (util_.GetPrice(doc, price))
        {
            UString docid;
            if (doc.getProperty(config_.docid_property_name, docid))
            {
                std::string docid_str;
                docid.convertString(docid_str, UString::UTF_8);
                if (timestamp == -1) GetTimestamp_(doc, timestamp);
//              std::map<std::string, std::string> group_prop_map;
//              GetGroupProperties_(doc, group_prop_map);
//              task_type task = boost::bind(&ProductPriceTrend::Update, price_trend_, docid_str, price, timestamp, group_prop_map);
                task_type task = boost::bind(&ProductPriceTrend::Insert, price_trend_, docid_str, price, timestamp);
                jobScheduler_.addTask(task);
            }
        }
    }
    ProductClustering* clustering = GetClustering_();
    if (clustering)
    {
        UString uuid;
        doc.property(config_.uuid_property_name) = uuid;
        if (!data_source_->SetUuid(index_document, uuid)) return false; // set a empty uuid for rtype update later
        clustering->Insert(doc);
    }
    return true;
}

bool ProductManager::HookUpdate(PMDocumentType& to, izenelib::ir::indexmanager::IndexerDocument& index_document, time_t timestamp, bool r_type)
{
    inhook_ = true;
    boost::mutex::scoped_lock lock(human_mutex_);

    uint32_t fromid = index_document.getId(); //oldid
    PMDocumentType from;
    if (!data_source_->GetDocument(fromid, from)) return false;
    UString from_uuid;
    if (!from.getProperty(config_.uuid_property_name, from_uuid) ) return false;
    ProductPrice from_price;
    ProductPrice to_price;
    util_.GetPrice(fromid, from_price);
    util_.GetPrice(to, to_price);

    if (has_price_trend_ && to_price.Valid() && from_price != to_price)
    {
        UString docid;
        if (to.getProperty(config_.docid_property_name, docid ) )
        {
            std::string docid_str;
            docid.convertString(docid_str, UString::UTF_8);
            if (timestamp == -1) GetTimestamp_(to, timestamp);
//          std::map<std::string, std::string> group_prop_map;
//          GetGroupProperties_(to, group_prop_map);
//          task_type task = boost::bind(&ProductPriceTrend::Update, price_trend_, docid_str, to_price, timestamp, group_prop_map);
            task_type task = boost::bind(&ProductPriceTrend::Insert, price_trend_, docid_str, to_price, timestamp);
            jobScheduler_.addTask(task);
        }
    }

    if (config_.enable_clustering_algo)
    {
        std::vector<uint32_t> docid_list;
        data_source_->GetDocIdList(from_uuid, docid_list, fromid); // except from.docid
        if (docid_list.empty()) // the from doc is unique, so delete it and insert 'to'
        {
            if (!data_source_->SetUuid(index_document, from_uuid)) return false;
            PMDocumentType new_doc(to);
            to.property(config_.uuid_property_name) = from_uuid;
            new_doc.property(config_.docid_property_name) = from_uuid;
            util_.SetItemCount(new_doc, 1);
            op_processor_->Append(2, new_doc);// if r_type, only numeric properties in 'to'
        }
        else
        {
            //need not to update(insert) to.uuid,
            if (!data_source_->SetUuid(index_document, from_uuid)) return false;
{'fraction_non_alphanumeric': 0.1179002447463921, 'fraction_numerical': 0.001856696767659718, 'mean_word_length': 2.742893240682249, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 45, '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': '2359233', 'n_tokens_mistral': 3723, 'n_tokens_neox': 3667, 'n_words': 689}