# Author: <NAME> <<EMAIL>>
# License: MIT
# Python 3.5

"""
This module handles the text feature preparation.
"""

import itertools
import spacy
import joblib
from pymongo import InsertOne, UpdateOne
from ..utils import settings
from ..utils import file_ops
from ..utils import notifiers
from ..utils import text_preprocessing
from ..db import mongo_base
from ..db import elasticsearch_base
from ..utils.CustomTwokenizer import CustomTwokenizer, EmbeddingTwokenizer
from ..pattern_classifier import SimpleClassifier, PatternVectorizer


def init_nlp_pipeline(parser, tokenizer=CustomTwokenizer):
    """Initialize spaCy nlp pipeline
    The params are boolean values that determine if that feature should
    be loaded with the pipeline.

    Returns:
        nlp: spaCy language model
    """
    if parser is False:
        nlp = spacy.load(settings.SPACY_EN_MODEL, create_make_doc=tokenizer,
                         parser=False)
    else:
        nlp = spacy.load(settings.SPACY_EN_MODEL,
                         create_make_doc=tokenizer)
    return nlp


def load_emotion_classifier():
    """Loads persisted classes for the emotion classifier

    Returns:
        SimpleClassifier, PatternVectorizer classes
    """

    cls_persistence = settings.MODEL_PATH + "simple_classifier_model.pkl.compressed"
    pv_persistence = settings.MODEL_PATH + "pattern_vectorizer.pkl.compressed"
    _cls = joblib.load(cls_persistence)
    _pv = joblib.load(pv_persistence)
    return (_cls, _pv)


def extract_lexical_features_test(nlp, tweet_list):
    """Provides tokenization, POS and dependency parsing
    Args:
        nlp  (spaCy model): Language processing pipeline
    """
    result = []
    texts = (tweet for tweet in tweet_list)
    for doc in nlp.pipe(texts, batch_size=10000, n_threads=3):
        settings.logger.info(doc)
        for parsed_doc in doc:
            result.append((parsed_doc.orth_, parsed_doc.tag_))
    return result


def run_parallel_pipeline(connection_params, method, job_details):
    """ Generic function for processing a collection in parallel.
    Args:
        connection_params  (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
        job_details (list): Info to insert in job complete notifcation

    """

    client = mongo_base.connect()
    query = {}

    projection = connection_params[2]
    # query["filter"] = {"conllFormat":{"$exists":False}}
    query["filter"] = {}
    query["projection"] = {projection: 1}
    query["limit"] = 0
    query["skip"] = 0
    query["no_cursor_timeout"] = True

    connection_params.insert(0, client)
    collection_size = mongo_base.finder(connection_params, query, True)
    del connection_params[0]
    client.close()

    if collection_size == 0:
        return

    num_cores = 4
    partition_size = collection_size // num_cores
    partitions = [(i, partition_size)
                  for i in range(0, collection_size, partition_size)]
    # Account for lists that aren't evenly divisible, update the last tuple to
    # retrieve the remainder of the items
    partitions[-1] = (partitions[-1][0], (collection_size - partitions[-1][0]))
    settings.logger.debug(
        "Starting parallel job with %s cores and batch size of %s", num_cores, partition_size)
    time1 = notifiers.time()
    joblib.Parallel(n_jobs=num_cores)(joblib.delayed(method)(
        connection_params, query, partition) for partition in partitions)
    time2 = notifiers.time()
    notifiers.send_job_completion(
        [time1, time2], [job_details[0], connection_params[1] + ": " + job_details[1]])


