<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

import re
from copy import copy

from mohg.repos.changesets import Changeset
from mohg.repos.pushs import Push
from mohg.repos.revisions import Revision
from pyLibrary import convert, strings
from pyLibrary.debugs.exceptions import Explanation, assert_no_exception
from pyLibrary.debugs.logs import Log
from pyLibrary.dot import set_default, Null, coalesce, unwraplist
from pyLibrary.env import http
from pyLibrary.meta import use_settings, cache
from pyLibrary.queries import jx
from pyLibrary.testing import elasticsearch
from pyLibrary.thread.threads import Thread, Lock, Queue
from pyLibrary.times.dates import Date
from pyLibrary.times.durations import SECOND, Duration, HOUR

_hg_branches = None
_OLD_BRANCH = None


def _late_imports():
    global _hg_branches
    global _OLD_BRANCH

    from mohg import hg_branches as _hg_branches
    from mohg.hg_branches import OLD_BRANCH as _OLD_BRANCH

    _ = _hg_branches
    _ = _OLD_BRANCH


DEFAULT_LOCALE = "en-US"
DEBUG = False

class HgMozillaOrg(object):
    """
    USE hg.mozilla.org FOR REPO INFORMATION
    USE ES AS A FASTER CACHE FOR THE SAME
    """

    @use_settings
    def __init__(
        self,
        hg=None,        # CONNECT TO hg
        repo=None,      # CONNECTION INFO FOR ES CACHE
        branches=None,  # CONNECTION INFO FOR ES CACHE
        use_cache=False,   # True IF WE WILL USE THE ES FOR DOWNLOADING BRANCHES
        timeout=30 * SECOND,
        settings=None
    ):
        if not _hg_branches:
            _late_imports()

        self.settings = settings
        self.timeout = Duration(timeout)

        if branches == None:
            self.branches = _hg_branches.get_branches(settings=settings)
            self.es = None
            return

        self.es = elasticsearch.Cluster(settings=repo).get_or_create_index(settings=repo)
        self.es.add_alias()
        try:
            self.es.set_refresh_interval(seconds=1)
        except Exception:
            pass

        self.branches = _hg_branches.get_branches(use_cache=use_cache, settings=settings)

        # TO ESTABLISH DATA
        self.es.add({"id": "b3649fd5cd7a-mozilla-inbound-en-US", "value": {
            "index": 247152,
            "branch": {
                "name": "mozilla-inbound",
                "locale": DEFAULT_LOCALE
            },
            "changeset": {
                "id": "b3649fd5cd7a76506d2cf04f45e39cbc972fb553",
                "id12": "b3649fd5cd7a",
                "author": "ryan",
                "description": "Backed out changeset 7d0d8d304cd8 (bug 1171357) for bustage.",
                "date": 1433429100,
                "files": ["gfx/thebes/gfxTextRun.cpp"]
            },
            "push": {
                "id": 60618,
                "user": "ryan",
                "date": 1433429138
            },
            "parents": ["7d0d8d304cd871f657effcc2d21d4eae5155fd1b"],
            "children": ["411a9af141781c3c8fa883287966a4af348dbca8"]
        }})
        self.es.flush()

    @cache(duration=HOUR, lock=True)
    def get_revision(self, revision, locale=None):
        """
        EXPECTING INCOMPLETE revision
        RETURNS revision
        """
        rev = revision.changeset.id
        if not rev:
            return Null
        elif rev == "None":
            return Null
        elif revision.branch.name == None:
            return Null
        locale = coalesce(locale, revision.branch.locale, DEFAULT_LOCALE)
        doc = self._get_from_elasticsearch(revision, locale=locale)
        if doc:
            Log.note("Got hg ({{branch}}, {{locale}}, {{revision}}) from ES", branch=doc.branch.name, locale=locale, revision=doc.changeset.id)
            return doc

        output = self._load_all_in_push(revision, locale=locale)
        return output

    def _get_from_elasticsearch(self, revision, locale=None):
        rev = revision.changeset.id
        query = {
            "query": {"filtered": {
                "query": {"match_all": {}},
                "filter": {"and": [
                    {"prefix": {"changeset.id": rev[0:12]}},
                    {"term": {"branch.name": revision.branch.name}},
                    {"term": {"branch.locale": coalesce(locale, revision.branch.locale, DEFAULT_LOCALE)}}
                ]}
            }},
            "size": 2000,
        }
        try:
            docs = self.es.search(query).hits.hits
            if len(docs) > 1:
                for d in docs:
                    if d._id.endswith(d._source.branch.locale):
                        return d._source
                Log.warning("expecting no more than one document")

            return docs[0]._source
        except Exception, e:
            Log.warning("Bad ES call, fall back to hg", e)
            return None
<commit_msg>more patience for hg too<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

import re
from copy import copy

