AiOpsWechatiPad/app.py

from flask import Flask, send_from_directory, request,jsonify
from flask_restful import Api,got_request_exception
from resources.messages_resource import MessagesResource
from resources.contacts_resources import DeleteFriendResource,GetFriendsInfoResource
from resources.config_reources import GetWxchatConfigResource ,SaveWxchatConfigResource
from resources.groups_resources import GetGroupsInfoResource,GetGroupMemberList
from resources.login_resources import GetLoginInfoResource,GetLoginWxQRCodeResource,LoginWxCaptchCodeResource
from resources.sns_resources import SendSNSTextResource,SendSNSImageResource, SendSNSVideoResource
from common.log import logger, log_exception 
from common.interceptors import before_request, after_request, handle_exception,check_login_status
import threading
from common import kafka_helper, redis_helper,utils
import random



from model import Models

import logging 
from config import load_config
  
from wechat.biz import start_kafka_consumer_thread


from wechat import gewe_chat

import os,time,json,time

from voice.ali.ali_voice import AliVoice



# 自定义错误消息
errors = {
    'UserAlreadyExistsError': {
        'message': "A user with that username already exists.",
        'status': 409,
    },
    'ResourceDoesNotExist': {
        'message': "A resource with that ID no longer exists.",
        'status': 410,
        'extra': "Any extra information you want.",
    },
}


def fetch_and_save_contacts():
    """
    获取联系人列表并保存到缓存
    """
    wxchat=gewe_chat.wxchat 
    while True:
        try:
            login_keys = list(redis_helper.redis_helper.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))
            logger.info(f"Fetching login keys: {login_keys}")

            # 遍历每一个获取到的登录键
            for k in login_keys:
                try:
                    r= redis_helper.redis_helper.get_hash(k)
                    # print(r)
                    token_id = r.get('tokenId')
                    app_id = r.get('appId')
                
                    wxid = r.get('wxid')
                    status=r.get('status')
                    if status=='1':   
                        ret,msg,contacts_list = wxchat.fetch_contacts_list(token_id, app_id)
                        if ret==200:
                            friend_wxids = [c for c in contacts_list['friends'] if c not in ['fmessage', 'medianote','weixin','weixingongzhong']] # 可以调整截取范围
                            print(f'{wxid}的好友数量 {len(friend_wxids)}')
                            wxchat.save_contacts_brief_to_cache(token_id, app_id, wxid, friend_wxids)
                            logger.info(f'微信ID {wxid} 登录APPID {app_id} 成功，联系人已定时保存')
                            chatrooms=contacts_list['chatrooms']
                            wxchat.save_groups_info_to_cache(token_id, app_id, wxid, chatrooms)
                            wxchat.save_groups_members_to_cache(token_id, app_id, wxid, chatrooms)
                            logger.info(f'微信ID {wxid} 登录APPID {app_id} 成功，群信息已定时保存')
                        else:
                            logger.warning(f'{msg}-微信ID {wxid} 登录APPID {app_id} 不能获取好友和群资料')
                    else:
                        logger.info(f'微信ID {wxid} 未登录 {app_id} ，联系人不能定时保存')
                    time.sleep(3)
                except Exception as e:
                    logger.error(f'处理好友和群资料出错 login key {k}: {str(e)}', exc_info=True)
        except Exception as e:
            logger.error(f'发送错误 {str(e)}', exc_info=True)
        time.sleep(3600*1)

def auto_add_contacts_from_chatrooms():
    '''
    从群中自动添加好友
    '''
    logger.info('自动从群添加好友')
    wxchat=gewe_chat.wxchat 
    while True:
        
        login_keys = list(redis_helper.redis_helper.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))
        #logger.info(f"Fetching login keys: {login_keys}")

        # 遍历每一个获取到的登录键
        for k in login_keys:
            r= redis_helper.redis_helper.get_hash(k)
            # print(r)
            token_id = r.get('tokenId')
            app_id = r.get('appId')
            nickname=r.get('nickName')
            wxid = r.get('wxid')
            status=r.get('status')

             # 启动线程处理登录信息
            thread = threading.Thread(target=process_add_contacts_from_chatrooms, args=(wxchat,status, wxid, token_id, app_id))
            thread.start()
            time.sleep(3)

        #time.sleep(60*10)
        time.sleep(3600*24)