def feature_extraction_pipeline(connection_params, query, partition, usage=None):
    """Handles the extraction of features needed for the model.
    Updates collection in place.
    Args:
        connection_params  (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
            3: usage (str): Operation to apply.
    """

    # Set skip limit values
    query["skip"] = partition[0]
    query["limit"] = partition[1]

    nlp = init_nlp_pipeline(True)
    client = mongo_base.connect()
    db_name = connection_params[0]
    collection = connection_params[1]
    projection = connection_params[2]
    usage = connection_params[3]
    connection_params.insert(0, client)
    # _cls, _pv = load_emotion_classifier()

    # Setup client object for bulk op
    bulk_client = mongo_base.connect()
    dbo = bulk_client[db_name]
    dbo.authenticate(settings.MONGO_USER, settings.MONGO_PW,
                     source=settings.DB_AUTH_SOURCE)

    cursor = mongo_base.finder(connection_params, query, False)

    hs_keywords = set(file_ops.read_csv_file(
        "refined_hs_keywords", settings.TWITTER_SEARCH_PATH))

    # Makes a copy of the MongoDB cursor, to the best of my
    # knowledge this does not attempt to exhaust the cursor
    cursor_1, cursor_2 = itertools.tee(cursor, 2)
    object_ids = (object_id["_id"] for object_id in cursor_1)
    tweet_texts = (tweet_text[projection] for tweet_text in cursor_2)

    operations = []
    staging = []
    count = 0

    # https://github.com/explosion/spaCy/issues/172
    docs = nlp.pipe(tweet_texts, batch_size=15000, n_threads=4)
    for object_id, doc in zip(object_ids, docs):
        # emotion_vector.append(doc.text)
        count += 1
        settings.logger.debug("Document %s out of %s", count, partition[1])

        # Construct a new tweet object to be appended
        parsed_tweet = {}
        parsed_tweet["_id"] = object_id
        if usage == "conll":
            parsed_tweet = text_preprocessing.prep_dependency_features(
                parsed_tweet, doc, usage)

        else:
            parsed_tweet["text"] = doc.text
            if usage == "analysis":
                parsed_tweet["tokens"] = list(set([token.lower_ for token in doc if not(
                    token.lower_ in text_preprocessing.STOP_LIST or token.lower_ == "user_mention"
                    or token.is_punct or token.lower_ == "rt" or token.is_digit
                    or token.prefix_ == "#")]))

                parsed_tweet = text_preprocessing.prep_linguistic_features(
                    parsed_tweet, hs_keywords, doc, usage)
                parsed_tweet = text_preprocessing.prep_dependency_features(
                    parsed_tweet, doc, usage)
            elif usage == "features":
                parsed_tweet["tokens"] = list(set([token.lower_ for token in doc if not(
                    token.lower_ in text_preprocessing.STOP_LIST or token.lower_ == "user_mention"
                    or token.is_punct or token.lower_ == "rt" or token.is_digit
                    or token.prefix_ == "#")]))

                parsed_tweet = text_preprocessing.prep_linguistic_features(
                    parsed_tweet, hs_keywords, doc, usage)
                parsed_tweet = text_preprocessing.prep_dependency_features(
                    parsed_tweet, doc, usage)

        # parsed_tweet["related_keywords"] = [[w.lower_ for w in text_preprocessing.get_similar_words(
        # nlp.vocab[token], settings.NUM_SYNONYMS)] for token in
        # text_preprocessing.get_keywords(doc)]

        staging.append(parsed_tweet)
        if len(staging) == settings.BULK_BATCH_SIZE:
            settings.logger.debug("Progress %s out of %s", count, partition[1])
            operations = update_schema(staging)
            _ = mongo_base.do_bulk_op(dbo, collection, operations)
            operations = []
            staging = []
    if staging:
        operations = update_schema(staging)
        _ = mongo_base.do_bulk_op(dbo, collection, operations)


def update_schema(staging):
    """ Short function for appending features"""
    operations = []
    for _idx, parsed_tweet in enumerate(staging):
        update_values = {}
        for key, val in parsed_tweet.items():
            update_values[key] = val

        update_values.pop("_id", None)
        operations.append(UpdateOne({"_id": parsed_tweet["_id"]}, {
            "$set": update_values}, upsert=False))
    return operations


def unpack_emotions(staging, emotion_vector, _pv, _cls):
    """Vectorize a list of tweets and return the emotion emotion_coverage
    for each entry.

    Args:
        staging (list): List of tweet objects.
        emotion_vector (list): List of tweet text.
        _pv (PatternVectorizer)
        _cls (SimpleClassifier)
    Returns:
        list of MongoDB InsertOne operations.
    """
    import pandas as pd
    emotion_vector = pd.Series(
        emotion_vector, index=range(0, len(emotion_vector)))
    emotion_vector = _pv.transform(emotion_vector)
    emotion_coverage = _cls.get_top_classes(
        emotion_vector, ascending=True, n=2)
    emotion_min_score = _cls.get_max_score_class(emotion_vector)

    operations = []
    for idx, parsed_tweet in enumerate(staging):
        parsed_tweet["emotions"] = {}
        parsed_tweet["emotions"]["first"] = emotion_coverage[idx][0]
        parsed_tweet["emotions"]["second"] = emotion_coverage[idx][1]
        parsed_tweet["emotions"]["min"] = emotion_min_score[idx]
        operations.append(InsertOne(parsed_tweet))
    return operations


