<commit_before>"""Discovers the mongo cluster and starts the daemon.
"""

import logging
import os
import re
import simplejson as json
import time

from backend_simulator import BackendSimulator
from doc_manager import DocManager
from pymongo import Connection
from oplog_manager import OplogThread
from optparse import OptionParser
from simplejson import JSONDecodeError
from sys import exit
from threading import Thread
from util import bson_ts_to_long


class Daemon(Thread):
    """Checks the cluster for shards to tail.
    """

    def __init__(self, address, oplog_checkpoint, backend_url, ns_set, u_key, auth_key):
        super(Daemon, self).__init__()
        self.can_run = True
        self.oplog_checkpoint = oplog_checkpoint
        self.address = address
        self.backend_url = backend_url
        self.ns_set = ns_set
        self.u_key = u_key
        self.auth_key = auth_key
        self.shard_set = {}
        self.oplog_progress_dict = {}

    def stop(self):
        self.can_run = False
        
    def write_oplog_progress(self):
    
        if self.oplog_checkpoint is None:
                return None
    
        ofile = file(self.oplog_checkpoint, 'r+')
        #fcntl.lockf(ofile.fileno(), fcntl.LOCK_EX)
    
        os.rename(self.oplog_checkpoint, self.oplog_checkpoint + '~')  # temp file
        dest = open(self.oplog_checkpoint, 'w')
        source = open(self.oplog_checkpoint + '~', 'r')
    
        for oplog, ts in self.oplog_progress_dict.items():
            print oplog
            print ts
            oplog_str = str(oplog)
            timestamp = bson_ts_to_long(ts)
            json_str = json.dumps([oplog_str, timestamp])
            dest.write(json_str) 
    
        dest.close()
        source.close()
        os.remove(self.oplog_checkpoint + '~')
        
    def read_oplog_progress(self):
        
        if self.oplog_checkpoint is None:
            return None

        source = open(self.oplog_checkpoint, 'r')
        try:
            data = json.load(source)
        except JSONDecodeError:                 # empty file              
            return None
        
        count = 0
        while count < len(data):
            oplog_str = data[count]
            ts = data[count+1]
            self.oplog_progress_dict[oplog_str] = long_to_bson_ts(ts)
            count = count + 2                        # skip to next set
        <commit_msg>removed print, added import<commit_after>"""Discovers the mongo cluster and starts the daemon.
"""

import logging
import os
import re
import simplejson as json
import time

from backend_simulator import BackendSimulator
from doc_manager import DocManager
from pymongo import Connection
from oplog_manager import OplogThread
from optparse import OptionParser
from simplejson import JSONDecodeError
from sys import exit
from threading import Thread
from util import bson_ts_to_long, long_to_bson_ts


class Daemon(Thread):
    """Checks the cluster for shards to tail.
    """

    def __init__(self, address, oplog_checkpoint, backend_url, ns_set, u_key, auth_key):
        super(Daemon, self).__init__()
        self.can_run = True
        self.oplog_checkpoint = oplog_checkpoint
        self.address = address
        self.backend_url = backend_url
        self.ns_set = ns_set
        self.u_key = u_key
        self.auth_key = auth_key
        self.shard_set = {}
        self.oplog_progress_dict = {}

    def stop(self):
        self.can_run = False
        
    def write_oplog_progress(self):
    
        if self.oplog_checkpoint is None:
                return None
    
        ofile = file(self.oplog_checkpoint, 'r+')
        #fcntl.lockf(ofile.fileno(), fcntl.LOCK_EX)
    
        os.rename(self.oplog_checkpoint, self.oplog_checkpoint + '~')  # temp file
        dest = open(self.oplog_checkpoint, 'w')
        source = open(self.oplog_checkpoint + '~', 'r')
    
        for oplog, ts in self.oplog_progress_dict.items():
            oplog_str = str(oplog)
            timestamp = bson_ts_to_long(ts)
            json_str = json.dumps([oplog_str, timestamp])
            dest.write(json_str) 
    
        dest.close()
        source.close()
        os.remove(self.oplog_checkpoint + '~')
        
    def read_oplog_progress(self):
        
        if self.oplog_checkpoint is None:
            return None

        source = open(self.oplog_checkpoint, 'r')
        try:
            data = json.load(source)
        except JSONDecodeError:                 # empty file              
            return None
        
        count = 0
        while count < len(data):
            oplog_str = data[count]
            ts = data[count+1]
            self.oplog_progress_dict[oplog_str] = long_to_bson_ts(ts)
            count = count + 2                        # skip to next set
        
{'fraction_non_alphanumeric': 0.07883817427385892, 'fraction_numerical': 0.0012448132780082987, 'mean_word_length': 2.088404868673927, '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': '6391339', 'n_tokens_mistral': 1482, 'n_tokens_neox': 1462, 'n_words': 352}