{"contents":"# -*- coding: utf-8 -*-#\n# -------------------------------------------------------------------------------\n# Name:         Favorite.py\n# Description:  负责处理wx收藏数据库\n# Author:       xaoyaoo\n# Date:         2024/05/18\n# -------------------------------------------------------------------------------\nfrom collections import defaultdict\n\nfrom .db_base import DatabaseBase\nfrom .utils import timestamp2str, xml2dict\n\n\n# * FavItems：收藏的消息条目列表\n# * FavDataItem：收藏的具体数据。大概可以确定以下两点\n#     * 即使只是简单收藏一篇公众号文章也会在 FavDataItem 中有一个对应的记录\n#     * 对于收藏的合并转发类型的消息，合并转发中的每一条消息在 FavDataItem 中都是一个独立的记录\n# * FavTags：为收藏内容添加的标签\n\n\nclass FavoriteHandler(DatabaseBase):\n    _class_name = \"Favorite\"\n    Favorite_required_tables = [\"FavItems\", \"FavDataItem\", \"FavTagDatas\", \"FavBindTagDatas\"]\n\n    def get_tags(self, LocalID):\n        \"\"\"\n        return: {LocalID: TagName}\n        \"\"\"\n        if not self.tables_exist(\"FavTagDatas\"):\n            return {}\n        if LocalID is None:\n            sql = \"select LocalID, TagName from FavTagDatas order by ServerSeq\"\n        else:\n            sql = \"select LocalID, TagName from FavTagDatas where LocalID = '%s' order by ServerSeq \" % LocalID\n        tags = self.execute(sql)  # [(1, 797940830, '程序语言类'), (2, 806153863, '账单')]\n        # 转换为字典\n        tags = {tag[0]: tag[1] for tag in tags}\n        return tags\n\n    def get_FavBindTags(self):\n        \"\"\"\n        return: [(FavLocalID, TagName)]\n        \"\"\"\n        sql = (\"select DISTINCT  A.FavLocalID, B.TagName \"\n               \"from FavBindTagDatas A, FavTagDatas B where A.TagLocalID = B.LocalID\")\n        FavBindTags = self.execute(sql)\n        return FavBindTags\n\n    def get_favorite(self):\n        \"\"\"\n        return: [{FavItemsFields}, {FavItemsFields}]\n        \"\"\"\n        FavItemsFields = {\n            \"FavLocalID\": \"本地收藏ID\",\n            \"SvrFavId\": \"服务器收藏ID\",\n            \"SourceId\": \"源ID\",\n            \"Type\": \"类型\",\n            \"SourceType\": \"源类型\",\n            \"LocalStatus\": \"本地状态\",\n            \"Flag\": \"标记\",\n            \"Status\": \"状态\",\n            \"FromUser\": \"源用户\",\n            \"RealChatName\": \"实际聊天名称\",\n            \"SearchKey\": \"搜索关键字\",\n            \"UpdateTime\": \"更新时间\",\n            \"reseverd0\": \"预留字段0\",\n            \"XmlBuf\": \"XML缓冲区\"\n        }\n        FavDataItemFields = {\n            \"FavLocalID\": \"本地收藏ID\",\n            \"Type\": \"类型\",\n            \"DataId\": \"数据ID\",\n            \"HtmlId\": \"HTML ID\",\n            \"Datasourceid\": \"数据源ID\",\n            \"Datastatus\": \"数据状态\",\n            \"Datafmt\": \"数据格式\",\n            \"Datatitle\": \"数据标题\",\n            \"Datadesc\": \"数据描述\",\n            \"Thumbfullmd5\": \"缩略图全MD5\",\n            \"Thumbhead256md5\": \"缩略图头256MD5\",\n            \"Thumbfullsize\": \"缩略图全尺寸\",\n            \"fullmd5\": \"全MD5\",\n            \"head256md5\": \"头256MD5\",\n            \"fullsize\": \"全尺寸\",\n            \"cdn_thumburl\": \"CDN缩略图URL\",\n            \"cdn_thumbkey\": \"CDN缩略图KEY\",\n            \"thumb_width\": \"缩略图宽度\",\n            \"thumb_height\": \"缩略图高度\",\n            \"cdn_dataurl\": \"CDN数据URL\",\n            \"cdn_datakey\": \"CDN数据KEY\",\n            \"cdn_encryver\": \"CDN加密版本\",\n            \"duration\": \"时长\",\n            \"stream_weburl\": \"流媒体WEB URL\",\n            \"stream_dataurl\": \"流媒体数据URL\",\n            \"stream_lowbandurl\": \"流媒体低带宽URL\",\n            \"sourcethumbpath\": \"源缩略图路径\",\n            \"sourcedatapath\": \"源数据路径\",\n            \"stream_videoid\": \"流媒体视频ID\",\n            \"Rerserved1\": \"保留字段1\",\n            \"Rerserved2\": \"保留字段2\",\n            \"Rerserved3\": \"保留字段3\",\n            \"Rerserved4\": \"保留字段4\",\n            \"Rerserved5\": \"保留字段5\",\n            \"Rerserved6\": \"保留字段6\",\n            \"Rerserved7\": \"保留字段7\"\n        }\n\n        if not self.tables_exist([\"FavItems\", \"FavDataItem\"]):\n            return False\n\n        sql1 = \"select \" + \",\".join(FavItemsFields.keys()) + \" from FavItems order by UpdateTime desc\"\n        sql2 = \"select \" + \",\".join(FavDataItemFields.keys()) + \" from FavDataItem B order by B.RecId asc\"\n\n        FavItemsList = self.execute(sql1)\n        FavDataItemList = self.execute(sql2)\n        if FavItemsList is None or len(FavItemsList) == 0:\n            return False\n\n        FavDataDict = {}\n        if FavDataItemList and len(FavDataItemList) \u003e= 0:\n            for item in FavDataItemList:\n                data_dict = {}\n                for i, key in enumerate(FavDataItemFields.keys()):\n                    data_dict[key] = item[i]\n                FavDataDict[item[0]] = FavDataDict.get(item[0], []) + [data_dict]\n        # 获取标签\n        FavTags = self.get_FavBindTags()\n        FavTagsDict = {}\n        for FavLocalID, TagName in FavTags:\n            FavTagsDict[FavLocalID] = FavTagsDict.get(FavLocalID, []) + [TagName]\n\n        rdata = []\n        for item in FavItemsList:\n            processed_item = {\n                key: item[i] for i, key in enumerate(FavItemsFields.keys())\n            }\n            processed_item['UpdateTime'] = timestamp2str(processed_item['UpdateTime'])\n            processed_item['XmlBuf'] = xml2dict(processed_item['XmlBuf'])\n            processed_item['TypeName'] = Favorite_type_converter(processed_item['Type'])\n            processed_item['FavData'] = FavDataDict.get(processed_item['FavLocalID'], [])\n            processed_item['Tags'] = FavTagsDict.get(processed_item['FavLocalID'], [])\n            rdata.append(processed_item)\n        try:\n            import pandas as pd\n        except ImportError:\n            return False\n        pf = pd.DataFrame(FavItemsList)\n        pf.columns = FavItemsFields.keys()  # set column names\n        pf[\"UpdateTime\"] = pf[\"UpdateTime\"].apply(timestamp2str)  # 处理时间\n        pf[\"XmlBuf\"] = pf[\"XmlBuf\"].apply(xml2dict)  # 处理xml\n        pf[\"TypeName\"] = pf[\"Type\"].apply(Favorite_type_converter)  # 添加类型名称列\n        pf[\"FavData\"] = pf[\"FavLocalID\"].apply(lambda x: FavDataDict.get(x, []))  # 添加数据列\n        pf[\"Tags\"] = pf[\"FavLocalID\"].apply(lambda x: FavTagsDict.get(x, []))  # 添加标签列\n        pf = pf.fillna(\"\")  # 去掉Nan\n        rdata = pf.to_dict(orient=\"records\")\n        return rdata\n\n\ndef Favorite_type_converter(type_id_or_name: [str, int]):\n    \"\"\"\n    收藏类型ID与名称转换\n    名称(str)=\u003eID(int)\n    ID(int)=\u003e名称(str)\n    :param type_id_or_name: 消息类型ID或名称\n    :return: 消息类型名称或ID\n    \"\"\"\n    type_name_dict = defaultdict(lambda: \"未知\", {\n        1: \"文本\",  # 文本 已测试\n        2: \"图片\",  # 图片 已测试\n        3: \"语音\",  # 语音\n        4: \"视频\",  # 视频 已测试\n        5: \"链接\",  # 链接 已测试\n        6: \"位置\",  # 位置\n        7: \"小程序\",  # 小程序\n        8: \"文件\",  # 文件 已测试\n        14: \"聊天记录\",  # 聊天记录 已测试\n        16: \"群聊视频\",  # 群聊中的视频 可能\n        18: \"笔记\"  # 笔记 已测试\n    })\n\n    if isinstance(type_id_or_name, int):\n        return type_name_dict[type_id_or_name]\n    elif isinstance(type_id_or_name, str):\n        return next((k for k, v in type_name_dict.items() if v == type_id_or_name), (0, 0))\n    else:\n        raise ValueError(\"Invalid input type\")\n","is_binary":false,"path":"wxdump_linux/db/db_favorite.py","ref":""}