def prep_word_embedding_text(connection_params, query, partition):
    """ Read a MongoDB collection and store the preprocessed text
    as a separate field. Text is preprocessed for training a word embedding model.
    Removes URLS, numbers, and stopwords, normalizes @usermentions. Updates the passed collection.
    Args:
        connection_params (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
        query (dict): Query to execute.
        partition   (tuple): Contains skip and limit values.

    """

    client = mongo_base.connect()
    db_name = connection_params[0]
    collection = connection_params[1]
    projection = connection_params[2]
    connection_params.insert(0, client)

    # Set skip limit values
    query["skip"] = partition[0]
    query["limit"] = partition[1]

    # Setup client object for bulk op
    bulk_client = mongo_base.connect()
    dbo = bulk_client[db_name]
    dbo.authenticate(settings.MONGO_USER, settings.MONGO_PW,
                     source=settings.DB_AUTH_SOURCE)

    operations = []
    nlp = init_nlp_pipeline(False, tokenizer=EmbeddingTwokenizer)
    cursor = mongo_base.finder(connection_params, query, False)

    # Makes a copy of the MongoDB cursor, to the best of my
    # knowledge this does not attempt to exhaust the cursor
    count = 0
    cursor_1, cursor_2 = itertools.tee(cursor, 2)
    object_ids = (object_id["_id"] for object_id in cursor_1)
    tweet_texts = (tweet_text[projection] for tweet_text in cursor_2)

    docs = nlp.pipe(tweet_texts, batch_size=15000, n_threads=4)
    for object_id, doc in zip(object_ids, docs):
        count += 1
        parsed_tweet = {}
        parsed_tweet["_id"] = object_id
        parsed_tweet["word_embedding_txt"] = str(doc.text).lower()

        operations.append(UpdateOne({"_id": parsed_tweet["_id"]}, {
            "$set": {"word_embedding_txt": parsed_tweet["word_embedding_txt"]}}, upsert=False))

        if len(operations) == settings.BULK_BATCH_SIZE:
            _ = mongo_base.do_bulk_op(dbo, collection, operations)
            operations = []
            settings.logger.debug("Progress %s out of %s", count, partition[1])

    if operations:
        _ = mongo_base.do_bulk_op(dbo, collection, operations)


def prep_preprocessed_text(connection_params, query, partition):
    """ Read a MongoDB collection and store the preprocessed text
    as a separate field. Text is preprocessed for generic usage.
    Removes URLS, normalizes @usermentions, and lowercases text. Updates the passed collection.
    Args:
        connection_params (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
        query (dict): Query to execute.
        partition   (tuple): Contains skip and limit values.

    """

    client = mongo_base.connect()
    db_name = connection_params[0]
    collection = connection_params[1]
    projection = connection_params[2]
    connection_params.insert(0, client)

    # Set skip limit values
    query["skip"] = partition[0]
    query["limit"] = partition[1]

    # Setup client object for bulk op
    bulk_client = mongo_base.connect()
    dbo = bulk_client[db_name]
    dbo.authenticate(settings.MONGO_USER, settings.MONGO_PW,
                     source=settings.DB_AUTH_SOURCE)

    operations = []
    nlp = init_nlp_pipeline(False)
    cursor = mongo_base.finder(connection_params, query, False)

    # Makes a copy of the MongoDB cursor, to the best of my
    # knowledge this does not attempt to exhaust the cursor
    count = 0
    cursor_1, cursor_2 = itertools.tee(cursor, 2)
    object_ids = (object_id["_id"] for object_id in cursor_1)
    tweet_texts = (tweet_text[projection] for tweet_text in cursor_2)

    docs = nlp.pipe(tweet_texts, batch_size=15000, n_threads=4)
    for object_id, doc in zip(object_ids, docs):
        count += 1
        parsed_tweet = {}
        parsed_tweet["_id"] = object_id
        parsed_tweet["preprocessed_txt"] = str(doc.text).lower()

        operations.append(UpdateOne({"_id": parsed_tweet["_id"]}, {
            "$set": {"preprocessed_txt": parsed_tweet["preprocessed_txt"]}}, upsert=False))

        if len(operations) == settings.BULK_BATCH_SIZE:
            _ = mongo_base.do_bulk_op(dbo, collection, operations)
            operations = []
            settings.logger.debug("Progress %s out of %s", count, partition[1])

    if operations:
        _ = mongo_base.do_bulk_op(dbo, collection, operations)


def run_fetch_es_tweets():
    """ Fetch tweets from elasticsearch
    """
    connection_params = ["twitter", "melvyn_hs_users"]
    lookup_list = file_ops.read_csv_file(
        'melvyn_hs_user_ids', settings.TWITTER_SEARCH_PATH)
    elasticsearch_base.migrate_es_tweets(connection_params, [
        "192.168.2.33", "tweets", "tweet", "user.id_str", lookup_list])


