<commit_before># encoding: utf-8
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
#
# Author: <NAME> (<EMAIL>)
#
from __future__ import division
from __future__ import unicode_literals

from collections import Mapping

from activedata_etl.etl import parse_id_argument
from mo_dots import coalesce, unwrap, Data, wrap
from mo_logs import Log, machine_metadata
from mo_logs import startup, constants
from mo_logs.exceptions import Explanation, WarnOnException
from mo_math import MAX
from mo_math.randoms import Random
from mo_threads import Process, Thread, Signal, Queue, THREAD_STOP
from mo_threads import Till
from mo_times.timer import Timer
from jx_base import container
from pyLibrary import aws
from pyLibrary.aws import s3
from pyLibrary.env import elasticsearch
from pyLibrary.env.rollover_index import RolloverIndex

split = {}
empty_bucket_complaint_sent = False


def splitter(work_queue, please_stop):
    global empty_bucket_complaint_sent

    for pair in iter(work_queue.pop_message, ""):
        if please_stop:
            for k, v in split.items():
                v.add(THREAD_STOP)
            return
        if pair == None:
            # ADD BACKFILLING HERE
            (Till(seconds=5) | please_stop).wait()
            continue

        message, payload = pair
        if not isinstance(payload, Mapping):
            Log.error("Not expecting a Mapping payload with `key` and `bucket` properties")

        key = payload.key
        with Explanation("Indexing records from {{bucket}}", bucket=payload.bucket):
            params = split.get(payload.bucket)
            if not params:
                message.delete()
                continue

        es = params.es
        source_bucket = params.bucket
        settings = params.settings

        extend_time = Timer("insert", silent=True)

        with extend_time:
            if settings.skip and Random.float() < settings.skip:
                Log.note("Skipping {{key}} from bucket {{bucket}}", key=key, bucket=source_bucket.name)
                work_queue.add(payload)
                message.delete()
                continue

            if settings.sample_only:
                sample_filter = {"terms": {"build.branch": settings.sample_only}}
            elif settings.sample_size:
                sample_filter = True
            else:
                sample_filter = None

            Log.note("Indexing {{key}} from bucket {{bucket}}", key=key, bucket=source_bucket.name)
            more_keys = source_bucket.keys(prefix=key)
            if not more_keys:
                # HAPPENS WHEN REPROCESSING (ETL WOULD HAVE CLEARED THE BUCKET OF THIS PREFIX FIRST)
                if not empty_bucket_complaint_sent:
                    empty_bucket_complaint_sent = True
                    Log.warning("No files found in bucket {{message|json}}. THIS WARNING WILL NOT BE SENT AGAIN!!", message=payload)
                message.delete()
                num_keys = 0
            else:
                num_keys = es.copy(more_keys, source_bucket, sample_filter, settings.sample_size, message.delete)

        if num_keys > 1:
            Log.note(
                "Added {{num}} keys from {{key}} in {{bucket}} to {{es}} in {{duration}} ({{rate|round(places=3)}} keys/second)",
                num=num_keys,
                key=key,
                bucket=source_bucket.name,
                es=es.settings.index,
                duration=extend_time.duration,
                rate=num_keys / MAX([extend_time.duration.seconds, 0.01])
            )


def safe_splitter(work_queue, please_stop):
    while not please_stop:
        try:
            with WarnOnException("Indexing records"):
                splitter(work_queue, please_stop)
        except Exception as e:
            Log.warning("problem", cause=e)


def add_message_confirmation(queue, payload_key, message):
    def _delete():
        # Log.note("confirming message for {{id}}", id=payload_key)
        message.delete()

    queue.add(_delete)


def shutdown_local_es_node():
    Log.warning("Shutdown ES on node {{node}}", node=machine_metadata)
    with Process("stop es", ["sudo", "supervisorctl", "stop", "es"]) as proc:
        while True:
            line = proc.stdout.pop().strip()
            if not line:
                continue
            Log.note("Shutdown es: {{note}}", note=line)


def main():
    try:
        settings = startup.read_settings(defs=[
            {
                "name": ["--id", "--key"],
                "help": "id to process (prefix is ok too) ",
                "type": str,
                "dest": "id",
                "required": False
            },
            {
                "name": ["--new", "--reset"],
                "help": "to make a new index (then exit immediately)",
                "type": str,
                "dest": "reset",
                "required": False
            }
        ])
        constants.set(settings.constants)
        Log.start(settings.debug)
<commit_msg>fix from es6<commit_after># encoding: utf-8
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
#
# Author: <NAME> (<EMAIL>)
#
from __future__ import division
from __future__ import unicode_literals

from collections import Mapping

