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}')
# 推送到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"