def start_store_preprocessed_text():
    """ Start the job for both word embedding and generic text preprocessing
    """
    job_list = [
        ["twitter_annotated_datasets", "NAACL_SRW_2016", "text"],
        ["twitter_annotated_datasets",
         "NLP_CSS_2016_expert", "text"],
        ["twitter_annotated_datasets", "crowdflower", "text"],
        ["dailystormer_archive", "d_stormer_documents", "article"],
        ["twitter", "melvyn_hs_users", "text"],
        ["manchester_event", "tweets", "text"],
        ["inauguration", "tweets", "text"],
        ["uselections", "tweets", "text"],
        ["twitter", "candidates_hs_exp6_combo_3_Mar_9813004", "text"],
        ["unfiltered_stream_May17", "tweets", "text"],
        ["twitter", "tweets", "text"],
        ["inauguration_no_filter", "tweets", "text"]
    ]
    for job in job_list:
        run_parallel_pipeline(job, prep_word_embedding_text, ["prep_word_embedding_text",
                                                              "Preprocess embedding text"])
        run_parallel_pipeline(
            job, prep_preprocessed_text, ["prep_preprocessed_text", "Preprocess Text"])


def start_feature_extraction():
    """Run operations"""
    job_list = [
        ["twitter_annotated_datasets", "NAACL_SRW_2016",
            "preprocessed_txt", "features"],
        ["twitter_annotated_datasets",
         "NLP_CSS_2016_expert", "preprocessed_txt", "features"],
        ["twitter_annotated_datasets", "crowdflower",
            "preprocessed_txt", "features"],
        ["dailystormer_archive", "d_stormer_documents",
            "preprocessed_txt", "features"],
        ["twitter", "melvyn_hs_users", "preprocessed_txt", "features"],
        ["manchester_event", "tweets", "preprocessed_txt", "features"],
        ["inauguration", "tweets",  "preprocessed_txt", "features"],
        ["unfiltered_stream_May17", "tweets", "preprocessed_txt", "features"],
        ["twitter", "tweets", "preprocessed_txt", "features"],
        ["inauguration_no_filter", "tweets", "preprocessed_txt", "features"],
        ["uselections", "tweets", "preprocessed_txt", "features"]
    ]

    conll_list = [
        # ["dailystormer_archive", "d_stormer_documents", "preprocessed_txt", "conll"]
        ["twitter", "melvyn_hs_users", "preprocessed_txt", "conll"],
        ["manchester_event", "tweets", "preprocessed_txt", "conll"],
        ["inauguration", "tweets", "preprocessed_txt", "conll"],
        ["unfiltered_stream_May17", "tweets", "preprocessed_txt", "conll"],
        ["twitter", "tweets", "preprocessed_txt", "conll"],
        ["inauguration_no_filter", "tweets", "preprocessed_txt", "conll"],
        ["uselections", "tweets", "preprocessed_txt", "conll"]
    ]

    # for job in job_list:
    #     run_parallel_pipeline(
    # job[0:4], feature_extraction_pipeline, [job[0] + "_" + job[2], "Extract
    # Features"])

    for job in conll_list:
        run_parallel_pipeline(
            job[0:4], feature_extraction_pipeline, [job[0] + "_" + job[2], "Extract Conll Features"])

# Author: <NAME> <<EMAIL>>
# License: MIT
# Python 3.5

"""
This module handles the text feature preparation.
"""

import itertools
import spacy
import joblib
from pymongo import InsertOne, UpdateOne
from ..utils import settings
from ..utils import file_ops
from ..utils import notifiers
from ..utils import text_preprocessing
from ..db import mongo_base
from ..db import elasticsearch_base
from ..utils.CustomTwokenizer import CustomTwokenizer, EmbeddingTwokenizer
from ..pattern_classifier import SimpleClassifier, PatternVectorizer


def init_nlp_pipeline(parser, tokenizer=CustomTwokenizer):
    """Initialize spaCy nlp pipeline
    The params are boolean values that determine if that feature should
    be loaded with the pipeline.

    Returns:
        nlp: spaCy language model
    """
    if parser is False:
        nlp = spacy.load(settings.SPACY_EN_MODEL, create_make_doc=tokenizer,
                         parser=False)
    else:
        nlp = spacy.load(settings.SPACY_EN_MODEL,
                         create_make_doc=tokenizer)
    return nlp


def load_emotion_classifier():
    """Loads persisted classes for the emotion classifier

    Returns:
        SimpleClassifier, PatternVectorizer classes
    """

    cls_persistence = settings.MODEL_PATH + "simple_classifier_model.pkl.compressed"
    pv_persistence = settings.MODEL_PATH + "pattern_vectorizer.pkl.compressed"
    _cls = joblib.load(cls_persistence)
    _pv = joblib.load(pv_persistence)
    return (_cls, _pv)


def extract_lexical_features_test(nlp, tweet_list):
    """Provides tokenization, POS and dependency parsing
    Args:
        nlp  (spaCy model): Language processing pipeline
    """
    result = []
    texts = (tweet for tweet in tweet_list)
    for doc in nlp.pipe(texts, batch_size=10000, n_threads=3):
        settings.logger.info(doc)
        for parsed_doc in doc:
            result.append((parsed_doc.orth_, parsed_doc.tag_))
    return result


