AiOpsWechat/tasks.py

from celery_app import celery_app
from fastapi import Request,FastAPI
import time,datetime
from celery import Celery
import celery.schedules
from redbeat import RedBeatSchedulerEntry
from datetime import timedelta

from services.redis_service import RedisService
from services.kafka_service import KafkaService
from services.gewe_service import GeWeService
# from common.log import logger
from common.utils import *
import asyncio,random
from model.models import AddGroupContactsHistory
import logging
from model.models import AgentConfig

import logging
import sys,traceback

logger = logging.getLogger('redbeat')


@celery_app.task(name='tasks.add_task', bind=True, acks_late=True)
def add_task(self, x, y):
    
    time.sleep(5)  # 模拟长时间计算
    logger.info('add')
    return x + y


@celery_app.task(name='tasks.mul_task', bind=True, acks_late=True)
def mul_task(self, x, y):
    time.sleep(5)  # 模拟长时间计算
    return x * y    


# @celery.task(name='app.tasks.sync_contacts', bind=True, acks_late=True)
# async def sync_contacts_task(self,app):
#     login_keys = list(await app.state.redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))
#     return login_keys
#     # for k in login_keys:
#     #     print(k)

@celery_app.task(name='tasks.sync_contacts', bind=True, acks_late=True)
async def sync_contacts_task(self, redis_service):
    # Use the redis_service passed as an argument
    login_keys = list(await redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))
    return login_keys


@celery_app.task(name='tasks.background_worker_task', bind=True, acks_late=True)
def background_worker_task(self, redis_config, kafka_config, gewe_config):
    async def task():
        redis_service = RedisService()
        await redis_service.init(**redis_config)
        login_keys = []
        async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):
            login_keys.append(key)
        print(login_keys)
    
    asyncio.run(task())

# @celery.task(name='tasks.background_worker_task', bind=True, acks_late=True)
# async def background_worker_task(self, redis_config, kafka_config, gewe_config):
#     # Initialize services inside the task
#     redis_service = RedisService()
#     await redis_service.init(**redis_config)

#     login_keys = []
#     async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):  # 使用 async for 遍历异步生成器
#         login_keys.append(key)

#     print(login_keys)

    # kafka_service = KafkaService(**kafka_config)
    # await kafka_service.start()

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

    # # Task logic
    # lock_name = "background_wxchat_thread_lock"
    # lock_identifier = str(time.time())
    # while True:
    #     if await redis_service.acquire_lock(lock_name, timeout=10):
    #         try:
    #             logger.info("分布式锁已成功获取")
    #             # Perform task logic
    #         finally:
    #             await redis_service.release_lock(lock_name, lock_identifier)
    #             break
    #     else:
    #         logger.info("获取分布式锁失败，等待10秒后重试...")
    #         await asyncio.sleep(10)



# @celery_app.task(name='tasks.scheduled_task', bind=True, acks_late=True)
# def scheduled_task(self):
#     print("定时任务执行成功！~~~~~~~~~~~~~~~~~")
#     return "Hello from Celery Beat + RedBeat!"


# @celery_app.task(name='tasks.scheduled_task_sync_wx', bind=True, acks_late=True)
# def scheduled_task_sync_wx(self,redis_service,kafka_service,gewe_service):
#     print("scheduled_task_sync_wx 定时任务执行成功！")
#     return "Hello from Celery Beat + RedBeat!"

