<commit_before>import os.path
import sys
import yaml
import signal
import requests
import subprocess
import simplejson as json
import logging
import shlex
import codecs
from datetime import datetime
from lxml.etree import iterparse
from functools import reduce
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent


def cleanup(*args):
    logger.info('Received stop signal')
    logger.info('Shutdown')
    sys.exit(0)

signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)

try:
    config = yaml.load(open(sys.argv[1]))
except IndexError:
    print('Error: not specify config file')
    exit(1)

logging.basicConfig(filename=config['logging']['file'],
                    level=logging.INFO,
                    format='[%(levelname)s] %(asctime)s %(message)s')
logger = logging.getLogger(__name__)
logging.getLogger("requests").setLevel(logging.WARNING)  # disable requests info logging

endpoint = 'http://{host}:{port}/{index}/{type}/_bulk'.format(host=config['elastic']['host'],
                                                              port=config['elastic']['port'],
                                                              index=config['elastic']['index'],
                                                              type=config['elastic']['type'])
                                                              # todo: supporting multi-index
MAPPING = config.get('mapping')
if MAPPING.get('_id'):
    id_key = MAPPING.pop('_id')
else:
    id_key = None

table_structure = {}

DUMP_CMD = 'mysqldump -h {host} -P {port} -u {user} --password={password} {db} {table} ' \
           '--default-character-set=utf8 -X'.format(**config['mysql'])

BINLOG_CFG = {key: config['mysql'][key] for key in ['host', 'port', 'user', 'password', 'db']}
BULK_SIZE = config.get('elastic').get('bulk_size')

log_file = None
log_pos = None

record_path = config['binlog_sync']['record_file']
if os.path.isfile(record_path):
    with open(record_path, 'r') as f:
        record = yaml.load(f)
        log_file = record.get('log_file')
        log_pos = record.get('log_pos')


def post_to_es(data):
    """
    send post requests to es restful api
    """
    resp = requests.post(endpoint, data=data)
    if resp.json().get('errors'):  # a boolean to figure error occurs
        for item in resp.json()['items']:
            if list(item.values())[0].get('error'):
                logging.error(item)
    else:
        save_binlog_record()


def bulker(bulk_size=BULK_SIZE):
    """
    Example:
        u = bulker()
        u.send(None)  #for generator initialize
        u.send(json_str)  # input json item
        u.send(another_json_str)  # input json item
        ...
        u.send(None) force finish bulk and post
    """
    while True:
        data = ""
        for i in range(bulk_size):
            item = yield
            if item:
                data = data + item + "\n"
            else:
                break
        # print(data)
        print('-'*10)
        if data:
            post_to_es(data)


def updater(data):
    """
    encapsulation of bulker
    """
    u = bulker()
    u.send(None)  # push the generator to first yield
    for item in data:
        u.send(item)
    u.send(None)  # tell the generator it's the end


def json_serializer(obj):
    """
    format the object which json not supported
    """
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError('Type not serializable')


def processor(data):
    """
    The action must be one of the following:
    create
        Create a document only if the document does not already exist.
    index
        Create a new document or replace an existing document.
    update
        Do a partial update on a document.
    delete
        Delete a document.
    """
    for item in data:
        if id_key:
            action_content = {'_id': item['doc'][id_key]}
        else:
            action_content = {}
        meta = json.dumps({item['action']: action_content})
        if item['action'] == 'index':
            body = json.dumps(item['doc'], default=json_serializer)
            rv = meta + '\n' + body
        elif item['action'] == 'update':
            body = json.dumps({'doc': item['doc']}, default=json_serializer)
            rv = meta + '\n' + body
        elif item['action'] == 'delete':
            rv = meta + '\n'
        elif item['action'] == 'create':
            body = json.dumps(item['doc'], default=json_serializer)
            rv = meta + '\n' + body
        else:
            logging.error('unknown action type')
            raise TypeError('unknown action type')
        yield rv


def mapper(data):
    """
    mapping old key to new key
    """
    for item in data:
        for k, v in MAPPING.items():
            item['doc'][k] = item['doc'][v]
            del item['doc'][v]
        # print(doc)
        yield item


def formatter(data):
    """
    format every field from xml, according to parsed table structure
    """
    for item in data:
        for field, serializer in table_structure.items():
            if item['doc'][field]:
                item['doc'][field] = serializer(item['doc'][field])
        # print(item)
        yield item


def binlog_loader():
    """
    read row from binlog
    """
    global log_file
    global log_pos
    if log_file and log_pos:
        resume_stream = True
        logging.info("Resume from binlog_file: {}  binlog_pos: {}".format(log_file, log_pos))
    else:
        resume_stream = False

    stream = BinLogStreamReader(connection_settings=BINLOG_CFG, server_id=config['mysql']['server_id'],
                                only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
                                only_tables=[config['mysql']['table']],
                                resume_stream=resume_stream,
                                blocking=True,
                                log_file=log_file,
                                log_pos=log_pos)
    for binlogevent in stream:
        log_file = stream.log_file<commit_msg>log format error and set null value<commit_after>import os.path
import sys
import yaml
import signal
import requests
import subprocess
import simplejson as json
import logging
import shlex
import codecs
from datetime import datetime
from lxml.etree import iterparse
from functools import reduce
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent


def cleanup(*args):
    logger.info('Received stop signal')
    logger.info('Shutdown')
    sys.exit(0)

signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)

try:
    config = yaml.load(open(sys.argv[1]))
except IndexError:
    print('Error: not specify config file')
    exit(1)

logging.basicConfig(filename=config['logging']['file'],
                    level=logging.INFO,
                    format='[%(levelname)s] %(asctime)s %(message)s')
logger = logging.getLogger(__name__)
logging.getLogger("requests").setLevel(logging.WARNING)  # disable requests info logging

endpoint = 'http://{host}:{port}/{index}/{type}/_bulk'.format(host=config['elastic']['host'],
                                                              port=config['elastic']['port'],
                                                              index=config['elastic']['index'],
                                                              type=config['elastic']['type'])
                                                              # todo: supporting multi-index
MAPPING = config.get('mapping')
if MAPPING.get('_id'):
    id_key = MAPPING.pop('_id')
else:
    id_key = None

table_structure = {}

DUMP_CMD = 'mysqldump -h {host} -P {port} -u {user} --password={password} {db} {table} ' \
           '--default-character-set=utf8 -X'.format(**config['mysql'])

BINLOG_CFG = {key: config['mysql'][key] for key in ['host', 'port', 'user', 'password', 'db']}
BULK_SIZE = config.get('elastic').get('bulk_size')

log_file = None
log_pos = None

record_path = config['binlog_sync']['record_file']
if os.path.isfile(record_path):
    with open(record_path, 'r') as f:
        record = yaml.load(f)
        log_file = record.get('log_file')
        log_pos = record.get('log_pos')


def post_to_es(data):
    """
    send post requests to es restful api
    """
    resp = requests.post(endpoint, data=data)
    if resp.json().get('errors'):  # a boolean to figure error occurs
        for item in resp.json()['items']:
            if list(item.values())[0].get('error'):
                logging.error(item)
    else:
        save_binlog_record()


def bulker(bulk_size=BULK_SIZE):
    """
    Example:
        u = bulker()
        u.send(None)  #for generator initialize
        u.send(json_str)  # input json item
        u.send(another_json_str)  # input json item
        ...
        u.send(None) force finish bulk and post
    """
    while True:
        data = ""
        for i in range(bulk_size):
            item = yield
            if item:
                data = data + item + "\n"
            else:
                break
        # print(data)
        print('-'*10)
        if data:
            post_to_es(data)


def updater(data):
    """
    encapsulation of bulker
    """
    u = bulker()
    u.send(None)  # push the generator to first yield
    for item in data:
        u.send(item)
    u.send(None)  # tell the generator it's the end


def json_serializer(obj):
    """
    format the object which json not supported
    """
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError('Type not serializable')


def processor(data):
    """
    The action must be one of the following:
    create
        Create a document only if the document does not already exist.
    index
        Create a new document or replace an existing document.
    update
        Do a partial update on a document.
    delete
        Delete a document.
    """
    for item in data:
        if id_key:
            action_content = {'_id': item['doc'][id_key]}
        else:
            action_content = {}
        meta = json.dumps({item['action']: action_content})
        if item['action'] == 'index':
            body = json.dumps(item['doc'], default=json_serializer)
            rv = meta + '\n' + body
        elif item['action'] == 'update':
            body = json.dumps({'doc': item['doc']}, default=json_serializer)
            rv = meta + '\n' + body
        elif item['action'] == 'delete':
            rv = meta + '\n'
        elif item['action'] == 'create':
            body = json.dumps(item['doc'], default=json_serializer)
            rv = meta + '\n' + body
        else:
            logging.error('unknown action type')
            raise TypeError('unknown action type')
        yield rv


def mapper(data):
    """
    mapping old key to new key
    """
    for item in data:
        for k, v in MAPPING.items():
            item['doc'][k] = item['doc'][v]
            del item['doc'][v]
        # print(doc)
        yield item


def formatter(data):
    """
    format every field from xml, according to parsed table structure
    """
    for item in data:
        for field, serializer in table_structure.items():
            if item['doc'][field]:
                try:
                    item['doc'][field] = serializer(item['doc'][field])
                except ValueError as e:
                    logger.error("Error occurred during format, ErrorMessage:{}, ErrorItem:{}".format(str(e),
                                                                                                      str(item)))
                    item['doc'][field] = None
        # print(item)
        yield item


def binlog_loader():
    """
    read row from binlog
    """
    global log_file
    global log_pos
    if log_file and log_pos:
        resume_stream = True
        logging.info("Resume from binlog_file: {}  binlog_pos: {}".format(log_file, log_pos))
    else:
        resume_stream = False

    stream = BinLogStreamReader(connection_settings=BINLOG_CFG, server_id=config['mysql']['server_id'],
                                only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
                                only_tables=[config['mysql']['table']],
                                resume_stream=resume_stream,
                                blocking=True,
                                log_file=log_file,
                                log_pos=log_pos)
    for binlogevent in stream:
        log_file = stream.log_file
{'fraction_non_alphanumeric': 0.12069101522442219, 'fraction_numerical': 0.0011043622308117063, 'mean_word_length': 2.021448999046711, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5666572', 'n_tokens_mistral': 3595, 'n_tokens_neox': 3538, 'n_words': 935}