def process_add_contacts_from_chatrooms(wxchat:gewe_chat.GeWeChatCom,status, wxid, token_id, app_id):
  
    if status == '1':
        c = wxchat.get_wxchat_config_from_cache(wxid)
        contacts = wxchat.get_contacts_brief_from_cache(wxid)
        contact_wxids = [c.get('userName') for c in contacts]
        chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])

        for chatroom_id in chatrooms:
            chatroom = wxchat.get_group_info_from_cache(wxid, chatroom_id)
            chatroom_member=wxchat.get_group_members_from_cache(wxid, chatroom_id)
            chatroom_nickname = chatroom.get('nickName')
            chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)
            admin_wxid = chatroom_member.get('adminWxid', None)
            logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxid}')
            contact_wxids_set = set(contact_wxids)
            if admin_wxid is not None:
                contact_wxids_set.add(admin_wxid)
            if chatroom_owner_wxid is not None:
                contact_wxids_set.add(chatroom_owner_wxid)
            
            contact_wxids_set.add(wxid)

            unavailable_wixds=wxchat.check_wixd_group_add_contacts_history(wxid,chatroom_id)
            contact_wxids_set.update(unavailable_wixds)

            chatroot_member_list = chatroom.get('memberList', [])
            remaining_chatroot_members = [x for x in chatroot_member_list if x.get('wxid') not in contact_wxids_set]
           
            nickname = next((member['nickName'] for member in chatroot_member_list if member['wxid'] == wxid), None)

            logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里还可以邀请的好友有：{[x.get("nickName") for x in remaining_chatroot_members]}')
            for m in remaining_chatroot_members:
                ret, msg, data = wxchat.add_group_member_as_friend(token_id, app_id, chatroom_id, m.get('wxid'), f'我是群聊"{chatroom_nickname}"群的{nickname}')
                if ret==200:
                    contact_wxids= m.get('wxid')
                    history=Models.AddGroupContactsHistory.model_validate({
                        "chatroomId":chatroom_id,
                        "wxid":wxid,
                        "contactWixd":contact_wxids,
                        "addTime":int(time.time())
                    })
                    wxchat.save_group_add_contacts_history(wxid,chatroom_id,contact_wxids,history)
                else:
                    logger.info(f'群好友邀请失败原因：{data}')
                logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg}')
                time.sleep(10)
            time.sleep(20)
    else:
        logger.info(f'微信ID {wxid} 未登录  {app_id} ，群成员不能定时定时')

def auto_add_contacts_from_to_add_contacts_queue():
    '''
    从待添加好友队列中自动添加好友
    '''
    logger.info('自动添加好友')
    wxchat=gewe_chat.wxchat 

    while True:  
        login_keys = list(redis_helper.redis_helper.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))
        for k in login_keys:
            r= redis_helper.redis_helper.get_hash(k)
            token_id = r.get('tokenId')
            app_id = r.get('appId')
            nickname=r.get('nickName')
            wxid = r.get('wxid')
            status=r.get('status')
            if status == '1':
                c = wxchat.get_wxchat_config_from_cache(wxid)
                contacts = wxchat.get_contacts_brief_from_cache(wxid)
                contact_wxids = [c.get('userName') for c in contacts]
                add_contacts_queue = c.get('addContactsQueue', [])
                for contact_wxid in add_contacts_queue:
                    if contact_wxid not in contact_wxids:        
                        ret, msg, data = wxchat.add_contacts(token_id, app_id, contact_wxid)
                        if ret==200:    
                            contact_wxids.append(contact_wxid)
                            wxchat.save_contacts_brief_to_cache(token_id, app_id, wxid, contact_wxids)
                            history=Models.AddContactsHistory.model_validate({
                                "wxid":wxid,
                                "contactWixd":contact_wxid,
                                "addTime":int(time.time())            
                            })

