def printable(data):

if 32 <= byte2int(data) < 127:

if isinstance(data, int):

return chr(data)

return data

return '.'

try:

print("packet length:", len(data))

for i in range(1, 7):

f = sys._getframe(i)

print("call[%d]: %s (line %d)" % (i, f.f_code.co_name, f.f_lineno))

print("-" * 66)

except ValueError:

pass

dump_data = [data[i:i+16] for i in range_type(0, min(len(data), 256), 16)]

for d in dump_data:

print(' '.join("{:02X}".format(byte2int(x)) for x in d) +

' ' * (16 - len(d)) + ' ' * 2 +

''.join(printable(x) for x in d))

print("-" * 66)

"""Representation of a MySQL response packet.

__slots__ = ('_position', '_data')

def __init__(self, data, encoding):

self._position = 0

self._data = data

def get_all_data(self):

return self._data

def read(self, size):

"""Read the first 'size' bytes in packet and advance cursor past them."""

result = self._data[self._position:(self._position+size)]

if len(result) != size:

error = ('Result length not requested length:\n'

'Expected=%s. Actual=%s. Position: %s. Data Length: %s'

% (size, len(result), self._position, len(self._data)))

if DEBUG:

print(error)

self.dump()

raise AssertionError(error)

self._position += size

return result

def read_all(self):

"""Read all remaining data in the packet.

result = self._data[self._position:]

self._position = None # ensure no subsequent read()

return result

def advance(self, length):

"""Advance the cursor in data buffer 'length' bytes."""

new_position = self._position + length

if new_position < 0 or new_position > len(self._data):

raise Exception('Invalid advance amount (%s) for cursor. '

'Position=%s' % (length, new_position))

self._position = new_position

def rewind(self, position=0):

"""Set the position of the data buffer cursor to 'position'."""

if position < 0 or position > len(self._data):

raise Exception("Invalid position to rewind cursor to: %s." % position)

self._position = position

def get_bytes(self, position, length=1):

"""Get 'length' bytes starting at 'position'.

return self._data[position:(position+length)]

if PY2:

def read_uint8(self):

result = ord(self._data[self._position])

self._position += 1

return result

else:

def read_uint8(self):

result = self._data[self._position]

self._position += 1

return result

def read_uint16(self):

result = struct.unpack_from('<H', self._data, self._position)[0]

self._position += 2

return result

def read_uint24(self):

low, high = struct.unpack_from('<HB', self._data, self._position)

self._position += 3

return low + (high << 16)

def read_uint32(self):

result = struct.unpack_from('<I', self._data, self._position)[0]

self._position += 4

return result

def read_uint64(self):

result = struct.unpack_from('<Q', self._data, self._position)[0]

self._position += 8

return result

def read_string(self):

end_pos = self._data.find(b'\0', self._position)

if end_pos < 0:

return None

result = self._data[self._position:end_pos]

self._position = end_pos + 1

return result

def read_length_encoded_integer(self):

"""Read a 'Length Coded Binary' number from the data buffer.

c = self.read_uint8()

if c == NULL_COLUMN:

return None

if c < UNSIGNED_CHAR_COLUMN:

return c

elif c == UNSIGNED_SHORT_COLUMN:

return self.read_uint16()

elif c == UNSIGNED_INT24_COLUMN:

return self.read_uint24()

elif c == UNSIGNED_INT64_COLUMN:

return self.read_uint64()

def read_length_coded_string(self):