# @celery_app.task(name='tasks.scheduled_task_sync_wx_info_1', bind=True, acks_late=True)
# def scheduled_task_sync_wx_info_1(self,redis_config, kafka_config, gewe_config):
#     '''
#     定时获取微信号资料
#     '''
#     loop = asyncio.new_event_loop()
#     asyncio.set_event_loop(loop)
#     async def task():
#         try:
#             redis_service = RedisService()
#             await redis_service.init(**redis_config)
#             # gewe_service = await GeWeService.get_instance(None, 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':
#             #         continue
#             #     ret, msg, profile = await gewe_service.get_profile_async(token_id, app_id)
#             #     if ret != 200:
#             #         logger.warning(f"同步微信号 {wxid} 资料失败: {ret}-{msg}")
#             #         continue
#             #     nickname=profile.get("nickName")
#             #     head_img_url=profile.get("smallHeadImgUrl")
#             #     r.update({"nickName":nickname,"headImgUrl":head_img_url,"modify_at":int(time.time())})
#             #     cleaned_login_info = {k: (v if v is not None else '') for k, v in r.items()}
#             #     await redis_service.set_hash(k, cleaned_login_info)
#             #     logger.info(f"同步微信号 {wxid} 资料 成功")
#                 # redis_service.update_hash_field(k,"nickName",nickname)
#                 # redis_service.update_hash_field(k,"headImgUrl",head_img_url)
#                 # redis_service.update_hash_field(k,"modify_at",int(time.time()))
#         except Exception as e:
#             logger.error(f"任务执行过程中发生异常: {e}")

#         print("scheduled_task_sync_wx_info 定时任务执行成功！")
#         return "Hello from Celery Beat + RedBeat!"
    
#     loop.run_until_complete(task())
#     loop.close()



# @celery_app.task(name='tasks.scheduled_task_sync_wx_info', bind=True, acks_late=True)
# def scheduled_task_sync_wx_info(self, redis_config, kafka_config, gewe_config):
#     '''
#     定时获取微信号资料
#     '''
#     async def task():
#         try:
#             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
#                 ret, msg, profile = await gewe_service.get_profile_async(token_id, app_id)
#                 if ret != 200:
#                     logger.warning(f"同步微信号 {wxid} 资料失败: {ret}-{msg}")
#                     continue
#                 nickname=profile.get("nickName")
#                 head_img_url=profile.get("smallHeadImgUrl")
#                 # print(nickname)

#                 nickname=profile.get("nickName")
#                 head_img_url=profile.get("smallHeadImgUrl")
#                 r.update({"nickName":nickname,"headImgUrl":head_img_url,"modify_at":int(time.time())})
#                 cleaned_login_info = {k: (v if v is not None else '') for k, v in r.items()}
#                 await redis_service.set_hash(k, cleaned_login_info)
#                 logger.info(f"定时同步微信号{wxid}-昵称{nickname} 资料成功")

#         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_sync_wx_info', bind=True, acks_late=True)
def scheduled_task_sync_wx_info(self, redis_config, kafka_config, gewe_config):
    '''
    定时获取微信号资料
    '''
    async def process_key(redis_service, gewe_service, semaphore, key):
        async with semaphore:  # 使用 Semaphore 控制并发
            try:
                r = await redis_service.get_hash(key)
                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} 已经离线")
                    return
                ret, msg, profile = await gewe_service.get_profile_async(token_id, app_id)
                if ret != 200:
                    logger.warning(f"同步微信号 {wxid} 资料失败: {ret}-{msg}")
                    return
                nickname = profile.get("nickName")
                head_img_url = profile.get("smallHeadImgUrl")
                r.update({"nickName": nickname, "headImgUrl": head_img_url, "modify_at": int(time.time())})
                cleaned_login_info = {k: (v if v is not None else '') for k, v in r.items()}
                await redis_service.set_hash(key, cleaned_login_info)
                logger.info(f"定时同步微信号{wxid}-昵称{nickname} 资料成功")
            except Exception as e:
                logger.error(f"处理键 {key} 时发生异常: {e}")

    async def task():
        try:
            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)

            # 设置 Semaphore，限制并发数为 10
            semaphore = asyncio.Semaphore(10)
            # 使用 asyncio.gather 并发处理所有键
            await asyncio.gather(*[process_key(redis_service, gewe_service, semaphore, key) for key in login_keys])
        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):