from mohg.repos.changesets import Changeset
from mohg.repos.pushs import Push
from mohg.repos.revisions import Revision
from pyLibrary import convert, strings
from pyLibrary.debugs.exceptions import Explanation, assert_no_exception, Except
from pyLibrary.debugs.logs import Log
from pyLibrary.dot import set_default, Null, coalesce, unwraplist
from pyLibrary.env import http
from pyLibrary.maths.randoms import Random
from pyLibrary.meta import use_settings, cache
from pyLibrary.queries import jx
from pyLibrary.testing import elasticsearch
from pyLibrary.thread.threads import Thread, Lock, Queue
from pyLibrary.times.dates import Date
from pyLibrary.times.durations import SECOND, Duration, HOUR

_hg_branches = None
_OLD_BRANCH = None


def _late_imports():
    global _hg_branches
    global _OLD_BRANCH

    from mohg import hg_branches as _hg_branches
    from mohg.hg_branches import OLD_BRANCH as _OLD_BRANCH

    _ = _hg_branches
    _ = _OLD_BRANCH


DEFAULT_LOCALE = "en-US"
DEBUG = False

class HgMozillaOrg(object):
    """
    USE hg.mozilla.org FOR REPO INFORMATION
    USE ES AS A FASTER CACHE FOR THE SAME
    """

    @use_settings
    def __init__(
        self,
        hg=None,        # CONNECT TO hg
        repo=None,      # CONNECTION INFO FOR ES CACHE
        branches=None,  # CONNECTION INFO FOR ES CACHE
        use_cache=False,   # True IF WE WILL USE THE ES FOR DOWNLOADING BRANCHES
        timeout=30 * SECOND,
        settings=None
    ):
        if not _hg_branches:
            _late_imports()

        self.settings = settings
        self.timeout = Duration(timeout)

        if branches == None:
            self.branches = _hg_branches.get_branches(settings=settings)
            self.es = None
            return

        self.es = elasticsearch.Cluster(settings=repo).get_or_create_index(settings=repo)
        self.es.add_alias()
        try:
            self.es.set_refresh_interval(seconds=1)
        except Exception:
            pass

        self.branches = _hg_branches.get_branches(use_cache=use_cache, settings=settings)

        # TO ESTABLISH DATA
        self.es.add({"id": "b3649fd5cd7a-mozilla-inbound-en-US", "value": {
            "index": 247152,
            "branch": {
                "name": "mozilla-inbound",
                "locale": DEFAULT_LOCALE
            },
            "changeset": {
                "id": "b3649fd5cd7a76506d2cf04f45e39cbc972fb553",
                "id12": "b3649fd5cd7a",
                "author": "ryan",
                "description": "Backed out changeset 7d0d8d304cd8 (bug 1171357) for bustage.",
                "date": 1433429100,
                "files": ["gfx/thebes/gfxTextRun.cpp"]
            },
            "push": {
                "id": 60618,
                "user": "ryan",
                "date": 1433429138
            },
            "parents": ["7d0d8d304cd871f657effcc2d21d4eae5155fd1b"],
            "children": ["411a9af141781c3c8fa883287966a4af348dbca8"]
        }})
        self.es.flush()

    @cache(duration=HOUR, lock=True)
    def get_revision(self, revision, locale=None):
        """
        EXPECTING INCOMPLETE revision
        RETURNS revision
        """
        rev = revision.changeset.id
        if not rev:
            return Null
        elif rev == "None":
            return Null
        elif revision.branch.name == None:
            return Null
        locale = coalesce(locale, revision.branch.locale, DEFAULT_LOCALE)
        doc = self._get_from_elasticsearch(revision, locale=locale)
        if doc:
            Log.note("Got hg ({{branch}}, {{locale}}, {{revision}}) from ES", branch=doc.branch.name, locale=locale, revision=doc.changeset.id)
            return doc

        output = self._load_all_in_push(revision, locale=locale)
        return output

    def _get_from_elasticsearch(self, revision, locale=None):
        rev = revision.changeset.id
        query = {
            "query": {"filtered": {
                "query": {"match_all": {}},
                "filter": {"and": [
                    {"prefix": {"changeset.id": rev[0:12]}},
                    {"term": {"branch.name": revision.branch.name}},
                    {"term": {"branch.locale": coalesce(locale, revision.branch.locale, DEFAULT_LOCALE)}}
                ]}
            }},
            "size": 2000,
        }

        for attempt in range(3):
            try:
                docs = self.es.search(query).hits.hits
                break
            except Exception, e:
                e = Except.wrap(e)
                if "NodeNotConnectedException" in e:
                    # WE LOST A NODE, THIS MAY TAKE A WHILE
                    Thread.sleep(seconds=Random.int(5 * 60))
                    continue
                elif "EsRejectedExecutionException[rejected execution (queue capacity" in e:
                    Thread.sleep(seconds=Random.int(30))
                    continue
                else:
                    Log.warning("Bad ES call, fall back to TH", cause=e)
                    return None

        if len(docs) > 1:
            for d in docs:
                if d._id.endswith(d._source.branch.locale):
                    return d._source
            Log.warning("expecting no more than one document")

        return docs[0]._source


{'fraction_non_alphanumeric': 0.09650505237786224, 'fraction_numerical': 0.02781125428756837, 'mean_word_length': 2.05436013590034, 'pattern_counts': {'":': 66, '<': 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': '3912003', 'n_tokens_mistral': 3489, 'n_tokens_neox': 3158, 'n_words': 812}