"""Read a 'Length Coded String' from the data buffer.

length = self.read_length_encoded_integer()

if length is None:

return None

return self.read(length)

def read_struct(self, fmt):

s = struct.Struct(fmt)

result = s.unpack_from(self._data, self._position)

self._position += s.size

return result

def is_ok_packet(self):

# https://dev.mysql.com/doc/internals/en/packet-OK_Packet.html

return self._data[0:1] == b'\0' and len(self._data) >= 7

def is_eof_packet(self):

# http://dev.mysql.com/doc/internals/en/generic-response-packets.html#packet-EOF_Packet

# Caution: \xFE may be LengthEncodedInteger.

# If \xFE is LengthEncodedInteger header, 8bytes followed.

return self._data[0:1] == b'\xfe' and len(self._data) < 9

def is_auth_switch_request(self):

# http://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::AuthSwitchRequest

return self._data[0:1] == b'\xfe'

def is_extra_auth_data(self):

# https://dev.mysql.com/doc/internals/en/successful-authentication.html

return self._data[0:1] == b'\x01'

def is_resultset_packet(self):

field_count = ord(self._data[0:1])

return 1 <= field_count <= 250

def is_load_local_packet(self):

return self._data[0:1] == b'\xfb'

def is_error_packet(self):

return self._data[0:1] == b'\xff'

def check_error(self):

if self.is_error_packet():

self.raise_for_error()

def raise_for_error(self):

self.rewind()

self.advance(1) # field_count == error (we already know that)

errno = self.read_uint16()

if DEBUG: print("errno =", errno)

err.raise_mysql_exception(self._data)

def dump(self):

def __init__(self, connection_settings,ctl_connection_settings=None,

only_events=None, log_file=None,

log_pos=None, filter_non_implemented_events=True,

ignored_events=None,

only_tables=None, ignored_tables=None,

only_schemas=None, ignored_schemas=None,

freeze_schema=False, skip_to_timestamp=None,

pymysql_wrapper=None):

self.__connection_settings = connection_settings

self.__connection_settings.setdefault("charset", "utf8")

self.__connected_stream = False

self.__connected_ctl = False

self._ctl_connection_settings = ctl_connection_settings

if ctl_connection_settings:

self._ctl_connection_settings.setdefault("charset", "utf8")

self.__only_tables = only_tables

self.__ignored_tables = ignored_tables

self.__only_schemas = only_schemas

self.__ignored_schemas = ignored_schemas

self.__freeze_schema = freeze_schema

self.__allowed_events = self._allowed_event_list(

only_events, ignored_events, filter_non_implemented_events)

self.__fail_on_table_metadata_unavailable = False

# We can't filter on packet level TABLE_MAP and rotate event because

# we need them for handling other operations

self.__allowed_events_in_packet = frozenset(

[TableMapEvent, RotateEvent]).union(self.__allowed_events)

# Store table meta information

self.table_map = {}

self.log_pos = log_pos

self.log_file = log_file

self.auto_position = True

self.skip_to_timestamp = skip_to_timestamp

if pymysql_wrapper:

self.pymysql_wrapper = pymysql_wrapper

else:

self.pymysql_wrapper = pymysql.connect

if self.log_file:

self._rfile = open(self.log_file, 'rb')

read_byte = self._read_bytes(4)

if read_byte != BINLOG_FILE_HEADER:

raise Exception("Is not a standard binlog file format")

else:

self._rfile = None

self.log_pos = 4

def close(self):

if self._rfile:

self._rfile.close()

def _force_close(self):

"""Close connection without QUIT message"""

self._rfile = None

def _read_bytes(self, num_bytes):

while True:

try:

data = self._rfile.read(num_bytes)

break

except (IOError, OSError) as e:

if e.errno == errno.EINTR:

continue

self._force_close()

raise err.OperationalError(

CR.CR_SERVER_LOST,

"Lost connection to MySQL server during query (%s)" % (e,))

except BaseException:

# Don't convert unknown exception to MySQLError.

self._force_close()

raise

if len(data) < num_bytes:

self._force_close()

raise err.OperationalError(

CR.CR_SERVER_LOST, "Lost connection to MySQL server during query")

return data

def read_header(self):

'''binlog_event_header_len = 19

read_byte = self._read_bytes(19)

if read_byte:

result = struct.unpack('=IBIIIH',read_byte)

return result[1],result[3],result[0]

else:

return None,None,None

def _read_packet(self, packet_type=FilePacket):

"""Read an entire "mysql packet" in its entirety from the network

buff = bytearray()

type_code,event_length,timestamp = self.read_header()

if DEBUG:

print(type_code,event_length,timestamp)

self._rfile.seek(self.log_pos)

if event_length:

buff += b'\0'

buff += self._read_bytes(event_length)

else:

buff += b'\xfe'

buff += b'\x00\x00\x02\x00'

packet = packet_type(bytes(buff), charset_by_name('utf8').encoding)

if packet.is_error_packet():

packet.raise_for_error()

return packet