#     '''
#     关于群加好友的请求规则: 每个智能体一次最多30人,间隔2小时做1次,即最多90人/天。
#     加好友规则:每天处理次数、间隔时间(分钟)、每次加好友人数这3个参数都可以设置。目前默认只是上面的设置。
#     '''
#     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_producer()

#             global_config=await gewe_service.get_global_config_from_cache_async()
#             scheduled_task_add_contacts_from_chatrooms_config=global_config.get('scheduledTaskAddContactsFromChatrooms',{})
            
#             oneday_add_contacts_total=90
#             once_add_contacts_total=30
#             #oneday_times=3

#             if scheduled_task_add_contacts_from_chatrooms_config:
#                 oneday_add_contacts_total=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayAddContactsTotal',90)
#                 once_add_contacts_total=scheduled_task_add_contacts_from_chatrooms_config.get('onceAddContactsTotal',30)
#                 #oneday_times=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayTimes',3)

#             # cache_task_run_time_logs= await gewe_service.get_task_run_time_async('scheduled_task_add_contacts_from_chatrooms')
#             # if cache_task_run_time_logs:
#             #     sorted_tasks = sorted(cache_task_run_time_logs, key=lambda x: x.get("runTime"), reverse=True)
#             #     last_run_time=sorted_tasks[0].get("runTime")

#             #     if last_run_time > 1e12:  # 毫秒级时间戳
#             #         last_run_time = last_run_time / 1000  # 转换为秒
                
#             #     # 将时间戳转换为 datetime 对象
#             #     last_run_time = datetime.fromtimestamp(last_run_time)
                
#             #     # 获取当前时间
#             #     current_time = datetime.now()
                
#             #     # 计算时间差
#             #     time_difference = current_time - last_run_time
                
#             #     # 判断是否相差2小时
#             #     if time_difference < timedelta(hours=2):
#             #         logger.info(f"上次定时群成员定时添好友任务在2小时内，不再执行")
#             #         return 

#             # time_difference_seconds = today_seconds_remaining()
#             # cache_task_run_time_logs.append({"runTime":int(time.time())})
#             # await gewe_service.save_task_run_time_async('scheduled_task_add_contacts_from_chatrooms',cache_task_run_time_logs,time_difference_seconds)
            
#             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 = {}

#             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

#                 config=await gewe_service.get_wxchat_config_from_cache_async(wxid)
#                 validated_config = AgentConfig.model_validate(config)
#                 if not validated_config.agentEnabled:
#                     logger.warning(f"微信号 {wxid} 取消了托管，群成员不能定时添加")
#                     continue


#                 # 判断是否过于频繁
#                 is_wx_expection = await gewe_service.get_wx_expection_async(wxid,"addGroupMemberAsFriend")
#                 if is_wx_expection:
#                     logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常, {is_wx_expection}，本次群好友邀请任务未开始,跳过任务。")   
#                     continue
                

#                 cache_task_run_time_wxid_logs= await gewe_service.get_task_run_time_by_wxid_async(wxid,'scheduled_task_add_contacts_from_chatrooms')
#                 if cache_task_run_time_wxid_logs:
#                     sorted_tasks = sorted(cache_task_run_time_wxid_logs, key=lambda x: x.get("runTime"), reverse=True)
#                     last_run_time=sorted_tasks[0].get("runTime")

#                     if last_run_time > 1e12:  # 毫秒级时间戳
#                         last_run_time = last_run_time / 1000  # 转换为秒
                    
#                     # 将时间戳转换为 datetime 对象
#                     last_run_time = datetime.fromtimestamp(last_run_time)
                    
#                     # 获取当前时间
#                     current_time = datetime.now()
                    
#                     # 计算时间差
#                     time_difference = current_time - last_run_time
                    
#                     # 判断是否相差2小时
#                     if time_difference < timedelta(hours=2):
#                         logger.info(f"{wxid}上次定时群成员定时添好友任务在2小时内，不再执行")
#                         continue 