def run_parallel_pipeline(connection_params, method, job_details):
    """ Generic function for processing a collection in parallel.
    Args:
        connection_params  (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
        job_details (list): Info to insert in job complete notifcation

    """

    client = mongo_base.connect()
    query = {}

    projection = connection_params[2]
    # query["filter"] = {"conllFormat":{"$exists":False}}
    query["filter"] = {}
    query["projection"] = {projection: 1}
    query["limit"] = 0
    query["skip"] = 0
    query["no_cursor_timeout"] = True

    connection_params.insert(0, client)
    collection_size = mongo_base.finder(connection_params, query, True)
    del connection_params[0]
    client.close()

    if collection_size == 0:
        return

    num_cores = 4
    partition_size = collection_size // num_cores
    partitions = [(i, partition_size)
                  for i in range(0, collection_size, partition_size)]
    # Account for lists that aren't evenly divisible, update the last tuple to
    # retrieve the remainder of the items
    partitions[-1] = (partitions[-1][0], (collection_size - partitions[-1][0]))
    settings.logger.debug(
        "Starting parallel job with %s cores and batch size of %s", num_cores, partition_size)
    time1 = notifiers.time()
    joblib.Parallel(n_jobs=num_cores)(joblib.delayed(method)(
        connection_params, query, partition) for partition in partitions)
    time2 = notifiers.time()
    notifiers.send_job_completion(
        [time1, time2], [job_details[0], connection_params[1] + ": " + job_details[1]])


def feature_extraction_pipeline(connection_params, query, partition, usage=None):
    """Handles the extraction of features needed for the model.
    Updates collection in place.
    Args:
        connection_params  (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
            3: usage (str): Operation to apply.
    """

    # Set skip limit values
    query["skip"] = partition[0]
    query["limit"] = partition[1]

    nlp = init_nlp_pipeline(True)
    client = mongo_base.connect()
    db_name = connection_params[0]
    collection = connection_params[1]
    projection = connection_params[2]
    usage = connection_params[3]
    connection_params.insert(0, client)
    # _cls, _pv = load_emotion_classifier()

    # Setup client object for bulk op
    bulk_client = mongo_base.connect()
    dbo = bulk_client[db_name]
    dbo.authenticate(settings.MONGO_USER, settings.MONGO_PW,
                     source=settings.DB_AUTH_SOURCE)

    cursor = mongo_base.finder(connection_params, query, False)

    hs_keywords = set(file_ops.read_csv_file(
        "refined_hs_keywords", settings.TWITTER_SEARCH_PATH))

    # Makes a copy of the MongoDB cursor, to the best of my
    # knowledge this does not attempt to exhaust the cursor
    cursor_1, cursor_2 = itertools.tee(cursor, 2)
    object_ids = (object_id["_id"] for object_id in cursor_1)
    tweet_texts = (tweet_text[projection] for tweet_text in cursor_2)

    operations = []
    staging = []
    count = 0

    # https://github.com/explosion/spaCy/issues/172
    docs = nlp.pipe(tweet_texts, batch_size=15000, n_threads=4)
    for object_id, doc in zip(object_ids, docs):
        # emotion_vector.append(doc.text)
        count += 1
        settings.logger.debug("Document %s out of %s", count, partition[1])

        # Construct a new tweet object to be appended
        parsed_tweet = {}
        parsed_tweet["_id"] = object_id
        if usage == "conll":
            parsed_tweet = text_preprocessing.prep_dependency_features(
                parsed_tweet, doc, usage)

        else:
            parsed_tweet["text"] = doc.text
            if usage == "analysis":
                parsed_tweet["tokens"] = list(set([token.lower_ for token in doc if not(
                    token.lower_ in text_preprocessing.STOP_LIST or token.lower_ == "user_mention"
                    or token.is_punct or token.lower_ == "rt" or token.is_digit
                    or token.prefix_ == "#")]))

                parsed_tweet = text_preprocessing.prep_linguistic_features(
                    parsed_tweet, hs_keywords, doc, usage)
                parsed_tweet = text_preprocessing.prep_dependency_features(
                    parsed_tweet, doc, usage)
            elif usage == "features":
                parsed_tweet["tokens"] = list(set([token.lower_ for token in doc if not(
                    token.lower_ in text_preprocessing.STOP_LIST or token.lower_ == "user_mention"
                    or token.is_punct or token.lower_ == "rt" or token.is_digit
                    or token.prefix_ == "#")]))

                parsed_tweet = text_preprocessing.prep_linguistic_features(
                    parsed_tweet, hs_keywords, doc, usage)
                parsed_tweet = text_preprocessing.prep_dependency_features(
                    parsed_tweet, doc, usage)

        # parsed_tweet["related_keywords"] = [[w.lower_ for w in text_preprocessing.get_similar_words(
        # nlp.vocab[token], settings.NUM_SYNONYMS)] for token in
        # text_preprocessing.get_keywords(doc)]

        staging.append(parsed_tweet)
        if len(staging) == settings.BULK_BATCH_SIZE:
            settings.logger.debug("Progress %s out of %s", count, partition[1])
            operations = update_schema(staging)
            _ = mongo_base.do_bulk_op(dbo, collection, operations)
            operations = []
            staging = []
    if staging:
        operations = update_schema(staging)
        _ = mongo_base.do_bulk_op(dbo, collection, operations)