def fetchone(self):

while True:

if not self.__connected_ctl:

self.__connect_to_ctl()

self.__use_checksum = True

try:

pkt = self._read_packet()

if DEBUG:

pkt.dump()

except :

raise

if pkt.is_eof_packet():

self.close()

return None

if not pkt.is_ok_packet():

continue

binlog_event = BinLogPacketWrapper(pkt, self.table_map,

self._ctl_connection,

self.__use_checksum,

self.__allowed_events_in_packet,

self.__only_tables,

self.__ignored_tables,

self.__only_schemas,

self.__ignored_schemas,

self.__freeze_schema,

self.__fail_on_table_metadata_unavailable)

if DEBUG:

if binlog_event.event:

binlog_event.event.dump()

print(binlog_event.log_pos)

if binlog_event.event_type == ROTATE_EVENT:

self.log_pos = binlog_event.event.position

self.log_file = binlog_event.event.next_binlog

# Table Id in binlog are NOT persistent in MySQL - they are in-memory identifiers

# that means that when MySQL master restarts, it will reuse same table id for different tables

# which will cause errors for us since our in-memory map will try to decode row data with

# wrong table schema.

# The fix is to rely on the fact that MySQL will also rotate to a new binlog file every time it

# restarts. That means every rotation we see *could* be a sign of restart and so potentially

# invalidates all our cached table id to schema mappings. This means we have to load them all

# again for each logfile which is potentially wasted effort but we can't really do much better

# without being broken in restart case

self.table_map = {}

elif binlog_event.log_pos:

self.log_pos = binlog_event.log_pos

# This check must not occur before clearing the ``table_map`` as a

# result of a RotateEvent.

#

# The first RotateEvent in a binlog file has a timestamp of

# zero. If the server has moved to a new log and not written a

# timestamped RotateEvent at the end of the previous log, the

# RotateEvent at the beginning of the new log will be ignored

# if the caller provided a positive ``skip_to_timestamp``

# value. This will result in the ``table_map`` becoming

# corrupt.

#

# https://dev.mysql.com/doc/internals/en/event-data-for-specific-event-types.html

# From the MySQL Internals Manual:

#

# ROTATE_EVENT is generated locally and written to the binary

# log on the master. It is written to the relay log on the

# slave when FLUSH LOGS occurs, and when receiving a

# ROTATE_EVENT from the master. In the latter case, there

# will be two rotate events in total originating on different

# servers.

#

# There are conditions under which the terminating

# log-rotation event does not occur. For example, the server

# might crash.

if self.skip_to_timestamp and binlog_event.timestamp < self.skip_to_timestamp:

continue

if binlog_event.event_type == TABLE_MAP_EVENT and \

binlog_event.event is not None:

self.table_map[binlog_event.event.table_id] = \

binlog_event.event.get_table()

# event is none if we have filter it on packet level

# we filter also not allowed events

if binlog_event.event is None or (binlog_event.event.__class__ not in self.__allowed_events):

continue

return binlog_event.event

def _allowed_event_list(self, only_events, ignored_events,

filter_non_implemented_events):

if only_events is not None:

events = set(only_events)

else:

events = set((

QueryEvent,

RotateEvent,

StopEvent,

FormatDescriptionEvent,

XidEvent,

GtidEvent,

BeginLoadQueryEvent,

ExecuteLoadQueryEvent,

UpdateRowsEvent,

WriteRowsEvent,

DeleteRowsEvent,

TableMapEvent,

HeartbeatLogEvent,

NotImplementedEvent,

))

if ignored_events is not None:

for e in ignored_events:

events.remove(e)

if filter_non_implemented_events:

try:

events.remove(NotImplementedEvent)

except KeyError:

pass

return frozenset(events)

def __connect_to_ctl(self):

if not self._ctl_connection_settings:

self._ctl_connection_settings = dict(self.__connection_settings)

self._ctl_connection_settings["db"] = "information_schema"

self._ctl_connection_settings["cursorclass"] = DictCursor

self._ctl_connection = self.pymysql_wrapper(**self._ctl_connection_settings)

self._ctl_connection._get_table_information = self.__get_table_information

self.__connected_ctl = True

def __get_table_information(self, schema, table):