#                 cache_task_run_time_wxid_logs.append({"runTime":int(time.time())})
#                 await gewe_service.save_task_run_time_by_wxid_async(wxid,'scheduled_task_add_contacts_from_chatrooms',cache_task_run_time_wxid_logs,3600*2)
                


#                 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)
#                     # contact_wxids_set.update(set(unavailable_wixds))

#                     # chatroom_member_list = chatroom.get('memberList', [])


#                     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)

#                     if not remaining_chatroot_members:
#                         logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里没有好友可以邀请')
#                         # 任务状态推送到kafka 
#                         k_message=wx_add_contacts_from_chatroom_task_status_message(wxid,chatroom_id,2)
#                         await kafka_service.send_message_async(k_message)
#                         continue

#                     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] == once_add_contacts_total:
#                             logger.info(f"{wxid} 本次任务已经邀请了{once_add_contacts_total}人,不再邀请")
#                             continue 
#                         # 判断当天群成员是否已经加了90个好友
#                         is_add_group_times = await gewe_service.is_group_add_contacts_history_one_day_async(wxid,oneday_add_contacts_total)
#                         if is_add_group_times:
#                             logger.info(f"当天 {wxid} 所有群的成员已经加了{oneday_add_contacts_total}个好友，不再添加")
#                             continue
                        
#                         # 判断是否过于频繁
#                         is_wx_expection = await gewe_service.get_wx_expection_async(wxid,"addGroupMemberAsFriend")
#                         if is_wx_expection:
#                             logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}")   
#                             continue

#                         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)
                        
#                         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}')
#                             if msg in '操作过于频繁，请稍后再试。':
#                                 await gewe_service.save_wx_expection_async(wxid,"addGroupMemberAsFriend",msg,today_seconds_remaining())
#                                 logger.warning(f'{nickname}-{wxid} 操作过于频繁，本次群好友邀请任务未完成跳过。')
#                                 continue 

#                         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(30,60)) 

#                     # 任务状态推送到kafka 
#                     task_status=await gewe_service.wx_add_contacts_from_chatroom_task_status_async(wxid,chatroom_id)
#                     k_message=wx_add_contacts_from_chatroom_task_status_message(wxid,chatroom_id,task_status)
#                     await kafka_service.send_message_async(k_message)
#                     # 下一个群
#                     await asyncio.sleep(random.uniform(1.5, 3)) 
#         except Exception as e:
#             # 获取当前的堆栈跟踪
#             tb = sys.exc_info()[2]
#             # 为异常附加堆栈跟踪
#             e = e.with_traceback(tb)
#             # 输出详细的错误信息
#             logger.error(f"任务执行过程中发生异常: {e}\n异常类型: {type(e).__name__}\n异常信息: {str(e)}\n堆栈跟踪: {traceback.format_exc()}")
#         finally:
#             await kafka_service.stop_producer()
    