from activedata_etl.etl import parse_id_argument
from mo_dots import coalesce, unwrap, Data, wrap
from mo_logs import Log, machine_metadata
from mo_logs import startup, constants
from mo_logs.exceptions import Explanation, WarnOnException
from mo_math import MAX
from mo_math.randoms import Random
from mo_threads import Process, Thread, Signal, Queue, THREAD_STOP
from mo_threads import Till
from mo_times.timer import Timer
from jx_base import container
from pyLibrary import aws
from pyLibrary.aws import s3
from pyLibrary.env import elasticsearch
from pyLibrary.env.rollover_index import RolloverIndex

split = {}
empty_bucket_complaint_sent = False


def splitter(work_queue, please_stop):
    global empty_bucket_complaint_sent

    for pair in iter(work_queue.pop_message, ""):
        if please_stop:
            for k, v in split.items():
                v.add(THREAD_STOP)
            return
        if pair == None:
            # ADD BACKFILLING HERE
            (Till(seconds=5) | please_stop).wait()
            continue

        message, payload = pair
        if not isinstance(payload, Mapping):
            Log.error("Not expecting a Mapping payload with `key` and `bucket` properties")

        key = payload.key
        with Explanation("Indexing records from {{bucket}}", bucket=payload.bucket):
            params = split.get(payload.bucket)
            if not params:
                message.delete()
                continue

        es = params.es
        source_bucket = params.bucket
        settings = params.settings

        extend_time = Timer("insert", silent=True)

        with extend_time:
            if settings.skip and Random.float() < settings.skip:
                Log.note("Skipping {{key}} from bucket {{bucket}}", key=key, bucket=source_bucket.name)
                work_queue.add(payload)
                message.delete()
                continue

            if settings.sample_only:
                sample_filter = {"terms": {"build.branch": settings.sample_only}}
            elif settings.sample_size:
                sample_filter = True
            else:
                sample_filter = None

            Log.note("Indexing {{key}} from bucket {{bucket}}", key=key, bucket=source_bucket.name)
            more_keys = source_bucket.keys(prefix=key)
            if not more_keys:
                # HAPPENS WHEN REPROCESSING (ETL WOULD HAVE CLEARED THE BUCKET OF THIS PREFIX FIRST)
                if not empty_bucket_complaint_sent:
                    empty_bucket_complaint_sent = True
                    Log.warning("No files found in bucket {{message|json}}. THIS WARNING WILL NOT BE SENT AGAIN!!", message=payload)
                message.delete()
                num_keys = 0
            else:
                num_keys = es.copy(more_keys, source_bucket, sample_filter, settings.sample_size, message.delete)

        if num_keys > 1:
            Log.note(
                "Added {{num}} keys from {{key}} in {{bucket}} to {{es}} in {{duration}} ({{rate|round(places=3)}} keys/second)",
                num=num_keys,
                key=key,
                bucket=source_bucket.name,
                es=es.settings.index,
                duration=extend_time.duration,
                rate=num_keys / MAX([extend_time.duration.seconds, 0.01])
            )


def safe_splitter(work_queue, please_stop):
    while not please_stop:
        try:
            with WarnOnException("Indexing records"):
                splitter(work_queue, please_stop)
        except Exception as e:
            Log.warning("problem", cause=e)


def add_message_confirmation(queue, payload_key, message):
    def _delete():
        # Log.note("confirming message for {{id}}", id=payload_key)
        message.delete()

    queue.add(_delete)


def shutdown_local_es_node():
    Log.warning("Shutdown ES on node {{node}}", node=machine_metadata)
    with Process("stop es", ["sudo", "supervisorctl", "stop", "es"]) as proc:
        while True:
            line = proc.stdout.pop().strip()
            if not line:
                continue
            if line == THREAD_STOP:
                break
            Log.note("Shutdown es: {{note}}", note=line)


def main():
    try:
        settings = startup.read_settings(defs=[
            {
                "name": ["--id", "--key"],
                "help": "id to process (prefix is ok too) ",
                "type": str,
                "dest": "id",
                "required": False
            },
            {
                "name": ["--new", "--reset"],
                "help": "to make a new index (then exit immediately)",
                "type": str,
                "dest": "reset",
                "required": False
            }
        ])
        constants.set(settings.constants)
        Log.start(settings.debug)

{'fraction_non_alphanumeric': 0.09964171589038442, 'fraction_numerical': 0.0026145056647622735, 'mean_word_length': 2.073809523809524, 'pattern_counts': {'":': 24, '<': 9, '<?xml version=': 0, '>': 9, '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': '2733288', 'n_tokens_mistral': 2922, 'n_tokens_neox': 2744, 'n_words': 814}