AiOpsWechat/services/gewe_service.py

import aiohttp
import asyncio
import json
import base64
import io
import json
import os
import threading
import time
import uuid
from fastapi import FastAPI, Depends
from common.singleton import singleton
from common.log import logger
from model.models import AddGroupContactsHistory

#@singleton
class GeWeService:
    _instance = None
    _lock = asyncio.Lock()  # 异步锁，确保单例初始化线程安全

    def __init__(self,app:FastAPI, base_url: str):
        if GeWeService._instance is not None:
            raise RuntimeError("请使用 get_instance() 获取单例！")
        self.base_url = base_url
        self.redis_service=app.state.redis_service

    @classmethod
    async def get_instance(cls, app:FastAPI,base_url: str = "http://api.geweapi.com/gewe"):
        """
        获取 GeWeChatCom 单例，确保只初始化一次。
        """
        async with cls._lock:  # 确保多个协程不会并发创建多个实例
            if cls._instance is None:
                cls._instance = cls(app,base_url)
        return cls._instance

    # _instance = None
    # def __new__(cls,app:FastAPI,base_url="http://api.geweapi.com/gewe"):
    #     if not cls._instance:
    #         cls._instance = super(GeWeService, cls).__new__(cls)
    #         cls._instance.client = None
    #         cls._instance.lock_renewal_thread = None
    #     return cls._instance

    # def __init__(self,app:FastAPI,base_url="http://api.geweapi.com/gewe"):
    #     if not hasattr(self, 'initialized'):
    #         #self.kafka_service =kafka_service # 获取 KafkaService 单例
    #         # self.kafka_service =app.state.kafka_service
    #         self.redis_service=app.state.redis_service
    #         self.base_url = base_url
    #         self.initialized = True