#     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):

    '''
    关于群加好友的请求规则: 每个智能体一次最多30人,间隔2小时做1次,即最多90人/天。
    加好友规则:每天处理次数、间隔时间(分钟)、每次加好友人数这3个参数都可以设置。目前默认只是上面的设置。
    '''
    async def process_login_key(redis_service:RedisService, gewe_service: GeWeService, kafka_service:KafkaService, k, gewe_config, oneday_add_contacts_total, once_add_contacts_total, semaphore):
        async with semaphore:  # 使用 Semaphore 控制并发
            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} 已经离线，群成员不能定时添加")
                return

            config = await gewe_service.get_wxchat_config_from_cache_async(wxid)
            validated_config = AgentConfig.model_validate(config)
            if not validated_config.agentEnabled:
                logger.warning(f"微信号 {wxid} 取消了托管，群成员不能定时添加")
                return

            # 判断是否过于频繁
            is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")
            if is_wx_expection:
                logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常, {is_wx_expection}，本次群好友邀请任务未开始,跳过任务。")
                return

            cache_task_run_time_wxid_logs = await gewe_service.get_task_run_time_by_wxid_async(wxid, 'scheduled_task_add_contacts_from_chatrooms')
            if cache_task_run_time_wxid_logs:
                sorted_tasks = sorted(cache_task_run_time_wxid_logs, key=lambda x: x.get("runTime"), reverse=True)
                last_run_time = sorted_tasks[0].get("runTime")
                if last_run_time > 1e12:  # 毫秒级时间戳
                    last_run_time = last_run_time / 1000  # 转换为秒

                # 将时间戳转换为 datetime 对象
                last_run_time = datetime.fromtimestamp(last_run_time)

                # 获取当前时间
                current_time = datetime.now()

                # 计算时间差
                time_difference = current_time - last_run_time

                # 判断是否相差2小时
                if time_difference < timedelta(hours=2):
                    logger.info(f"{wxid}上次定时群成员定时添好友任务在2小时内，不再执行")
                    return

            cache_task_run_time_wxid_logs.append({"runTime": int(time.time())})
            await gewe_service.save_task_run_time_by_wxid_async(wxid, 'scheduled_task_add_contacts_from_chatrooms', cache_task_run_time_wxid_logs, 3600 * 2)
            c:dict = 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)
                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)
                if not remaining_chatroot_members:
                    logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname}-{chatroom_id} 群里没有好友可以邀请')
                    # 任务状态推送到kafka
                    k_message = wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, 2)
                    await kafka_service.send_message_async(k_message)
                    continue 
                
                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] == once_add_contacts_total:
                        logger.info(f"{wxid} 本次任务已经邀请了{once_add_contacts_total}人,不再邀请")
                        return

                    # 判断当天群成员是否已经加了90个好友
                    is_add_group_times = await gewe_service.is_group_add_contacts_history_one_day_async(wxid, oneday_add_contacts_total)
                    if is_add_group_times:
                        logger.info(f"当天 {wxid} 所有群的成员已经加了{oneday_add_contacts_total}个好友，不再添加")
                        return

                    # 判断是否过于频繁
                    is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")
                    if is_wx_expection:
                        logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}")
                        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)

                    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
                        
                        # 24小时邀请过，不再邀请
                        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} 已经24小时邀请，不再邀请')
                                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}')
                    
                    history = AddGroupContactsHistory.model_validate({
                        "chatroomId": chatroom_id,
                        "wxid": wxid,
                        "contactWixd": contact_wxid,
                        "addTime": int(time.time())
                    }) 

                    if ret != 200:
                        logger.warning(f'{nickname}-{wxid} 在 {chatroom_nickname}-{chatroom_id} 群好友 {member_nickname}-{contact_wxid} 邀请失败原因：{ret} {msg} {data}')
                        if '操作过于频繁' in data.get('msg'):
                            await gewe_service.save_wx_expection_async(wxid, "addGroupMemberAsFriend", msg, today_seconds_remaining())
                            await gewe_service.save_group_add_contacts_history_async(wxid, chatroom_id, contact_wxid, history)
                            logger.warning(f'{nickname}-{wxid} 在 {chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 操作过于频繁，本次群好友邀请任务未完成跳过。当天不再处理该号群好友邀请任务')
                            return 
                        

                    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} \n {data} \n 当前已邀请好友数:{wixd_add_contacts_from_chatrooms_times[wxid]}')
                    # 推送到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(30, 60))
                    await asyncio.sleep(random.uniform(270,300))
                # 任务状态推送到kafka
                task_status = await gewe_service.wx_add_contacts_from_chatroom_task_status_async(wxid, chatroom_id)
                k_message=wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, task_status)
                await kafka_service.send_message_async(k_message)
                # 下一个群
                await asyncio.sleep(random.uniform(1.5, 3))

    async def task():
        try:
            now = datetime.now()
            # if 10> now.hour < 8:
            #     logger.info(f"定时群成员定时添好友任务不启动,当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，早于8点")
            #     return

            if now.hour < 8 or now.hour > 22:
                logger.info(f"定时群成员定时添好友任务不启动, 当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，不在8点到22点之间")
                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_producer()
            global_config = await gewe_service.get_global_config_from_cache_async()
            scheduled_task_add_contacts_from_chatrooms_config = global_config.get('scheduledTaskAddContactsFromChatrooms', {})

            oneday_add_contacts_total = 90
            once_add_contacts_total = 30
            # oneday_times=3
            if scheduled_task_add_contacts_from_chatrooms_config:
                oneday_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('oneDayAddContactsTotal', 90)
                once_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('onceAddContactsTotal', 30)
                # oneday_times=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayTimes',3)

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

            # 设置 Semaphore，限制并发任务数量
            semaphore = asyncio.Semaphore(10)  # 例如，限制同时运行的任务数量为 10
            # 使用 asyncio.gather 并发处理每个 login_key
            await asyncio.gather(*[process_login_key(redis_service, gewe_service, kafka_service, k, gewe_config, oneday_add_contacts_total, once_add_contacts_total, semaphore) for k in login_keys])

        except Exception as e:
            # 获取当前的堆栈跟踪
            tb = sys.exc_info()[2]
            # 为异常附加堆栈跟踪
            e = e.with_traceback(tb)
            # 输出详细的错误信息
            logger.error(f"任务执行过程中发生异常: {e}\n异常类型: {type(e).__name__}\n异常信息: {str(e)}\n堆栈跟踪: {traceback.format_exc()}")
        finally:
            await kafka_service.stop_producer()

    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):

