引言：

比较两个数据库中多个表的结构和数据差异，并将比较结果保存到一个文本文件中。

代码如下：

import pymssql
import pandas as pd

class DatabaseComparer:
    def __init__(self, conn_params_db1, conn_params_db2):
        """
        初始化DatabaseComparer对象，接受两个数据库的连接参数。
        :param conn_params_db1: 第一个数据库的连接参数
        :param conn_params_db2: 第二个数据库的连接参数
        """
        self.conn_params_db1 = conn_params_db1
        self.conn_params_db2 = conn_params_db2

    def get_connection(self, params):
        """ 获取数据库连接 """
        return pymssql.connect(
            server=params['server'],
            user=params['user'],
            password=params['password'],
            database=params['database']
        )

    def get_table_columns(self, conn, table_name):
        """ 获取表字段信息 """
        query = f"""
            SELECT COLUMN_NAME, DATA_TYPE
            FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_NAME = '{table_name}'
        """
        return pd.read_sql(query, conn)

    def compare_table_columns(self, table_name):
        """ 比较两个数据库表的结构 """
        conn1 = self.get_connection(self.conn_params_db1)
        conn2 = self.get_connection(self.conn_params_db2)
        
        columns_db1 = self.get_table_columns(conn1, table_name)
        columns_db2 = self.get_table_columns(conn2, table_name)
        
        conn1.close()
        conn2.close()

        #todo 找出差异
        diff_columns_db1 = columns_db1[~columns_db1['COLUMN_NAME'].isin(columns_db2['COLUMN_NAME'])]
        diff_columns_db2 = columns_db2[~columns_db2['COLUMN_NAME'].isin(columns_db1['COLUMN_NAME'])]
        
        return diff_columns_db1, diff_columns_db2

    def compare_table_data(self, table_name):
        """ 比较两个数据库表的数据 """
        conn1 = self.get_connection(self.conn_params_db1)
        conn2 = self.get_connection(self.conn_params_db2)
        
        query_db1 = f"SELECT * FROM {table_name}"
        query_db2 = f"SELECT * FROM {table_name}"
        
        df1 = pd.read_sql(query_db1, conn1)
        df2 = pd.read_sql(query_db2, conn2)
        
        conn1.close()
        conn2.close()

        #todo 使用 concat 找出不同的行
        diff = pd.concat([df1, df2]).drop_duplicates(keep=False)
        
        return diff

    def compare_and_save_to_txt(self, table_names, output_filename):
        """
        比较多个表的结构和数据，并将比较结果保存到 txt 文件中。
        :param table_names: 需要比较的表名列表
        :param output_filename: 输出的txt文件名
        """
        with open(output_filename, 'w') as f:
            for table_name in table_names:
                print(f"正在比较表: {table_name}")

                #todo 比较表结构
                diff_columns_db1, diff_columns_db2 = self.compare_table_columns(table_name)

                #todo 比较表数据
                diff_data = self.compare_table_data(table_name)

                #todo 写入表名（仅在存在差异时）
                changes_found = False

                #todo 写入表结构差异
                if not diff_columns_db1.empty or not diff_columns_db2.empty:
                    changes_found = True
                    f.write(f"\n\n=== 正在比较表: {table_name} ===\n")
                    
                    if not diff_columns_db1.empty:
                        f.write("\n--- DB1 中有，但 DB2 中没有的列 ---\n")
                        for _, row in diff_columns_db1.iterrows():
                            f.write(f"列名: {row['COLUMN_NAME']}, 数据类型: {row['DATA_TYPE']} (DB1)\n")
                    
                    if not diff_columns_db2.empty:
                        f.write("\n--- DB2 中有，但 DB1 中没有的列 ---\n")
                        for _, row in diff_columns_db2.iterrows():
                            f.write(f"列名: {row['COLUMN_NAME']}, 数据类型: {row['DATA_TYPE']} (DB2)\n")

                #todo 写入表数据差异
                if not diff_data.empty:
                    changes_found = True
                    f.write("\n--- DB1 和 DB2 之间的不同数据行 ---\n")
                    diff_data_str = diff_data.to_string(index=False)
                    f.write(diff_data_str + "\n")
                
                #todo 如果没有任何差异，则不输出该表的内容
                if not changes_found:
                    f.write(f"\n表 {table_name} 没有发现差异\n")

        print(f"比较结果已保存到 {output_filename}")

#todo 使用示例
if __name__ == "__main__":
    #todo 设置数据库连接参数
    conn_params_db1 = {
        'server': '127.0.0.1',
        'user': 'user',
        'password': 'password',
        'database': 'db1'
    }
    
    conn_params_db2 = {
        'server': '127.0.0.2',
        'user': 'user',
        'password': 'password',
        'database': 'db2'
    }
    
    comparer = DatabaseComparer(conn_params_db1, conn_params_db2)
    
    #todo 指定多个表名进行比较
    table_names = ['table1', 'table2', 'table3']
    
    #todo 输出文件名
    output_filename = 'comparison_results_filtered.txt'

    #todo 执行比较并保存结果
    comparer.compare_and_save_to_txt(table_names, output_filename)

注意事项：

• 结果格式：当前代码将比较结果输出到文本文件中，格式较为简单。对于大量数据的比较，可能需要考虑更为结构化的输出格式（如 CSV、JSON 或 Excel），这样可以方便后续的分析或处理。
• 性能问题：pandas.read_sql() 会一次性从数据库读取表的所有数据，如果表数据量很大，可能会导致内存问题或性能瓶颈。可以考虑限制查询的数据量（如分页查询，或者通过 WHERE 子句指定过滤条件）。