def background_wxchat_thread():
    lock_name = "background_wxchat_thread_lock"
    lock_identifier = str(time.time())  # 使用时间戳作为唯一标识

    # 尝试获取分布式锁
    if redis_helper.redis_helper.acquire_lock(lock_name, timeout=60):
        try:
            logger.info("分布式锁已成功获取")
            # 启动任务
            scan_wx_login_info()
            #threading.Thread(target=fetch_and_save_contacts).start()
            #threading.Thread(target=auto_add_contacts_from_chatrooms).start()
            
            threading.Thread(target=wx_thread_manager.check_and_manage_threads()).start()

            # 保持锁的续期
            while True:
                time.sleep(30)  # 每30秒检查一次锁的状态
                if not redis_helper.redis_helper.renew_lock(lock_name, lock_identifier, timeout=60):
                    break  # 如果无法续期锁，退出循环

        finally:
            # 释放锁
            redis_helper.redis_helper.release_lock(lock_name, lock_identifier)
    else:
        # 如果获取锁失败，等待一段时间后重试
        time.sleep(10)
        background_wxchat_thread()

def scan_wx_login_info():
    wxchat = gewe_chat.wxchat
    cursor = 0
    while True:
        cursor, login_keys = redis_helper.redis_helper.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*')
        
        for k in login_keys:
            r = redis_helper.redis_helper.get_hash(k)
            app_id=r.get("appId")
            #tel=r.get("mobile")
            token_id=r.get("tokenId")
            wxid=r.get("wxid")

            is_online = wxchat.check_online(token_id, app_id)
            if is_online:
                logger.info(f'微信ID {wxid} 在APPID {app_id} 已经在线')
            else:
                # 尝试重连
                res = wxchat.reconnection(token_id, app_id)
                if res.get('ret') == 200:
                    logger.info(f'微信ID {wxid} 在APPID {app_id} 重连成功')
                    # 同步联系人
                    #fetch_and_save_contacts(wxchat, token_id, app_id, k)  
                else:
                    print("重连失败，重新登录...")
                    print("发送离线消息到kafka")
                    redis_helper.redis_helper.update_hash_field(k,'status',0)  


            time.sleep(3)
            
        
        # 如果游标为 0，则表示扫描完成
        if cursor == 0:
            break
    