#     '''
#     关于群加好友的请求规则: 每个智能体一次最多30人,间隔2小时做1次,即最多90人/天。
#     加好友规则:每天处理次数、间隔时间(分钟)、每次加好友人数这3个参数都可以设置。目前默认只是上面的设置。
#     '''
#     async def process_login_key(redis_service:RedisService, gewe_service: GeWeService, kafka_service:KafkaService, k, gewe_config, oneday_add_contacts_total, once_add_contacts_total):
#         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} 已经离线，群成员不能定时添加")
#             return

#         config = await gewe_service.get_wxchat_config_from_cache_async(wxid)
#         validated_config = AgentConfig.model_validate(config)
#         if not validated_config.agentEnabled:
#             logger.warning(f"微信号 {wxid} 取消了托管，群成员不能定时添加")
#             return

#         # 判断是否过于频繁
#         is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")
#         if is_wx_expection:
#             logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常, {is_wx_expection}，本次群好友邀请任务未开始,跳过任务。")
#             return

#         cache_task_run_time_wxid_logs = await gewe_service.get_task_run_time_by_wxid_async(wxid, 'scheduled_task_add_contacts_from_chatrooms')
#         if cache_task_run_time_wxid_logs:
#             sorted_tasks = sorted(cache_task_run_time_wxid_logs, key=lambda x: x.get("runTime"), reverse=True)
#             last_run_time = sorted_tasks[0].get("runTime")
#             if last_run_time > 1e12:  # 毫秒级时间戳
#                 last_run_time = last_run_time / 1000  # 转换为秒

#             # 将时间戳转换为 datetime 对象
#             last_run_time = datetime.fromtimestamp(last_run_time)

#             # 获取当前时间
#             current_time = datetime.now()

#             # 计算时间差
#             time_difference = current_time - last_run_time

#             # 判断是否相差2小时
#             if time_difference < timedelta(hours=2):
#                 logger.info(f"{wxid}上次定时群成员定时添好友任务在2小时内，不再执行")
#                 return