for i in range(1, 3):

try:

if not self.__connected_ctl:

self.__connect_to_ctl()

cur = self._ctl_connection.cursor()

cur.execute("""

""", (schema, table))

return cur.fetchall()

except pymysql.OperationalError as error:

code, message = error.args

if code in MYSQL_EXPECTED_ERROR_CODES:

self.__connected_ctl = False

continue

else:

raise error

def __iter__(self):

def __init__(self, connection_settings, start_file=None, start_pos=None, end_file=None, end_pos=None,

start_time=None, stop_time=None, only_schemas=None, only_tables=None, no_pk=False,

flashback=False, stop_never=False, back_interval=1.0, only_dml=True, sql_type=None):

"""

if not start_file:

raise ValueError('Lack of parameter: start_file')

self.conn_setting = connection_settings

self.local_file = ""

self.start_file = start_file

self.start_pos = start_pos if start_pos else 4 # use binlog v4

self.end_file = end_file if end_file else start_file

self.end_pos = end_pos

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.only_schemas = only_schemas if only_schemas else None

self.only_tables = only_tables if only_tables else None

self.no_pk, self.flashback, self.stop_never, self.back_interval = (no_pk, flashback, stop_never, back_interval)

self.only_dml = only_dml

self.sql_type = [t.upper() for t in sql_type] if sql_type else []

self.binlogList = []

self.connection = pymysql.connect(**self.conn_setting)

with self.connection as cursor:

cursor.execute("SHOW MASTER STATUS")

self.eof_file, self.eof_pos = cursor.fetchone()[:2]

cursor.execute("SHOW MASTER LOGS")

bin_index = [row[0] for row in cursor.fetchall()]

if self.start_file not in bin_index:

if os.path.isfile(self.start_file):

self.local_file = self.start_file

self.binlogList.append(self.local_file)

self.start_file = ""

return

else:

raise ValueError('parameter error: start_file %s not in mysql server' % self.start_file)

binlog2i = lambda x: x.split('.')[1]

for binary in bin_index:

if binlog2i(self.start_file) <= binlog2i(binary) <= binlog2i(self.end_file):

self.binlogList.append(binary)

cursor.execute("SELECT @@server_id")

self.server_id = cursor.fetchone()[0]

if not self.server_id:

raise ValueError('missing server_id in %s:%s' % (self.conn_setting['host'], self.conn_setting['port']))

def process_binlog(self):

if self.local_file:

stream = BinLogFileReader(connection_settings=self.conn_setting,log_file=self.local_file, log_pos=self.start_pos)

else:

stream = BinLogStreamReader(connection_settings=self.conn_setting, server_id=self.server_id,

log_file=self.start_file, log_pos=self.start_pos, only_schemas=self.only_schemas,

only_tables=self.only_tables, resume_stream=True, blocking=True)

flag_last_event = False

e_start_pos, last_pos = stream.log_pos, stream.log_pos

# to simplify code, we do not use flock for tmp_file.

tmp_file = create_unique_file('%s.%s' % (self.conn_setting['host'], self.conn_setting['port']))

with temp_open(tmp_file, "w") as f_tmp, self.connection as cursor:

for binlog_event in stream:

if not self.stop_never:

try:

event_time = datetime.datetime.fromtimestamp(binlog_event.timestamp)

except OSError:

event_time = datetime.datetime(1980, 1, 1, 0, 0)

if (stream.log_file == self.end_file and stream.log_pos == self.end_pos) or \

(stream.log_file == self.eof_file and stream.log_pos == self.eof_pos):

flag_last_event = True

elif event_time < self.start_time:

if not (isinstance(binlog_event, RotateEvent)

or isinstance(binlog_event, FormatDescriptionEvent)):

last_pos = binlog_event.packet.log_pos

continue

elif (stream.log_file not in self.binlogList) or \

(self.end_pos and stream.log_file == self.end_file and stream.log_pos > self.end_pos) or \

(stream.log_file == self.eof_file and stream.log_pos > self.eof_pos) or \

(event_time >= self.stop_time):

break

# else:

# raise ValueError('unknown binlog file or position')

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

stream.close()

f_tmp.close()

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(%s);' % self.back_interval)

else:

i += 1

def __del__(self):