1 week ago · 64f15c21e7
--- a/tasks.py
+++ b/tasks.py
@@ -207,324 +207,6 @@ def scheduled_task_sync_wx_info(self, redis_config, kafka_config, gewe_config):
    loop.run_until_complete(task())  # 在现有事件循环中运行任务



 #@celery_app.task(name='tasks.scheduled_task_add_contacts_from_chatrooms_p', bind=True, acks_late=True)
 def scheduled_task_add_contacts_from_chatrooms_p(self, redis_config, kafka_config, gewe_config):
    async def task():
        try:
            now = datetime.now()
            if now.hour < 8:
                logger.info(f"定时群成员定时添好友任务不启动,当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，早于8点")
                return
            logger.info('定时群成员定时添好友任务开始')
            redis_service = RedisService()
            await redis_service.init(**redis_config)
            gewe_service = await GeWeService.get_instance(redis_service,gewe_config['api_url'])
            login_keys = []
            async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):
                login_keys.append(key)

            #print(login_keys)
            for k in login_keys:
                r = await redis_service.get_hash(k)
                app_id = r.get("appId")
                token_id = r.get("tokenId")
                wxid = r.get("wxid")
                status = r.get('status')
                if status == '0':
                    logger.warning(f"微信号 {wxid} 已经离线，群成员不能定时添加")
                    continue

                c = await gewe_service.get_wxchat_config_from_cache_async(wxid)
                contacts = await gewe_service.get_contacts_brief_from_cache_async(wxid)
            
                contact_wxids = [c.get('userName') for c in contacts]
                chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])
                logger.info(f'{wxid} 定时群成员定时添好友任务开始')
                for chatroom_id in chatrooms:
                    chatroom = await gewe_service.get_group_info_from_cache_async(wxid, chatroom_id)
                    chatroom_member=await gewe_service.get_group_members_from_cache_async(wxid, chatroom_id)
                    
                    chatroom_nickname = chatroom.get('nickName')
                    chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)

                    admin_wxids = chatroom_member.get('adminWxid', []) 

                    admin_wxids = chatroom_member.get('adminWxid')
                    if admin_wxids is None:
                        admin_wxids = []  # 如果 admin_wxids 是 None，将其初始化为空列表
                    
                    # 判断当天群成员是否已经加了200个好友
                    is_add_group_200_times = await gewe_service.is_group_add_contacts_history_one_day_200_async(wxid, chatroom_id)
                    if is_add_group_200_times:
                        logger.info(f"{wxid}在 {chatroom_nickname} 群成员已经加了200个好友，不再添加，群id：{chatroom_id}")
                        continue

                    logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxids}')
                    contact_wxids_set = set(contact_wxids)
                    # for admin_wxid in admin_wxids:
                    #     contact_wxids_set.add(admin_wxid)
                    if admin_wxids:
                        contact_wxids_set.update(set(admin_wxids))

                    contact_wxids_set.update(set(admin_wxids))
                    if chatroom_owner_wxid is not None:
                        contact_wxids_set.add(chatroom_owner_wxid)
                    
                    contact_wxids_set.add(wxid)

                    # unavailable_wixds=await gewe_service.check_wixd_group_add_contacts_history_async(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:
                        contact_wxid= m.get('wxid')
                        member_nickname=m.get("nickName")
                        group_add_contacts_history = await gewe_service.get_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid)
                        sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)
                
                        # 已经邀请过两次，不再邀请
                        if len(sorted_history)==2:
                            logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')
                            continue

                        # 当天邀请过，不再邀请
                        if len(sorted_history) > 0:
                            last_add_time = sorted_history[0].addTime
                            def is_add_time_more_than_one_day(addTime: int) -> bool:
                                """
                                判断 addTime 是否与当前时间相隔大于 3600 × 24 秒
                                :param addTime: Unix 时间戳
                                :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False
                                """
                                # 获取当前时间的时间戳
                                current_time = time.time()
                                
                                # 计算时间戳差值
                                time_difference = abs(current_time - addTime)
                                
                                # 检查是否大于 3600 × 24 秒
                                return time_difference > 3600 * 24
                        
                            is_more_than_one_day= is_add_time_more_than_one_day(last_add_time)

                            if not is_more_than_one_day:
                                logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的{member_nickname}-{contact_wxid}已经当天邀请，不再邀请')
                                continue



                        ret, msg, data = await gewe_service.add_group_member_as_friend_async(token_id, app_id, chatroom_id, m.get('wxid'), f'我是群聊"{chatroom_nickname}"群的{nickname}')
                        # if ret==200:
                        #     history=AddGroupContactsHistory.model_validate({
                        #         "chatroomId":chatroom_id,
                        #         "wxid":wxid,
                        #         "contactWixd":contact_wxid,
                        #         "addTime":int(time.time())
                        #     })
                        #     await gewe_service.save_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid,history)
                        # else:
                        #     logger.info(f'群好友邀请失败原因：{data}')
                        if ret!=200:
                            logger.warning(f'群好友邀请失败原因：{ret} {data}')

                        history=AddGroupContactsHistory.model_validate({
                            "chatroomId":chatroom_id,
                            "wxid":wxid,
                            "contactWixd":contact_wxid,
                            "addTime":int(time.time())
                        })
                        await gewe_service.save_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid,history)

                        logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg}')
                        await asyncio.sleep(random.uniform(1.5, 3)) 
                    await asyncio.sleep(random.uniform(1.5, 3)) 


        except Exception as e:
            logger.error(f"任务执行过程中发生异常: {e}")
    
    loop = asyncio.get_event_loop()
    if loop.is_closed():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    loop.run_until_complete(task())  # 在现有事件循环中运行任务



 #@celery_app.task(name='tasks.scheduled_task_add_contacts_from_chatrooms_p2', bind=True, acks_late=True)
 def scheduled_task_add_contacts_from_chatrooms_p2(self, redis_config, kafka_config, gewe_config):

    '''
    关于群加好友的请求规则:一次30人,间隔2小时做1次,一天做3次,即最多90人/天。
    '''
    async def task():
        try:
            now = datetime.now()
            if now.hour < 8:
                logger.info(f"定时群成员定时添好友任务不启动,当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，早于8点")
                return
                        
            logger.info('定时群成员定时添好友任务开始')
            redis_service = RedisService()
            await redis_service.init(**redis_config)

            gewe_service = await GeWeService.get_instance(redis_service,gewe_config['api_url'])

            KAFKA_BOOTSTRAP_SERVERS=kafka_config['bootstrap_servers']
            KAFKA_TOPIC=kafka_config['topic']
            KAFKA_GROUP_ID=kafka_config['group_id']
            
            kafka_service= KafkaService(KAFKA_BOOTSTRAP_SERVERS, KAFKA_TOPIC, KAFKA_TOPIC,KAFKA_GROUP_ID)
            await kafka_service.start()

            login_keys = []
            async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):
                login_keys.append(key)

            wixd_add_contacts_from_chatrooms_times = {}
            #print(login_keys)
            for k in login_keys:
                r = await redis_service.get_hash(k)
                app_id = r.get("appId")
                token_id = r.get("tokenId")
                wxid = r.get("wxid")
                status = r.get('status')
                if status == '0':
                    logger.warning(f"微信号 {wxid} 已经离线，群成员不能定时添加")
                    continue

                c = await gewe_service.get_wxchat_config_from_cache_async(wxid)
                contacts = await gewe_service.get_contacts_brief_from_cache_async(wxid)
            
                contact_wxids = [c.get('userName') for c in contacts]
                chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])
                logger.info(f'{wxid} 定时群成员定时添好友任务开始')
                wixd_add_contacts_from_chatrooms_times[wxid] = 0
                for chatroom_id in chatrooms:
                    chatroom = await gewe_service.get_group_info_from_cache_async(wxid, chatroom_id)
                    chatroom_member=await gewe_service.get_group_members_from_cache_async(wxid, chatroom_id)
                    
                    chatroom_nickname = chatroom.get('nickName')
                    chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)

                    admin_wxids = chatroom_member.get('adminWxid', []) 

                    admin_wxids = chatroom_member.get('adminWxid')
                    if admin_wxids is None:
                        admin_wxids = []  # 如果 admin_wxids 是 None，将其初始化为空列表
                    
                    logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxids}')
                    contact_wxids_set = set(contact_wxids)
                    # for admin_wxid in admin_wxids:
                    #     contact_wxids_set.add(admin_wxid)
                    if admin_wxids:
                        contact_wxids_set.update(set(admin_wxids))
                    if chatroom_owner_wxid is not None:
                        contact_wxids_set.add(chatroom_owner_wxid)
                    
                    contact_wxids_set.add(wxid)

                    unavailable_wixds=await gewe_service.check_wixd_group_add_contacts_history_async(wxid,chatroom_id)
                    if unavailable_wixds:
                        contact_wxids_set.update(set(unavailable_wixds))
                        
                    chatroom_member_list = chatroom.get('memberList', [])
                    if chatroom_member_list is None:
                        chatroom_member_list = []  # 如果 memberList 是 None，将其初始化为空列表
                    elif not isinstance(chatroom_member_list, list):
                        chatroom_member_list = list(chatroom_member_list)  # 如果 memberList 不是列表，将其转换为列表

                    remaining_chatroot_members = [x for x in chatroom_member_list if x.get('wxid') not in contact_wxids_set]
                
                    nickname = next((member['nickName'] for member in chatroom_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:
                        # 判断本次任务是否已经邀请了30个好友
                        if wixd_add_contacts_from_chatrooms_times[wxid] == 30:
                            logger.info(f"{wxid} 本次任务已经邀请了30人,不再邀请")
                            return 
                        # 判断当天群成员是否已经加了90个好友
                        is_add_group_90_times = await gewe_service.is_group_add_contacts_history_one_day_90_async(wxid)
                        if is_add_group_90_times:
                            logger.info(f"当天 {wxid} 所有群的成员已经加了90个好友，不再添加")
                            return

                        contact_wxid= m.get('wxid')
                        member_nickname=m.get("nickName")
                        group_add_contacts_history = await gewe_service.get_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid)
                        sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)
                
                        # 已经邀请过两次，不再邀请
                        if len(sorted_history)==2:
                            logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')
                            continue

                        # 当天邀请过，不再邀请
                        if len(sorted_history) > 0:
                            last_add_time = sorted_history[0].addTime
                            def is_add_time_more_than_one_day(addTime: int) -> bool:
                                """
                                判断 addTime 是否与当前时间相隔大于 3600 × 24 秒
                                :param addTime: Unix 时间戳
                                :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False
                                """
                                # 获取当前时间的时间戳
                                current_time = time.time()
                                
                                # 计算时间戳差值
                                time_difference = abs(current_time - addTime)
                                
                                # 检查是否大于 3600 × 24 秒
                                return time_difference > 3600 * 24
                        
                            is_more_than_one_day= is_add_time_more_than_one_day(last_add_time)

                            if not is_more_than_one_day:
                                logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的{member_nickname}-{contact_wxid}已经当天邀请，不再邀请')
                                continue



                        ret, msg, data = await gewe_service.add_group_member_as_friend_async(token_id, app_id, chatroom_id, m.get('wxid'), f'我是群聊"{chatroom_nickname}"群的{nickname}')
                        if ret!=200:
                            logger.warning(f'群好友邀请失败原因：{ret} {data}')

                        history=AddGroupContactsHistory.model_validate({
                            "chatroomId":chatroom_id,
                            "wxid":wxid,
                            "contactWixd":contact_wxid,
                            "addTime":int(time.time())
                        })
                        await gewe_service.save_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid,history)
                        wixd_add_contacts_from_chatrooms_times[wxid]+=1
                        logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg}')
                        # 推送到kafka

                        k_message = wx_add_contacts_from_chatroom_message(history.wxid,history.chatroomId,history.contactWixd,history.addTime)
                        await kafka_service.send_message_async(k_message)

                        await asyncio.sleep(random.uniform(1.5, 3)) 
                    await asyncio.sleep(random.uniform(1.5, 3)) 


        except Exception as e:
            logger.error(f"任务执行过程中发生异常: {e}")
    
    loop = asyncio.get_event_loop()
    if loop.is_closed():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    loop.run_until_complete(task())  # 在现有事件循环中运行任务