#         cache_task_run_time_wxid_logs.append({"runTime": int(time.time())})
#         await gewe_service.save_task_run_time_by_wxid_async(wxid, 'scheduled_task_add_contacts_from_chatrooms', cache_task_run_time_wxid_logs, 3600 * 2)
#         c:dict = 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)
#             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)
#             if not remaining_chatroot_members:
#                 logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里没有好友可以邀请')
#                 # 任务状态推送到kafka
#                 k_message = wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, 2)
#                 await kafka_service.send_message_async(k_message)
#                 return 
            
#             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] == once_add_contacts_total:
#                     logger.info(f"{wxid} 本次任务已经邀请了{once_add_contacts_total}人,不再邀请")
#                     return

#                 # 判断当天群成员是否已经加了90个好友
#                 is_add_group_times = await gewe_service.is_group_add_contacts_history_one_day_async(wxid, oneday_add_contacts_total)
#                 if is_add_group_times:
#                     logger.info(f"当天 {wxid} 所有群的成员已经加了{oneday_add_contacts_total}个好友，不再添加")
#                     return

#                 # 判断是否过于频繁
#                 is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")
#                 if is_wx_expection:
#                     logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}")
#                     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)

#                 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次，不再邀请')
#                         return
                    
#                     # 当天邀请过，不再邀请
#                     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} 已经当天邀请，不再邀请')
#                             return 

#                 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} {msg} {data}')
#                     if '操作过于频繁' in data.get('msg'):
#                         await gewe_service.save_wx_expection_async(wxid, "addGroupMemberAsFriend", msg, today_seconds_remaining())
#                         logger.warning(f'{nickname}-{wxid} 操作过于频繁，本次群好友邀请任务未完成跳过。')
#                         return 
                    
#                 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(30, 60))
#             # 任务状态推送到kafka
#             task_status = await gewe_service.wx_add_contacts_from_chatroom_task_status_async(wxid, chatroom_id)
#             wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, task_status)
#             await kafka_service.send_message_async(k_message)
#             # 下一个群
#             await asyncio.sleep(random.uniform(1.5, 3))

#     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_producer()
#             global_config = await gewe_service.get_global_config_from_cache_async()
#             scheduled_task_add_contacts_from_chatrooms_config = global_config.get('scheduledTaskAddContactsFromChatrooms', {})

#             oneday_add_contacts_total = 90
#             once_add_contacts_total = 30
#             # oneday_times=3
#             if scheduled_task_add_contacts_from_chatrooms_config:
#                 oneday_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('oneDayAddContactsTotal', 90)
#                 once_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('onceAddContactsTotal', 30)
#                 # oneday_times=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayTimes',3)

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

#             # 使用 asyncio.gather 并发处理每个 login_key
#             await asyncio.gather(*[process_login_key(redis_service, gewe_service, kafka_service, k, gewe_config, oneday_add_contacts_total, once_add_contacts_total) for k in login_keys])

#         except Exception as e:
#             # 获取当前的堆栈跟踪
#             tb = sys.exc_info()[2]
#             # 为异常附加堆栈跟踪
#             e = e.with_traceback(tb)
#             # 输出详细的错误信息
#             logger.error(f"任务执行过程中发生异常: {e}\n异常类型: {type(e).__name__}\n异常信息: {str(e)}\n堆栈跟踪: {traceback.format_exc()}")
#         finally:
#             await kafka_service.stop_producer()

#     loop = asyncio.get_event_loop()
#     if loop.is_closed():
#         loop = asyncio.new_event_loop()
#         asyncio.set_event_loop(loop)
#     loop.run_until_complete(task())  # 在现有事件循环中运行任务






REDIS_KEY_PATTERN = "friend_add_limit:{date}"
REDIS_LAST_RUN_KEY = "last_run_time:add_friends_task"

