pip install pymysqlreplication

import json
import sys
from datetime import datetime
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
    DeleteRowsEvent,
    UpdateRowsEvent,
    WriteRowsEvent,
)
import pandas as pd

MYSQL_SETTINGS = {
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'root',
    'password': '123456'
}
# 要監(jiān)控的數(shù)據(jù)庫和表 ssc_sjzz2
database_name = 'ssc_wfg'
table_name = 't_sys_user'

def default(o):
    if isinstance(o, datetime):
        return o.isoformat()
    raise TypeError("Unserializable object {}".format(o))
def main():
    stream = BinLogStreamReader(
        connection_settings=MYSQL_SETTINGS,
        server_id=6, # 必須與MySQL服務器上的其他復制客戶端不同
        only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
        only_tables=[table_name],
        only_schemas=[database_name]
        ) 
    df = pd.DataFrame()
    for binlogevent in stream:
        if binlogevent.table == table_name and binlogevent.schema == database_name:
            time = binlogevent.formatted_timestamp.replace('T', ' ')
            timestamp= binlogevent.timestamp
            for row in binlogevent.rows:
                event = {
                    "schema": binlogevent.schema, 
                    "table": binlogevent.table,
                    "time": time,
                    "timestamp": timestamp,
                    }
            
                if isinstance(binlogevent, DeleteRowsEvent):
                    event["action"] = "delete"
                    event["value"] = json.dumps(list(row["values"].items()), default=default)
                    # event = dict(event.items() + row["values"].items())
                elif isinstance(binlogevent, UpdateRowsEvent):
                    event["action"] = "update"
                    event["value"] = json.dumps(list(row["after_values"].items()), default=default)
                    # event = dict(event.items() + row["after_values"].items())
                elif isinstance(binlogevent, WriteRowsEvent):
                    event["action"] = "insert"
                    event["value"] = json.dumps(list(row["values"].items()), default=default)
                print(json.dumps(event, default=default))
                df = pd.concat([df, pd.DataFrame(event, index=[0])], ignore_index=True)
    stream.close()
    df.to_excel('binlog.xlsx', index=False)
if __name__ == "__main__":
    main()