class WechatThreadManager:
    def __init__(self):
        # 同步好友服务线程
        self.sync_contacts_worker_threads = {}
        self.sync_contacts_task_threads = {}
        # 群加好友服务线程
        self.add_contacts_from_group_worker_threads = {}
        # 群加好友任务线程
        self.add_group_member_as_friend_task_threads = {}

        # 自动加好友线程
        self.add_contacts_worker_threads = {}


        self.wxchat=gewe_chat.wxchat

    def check_and_manage_threads(self):
        
        while True:  
            login_keys = list(redis_helper.redis_helper.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))
            for k in login_keys:
                r= redis_helper.redis_helper.get_hash(k)
                token_id = r.get('tokenId')
                app_id = r.get('appId')
                nickname=r.get('nickName')
                wxid = r.get('wxid')
                status=r.get('status')
                #print(status)
                if status == '1':
                    #print(wxid not in self.add_contacts_from_group_worker_threads)
                    if wxid not in self.add_contacts_from_group_worker_threads:
                        print(f"微信 {wxid} 在线，创建群好友添加线程...")
                        self.add_contacts_from_group_worker_threads[wxid]  = threading.Thread(
                                    target=self.auto_add_contacts_from_group_worker,
                                    kwargs={"wxid":wxid, "token_id": token_id, "app_id": app_id}  # wxid 作为关键字参数
                                )

                        self.add_contacts_from_group_worker_threads[wxid].start()
                        

                    if wxid not in self.sync_contacts_worker_threads:
                        print(f"微信 {wxid} 在线，创建同步联系人任务...")
                        self.sync_contacts_worker_threads[wxid]  = threading.Thread(
                            target=self.sync_contacts_processer,
                            kwargs={"wxid":wxid, "token_id": token_id, "app_id": app_id} 
                        )
                        self.sync_contacts_worker_threads[wxid].start()

                else:
                    if wxid in self.add_contacts_from_group_worker_threads:
                        print(f"微信 {wxid} 离线，关闭群好友添加线程...")
                        self.add_contacts_from_group_worker_threads[wxid].join()
                        del self.add_contacts_from_group_worker_threads[wxid]
                    if wxid in self.sync_contacts_worker_threads:
                        print(f"微信 {wxid} 离线，关闭同步联系线程线程...")
                        self.sync_contacts_worker_threads[wxid].join()
                        del self.sync_contacts_worker_threads[wxid]
                        print('关闭线程')
            time.sleep(0)
    
    def auto_add_contacts_worker(self,wxid):
        while True:
            try:
                to_add_contact=  self.wxchat.dequeue_to_add_contacts(wxid)
                if not to_add_contact:
                    continue
                
            except Exception as e:
                logger.error(f'{wxid} 自动添加好友出错')
                
    def add_friends2(self,wxid):
        '''
        24 小时只能加 5-15 位好友，每 2 小时不要超过 8 人，每个好友添加间隔要做随机间隔
        '''
        total_friends = 0
        start_time = time.time()
        last_friend_time = start_time
        
        while time.time() - start_time < 24 * 60 * 60:  # 24 小时内
            # 每 2 小时内不超过 8 个好友
            if time.time() - last_friend_time < 2 * 60 * 60:
                if len([t for t in [last_friend_time] if t > time.time() - 2 * 60 * 60]) >= 8:
                    time.sleep(1)
                    continue

            # 添加好友
            print(f"添加好友 {total_friends + 1}")
            total_friends += 1

            # 随机间隔
            interval = random.randint(1, 10) * 60  # 1到10分钟的随机间隔
            time.sleep(interval)

            # 更新最后添加好友的时间
            last_friend_time = time.time()

            # 如果已经添加了 15 个好友，提前退出
            if total_friends >= 15:
                break

    def add_friends():
        total_friends = 0
        start_time = time.time()
        friend_times = []  # 用于记录每次添加好友的时间

        while time.time() - start_time < 24 * 60 * 60:  # 24 小时内
            # 检查最近 2 小时内是否已经添加了 8 个好友
            current_time = time.time()
            recent_friends = [t for t in friend_times if t > current_time - 2 * 60 * 60]
            if len(recent_friends) >= 8:
                # 如果已经添加了 8 个好友，等待一段时间再检查
                time.sleep(1)
                continue

            # 添加好友
            print(f"添加好友 {total_friends + 1}")
            total_friends += 1

            # 记录添加好友的时间
            friend_times.append(current_time)

            # 随机间隔
            interval = random.randint(1, 10) * 60  # 1到10分钟的随机间隔
            time.sleep(interval)

            # 如果已经添加了 15 个好友，提前退出
            if total_friends >= 15:
                break
    
    def auto_add_contacts_from_group_worker(self,wxid, token_id, app_id):
        '''
        从群添加好友任务
        '''
        while True:
            k, login_info = utils.get_login_info_by_wxid(wxid)
            if login_info.get('status') == "0":
                return             
            c = self.wxchat.get_wxchat_config_from_cache(wxid)
            contacts = self.wxchat.get_contacts_brief_from_cache(wxid)
            contact_wxids = [c.get('userName') for c in contacts]
            chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])
            for chatroom_id in chatrooms:
                chatroom = self.wxchat.get_group_info_from_cache(wxid, chatroom_id)
                chatroom_member=self.wxchat.get_group_members_from_cache(wxid, chatroom_id)
                chatroom_nickname = chatroom.get('nickName')
                chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)
                admin_wxid = chatroom_member.get('adminWxid', None)
                #logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxid}')
                contact_wxids_set = set(contact_wxids)
                if admin_wxid is not None:
                    contact_wxids_set.add(admin_wxid)
                if chatroom_owner_wxid is not None:
                    contact_wxids_set.add(chatroom_owner_wxid)
                
                contact_wxids_set.add(wxid)

                unavailable_wixds=self.wxchat.check_wixd_group_add_contacts_history(wxid,chatroom_id)
                contact_wxids_set.update(unavailable_wixds)

                chatroot_member_list = chatroom.get('memberList', [])
                remaining_chatroot_members = [x for x in chatroot_member_list if x.get('wxid') not in contact_wxids_set]
            
                nickname = next((member['nickName'] for member in chatroot_member_list if member['wxid'] == wxid), None)

                #logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里还可以邀请的好友有：{[x.get("nickName") for x in remaining_chatroot_members]}')
                for m in remaining_chatroot_members:
                    tasks_id=f'{wxid}-{chatroom_id}-{m.get("wxid")}'
                    if tasks_id not in self.add_group_member_as_friend_task_threads:
                        self.add_group_member_as_friend_task_threads[tasks_id] =threading.Thread(target=self.add_group_member_as_friend_thread,                                                       
                                                    args=(token_id, app_id, wxid,chatroom_id,m,nickname,chatroom_nickname,)
                                                    )
                        #self.add_group_member_as_friend_task_threads[tasks_id].daemon = True  # 设置为守护线程
                        self.add_group_member_as_friend_task_threads[tasks_id].start()
            time.sleep(0)

    def sync_contacts_processer(self,wxid, token_id, app_id):
        while True:
            k, login_info = utils.get_login_info_by_wxid(wxid)
            if login_info.get('status') == "0":
                print('sync_contacts_processer exit')
                #self.sync_contacts_task_threads[wxid].join()
                del self.sync_contacts_task_threads[wxid]
                return
            if wxid not in self.sync_contacts_task_threads:                 
                self.sync_contacts_task_threads[wxid]=threading.Thread(target=self.sync_contacts_threads_processer, args=(wxid,token_id,app_id))
                #self.sync_contacts_task_threads[wxid].daemon = True 
                self.sync_contacts_task_threads[wxid].start()
            time.sleep(0)

    def sync_contacts_threads_processer(self,wxid, token_id, app_id):
        ret,msg,contacts_list = self.wxchat.fetch_contacts_list(token_id, app_id)
        if ret==200:
            friend_wxids = [c for c in contacts_list['friends'] if c not in ['fmessage', 'medianote','weixin','weixingongzhong']] # 可以调整截取范围
            print(f'{wxid}的好友数量 {len(friend_wxids)}')
            self.wxchat.save_contacts_brief_to_cache(token_id, app_id, wxid, friend_wxids)
            logger.info(f'微信ID {wxid} 登录APPID {app_id} 成功，联系人已定时保存')
            chatrooms=contacts_list['chatrooms']
            self.wxchat.save_groups_info_to_cache(token_id, app_id, wxid, chatrooms)
            self.wxchat.save_groups_members_to_cache(token_id, app_id, wxid, chatrooms)
            logger.info(f'微信ID {wxid} 登录APPID {app_id} 成功，群信息已定时保存')
        else:
            logger.warning(f'{msg}-微信ID {wxid} 登录APPID {app_id} 不能获取好友和群资料')
        time.sleep(random.uniform(3000, 3600))

    def add_group_member_as_friend_thread(self,token_id, app_id, wxid,chatroom_id,chatroot_member,nickname,chatroom_nickname):
        '''
        从群添加好友线程，
        '''
        tasks_id=f'{wxid}-{chatroom_id}-{chatroot_member.get("wxid")}'
        k, login_info = utils.get_login_info_by_wxid(wxid)
        if login_info.get('status') == "0":
            del self.add_group_member_as_friend_task_threads[tasks_id]
            return  
        
        contact_history=self.wxchat.get_group_add_contacts_history(wxid,chatroom_id,chatroot_member.get('wxid'))
        if len(contact_history)==1:
            last_contact_history=contact_history[0]
            diff_time=int(time.time())-last_contact_history.addTime
            oneday_seconds=60 * 60 * 24
            if diff_time<oneday_seconds:
                logging.info(f'{nickname}-{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {chatroot_member.get("nickName")}-{ chatroot_member.get("wxid")} 第二次邀请好友，等待{oneday_seconds-diff_time}秒再执行')
                time.sleep(oneday_seconds-diff_time)
        ret, msg, data = self.wxchat.add_group_member_as_friend(token_id, app_id, chatroom_id, chatroot_member.get('wxid'), f'我是群聊"{chatroom_nickname}"群的{nickname}')
        if ret!=200:
                logger.info(f'群好友邀请失败原因：{data}')
        contact_wxids= chatroot_member.get('wxid')
        history=Models.AddGroupContactsHistory.model_validate({
            "chatroomId":chatroom_id,
            "wxid":wxid,
            "contactWixd":contact_wxids,
            "addTime":int(time.time())
        })
        self.wxchat.save_group_add_contacts_history(wxid,chatroom_id,contact_wxids,history)                
        logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {chatroot_member.get("nickName")}-{chatroot_member.get("wxid")} 发送好友邀请  {msg}')
        time.sleep(random.uniform(5, 10))
        del self.add_group_member_as_friend_task_threads[tasks_id]


