#!/usr/bin/env python # -*- coding: utf-8 -*- import sys import datetime import pymysql from pymysqlreplication import BinLogStreamReader from pymysqlreplication.event import QueryEvent, RotateEvent, FormatDescriptionEvent from binlog2sql_util import command_line_args, concat_sql_from_binlog_event, create_unique_file, temp_open, \ reversed_lines, is_dml_event, event_type class Binlog2sql(object): def __init__(self, connection_settings, start_file, start_pos, end_file, end_pos, start_time, stop_time, only_schemas, only_tables, sql_type, no_pk=False, flashback=False, stop_never=False, only_dml=True, back_interval=0.1): """ connection_settings: {'host': 127.0.0.1, 'port': 3306, 'user': user, 'passwd': passwd, 'charset': 'utf8'} """ if not start_file: raise ValueError('Lack of parameter: start_file') if start_time: self.start_time = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S") else: self.start_time = datetime.datetime.strptime('1980-01-01 00:00:00', "%Y-%m-%d %H:%M:%S") if stop_time: self.stop_time = datetime.datetime.strptime(stop_time, "%Y-%m-%d %H:%M:%S") else: self.stop_time = datetime.datetime.strptime('2999-12-31 00:00:00', "%Y-%m-%d %H:%M:%S") self.conn = pymysql.connect(**connection_settings) self.conn.autocommit(True) self.instance = "{0}:{1}".format(connection_settings['host'], connection_settings['port']) self.eof_file, self.eof_pos = self.get_master_status() self.end_file = end_file if end_file else start_file self.end_pos = end_pos self.binlog_list = self.get_binlog_list(start_file, self.end_file) server_id = self.get_server_id() start_pos = start_pos if start_pos else 4 # use binlog v4 self.stream = BinLogStreamReader(connection_settings=connection_settings, server_id=server_id, log_file=start_file, log_pos=start_pos, only_schemas=only_schemas, only_tables=only_tables, resume_stream=True, blocking=True) self.no_pk, self.flashback, self.stop_never = (no_pk, flashback, stop_never) self.only_dml, self.back_interval = (only_dml, back_interval) self.sql_type = [t.upper() for t in sql_type] if sql_type else ['INSERT', 'UPDATE', 'DELETE'] def get_binlog_list(self, start_file, end_file): binlog_list = [] with self.conn as cursor: cursor.execute("SHOW MASTER LOGS") bin_index = [row[0] for row in cursor.fetchall()] if start_file not in bin_index: raise ValueError('parameter error: start_file {0} not in mysql instance'.format(start_file)) binlog2i = lambda x: x.split('.')[1] for binary in bin_index: if binlog2i(start_file) <= binlog2i(binary) <= binlog2i(end_file): binlog_list.append(binary) return binlog_list def get_master_status(self): with self.conn as cursor: cursor.execute("SHOW MASTER STATUS") return cursor.fetchone()[:2] def get_server_id(self): with self.conn as cursor: cursor.execute("SELECT @@server_id") server_id = cursor.fetchone()[0] if not server_id: raise ValueError('missing server_id in mysql instance') return server_id def position(self, log_file, log_pos, event_time): """position compared to start-stop section""" if (log_file not in self.binlog_list) or (event_time >= self.stop_time) or \ (self.end_pos and log_file == self.end_file and log_pos > self.end_pos) or \ (log_file == self.eof_file and log_pos > self.eof_pos): position = 'after' elif (log_file == self.end_file and log_pos == self.end_pos) or \ (log_file == self.eof_file and log_pos == self.eof_pos): position = 'end' elif event_time < self.start_time: position = 'before' elif event_time >= self.start_time: position = 'in' return position def process_binlog(self): flag_last_event = False e_start_pos, last_pos = self.stream.log_pos, self.stream.log_pos # to simplify code, I do not use flock for tmp_file. tmp_file = create_unique_file(self.instance) with temp_open(tmp_file, "w") as f_tmp, self.conn as cursor: for binlog_event in self.stream: print(binlog_event) log_file = self.stream.log_file log_pos = self.stream.log_pos try: event_time = datetime.datetime.fromtimestamp(binlog_event.timestamp) except OSError: event_time = datetime.datetime(1980, 1, 1, 0, 0) position = self.position(log_file, log_pos, event_time) if position == 'after' and not self.stop_never: break elif position == 'end' and not self.stop_never: flag_last_event = True elif position == 'before': if not (isinstance(binlog_event, RotateEvent) or isinstance(binlog_event, FormatDescriptionEvent)): last_pos = binlog_event.packet.log_pos continue # 此处有问题，updateevent变成了queryevent if isinstance(binlog_event, QueryEvent) and binlog_event.query == 'BEGIN': e_start_pos = last_pos if isinstance(binlog_event, QueryEvent) and not self.only_dml: sql = concat_sql_from_binlog_event(cursor=cursor, binlog_event=binlog_event, flashback=self.flashback, no_pk=self.no_pk) if sql: print(sql) elif is_dml_event(binlog_event) and event_type(binlog_event) in self.sql_type: for row in binlog_event.rows: sql = concat_sql_from_binlog_event(cursor=cursor, binlog_event=binlog_event, no_pk=self.no_pk, row=row, flashback=self.flashback, e_start_pos=e_start_pos) if self.flashback: f_tmp.write(sql + '\n') else: print(sql) if not (isinstance(binlog_event, RotateEvent) or isinstance(binlog_event, FormatDescriptionEvent)): last_pos = binlog_event.packet.log_pos if flag_last_event: break if self.flashback: self.print_rollback_sql(filename=tmp_file) return True def print_rollback_sql(self, filename): """print rollback sql from tmp_file""" with open(filename, "rb") as f_tmp: batch_size = 1000 i = 0 for line in reversed_lines(f_tmp): print(line.rstrip()) if i >= batch_size: i = 0 if self.back_interval: print('SELECT SLEEP({0});'.format(self.back_interval)) else: i += 1 def __del__(self): self.stream.close() self.conn.close() if __name__ == '__main__': args = command_line_args(sys.argv[1:]) conn_setting = {'host': args.host, 'port': args.port, 'user': args.user, 'passwd': args.password, 'charset': 'utf8'} binlog2sql = Binlog2sql(connection_settings=conn_setting, start_file=args.start_file, start_pos=args.start_pos, end_file=args.end_file, end_pos=args.end_pos, start_time=args.start_time, stop_time=args.stop_time, only_schemas=args.databases, only_tables=args.tables, no_pk=args.no_pk, flashback=args.flashback, stop_never=args.stop_never, back_interval=args.back_interval, only_dml=args.only_dml, sql_type=args.sql_type) binlog2sql.process_binlog()