def update_schema(staging):
    """ Short function for appending features"""
    operations = []
    for _idx, parsed_tweet in enumerate(staging):
        update_values = {}
        for key, val in parsed_tweet.items():
            update_values[key] = val

        update_values.pop("_id", None)
        operations.append(UpdateOne({"_id": parsed_tweet["_id"]}, {
            "$set": update_values}, upsert=False))
    return operations


def unpack_emotions(staging, emotion_vector, _pv, _cls):
    """Vectorize a list of tweets and return the emotion emotion_coverage
    for each entry.

    Args:
        staging (list): List of tweet objects.
        emotion_vector (list): List of tweet text.
        _pv (PatternVectorizer)
        _cls (SimpleClassifier)
    Returns:
        list of MongoDB InsertOne operations.
    """
    import pandas as pd
    emotion_vector = pd.Series(
        emotion_vector, index=range(0, len(emotion_vector)))
    emotion_vector = _pv.transform(emotion_vector)
    emotion_coverage = _cls.get_top_classes(
        emotion_vector, ascending=True, n=2)
    emotion_min_score = _cls.get_max_score_class(emotion_vector)

    operations = []
    for idx, parsed_tweet in enumerate(staging):
        parsed_tweet["emotions"] = {}
        parsed_tweet["emotions"]["first"] = emotion_coverage[idx][0]
        parsed_tweet["emotions"]["second"] = emotion_coverage[idx][1]
        parsed_tweet["emotions"]["min"] = emotion_min_score[idx]
        operations.append(InsertOne(parsed_tweet))
    return operations


def prep_word_embedding_text(connection_params, query, partition):
    """ Read a MongoDB collection and store the preprocessed text
    as a separate field. Text is preprocessed for training a word embedding model.
    Removes URLS, numbers, and stopwords, normalizes @usermentions. Updates the passed collection.
    Args:
        connection_params (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
        query (dict): Query to execute.
        partition   (tuple): Contains skip and limit values.

    """

    client = mongo_base.connect()
    db_name = connection_params[0]
    collection = connection_params[1]
    projection = connection_params[2]
    connection_params.insert(0, client)

    # Set skip limit values
    query["skip"] = partition[0]
    query["limit"] = partition[1]

    # Setup client object for bulk op
    bulk_client = mongo_base.connect()
    dbo = bulk_client[db_name]
    dbo.authenticate(settings.MONGO_USER, settings.MONGO_PW,
                     source=settings.DB_AUTH_SOURCE)

    operations = []
    nlp = init_nlp_pipeline(False, tokenizer=EmbeddingTwokenizer)
    cursor = mongo_base.finder(connection_params, query, False)

    # Makes a copy of the MongoDB cursor, to the best of my
    # knowledge this does not attempt to exhaust the cursor
    count = 0
    cursor_1, cursor_2 = itertools.tee(cursor, 2)
    object_ids = (object_id["_id"] for object_id in cursor_1)
    tweet_texts = (tweet_text[projection] for tweet_text in cursor_2)

    docs = nlp.pipe(tweet_texts, batch_size=15000, n_threads=4)
    for object_id, doc in zip(object_ids, docs):
        count += 1
        parsed_tweet = {}
        parsed_tweet["_id"] = object_id
        parsed_tweet["word_embedding_txt"] = str(doc.text).lower()

        operations.append(UpdateOne({"_id": parsed_tweet["_id"]}, {
            "$set": {"word_embedding_txt": parsed_tweet["word_embedding_txt"]}}, upsert=False))

        if len(operations) == settings.BULK_BATCH_SIZE:
            _ = mongo_base.do_bulk_op(dbo, collection, operations)
            operations = []
            settings.logger.debug("Progress %s out of %s", count, partition[1])

    if operations:
        _ = mongo_base.do_bulk_op(dbo, collection, operations)