app = Flask(__name__)

# api = Api(app)
flask_api = Api(app,errors=errors, catch_all_404s=True)

# 设置日志（logger 已在 log.py 中配置）
app.logger.handlers.clear()  # 清除 Flask 默认的日志处理器
app.logger.addHandler(logger.handlers[1])  # 使用文件日志处理器
app.logger.setLevel(logging.DEBUG)  # 设置日志级别

# 添加拦截器
app.before_request(before_request)
app.after_request(after_request)
app.register_error_handler(Exception, handle_exception)

# 定义路由

flask_api.add_resource(MessagesResource, '/messages')

flask_api.add_resource(DeleteFriendResource, '/api/contacts/deletefriend')
flask_api.add_resource(GetFriendsInfoResource, '/api/contacts/getfriends')

flask_api.add_resource(GetWxchatConfigResource, '/api/wxchat/getconfig') 
flask_api.add_resource(SaveWxchatConfigResource, '/api/wxchat/saveconfig') 

flask_api.add_resource(GetGroupsInfoResource, '/api/groups/getchatroominfo') 
flask_api.add_resource(GetGroupMemberList,'/api/groups/getmenberlist')

flask_api.add_resource(GetLoginInfoResource, '/api/agent/getlogin')
flask_api.add_resource(GetLoginWxQRCodeResource, '/api/agent/getwxqrcode')
flask_api.add_resource(LoginWxCaptchCodeResource, '/api/agent/logincaptchcode')

flask_api.add_resource(SendSNSTextResource, '/api/sns/sendtext')
flask_api.add_resource(SendSNSImageResource, '/api/sns/sendimages')
flask_api.add_resource(SendSNSVideoResource, '/api/sns/sendvideo')


load_config()
kafka_helper.start()
redis_helper.start()
gewe_chat.start()
wx_thread_manager=WechatThreadManager()
#wx_thread_manager.check_and_manage_threads()
start_kafka_consumer_thread()
# background_wxchat_thread()
threading.Thread(target=background_wxchat_thread).start()

if __name__ == '__main__':
  # 获取环境变量
    environment = os.environ.get('environment', 'default')
    port = 80 if environment == 'default' else 5000  

    app.run(debug=False, host='0.0.0.0', port=port)