PyWxDump/pywxdump/analyzer/utils.py

# -*- coding: utf-8 -*-#
# -------------------------------------------------------------------------------
# Name:         utils.py
# Description:  
# Author:       xaoyaoo
# Date:         2023/12/03
# -------------------------------------------------------------------------------
import hashlib
import os
import re
import sqlite3
import time


def time_int2str(time_int):
    """
    时间戳转换为时间字符串
    :param time_int: 时间戳
    :return: 时间字符串
    """
    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time_int))


def read_dict_all_values(data):
    """
    读取字典中所有的值（单层）
    :param dict_data: 字典
    :return: 所有值的list
    """
    result = []
    if isinstance(data, list):
        for item in data:
            result.extend(read_dict_all_values(item))
    elif isinstance(data, dict):
        for key, value in data.items():
            result.extend(read_dict_all_values(value))
    else:
        if isinstance(data, bytes):
            tmp = data.decode("utf-8")
        else:
            tmp = str(data) if isinstance(data, int) else data
        result.append(tmp)

    for i in range(len(result)):
        if isinstance(result[i], bytes):
            result[i] = result[i].decode("utf-8")
    return result


def match_BytesExtra(BytesExtra, pattern=r"FileStorage(.*?)'"):
    """
    匹配 BytesExtra
    :param BytesExtra: BytesExtra
    :param pattern: 匹配模式
    :return:
    """
    if not BytesExtra:
        return False
    BytesExtra = read_dict_all_values(BytesExtra)
    BytesExtra = "'" + "'".join(BytesExtra) + "'"
    # print(BytesExtra)

    match = re.search(pattern, BytesExtra)
    if match:
        video_path = match.group(0).replace("'", "")
        return video_path
    else:
        return ""


def get_type_name(type_id: tuple):
    """
    获取消息类型名称
    :param type_id: 消息类型ID 元组 eg: (1, 0)
    :return:
    """
    type_name_dict = {
        (1, 0): "文本",
        (3, 0): "图片",
        (34, 0): "语音",
        (43, 0): "视频",
        (47, 0): "动画表情",

        (37, 0): "添加好友",  # 感谢 https://github.com/zhyc9de
        (42, 0): "推荐公众号",  # 感谢 https://github.com/zhyc9de
        (48, 0): "地图信息",  # 感谢 https://github.com/zhyc9de
        (49, 40): "分享收藏夹",  # 感谢  https://github.com/zhyc9de
        (49, 53): "接龙",  # 感谢  https://github.com/zhyc9de

        (49, 0): "文件",
        (49, 1): "类似文字消息而不一样的消息",
        (49, 5): "卡片式链接",
        (49, 6): "文件",
        (49, 8): "用户上传的GIF表情",
        (49, 19): "合并转发的聊天记录",
        (49, 33): "分享的小程序",
        (49, 36): "分享的小程序",
        (49, 57): "带有引用的文本消息",
        (49, 63): "视频号直播或直播回放等",
        (49, 87): "群公告",
        (49, 88): "视频号直播或直播回放等",
        (49, 2000): "转账消息",
        (49, 2003): "赠送红包封面",

        (50, 0): "语音通话",
        (10000, 0): "系统通知",
        (10000, 4): "拍一拍",
        (10000, 8000): "系统通知"
    }

    if type_id in type_name_dict:
        return type_name_dict[type_id]
    else:
        return "未知"


def get_name_typeid(type_name: str):
    """
    获取消息类型名称
    :param type_id: 消息类型ID 元组 eg: (1, 0)
    :return:
    """
    type_name_dict = {
        (1, 0): "文本",
        (3, 0): "图片",
        (34, 0): "语音",
        (43, 0): "视频",
        (47, 0): "动画表情",

        (37, 0): "添加好友",  # 感谢 https://github.com/zhyc9de
        (42, 0): "推荐公众号",  # 感谢 https://github.com/zhyc9de
        (48, 0): "地图信息",  # 感谢 https://github.com/zhyc9de
        (49, 40): "分享收藏夹",  # 感谢  https://github.com/zhyc9de
        (49, 53): "接龙",  # 感谢  https://github.com/zhyc9de

        (49, 0): "文件",
        (49, 1): "类似文字消息而不一样的消息",
        (49, 5): "卡片式链接",
        (49, 6): "文件",
        (49, 8): "用户上传的GIF表情",
        (49, 19): "合并转发的聊天记录",
        (49, 33): "分享的小程序",
        (49, 36): "分享的小程序",
        (49, 57): "带有引用的文本消息",
        (49, 63): "视频号直播或直播回放等",
        (49, 87): "群公告",
        (49, 88): "视频号直播或直播回放等",
        (49, 2000): "转账消息",
        (49, 2003): "赠送红包封面",

        (50, 0): "语音通话",
        (10000, 0): "系统通知",
        (10000, 4): "拍一拍",
        (10000, 8000): "系统通知"
    }
    type_tup = []
    for k, v in type_name_dict.items():
        if v == type_name:
            type_tup.append(k)
    return type_tup


def get_md5(data):
    """
    获取数据的 MD5 值
    :param data: 数据（bytes）
    :return:
    """
    md5 = hashlib.md5()
    md5.update(data)
    return md5.hexdigest()


import threading


def get_thread_id():
    current_thread = threading.current_thread()
    thread_id = current_thread.ident
    return thread_id


class DBPool:
    __db_pool = {}
    __thread_pool = {}

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, '_instance'):
            cls._instance = super(DBPool, cls).__new__(cls)
        return cls._instance

    @classmethod
    def create_connection(cls, db_path):
        if db_path == "DBPOOL_INIT":
            return
        if not os.path.exists(db_path):
            raise FileNotFoundError(f"数据库文件不存在：{db_path}")

        if db_path not in cls.__db_pool:
            cls.__db_pool[db_path] = sqlite3.connect(db_path, check_same_thread=False)
        cls.connection = cls.__db_pool[db_path]

    def __init__(self, db_path):
        if db_path == "DBPOOL_INIT":
            return
        self.db_path = db_path
        if db_path not in self.__db_pool:
            self.create_connection(db_path)
        self.connection = self.__db_pool.get(db_path)
        # 检查数据库是否关闭
        if not self.connection:
            self.create_connection(db_path)
            self.connection = self.__db_pool.get(db_path)

    def __enter__(self):
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.connection = None

    def close_all(self):
        for db_path, connection in self.__db_pool.items():
            connection.close()
        self.__db_pool.clear()


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 e1:
        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 e2:
            print(f"**********\nSQL: {sql}\nparams: {params}\n{e1}\n{e2}\n**********")
            return None


if __name__ == '__main__':
    pass