# -*- coding: utf-8 -*-# # ------------------------------------------------------------------------------- # Name: utils.py # Description: # Author: xaoyaoo # Date: 2023/12/03 # ------------------------------------------------------------------------------- import hashlib def get_md5(data): """ 获取数据的 MD5 值 :param data: 数据(bytes) :return: """ md5 = hashlib.md5() md5.update(data) return md5.hexdigest() def attach_databases(connection, databases): """ 将多个数据库附加到给定的SQLite连接。 参数: -连接:SQLite连接 -数据库:包含数据库别名和文件路径的词典 """ cursor = connection.cursor() for alias, file_path in databases.items(): attach_command = f"ATTACH DATABASE '{file_path}' AS {alias};" cursor.execute(attach_command) connection.commit() def detach_databases(connection, aliases): """ 从给定的 SQLite 连接中分离多个数据库。 参数: - connection: SQLite连接 - aliases:要分离的数据库别名列表 """ cursor = connection.cursor() for alias in aliases: detach_command = f"DETACH DATABASE {alias};" cursor.execute(detach_command) connection.commit() def execute_sql(connection, sql, params=None): """ 执行给定的SQL语句,返回结果。 参数: - connection: SQLite连接 - sql:要执行的SQL语句 - params:SQL语句中的参数 """ try: # connection.text_factory = bytes cursor = connection.cursor() if params: cursor.execute(sql, params) else: cursor.execute(sql) return cursor.fetchall() except Exception as e: try: connection.text_factory = bytes cursor = connection.cursor() if params: cursor.execute(sql, params) else: cursor.execute(sql) rdata = cursor.fetchall() connection.text_factory = str return rdata except Exception as e: print(f"**********\nSQL: {sql}\nparams: {params}\n{e}\n**********") return None if __name__ == '__main__': pass