@celery_app.task(name='tasks.scheduled_task_add_contacts_from_chatrooms', bind=True, acks_late=True)
 def scheduled_task_add_contacts_from_chatrooms(self, redis_config, kafka_config, gewe_config):

@@ -731,39 +413,41 @@ def scheduled_task_add_contacts_from_chatrooms(self, redis_config, kafka_config,
                        contact_wxid= m.get('wxid')
                        member_nickname=m.get("nickName")
                        group_add_contacts_history = await gewe_service.get_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid)
                        sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)
                
                        # 已经邀请过两次，不再邀请
                        if len(sorted_history)==2:
                            logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')
                            continue

                        # 当天邀请过，不再邀请
                        if len(sorted_history) > 0:
                            last_add_time = sorted_history[0].addTime
                            def is_add_time_more_than_one_day(addTime: int) -> bool:
                                """
                                判断 addTime 是否与当前时间相隔大于 3600 × 24 秒
                                :param addTime: Unix 时间戳
                                :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False
                                """
                                # 获取当前时间的时间戳
                                current_time = time.time()
                                
                                # 计算时间戳差值
                                time_difference = abs(current_time - addTime)
                                
                                # 检查是否大于 3600 × 24 秒
                                return time_difference > 3600 * 24
                        
                            is_more_than_one_day= is_add_time_more_than_one_day(last_add_time)

                            if not is_more_than_one_day:
                                logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经当天邀请，不再邀请')
                        if group_add_contacts_history:
                            sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)
                            
                            # 已经邀请过两次，不再邀请
                            if len(sorted_history)==2:
                                logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')
                                continue



                            # 当天邀请过，不再邀请
                            if len(sorted_history) > 0:
                                last_add_time = sorted_history[0].addTime
                                def is_add_time_more_than_one_day(addTime: int) -> bool:
                                    """
                                    判断 addTime 是否与当前时间相隔大于 3600 × 24 秒
                                    :param addTime: Unix 时间戳
                                    :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False
                                    """
                                    # 获取当前时间的时间戳
                                    current_time = time.time()
                                    
                                    # 计算时间戳差值
                                    time_difference = abs(current_time - addTime)
                                    
                                    # 检查是否大于 3600 × 24 秒
                                    return time_difference > 3600 * 24
                            
                                is_more_than_one_day= is_add_time_more_than_one_day(last_add_time)

                                if not is_more_than_one_day:
                                    logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经当天邀请，不再邀请')
                                    continue


                       
                        ret, msg, data = await gewe_service.add_group_member_as_friend_async(token_id, app_id, chatroom_id, m.get('wxid'), f'我是群聊"{chatroom_nickname}"群的{nickname}')
                        if ret!=200:
                            logger.warning(f'群好友邀请失败原因：{ret} {data}')