<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 pyLibrary import queries, aws
from pyLibrary.aws import s3
from pyLibrary.debugs import startup, constants
from pyLibrary.debugs.exceptions import Explanation, WarnOnException
from pyLibrary.debugs.logs import Log
from pyLibrary.dot import coalesce, unwrap, Dict
from pyLibrary.env import elasticsearch
from pyLibrary.maths import Math
from pyLibrary.thread.threads import Thread, Signal, Queue
from pyLibrary.times.timer import Timer
from testlog_etl.etl import parse_id_argument
from testlog_etl.sinks.multi_day_index import MultiDayIndex

split = {}


def splitter(work_queue, please_stop):
    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:
            continue

        message, payload = pair
        if not isinstance(payload, Mapping):
            Log.error("Not expecting a non-Mapping payload")

        key = payload.key
        with Explanation("Indexing records from {{bucket}}", bucket=payload.bucket):
            params = split[payload.bucket]

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

        extend_time = Timer("insert", silent=True)
        Log.note("Indexing {{key}} from bucket {{bucket}}", key=key, bucket=bucket.name)
        with extend_time:
            if settings.sample_only:
                sample_filter = {"terms": {"build.branch": settings.sample_only}}
            elif settings.sample_size:
                sample_filter = True
            else:
                sample_filter = None

            more_keys = bucket.keys(prefix=key)
            num_keys = es.copy(more_keys, bucket, sample_filter, settings.sample_size)

            add_message_to_queue(es.queue, payload.key, message)

        if num_keys > 1:
            Log.note(
                "Added {{num}} keys from {{key}} block to {{bucket}} in {{duration}} ({{rate|round(places=3)}} keys/second)",
                num=num_keys,
                key=key,
                bucket=bucket.name,
                duration=extend_time.duration,
                rate=num_keys / Math.max(extend_time.duration.seconds, 0.01)
            )


def safe_splitter(work_queue, please_stop):
    while not please_stop:
        with WarnOnException("Indexing records"):
            splitter(work_queue, please_stop)


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

    queue.add(_delete)


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

        queries.config.default = {
            "type": "elasticsearch",
            "settings": settings.elasticsearch.copy()
        }

        if settings.args.reset:
            Log.error("not working, multiple indexes involved")
            cluster = elasticsearch.Cluster(settings.elasticsearch)
            alias = coalesce(settings.elasticsearch.alias, settings.elasticsearch.index)
            index = cluster.get_prototype(alias)[0]
            if index:
                Log.error("Index {{index}} has prefix={{alias|quote}}, and has no alias.  Can not make another.", alias=alias, index=index)
            else:
                Log.alert("Creating index for alias={{alias}}", alias=alias)
                cluster.create_index(settings=settings.elasticsearch)
                Log.alert("Done.  Exiting.")
                return

        if settings.args.id:
            main_work_queue = Queue("local work queue")
            main_work_queue.extend(parse_id_argument(settings.args.id))
        else:
            main_work_queue = aws.Queue(settings=settings.work_queue)
        Log.note("Listen to queue {{queue}}, and read off of {{s3}}", queue=settings.work_queue.name, s3=settings.workers.source.bucket)

        for w in settings.workers:
            split[w.source.bucket] = Dict(
                es=MultiDayIndex(w.elasticsearch, queue_size=coalesce(w.queue_size, 1000), batch_size=unwrap(w.batch_size)),
                bucket=s3.Bucket(w.source),
                settings=settings
            )

        please_stop=Signal()<commit_msg>better logging<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 pyLibrary import queries, aws
from pyLibrary.aws import s3
from pyLibrary.debugs import startup, constants
from pyLibrary.debugs.exceptions import Explanation, WarnOnException
from pyLibrary.debugs.logs import Log
from pyLibrary.dot import coalesce, unwrap, Dict
from pyLibrary.env import elasticsearch
from pyLibrary.maths import Math
from pyLibrary.thread.threads import Thread, Signal, Queue
from pyLibrary.times.timer import Timer
from testlog_etl.etl import parse_id_argument
from testlog_etl.sinks.multi_day_index import MultiDayIndex

split = {}


def splitter(work_queue, please_stop):
    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:
            continue

        message, payload = pair
        if not isinstance(payload, Mapping):
            Log.error("Not expecting a non-Mapping payload")

        key = payload.key
        with Explanation("Indexing records from {{bucket}}", bucket=payload.bucket):
            params = split[payload.bucket]

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

        extend_time = Timer("insert", silent=True)
        Log.note("Indexing {{key}} from bucket {{bucket}}", key=key, bucket=bucket.name)
        with extend_time:
            if settings.sample_only:
                sample_filter = {"terms": {"build.branch": settings.sample_only}}
            elif settings.sample_size:
                sample_filter = True
            else:
                sample_filter = None

            more_keys = bucket.keys(prefix=key)
            num_keys = es.copy(more_keys, bucket, sample_filter, settings.sample_size)

            add_message_to_queue(es.queue, payload.key, message)

        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=bucket.name,
                es=es.settings.index,
                duration=extend_time.duration,
                rate=num_keys / Math.max(extend_time.duration.seconds, 0.01)
            )


def safe_splitter(work_queue, please_stop):
    while not please_stop:
        with WarnOnException("Indexing records"):
            splitter(work_queue, please_stop)


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

    queue.add(_delete)


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

        queries.config.default = {
            "type": "elasticsearch",
            "settings": settings.elasticsearch.copy()
        }

        if settings.args.reset:
            Log.error("not working, multiple indexes involved")
            cluster = elasticsearch.Cluster(settings.elasticsearch)
            alias = coalesce(settings.elasticsearch.alias, settings.elasticsearch.index)
            index = cluster.get_prototype(alias)[0]
            if index:
                Log.error("Index {{index}} has prefix={{alias|quote}}, and has no alias.  Can not make another.", alias=alias, index=index)
            else:
                Log.alert("Creating index for alias={{alias}}", alias=alias)
                cluster.create_index(settings=settings.elasticsearch)
                Log.alert("Done.  Exiting.")
                return

        if settings.args.id:
            main_work_queue = Queue("local work queue")
            main_work_queue.extend(parse_id_argument(settings.args.id))
        else:
            main_work_queue = aws.Queue(settings=settings.work_queue)
        Log.note("Listen to queue {{queue}}, and read off of {{s3}}", queue=settings.work_queue.name, s3=settings.workers.source.bucket)

        for w in settings.workers:
            split[w.source.bucket] = Dict(
                es=MultiDayIndex(w.elasticsearch, queue_size=coalesce(w.queue_size, 1000), batch_size=unwrap(w.batch_size)),
                bucket=s3.Bucket(w.source),
                settings=w
            )

        please_stop = Signal()
{'fraction_non_alphanumeric': 0.10655259576122886, 'fraction_numerical': 0.0036943418238382267, 'mean_word_length': 2.1789245982694685, 'pattern_counts': {'":': 28, '<': 7, '<?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': '5417583', 'n_tokens_mistral': 2880, 'n_tokens_neox': 2770, 'n_words': 715}