def prep_preprocessed_text(connection_params, query, partition):
    """ Read a MongoDB collection and store the preprocessed text
    as a separate field. Text is preprocessed for generic usage.
    Removes URLS, normalizes @usermentions, and lowercases text. Updates the passed collection.
    Args:
        connection_params (list): Contains connection objects and params as follows:
            0: db_name     (str): Name of database to query.
            1: collection  (str): Name of collection to use.
            2: projection (str): Document field name to return.
        query (dict): Query to execute.
        partition   (tuple): Contains skip and limit values.

    """

    client = mongo_base.connect()
    db_name = connection_params[0]
    collection = connection_params[1]
    projection = connection_params[2]
    connection_params.insert(0, client)

    # Set skip limit values
    query["skip"] = partition[0]
    query["limit"] = partition[1]

    # Setup client object for bulk op
    bulk_client = mongo_base.connect()
    dbo = bulk_client[db_name]
    dbo.authenticate(settings.MONGO_USER, settings.MONGO_PW,
                     source=settings.DB_AUTH_SOURCE)

    operations = []
    nlp = init_nlp_pipeline(False)
    cursor = mongo_base.finder(connection_params, query, False)

    # Makes a copy of the MongoDB cursor, to the best of my
    # knowledge this does not attempt to exhaust the cursor
    count = 0
    cursor_1, cursor_2 = itertools.tee(cursor, 2)
    object_ids = (object_id["_id"] for object_id in cursor_1)
    tweet_texts = (tweet_text[projection] for tweet_text in cursor_2)

    docs = nlp.pipe(tweet_texts, batch_size=15000, n_threads=4)
    for object_id, doc in zip(object_ids, docs):
        count += 1
        parsed_tweet = {}
        parsed_tweet["_id"] = object_id
        parsed_tweet["preprocessed_txt"] = str(doc.text).lower()

        operations.append(UpdateOne({"_id": parsed_tweet["_id"]}, {
            "$set": {"preprocessed_txt": parsed_tweet["preprocessed_txt"]}}, upsert=False))

        if len(operations) == settings.BULK_BATCH_SIZE:
            _ = mongo_base.do_bulk_op(dbo, collection, operations)
            operations = []
            settings.logger.debug("Progress %s out of %s", count, partition[1])

    if operations:
        _ = mongo_base.do_bulk_op(dbo, collection, operations)


def run_fetch_es_tweets():
    """ Fetch tweets from elasticsearch
    """
    connection_params = ["twitter", "melvyn_hs_users"]
    lookup_list = file_ops.read_csv_file(
        'melvyn_hs_user_ids', settings.TWITTER_SEARCH_PATH)
    elasticsearch_base.migrate_es_tweets(connection_params, [
        "192.168.2.33", "tweets", "tweet", "user.id_str", lookup_list])


def start_store_preprocessed_text():
    """ Start the job for both word embedding and generic text preprocessing
    """
    job_list = [
        ["twitter_annotated_datasets", "NAACL_SRW_2016", "text"],
        ["twitter_annotated_datasets",
         "NLP_CSS_2016_expert", "text"],
        ["twitter_annotated_datasets", "crowdflower", "text"],
        ["dailystormer_archive", "d_stormer_documents", "article"],
        ["twitter", "melvyn_hs_users", "text"],
        ["manchester_event", "tweets", "text"],
        ["inauguration", "tweets", "text"],
        ["uselections", "tweets", "text"],
        ["twitter", "candidates_hs_exp6_combo_3_Mar_9813004", "text"],
        ["unfiltered_stream_May17", "tweets", "text"],
        ["twitter", "tweets", "text"],
        ["inauguration_no_filter", "tweets", "text"]
    ]
    for job in job_list:
        run_parallel_pipeline(job, prep_word_embedding_text, ["prep_word_embedding_text",
                                                              "Preprocess embedding text"])
        run_parallel_pipeline(
            job, prep_preprocessed_text, ["prep_preprocessed_text", "Preprocess Text"])