###############################  登录模块   ###############################
    async def check_login_async(self, token_id: str, app_id: str, uuid: str, captch_code: str = ""):
        """
        执行登录(步骤3)
        获取到登录二维码后需每间隔5s调用本接口来判断是否登录成功。
        """
        api_url = f"{self.base_url}/v2/api/login/checkLogin"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {"appId": app_id, "uuid": uuid}
        if captch_code:
            data["captchCode"] = captch_code

        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()

        logger.info(f'登录验证码-请求参数 {json.dumps(data)}')
        logger.info(f'登录验证码-返回响应 {response_object}')

        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)


    async def qrCallback(self, uuid: str, base64_string: str):
        """
        处理二维码回调，显示二维码图片，并提供多种二维码访问链接
        """
        try:
            from PIL import Image
            base64_string = base64_string.split(',')[1]
            img_data = base64.b64decode(base64_string)
            img = Image.open(io.BytesIO(img_data))
            _thread = threading.Thread(target=img.show, args=("QRCode",))
            _thread.setDaemon(True)
            _thread.start()
        except Exception as e:
            logger.error(f"处理二维码回调时发生错误: {e}")
        
        url = f"http://weixin.qq.com/x/{uuid}"
        qr_api1 = f"https://api.isoyu.com/qr/?m=1&e=L&p=20&url={url}"
        qr_api2 = f"https://api.qrserver.com/v1/create-qr-code/?size=400×400&data={url}"
        qr_api3 = f"https://api.pwmqr.com/qrcode/create/?url={url}"
        qr_api4 = f"https://my.tv.sohu.com/user/a/wvideo/getQRCode.do?text={url}"
        logger.info("您也可以通过以下网站扫描二维码：")
        logger.info(f"{qr_api3}\n{qr_api4}\n{qr_api2}\n{qr_api1}")
        return [qr_api1, qr_api2, qr_api3, qr_api4]    

    async def get_login_qr_code_async(self, token_id: str, app_id: str = "", region_id: str = "440000"):
        """
        获取登录二维码(步骤2)
        第一次登录时传空，后续登录时需传appId。
        """
        api_url = f"{self.base_url}/v2/api/login/getLoginQrCode"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "regionId": region_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_data = await response.json()
        logger.info(f'{token_id} 的登录APP信息：{json.dumps(response_data, separators=(",", ":"), ensure_ascii=False)}')
        return response_data.get('data')

    ###############################  账号管理   ###############################
    async def reconnection(self, token_id, app_id):
        '''
        断线重连
        当系统返回账号已离线，但是手机顶部还显示ipad在线，可用此接口尝试重连，若返回错误/失败则必须重新调用步骤一登录
        本接口非常用接口，可忽略
        '''
        api_url = f"{self.base_url}/v2/api/login/reconnection"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object

    async def logout_async(self, token_id, app_id):
        '''
        退出
        '''
        api_url = f"{self.base_url}/v2/api/login/logout"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_data = await response.json()
        return response_data.get('data')        
    
    async def check_online(self, token_id, app_id):
        '''
        检查是否在线
        响应结果的data=true则是在线，反之为离线
        '''
        api_url = f"{self.base_url}/v2/api/login/checkOnline"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_data = await response.json()
        return response_data.get('data')        
               
    ###############################  联系人模块   ###############################  
    async def fetch_contacts_list_async(self, token_id, app_id):
        '''
        获取通讯录列表
        
        本接口为长耗时接口，耗时时间根据好友数量递增，若接口返回超时可通过获取通讯录列表缓存接口获取响应结果
        本接口返回的群聊仅为保存到通讯录中的群聊，若想获取会话列表中的所有群聊，需要通过消息订阅做二次处理。
        原因：当未获取的群有成员在群内发消息的话会有消息回调， 开发者此刻调用获取群详情接口查询群信息入库保存即可，
        比如说手机上三年前不说话的群，侧滑删除了，用户手机上也不会看到被删除的群聊的 ，但是有群成员说了话他会显示，
        原理就是各个终端（Android、IOS、桌面版微信）取得了消息回调，又去获取群详情信息，本地数据库缓存了下来，显示的手机群聊，让用户感知的。
        '''
        api_url = f"{self.base_url}/v2/api/contacts/fetchContactsList"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def fetch_contacts_list_cache_async(self, token_id, app_id):
        '''
        获取通讯录列表缓存
        通讯录列表数据缓存10分钟，超时则需要重新调用获取通讯录列表接口
        '''
        api_url = f"{self.base_url}/v2/api/contacts/fetchContactsListCache"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_data = await response.json()
        return response_data.get('data')
    
    async def get_brief_info_async(self, token_id, app_id, wxids: list):
        '''
        获取群/好友简要信息
         1<= wxids <=100
        '''
        api_url = f"{self.base_url}/v2/api/contacts/getBriefInfo"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "wxids": wxids  # list 1<= wxids <=100
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_data = await response.json()
        return response_data.get('data')
    
    async def get_detail_info(self, token_id, app_id, wxids):
        '''
        获取群/好友详细信息
         1<= wxids <=20
        '''
        api_url = f"{self.base_url}/v2/api/contacts/getDetailInfo"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "wxids": wxids  # list 1<= wxids <=20
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_data = await response.json()
        return response_data.get('data')
    
    async def delete_friend_async(self, token_id, app_id, friend_wxid):
        '''
        删除好友
        '''
        api_url = f"{self.base_url}/v2/api/contacts/deleteFriend"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "wxid": friend_wxid
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def set_friend_remark(self, token_id, app_id, friend_wxid, remark):    
        '''
        设置好友备注
        '''
        api_url = f"{self.base_url}/v2/api/contacts/setFriendRemark"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "wxid": friend_wxid,
            "remark": remark
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    ###############################  消息模块   ############################### 
    async def post_text_async(self, token_id, app_id, to_wxid, content):
        api_url = f"{self.base_url}/v2/api/message/postText"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "toWxid": to_wxid,
            "content": content
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def post_image_async(self, token_id, app_id, to_wxid, img_url):
        api_url = f"{self.base_url}/v2/api/message/postImage"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "toWxid": to_wxid,
            "imgUrl": img_url
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers= headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def post_voice_async(self, token_id, app_id, to_wxid, voice_url, voice_duration):
        api_url = f"{self.base_url}/v2/api/message/postVoice"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "toWxid": to_wxid,
            "voiceUrl": voice_url,
            "voiceDuration": voice_duration
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def post_video_async(self, token_id, app_id, to_wxid, video_url, video_thumb_url, video_duration):
        api_url = f"{self.base_url}/v2/api/message/postVideo"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "toWxid": to_wxid,
            "videoUrl": video_url,
            "videoDuration": video_duration,
            "thumbUrl": video_thumb_url
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def forward_image_async(self, token_id, app_id, to_wxid, aeskey, cdnthumburl, cdnthumblength, cdnthumbheight, cdnthumbwidth, length, md5):
        
        api_url = f"{self.base_url}/v2/api/message/forwardImage"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "toWxid": to_wxid,
            "xml": f"<?xml version=\"1.0\"?>\n<msg>\n\t<img aeskey=\"{aeskey}\" encryver=\"1\" cdnthumbaeskey=\"{aeskey}\" cdnthumburl=\"{cdnthumburl}\" cdnthumblength=\"{cdnthumblength}\" cdnthumbheight=\"{cdnthumbheight}\" cdnthumbwidth=\"{cdnthumbwidth}\" cdnmidheight=\"0\" cdnmidwidth=\"0\" cdnhdheight=\"0\" cdnhdwidth=\"0\" cdnmidimgurl=\"{cdnthumburl}\" length=\"{length}\" md5=\"{md5}\" />\n\t<platform_signature></platform_signature>\n\t<imgdatahash></imgdatahash>\n</msg>"
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('data', None), response_object.get('ret', None), response_object.get('msg', None)
    
    async def forward_video_async(self, token_id, app_id, to_wxid, aeskey, cdnvideourl, length):
        api_url = f"{self.base_url}/v2/api/message/forwardVideo"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "toWxid": to_wxid,
            "xml": f"<?xml version=\"1.0\"?>\n<msg>\n\t<videomsg aeskey=\"{aeskey}\" cdnvideourl=\"{cdnvideourl}\" cdnthumbaeskey=\"{aeskey}\" cdnthumburl=\"{cdnvideourl}\" length=\"{length}\" playlength=\"7\" cdnthumblength=\"8192\" cdnthumbwidth=\"135\" cdnthumbheight=\"240\" fromusername=\"zhangchuan2288\" md5=\"8804c121e9db91dd844f7a34035beb88\" newmd5=\"\" isplaceholder=\"0\" rawmd5=\"\" rawlength=\"0\" cdnrawvideourl=\"\" cdnrawvideoaeskey=\"\" overwritenewmsgid=\"0\" originsourcemd5=\"\" isad=\"0\" />\n</msg>"
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def add_contacts_async(self, token_id: str, app_id: str, scene: int, option: int, v3: str, v4: str, content: str):
        api_url = f"{self.base_url}/v2/api/contacts/addContacts"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "scene": scene,
            "option": option,
            "v3": v3,
            "v4": v4,
            "content": content
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None)
    
    async def check_relation(self, token_id, app_id, wxids: list):
        '''
        检查好友关系
        '''
        api_url = f"{self.base_url}/v2/api/contacts/checkRelation"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "wxids": wxids  # list 1<= wxids <=20
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    ###############################  下载模块   ############################### 
    async def download_audio_msg_async(self, token_id: str, app_id: str, msg_id: int, xml: str):
        data = {
            "appId": app_id,
            "msgId": msg_id,
            "xml": xml
        }
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        url = 'http://api.geweapi.com/gewe/v2/api/message/downloadVoice'
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=data, headers=headers) as response:
                if response.ok:
                    data = await response.json()
                    if data['ret'] == 200:
                        return data['data']['fileUrl']
                    else:
                        return False
                else:
                    return False
    
    async def download_image_msg_async(self, token_id: str, app_id: str, xml: str):
        data = {
            "appId": app_id,
            "type": 2,
            "xml": xml
        }
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(f"{self.base_url}/v2/api/message/downloadImage", json=data, headers=headers) as response:
                if response.ok:
                    data = await response.json()
                    if data['ret'] == 200:
                        return data['data']['fileUrl']
                    else:
                        return False
                else:
                    return False
    
    async def download_audio_file(self, fileUrl: str, file_name: str):
        local_filename = f'./silk/{file_name}.silk'
        async with aiohttp.ClientSession() as session:
            async with session.get(fileUrl) as response:
                if response.status == 200:
                    with open(local_filename, 'wb') as f:
                        while True:
                            chunk = await response.content.read(1024)
                            if not chunk:
                                break
                            f.write(chunk)
                    return local_filename
                else:
                    return None

    ###############################  群模块   ###############################
    async def get_chatroom_info_async(self, token_id, app_id, chatroom_id):    
        '''
        获取群信息
        '''
        api_url = f"{self.base_url}/v2/api/group/getChatroomInfo"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "chatroomId": chatroom_id
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def add_group_member_as_friend(self, token_id, app_id, chatroom_id, member_wxid, content):  
        '''
        添加群成员为好友
        '''
        api_url = f"{self.base_url}/v2/api/group/addGroupMemberAsFriend"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "chatroomId": chatroom_id,
            "content": content,
            "memberWxid": member_wxid,
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def save_contract_list_async(self, token_id, app_id, chatroom_id, oper_type):
        '''
        群保存到通讯录
        操作类型 3保存到通讯录 2从通讯录移除
        '''
        api_url = f"{self.base_url}/v2/api/group/saveContractList"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "chatroomId": chatroom_id,
            "operType": oper_type
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def get_group_memberlist_async(self, token_id, app_id, chatroom_id):
        '''
        获取群成员列表
        '''
        api_url = f"{self.base_url}/v2/api/group/getChatroomMemberList"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "chatroomId": chatroom_id,      
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def agree_join_room_async(self, token_id, app_id, url):
        '''
        同意入群
        '''
        api_url = f"{self.base_url}/v2/api/group/agreeJoinRoom"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "url": url,
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    ###############################  朋友圈模块 ###################################
    # 在新设备登录后的1-3天内，您将无法使用朋友圈发布、点赞、评论等功能。在此期间，如果尝试进行这些操作，您将收到来自微信团队的提醒。请注意遵守相关规定。
    
    async def sns_visible_scope(self, token_id, app_id, option): 
        '''
        朋友圈可见范围 option 可选项
        1:全部
        2:最近半年
        3:最近一个月
        4:最近三天
        '''
        api_url = f"{self.base_url}/v2/api/sns/snsVisibleScope"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "option": option,
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def stranger_visibility_enabled(self, token_id, app_id, enabled: bool):
        '''
        是否允许陌生人查看朋友圈
        '''
        api_url = f"{self.base_url}/v2/api/sns/strangerVisibilityEnabled"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "enabled": enabled
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)       
    
    async def send_text_sns_async(self, token_id, app_id, content):  
        '''
        发送文字朋友圈
        '''
        api_url = f"{self.base_url}/v2/api/sns/sendTextSns"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "content": content
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def send_image_sns_async(self, token_id, app_id, content, img_infos: list): 
        '''
        发送图片朋友圈
        '''
        api_url = f"{self.base_url}/v2/api/sns/sendImgSns"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "allowWxIds": [],
            "atWxIds": [],
            "disableWxIds": [],
            "content": content,
            "imgInfos": img_infos,  # 通过上传朋友圈图片接口获取
            "privacy": False
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
    
    async def send_video_sns_async(self, token_id, app_id, content: str, video_info: dict):   
        '''
        发送视频朋友圈
        '''
        api_url = f"{self.base_url}/v2/api/sns/sendVideoSns"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "content": content,
            "allowWxIds": [],
            "atWxIds": [],
            "disableWxIds": [],
            "videoInfo": video_info,
            "privacy": False
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def upload_sns_image_async(self, token_id, app_id, img_urls: list):
        '''
        上传朋友圈图片
        '''
        api_url = f"{self.base_url}/v2/api/sns/uploadSnsImage"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "imgUrls": img_urls
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    async def upload_sns_video_async(self, token_id, app_id, video_url: str, video_thumb_url: str):
        '''
        上传朋友圈视频
        '''
        api_url = f"{self.base_url}/v2/api/sns/uploadSnsVideo"
        headers = {
            'X-GEWE-TOKEN': token_id,
            'Content-Type': 'application/json'
        }
        data = {
            "appId": app_id,
            "thumbUrl": video_thumb_url,
            "videoUrl": video_url,
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, headers=headers, json=data) as response:
                response_object = await response.json()
        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)

    ###############################     其他   ###############################
    
    async def save_session_messages_to_cache_async(self, hash_key,item:object)->list:
        '''
        对话列表
        '''
        messages = await self.redis_service.get_hash(hash_key)
        wxid=hash_key.split(':')[-1]
        if not messages:
            messages=[{"role": "system", "content": ""}]
            messages.append(item)
            await self.redis_service.set_hash(hash_key, {"data": json.dumps(messages, ensure_ascii=False)}, 600)
        else:
            messages_str = await self.redis_service.get_hash_field(hash_key, "data")
            messages = json.loads(messages_str) if messages_str else []
            #判断是否含有图片
            last_message = messages[-1]
            content = last_message.get("content", [])
            if isinstance(content, list) and content:
                last_content_type = content[-1].get("type")
                if last_content_type == 'image_url':
                    content.append(item['content'][0])
                    messages[-1]['content']=content
                else:
                    messages.append(item)
            else:
                if last_message!= item:
                    messages.append(item)
            await self.redis_service.set_hash(hash_key,{"data":json.dumps(messages,ensure_ascii=False)},600)
        return messages    
 
    async def get_contacts_brief_from_cache_async(self, wxid)->list:
        """
        获取联系人信息保存到 Redis 缓存。
        """
        hash_key = f"__AI_OPS_WX__:CONTACTS_BRIEF:{wxid}"
        cache_str = await self.redis_service.get_hash_field(hash_key, "data")
        return json.loads(cache_str) if cache_str else []
        
    async def save_contacts_brief_to_cache_async(self, token_id, app_id, wxid, contacts_wxids: list)->list:
        """
        将联系人信息保存到 Redis 缓存。
        """
        # Redis 缓存的 key
        hash_key = f"__AI_OPS_WX__:CONTACTS_BRIEF:{wxid}"
        
        # 获取缓存中的数据
        cache_str = await self.redis_service.get_hash_field(hash_key, "data")
        cache = json.loads(cache_str) if cache_str else []
        # 回调处理
        if len(contacts_wxids) == 1:
            cache_wxids = [f['userName'] for f in cache]
            friends_brief = await self.get_brief_info_async(token_id, app_id, contacts_wxids)
            if contacts_wxids[0] in cache_wxids:
                # 替换已经存在的数据
                for i in range(len(cache)):
                    if cache[i]['userName'] == contacts_wxids[0]:
                        cache[i] = friends_brief[0]
            else:     
                cache.extend(f for f in friends_brief if f["nickName"])
                friends_no_brief_wxid = [f['userName'] for f in friends_brief if not f["nickName"]]
                if friends_no_brief_wxid:
                    detailed_info = self.get_detail_info(token_id, app_id, friends_no_brief_wxid)
                    cache.extend(detailed_info)
            
        # 分批处理
        else:
            cache=[]
            # 缓存为空，分批处理 contacts_wxids
            batch_size = 100
            for i in range(0, len(contacts_wxids), batch_size):
                batch = contacts_wxids[i:i + batch_size]
                friends_brief = await self.get_brief_info_async(token_id, app_id, batch)

                cache.extend(f for f in friends_brief if f["nickName"])

                friends_no_brief_wxid = [f['userName'] for f in friends_brief if not f["nickName"]]
                if friends_no_brief_wxid:
                    detailed_info = self.get_detail_info(token_id, app_id, friends_no_brief_wxid)
                    cache.extend(detailed_info)
            
            # 更新缓存
        
        await self.redis_service.update_hash_field(hash_key, "data", json.dumps(cache, ensure_ascii=False))
        return cache

    async def delete_contacts_brief_from_cache_async(self, wxid, contacts_wxids: list):
        """
        删除联系人信息保存到 Redis 缓存。
        """
        hash_key = f"__AI_OPS_WX__:CONTACTS_BRIEF:{wxid}"
        cache_str = await self.redis_service.get_hash_field(hash_key, "data")
        cache = json.loads(cache_str) if cache_str else []


        # 将 contacts_wxids 转换为集合，提高查找效率
        wxids_set = set(contacts_wxids)

        # 过滤 cache：保留 userName 不在集合中的对象
        filtered_cache_async = [contact for contact in cache if contact["userName"] not in wxids_set]

        # # 如果需要原地修改原 cache 列表：
        # cache[:] = filtered_cache_async
        await self.redis_service.update_hash_field(hash_key, "data", json.dumps(filtered_cache_async, ensure_ascii=False))

    async def save_groups_info_to_cache_async(self, token_id, app_id, wxid, chatroom_ids: list)->list:
        """
        将群信息保存到 Redis 缓存。
        """
        # Redis 缓存的 key
        hash_key = f"__AI_OPS_WX__:GROUPS_INFO:{wxid}"
        
         # 获取当前缓存中所有的 chatroom_id
        existing_chatrooms = await self.redis_service.get_hash(hash_key)

        # 找出需要删除的 chatroom_ids
        chatrooms_to_delete = set(existing_chatrooms.keys()) - set(chatroom_ids)

        # 删除缓存中不再需要的 chatroom_id 数据
        for chatroom_id in chatrooms_to_delete:
            await self.redis_service.delete_hash_field(hash_key, chatroom_id)
        for chatroom_id in chatroom_ids:
            # 获取群信息
            ret, msg, data =await self.get_chatroom_info_async(token_id, app_id, chatroom_id)
            if ret != 200:
                continue
            
            # 更新缓存
            await self.redis_service.update_hash_field(hash_key, chatroom_id, json.dumps(data, ensure_ascii=False))
            await asyncio.sleep(0.1)
                

    async def save_groups_members_to_cache_async(self, token_id, app_id, wxid, chatroom_ids: list):
        """
        将群成员保存到 Redis 缓存。
        """
        # Redis 缓存的 key
        hash_key = f"__AI_OPS_WX__:GROUPS_MEMBERS:{wxid}"
        
         # 获取当前缓存中所有的 chatroom_id
        existing_chatrooms = await self.redis_service.get_hash(hash_key)

        # 找出需要删除的 chatroom_ids
        chatrooms_to_delete = set(existing_chatrooms.keys()) - set(chatroom_ids)

        # 删除缓存中不再需要的 chatroom_id 数据
        for chatroom_id in chatrooms_to_delete:
            await self.redis_service.delete_hash_field(hash_key, chatroom_id)
        
        for chatroom_id in chatroom_ids:
            # 获取群信息
            ret, msg, data = await self.get_group_memberlist_async(token_id, app_id, chatroom_id)
            if ret != 200:
                continue
            # 更新缓存
            await self.redis_service.update_hash_field(hash_key, chatroom_id, json.dumps(data, ensure_ascii=False))
            await asyncio.sleep(0.1)
        
    async def update_group_members_to_cache_async(self, token_id, app_id, wxid, chatroom_id: str):
        """
        更新将群信息保存到 Redis 缓存。
        """
        # Redis 缓存的 key
        hash_key = f"__AI_OPS_WX__:GROUPS_MEMBERS:{wxid}"

        # 获取群信息
        ret, msg, data = await self.get_group_memberlist_async(token_id, app_id, chatroom_id)   
        await self.redis_service.update_hash_field(hash_key, chatroom_id, json.dumps(data, ensure_ascii=False))

    async def get_group_members_from_cache_async(self, wxid,chatroom_id)->dict:
        """
        获取缓存中单个群成员。
        """
        hash_key = f"__AI_OPS_WX__:GROUPS_MEMBERS:{wxid}"
        cache = await self.redis_service.get_hash_field(hash_key,chatroom_id)
        groups=json.loads(cache) if cache else {}
        return groups
    
    async def get_groups_members_from_cache_async(self, wxid)->list:
        """
        获取缓存中所有群成员。
        """
        hash_key = f"__AI_OPS_WX__:GROUPS_MEMBERS:{wxid}"
        cache = await self.redis_service.get_hash(hash_key)
        groups=[json.loads(v) for v in cache.values()]
        return groups
       
    async def get_groups_info_members_from_cache_async(self, wxid)->list:
        groups_info_hash_key = f"__AI_OPS_WX__:GROUPS_INFO:{wxid}"
        groups_info_cache = await self.redis_service.get_hash(groups_info_hash_key)
        
        groups_menbers_hash_key= f"__AI_OPS_WX__:GROUPS_MEMBERS:{wxid}"
        groups_menbers_cache = await self.redis_service.get_hash(groups_menbers_hash_key)

        groups_info_cache_list=[{"chatroom_id": field, "value": json.loads(value)} for field, value in groups_info_cache.items()]
        groups_menbers_cache_list=[{"chatroom_id": field, "value": json.loads(value)} for field, value in groups_menbers_cache.items()]

        # 合并逻辑
        merged_data = []

        # 遍历 group_info
        for info in groups_info_cache_list:
            chatroom_id = info['chatroom_id']
            group_data = info['value']
            
            # 查找对应的 group_members
            members = next((m['value'] for m in groups_menbers_cache_list if m['chatroom_id'] == chatroom_id), None)
            
            if members:
                # 合并数据
                merged_group = {
                    "nickName": group_data['nickName'],
                    "chatroomId": group_data['chatroomId'],
                    "memberList": members['memberList'],
                    "chatroomOwner": members.get('chatroomOwner', None),
                    "adminWxid": members.get('adminWxid',None)
                }
                merged_data.append(merged_group)

        return merged_data

    async def get_group_info_members_from_cache_async(self, wxid,chatroom_id)->dict:
        group_info_cache= await self.get_group_info_from_cache_async(wxid,chatroom_id)
        group_menbers_cache= await self.get_group_members_from_cache_async(wxid,chatroom_id)
        group_info_members = {
            "nickName": group_info_cache['nickName'],
            "chatroomId": group_info_cache['chatroomId'],
            "memberList": group_menbers_cache['memberList'],
            "chatroomOwner": group_menbers_cache['chatroomOwner'],
            "adminWxid": group_menbers_cache['adminWxid']
        }
        return group_info_members

    async def update_group_info_to_cache_async(self, token_id, app_id, wxid, chatroom_id: str):
        """
        更新将群信息保存到 Redis 缓存。
        """
        # Redis 缓存的 key
        hash_key = f"__AI_OPS_WX__:GROUPS_INFO:{wxid}"

        # 获取群信息
        ret, msg, data =await self.get_chatroom_info_async(token_id, app_id, chatroom_id)   
        await self.redis_service.update_hash_field(hash_key, chatroom_id, json.dumps(data, ensure_ascii=False))

    async def get_groups_info_from_cache_async(self, wxid)->list:
        """
        获取群信息保存到 Redis 缓存。
        """
        hash_key = f"__AI_OPS_WX__:GROUPS_INFO:{wxid}"
        cache = await self.redis_service.get_hash(hash_key)
        groups=[json.loads(v) for v in cache.values()]
        return groups
    
    async def get_group_info_from_cache_async(self, wxid,chatroom_id)->dict:
        """
        获取群信息保存到 Redis 缓存。
        """
        hash_key = f"__AI_OPS_WX__:GROUPS_INFO:{wxid}"
        cache = await self.redis_service.get_hash_field(hash_key,chatroom_id)
        groups=json.loads(cache) if cache else {}
        return groups

    async def get_wxchat_config_from_cache_async(self, wxid):
        """
        获取配置信息
        """
        hash_key = f"__AI_OPS_WX__:WXCHAT_CONFIG"
        config = await self.redis_service.get_hash_field(hash_key, wxid)
        return json.loads(config) if config else {}
    
    async def save_wxchat_config_async(self, wxid, config:dict):
        """
        保存配置信息
        """
        hash_key = f"__AI_OPS_WX__:WXCHAT_CONFIG"
        await self.redis_service.update_hash_field(hash_key, wxid, json.dumps(config, ensure_ascii=False))

    async def get_login_info_from_cache_async(self,tel):
        hash_key = f"__AI_OPS_WX__:LOGININFO:{tel}"
        cache = await self.redis_service.get_hash(hash_key)
        return cache
    
    async def get_login_info_by_wxid_async(self,wxid:str)->tuple:
        cursor = 0
        while True:
            cursor, login_keys =await self.redis_service.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*')
            # 批量获取所有键的 hash 数据
            for k in login_keys:
                r = await self.redis_service.get_hash(k)
                if r.get("wxid") == wxid:
                    return k,r
            
            # 如果游标为 0，则表示扫描完成
            if cursor == 0:
                break
        
        return None,None
    
    async def get_login_info_by_app_id_async(self,app_id:str)->tuple:
        cursor = 0
        while True:
            cursor, login_keys =await self.redis_service.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*')
            print(f'login_keys:{login_keys}')
            # 批量获取所有键的 hash 数据
            for k in login_keys:
                r = await self.redis_service.get_hash(k)
                print(r)
                if r.get("appId") == app_id:
                    return k,r
            
            # 如果游标为 0，则表示扫描完成
            if cursor == 0:
                break
        
        return None,None

    async def get_token_id_by_app_id_async(self,app_id: str) -> str:
        # 使用 SCAN 避免一次性返回所有的匹配键，逐步扫描
        cursor = 0
        while True:
            cursor, login_keys =await self.redis_service.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*')
            
            # 批量获取所有键的 hash 数据
            for k in login_keys:
                r = await self.redis_service.get_hash(k)
                if r.get("appId") == app_id:
                    return r.get("tokenId", "")
            
            # 如果游标为 0，则表示扫描完成
            if cursor == 0:
                break
        
        return ""    


    async def save_login_wx_captch_code_to_cache_async(self,token_id,captch_code):
        hash_key = f"__AI_OPS_WX__:WXCAPTCHCODE:{token_id}"
        await self.redis_service.set_hash(hash_key,{"data":captch_code},30) 

    async def get_login_wx_captch_code_from_cache_async(self,token_id)->str:
        hash_key = f"__AI_OPS_WX__:WXCAPTCHCODE:{token_id}"
        r=await self.redis_service.get_hash_field(hash_key,"data")
        return r

    async def acquire_login_lock_async(self, token_id, expire_time=10):
        hash_key = f"__AI_OPS_WX__:LOGINLOCK:{token_id}"
        identifier=str(uuid.uuid4())
        if await self.redis_service.client.setnx(hash_key, identifier):
                await self.redis_service.client.expire(hash_key, expire_time)
                return True
        return False

    async def release_login_lock_async(self, token_id):
        hash_key = f"__AI_OPS_WX__:LOGINLOCK:{token_id}"
        await self.redis_service.client.delete(hash_key)

    async def save_group_add_contacts_history_async(self,wxid,chatroom_id,contact_wxid,history:AddGroupContactsHistory):
        '''
        保存群加好友历史
        '''
        hash_key = f"__AI_OPS_WX__:GROUPS_ADD_CONTACT_HISTORY:{wxid}:{chatroom_id}"
        data_str=await self.redis_service.get_hash_field(hash_key,contact_wxid)
        data=json.loads(data_str) if data_str else []
        data.append(history.model_dump())
        await self.redis_service.update_hash_field(hash_key, contact_wxid, json.dumps(data, ensure_ascii=False))

    async def get_group_add_contacts_history_async(self,wxid,chatroom_id,contact_wxid)->list[AddGroupContactsHistory]:
        '''
        获取群加好友历史
        '''
        hash_key = f"__AI_OPS_WX__:GROUPS_ADD_CONTACT_HISTORY:{wxid}:{chatroom_id}"
        data_str=await self.redis_service.get_hash_field(hash_key,contact_wxid)
        data=json.loads(data_str) if data_str else []
        #TypeAdapter.validate_python(List[Models.AddGroupContactsHistory], data)
        return [AddGroupContactsHistory.model_validate(item) for item in data]
    
    async def check_wixd_group_add_contacts_history_async(self,wxid,chatroom_id):
        '''
        返回群发送好友达到2次的wxid
        '''
        hash_key = f"__AI_OPS_WX__:GROUPS_ADD_CONTACT_HISTORY:{wxid}:{chatroom_id}"
        cache = await self.redis_service.get_hash(hash_key)
        wxids = [key for key, value in cache.items() if len(json.loads(value)) == 2]
        return wxids

    async def enqueue_to_add_contacts_async(self,wxid,scene:int,v3,v4):
        """
        入列待添加好友
        """
        hash_key = f'__AI_OPS_WX__:TO_ADD_CONTACTS:{wxid}'
        data_str=json.dumps({"wxid":wxid,"scene":scene,"v3":v3,"v4":v4},ensure_ascii=False)
        await self.redis_service.enqueue(hash_key,data_str)
    
    async def dequeue_to_add_contacts_async(self,wxid)->dict:
        """
        出列待添加好友
        """
        hash_key = f'__AI_OPS_WX__:TO_ADD_CONTACTS:{wxid}'
        data=await self.redis_service.dequeue(hash_key)
        return json.loads(data) if data else {}




# 依赖项：获取 GeWeChatCom 单例
async def get_gewe_service(app: FastAPI = Depends()) -> GeWeService:
    return app.state.gewe_service