<commit_before>"""
blockfeed: sync with and process new blocks from counterparty-server
"""
import re
import os
import sys
import json
import copy
import logging
import datetime
import decimal
import configparser
import time
import itertools
import pymongo
import gevent

from counterblock.lib import config, util, blockchain, cache, database
from counterblock.lib.processor import MessageProcessor, MempoolMessageProcessor, BlockProcessor, CaughtUpProcessor

D = decimal.Decimal
logger = logging.getLogger(__name__)


def fuzzy_is_caught_up():
    """We don't want to give users 525 errors or login errors if counterblockd/counterparty-server is in the process of
    getting caught up, but we DO if counterblockd is either clearly out of date with the blockchain, or reinitializing its database"""
    if not config.state['caught_up']:
        return False
    if not config.state['cp_backend_block_index'] or not config.state['my_latest_block']['block_index']:
        return False
    if config.state['my_latest_block']['block_index'] < config.state['cp_backend_block_index'] - 1:  # "fuzzy" part
        return False
    return True


def process_cp_blockfeed():
    # initialize state
    config.state['cur_block'] = {'block_index': 0, }  # block being currently processed
    config.state['my_latest_block'] = {'block_index': 0}  # last block that was successfully processed by counterblockd
    config.state['last_message_index'] = -1  # initialize (last processed message index)
    config.state['cp_latest_block_index'] = 0  # last block that was successfully processed by counterparty
    config.state['cp_backend_block_index'] = 0  # the latest block height as reported by the cpd blockchain backend
    config.state['cp_caught_up'] = False  # whether counterparty-server is caught up to the backend (e.g. bitcoind)
    config.state['caught_up_started_events'] = False
    #^ set after we are caught up and start up the recurring events that depend on us being caught up with the blockchain

    # enabled processor functions
    logger.debug("Enabled Message Processor Functions {0}".format(MessageProcessor.active_functions()))
    logger.debug("Enabled Block Processor Functions {0}".format(BlockProcessor.active_functions()))

    def publish_mempool_tx():
        """fetch new tx from mempool"""
        mempool_txs = config.mongo_db.mempool.find(projection={'tx_hash': True})
        tx_hashes = {t['tx_hash'] for t in mempool_txs}

        params = { # get latest 1000 entries from mempool
            'order_by': 'timestamp',
            'order_dir': 'DESC'
        }
        new_txs = util.jsonrpc_api("get_mempool", params, abort_on_error=True)
        num_skipped_tx = 0
        if new_txs:
            for new_tx in new_txs['result']:
                # skip if it's already in our mempool table
                if new_tx['tx_hash'] in tx_hashes:
                    num_skipped_tx += 1
                    continue

                tx = {
                    'tx_hash': new_tx['tx_hash'],
                    'command': new_tx['command'],
                    'category': new_tx['category'],
                    'bindings': new_tx['bindings'],
                    'timestamp': new_tx['timestamp'],
                    'viewed_in_block': config.state['my_latest_block']['block_index']
                }
                config.mongo_db.mempool.insert(tx)
                del(tx['_id'])
                tx['_category'] = tx['category']
                tx['_message_index'] = 'mempool'
                logger.debug("Spotted mempool tx: %s" % tx)
                for function in MempoolMessageProcessor.active_functions():
                    logger.debug('starting {} (mempool)'.format(function['function']))
                    # TODO: Better handling of double parsing
                    try:
                        result = function['function'](tx, json.loads(tx['bindings'])) or None
                    except pymongo.errors.DuplicateKeyError as e:
                        logging.exception(e)
                    if result == 'ABORT_THIS_MESSAGE_PROCESSING' or result == 'continue':
                        break
                    elif result:
                        raise Exception(
                            "Message processor returned unknown code -- processor: '%s', result: '%s'" %
                            (function, result))
        logger.debug("Mempool refresh: {} entries retrieved from counterparty-server, {} new".format(len(new_txs['result']), len(new_txs['result']) - num_skipped_tx))

    def clean_mempool_tx():
        """clean mempool transactions older than MAX_REORG_NUM_BLOCKS blocks"""
        config.mongo_db.mempool.remove(<commit_msg>fix exception in certain conditions<commit_after>"""
blockfeed: sync with and process new blocks from counterparty-server
"""
import re
import os
import sys
import json
import copy
import logging
import datetime
import decimal
import configparser
import time
import itertools
import pymongo
import gevent

from counterblock.lib import config, util, blockchain, cache, database
from counterblock.lib.processor import MessageProcessor, MempoolMessageProcessor, BlockProcessor, CaughtUpProcessor

D = decimal.Decimal
logger = logging.getLogger(__name__)


def fuzzy_is_caught_up():
    """We don't want to give users 525 errors or login errors if counterblockd/counterparty-server is in the process of
    getting caught up, but we DO if counterblockd is either clearly out of date with the blockchain, or reinitializing its database"""
    if not config.state['caught_up']:
        return False
    if not config.state['cp_backend_block_index'] or not config.state['my_latest_block']['block_index']:
        return False
    if config.state['my_latest_block']['block_index'] < config.state['cp_backend_block_index'] - 1:  # "fuzzy" part
        return False
    return True


def process_cp_blockfeed():
    # initialize state
    config.state['cur_block'] = {'block_index': 0, }  # block being currently processed
    config.state['my_latest_block'] = {'block_index': 0}  # last block that was successfully processed by counterblockd
    config.state['last_message_index'] = -1  # initialize (last processed message index)
    config.state['cp_latest_block_index'] = 0  # last block that was successfully processed by counterparty
    config.state['cp_backend_block_index'] = 0  # the latest block height as reported by the cpd blockchain backend
    config.state['cp_caught_up'] = False  # whether counterparty-server is caught up to the backend (e.g. bitcoind)
    config.state['caught_up_started_events'] = False
    #^ set after we are caught up and start up the recurring events that depend on us being caught up with the blockchain

    # enabled processor functions
    logger.debug("Enabled Message Processor Functions {0}".format(MessageProcessor.active_functions()))
    logger.debug("Enabled Block Processor Functions {0}".format(BlockProcessor.active_functions()))

    def publish_mempool_tx():
        """fetch new tx from mempool"""
        mempool_txs = config.mongo_db.mempool.find(projection={'tx_hash': True})
        tx_hashes = {t['tx_hash'] for t in mempool_txs}

        params = { # get latest 1000 entries from mempool
            'order_by': 'timestamp',
            'order_dir': 'DESC'
        }
        new_txs = util.jsonrpc_api("get_mempool", params, abort_on_error=True)
        num_skipped_tx = 0
        if new_txs:
            for new_tx in new_txs['result']:
                # skip if it's already in our mempool table
                if new_tx['tx_hash'] in tx_hashes:
                    num_skipped_tx += 1
                    continue

                tx = {
                    'tx_hash': new_tx['tx_hash'],
                    'command': new_tx['command'],
                    'category': new_tx['category'],
                    'bindings': new_tx['bindings'],
                    'timestamp': new_tx['timestamp'],
                    'viewed_in_block': config.state['my_latest_block']['block_index']
                }
                config.mongo_db.mempool.insert(tx)
                del(tx['_id'])
                tx['_category'] = tx['category']
                tx['_message_index'] = 'mempool'
                logger.debug("Spotted mempool tx: %s" % tx)
                for function in MempoolMessageProcessor.active_functions():
                    logger.debug('starting {} (mempool)'.format(function['function']))
                    # TODO: Better handling of double parsing
                    try:
                        result = function['function'](tx, json.loads(tx['bindings'])) or None
                    except pymongo.errors.DuplicateKeyError as e:
                        logging.exception(e)
                    if result == 'ABORT_THIS_MESSAGE_PROCESSING' or result == 'continue':
                        break
                    elif result:
                        raise Exception(
                            "Message processor returned unknown code -- processor: '%s', result: '%s'" %
                            (function, result))
        logger.debug("Mempool refresh: {} entries retrieved from counterparty-server, {} new".format(len(new_txs['result']) if new_txs else '??', (len(new_txs['result']) - num_skipped_tx) if new_txs else '??'))

    def clean_mempool_tx():
        """clean mempool transactions older than MAX_REORG_NUM_BLOCKS blocks"""
        config.mongo_db.mempool.remove(
{'fraction_non_alphanumeric': 0.0973133065369156, 'fraction_numerical': 0.0035963613285381847, 'mean_word_length': 2.6074017550553226, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, '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': '6586467', 'n_tokens_mistral': 2598, 'n_tokens_neox': 2466, 'n_words': 733}