def start_feature_extraction():
    """Run operations"""
    job_list = [
        ["twitter_annotated_datasets", "NAACL_SRW_2016",
            "preprocessed_txt", "features"],
        ["twitter_annotated_datasets",
         "NLP_CSS_2016_expert", "preprocessed_txt", "features"],
        ["twitter_annotated_datasets", "crowdflower",
            "preprocessed_txt", "features"],
        ["dailystormer_archive", "d_stormer_documents",
            "preprocessed_txt", "features"],
        ["twitter", "melvyn_hs_users", "preprocessed_txt", "features"],
        ["manchester_event", "tweets", "preprocessed_txt", "features"],
        ["inauguration", "tweets",  "preprocessed_txt", "features"],
        ["unfiltered_stream_May17", "tweets", "preprocessed_txt", "features"],
        ["twitter", "tweets", "preprocessed_txt", "features"],
        ["inauguration_no_filter", "tweets", "preprocessed_txt", "features"],
        ["uselections", "tweets", "preprocessed_txt", "features"]
    ]

    conll_list = [
        # ["dailystormer_archive", "d_stormer_documents", "preprocessed_txt", "conll"]
        ["twitter", "melvyn_hs_users", "preprocessed_txt", "conll"],
        ["manchester_event", "tweets", "preprocessed_txt", "conll"],
        ["inauguration", "tweets", "preprocessed_txt", "conll"],
        ["unfiltered_stream_May17", "tweets", "preprocessed_txt", "conll"],
        ["twitter", "tweets", "preprocessed_txt", "conll"],
        ["inauguration_no_filter", "tweets", "preprocessed_txt", "conll"],
        ["uselections", "tweets", "preprocessed_txt", "conll"]
    ]

    # for job in job_list:
    #     run_parallel_pipeline(
    # job[0:4], feature_extraction_pipeline, [job[0] + "_" + job[2], "Extract
    # Features"])

    for job in conll_list:
        run_parallel_pipeline(
            job[0:4], feature_extraction_pipeline, [job[0] + "_" + job[2], "Extract Conll Features"])

# Author: <NAME> <<EMAIL>> # License: MIT # Python 3.5 This module handles the text feature preparation. Initialize spaCy nlp pipeline     The params are boolean values that determine if that feature should     be loaded with the pipeline.      Returns:         nlp: spaCy language model Loads persisted classes for the emotion classifier      Returns:         SimpleClassifier, PatternVectorizer classes Provides tokenization, POS and dependency parsing     Args:         nlp  (spaCy model): Language processing pipeline Generic function for processing a collection in parallel.     Args:         connection_params  (list): Contains connection objects and params as follows:             0: db_name     (str): Name of database to query.             1: collection  (str): Name of collection to use.             2: projection (str): Document field name to return.         job_details (list): Info to insert in job complete notifcation # query["filter"] = {"conllFormat":{"$exists":False}} # Account for lists that aren't evenly divisible, update the last tuple to # retrieve the remainder of the items Handles the extraction of features needed for the model.     Updates collection in place.     Args:         connection_params  (list): Contains connection objects and params as follows:             0: db_name     (str): Name of database to query.             1: collection  (str): Name of collection to use.             2: projection (str): Document field name to return.             3: usage (str): Operation to apply. # Set skip limit values # _cls, _pv = load_emotion_classifier() # Setup client object for bulk op # Makes a copy of the MongoDB cursor, to the best of my # knowledge this does not attempt to exhaust the cursor # https://github.com/explosion/spaCy/issues/172 # emotion_vector.append(doc.text) # Construct a new tweet object to be appended # parsed_tweet["related_keywords"] = [[w.lower_ for w in text_preprocessing.get_similar_words( # nlp.vocab[token], settings.NUM_SYNONYMS)] for token in # text_preprocessing.get_keywords(doc)] Short function for appending features Vectorize a list of tweets and return the emotion emotion_coverage     for each entry.      Args:         staging (list): List of tweet objects.         emotion_vector (list): List of tweet text.         _pv (PatternVectorizer)         _cls (SimpleClassifier)     Returns:         list of MongoDB InsertOne operations. Read a MongoDB collection and store the preprocessed text     as a separate field. Text is preprocessed for training a word embedding model.     Removes URLS, numbers, and stopwords, normalizes @usermentions. Updates the passed collection.     Args:         connection_params (list): Contains connection objects and params as follows:             0: db_name     (str): Name of database to query.             1: collection  (str): Name of collection to use.             2: projection (str): Document field name to return.         query (dict): Query to execute.         partition   (tuple): Contains skip and limit values. # Set skip limit values # Setup client object for bulk op # Makes a copy of the MongoDB cursor, to the best of my # knowledge this does not attempt to exhaust the cursor Read a MongoDB collection and store the preprocessed text     as a separate field. Text is preprocessed for generic usage.     Removes URLS, normalizes @usermentions, and lowercases text. Updates the passed collection.     Args:         connection_params (list): Contains connection objects and params as follows:             0: db_name     (str): Name of database to query.             1: collection  (str): Name of collection to use.             2: projection (str): Document field name to return.         query (dict): Query to execute.         partition   (tuple): Contains skip and limit values. # Set skip limit values # Setup client object for bulk op # Makes a copy of the MongoDB cursor, to the best of my # knowledge this does not attempt to exhaust the cursor Fetch tweets from elasticsearch Start the job for both word embedding and generic text preprocessing Run operations # ["dailystormer_archive", "d_stormer_documents", "preprocessed_txt", "conll"] # for job in job_list: #     run_parallel_pipeline( # job[0:4], feature_extraction_pipeline, [job[0] + "_" + job[2], "Extract # Features"])