@celery_app.task(name='tasks.add_friends_task', bind=True, acks_late=True)
def add_friends_task(self,redis_config):
    """
    限制每天最多 15 个，每 2 小时最多 8 个
    """
    async def task():
        redis_service = RedisService()
        await redis_service.init(**redis_config)
        today_str = datetime.now().strftime("%Y%m%d")
        redis_key = REDIS_KEY_PATTERN.format(date=today_str)

        # 获取当前总添加数量
        total_added = await redis_service.get_hash_field(redis_key, "total") or 0
        last_2h_added =await redis_service.get_hash_field(redis_key, "last_2h") or 0

        total_added = int(total_added)
        last_2h_added = int(last_2h_added)

        logger.info(f"当前添加好友总数: {total_added}, 过去2小时添加: {last_2h_added}")

        # 判断是否超过限制
        if total_added >= 15:
            logger.warning("今日好友添加已达上限！")
            return

        if last_2h_added >= 8:
            logger.warning("过去2小时添加已达上限！")
            return

        # 计算本次要添加的好友数量 (控制每天 5-15 个)
        max_add = min(15 - total_added, 8 - last_2h_added)
        if max_add <= 0:
            return

        num_to_add = min(max_add, 1)  # 每次最多加 1 个
        logger.info(f"本次添加 {num_to_add} 位好友")

        # TODO: 调用好友添加逻辑 (接口 or 业务逻辑)
        # success = add_friends(num_to_add)

        success = num_to_add  # 假设成功添加 num_to_add 个

        # 更新 Redis 计数
        if success > 0:
            await redis_service.increment_hash_field(redis_key, "total", success)
            await redis_service.increment_hash_field(redis_key, "last_2h", success)

            # 设置 Redis 过期时间 (每日记录存 1 天, 2 小时记录存 2 小时)
            await redis_service.expire(redis_key, 86400)  # 24小时
            await redis_service.expire_field(redis_key, "last_2h", 7200)  # 2小时

        logger.info(f"成功添加 {success} 位好友, 今日总数 {total_added + success}")

        # 生成一个新的随机时间（5-15 分钟之间）
        # next_interval = random.randint(10, 20)

        # # 计算新的执行时间
        # next_run_time = datetime.datetime.now() + timedelta(seconds=next_interval)

        # # 重新注册 RedBeat 任务，确保下次执行时间不同
        # redbeat_entry = RedBeatSchedulerEntry(
        #     name="redbeat:add_friends_task",
        #     task="tasks.add_friends_task",
        #     schedule=celery.schedules.schedule(timedelta(seconds=next_interval)),
        #     args=[redis_config],
        #     app=celery_app
        # )

        # # 设置任务的下次执行时间
        # redbeat_entry.last_run_at = next_run_time
        # redbeat_entry.save()

        # logger.info(f"下次任务将在 {next_run_time} 执行（间隔 {next_interval} 秒）")

    loop = asyncio.get_event_loop()
    
    if loop.is_closed():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

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

def today_seconds_remaining()->int:
    current_time = datetime.now()

    # 计算当天的结束时间（23:59:59）
    end_of_day = datetime(current_time.year, current_time.month, current_time.day, 23, 59, 59)

    # 计算时间差
    time_difference = end_of_day - current_time

    # 将时间差转换为秒数
    time_difference_seconds = int(time_difference.total_seconds())

    return time_difference_seconds


@celery_app.task(name='tasks.random_scheduled_task', bind=True, acks_late=True)
def random_scheduled_task(self,):
    print(f"Task executed at {datetime.now()}")
    # 随机生成下次执行时间（例如：10-60秒内的随机时间）
    next_run_in = random.randint(10, 60)
    print(f"Next execution will be in {next_run_in} seconds")
    
    # 设置下次执行时间
    entry = RedBeatSchedulerEntry(
        name='random-task',
        task='tasks.random_scheduled_task',
        schedule=timedelta(seconds=next_run_in),
        app=celery_app
    )
    entry.save()
    return f"Scheduled next run in {next_run_in} seconds"