J1147
/
AiOpsWechat
Следить 1
В избранное 0
Форкнуть 0
Код Задачи 0 Pull Request'ы 0 Релизы 0 Вики Активность
Вы не можете выбрать более 25 тем Темы должны начинаться с буквы или цифры, могут содержать дефисы(-) и должны содержать не более 35 символов.
103 коммитов
5 Ветки
2.1MB
Дерево: 53205d17cc
Ветки Теги
${ item.name }
Создать ветку ${ searchTerm }
из '53205d17cc'
${ noResults }
AiOpsWechat/tasks.py

1150 lines
60KB
Исходник Blame История 

  1. from celery_app import celery_app

  2. from fastapi import Request,FastAPI

  3. import time,datetime

  4. from celery import Celery

  5. import celery.schedules

  6. from redbeat import RedBeatSchedulerEntry

  7. from datetime import timedelta

  8. 

  9. from services.redis_service import RedisService

  10. from services.kafka_service import KafkaService

  11. from services.gewe_service import GeWeService

  12. # from common.log import logger

  13. from common.utils import *

  14. import asyncio,random

  15. from model.models import AddGroupContactsHistory

  16. import logging

  17. from model.models import AgentConfig

  18. 

  19. import logging

  20. import sys,traceback

  21. 

  22. logger = logging.getLogger('redbeat')

  23. 

  24. 

  25. @celery_app.task(name='tasks.add_task', bind=True, acks_late=True)

  26. def add_task(self, x, y):

  27.     

  28.     time.sleep(5)  # 模拟长时间计算

  29.     logger.info('add')

  30.     return x + y

  31. 

  32. 

  33. @celery_app.task(name='tasks.mul_task', bind=True, acks_late=True)

  34. def mul_task(self, x, y):

  35.     time.sleep(5)  # 模拟长时间计算

  36.     return x * y    

  37. 

  38. 

  39. # @celery.task(name='app.tasks.sync_contacts', bind=True, acks_late=True)

  40. # async def sync_contacts_task(self,app):

  41. #     login_keys = list(await app.state.redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))

  42. #     return login_keys

  43. #     # for k in login_keys:

  44. #     #     print(k)

  45. 

  46. @celery_app.task(name='tasks.sync_contacts', bind=True, acks_late=True)

  47. async def sync_contacts_task(self, redis_service):

  48.     # Use the redis_service passed as an argument

  49.     login_keys = list(await redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'))

  50.     return login_keys

  51. 

  52. 

  53. @celery_app.task(name='tasks.background_worker_task', bind=True, acks_late=True)

  54. def background_worker_task(self, redis_config, kafka_config, gewe_config):

  55.     async def task():

  56.         redis_service = RedisService()

  57.         await redis_service.init(**redis_config)

  58.         login_keys = []

  59.         async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  60.             login_keys.append(key)

  61.         print(login_keys)

  62.     

  63.     asyncio.run(task())

  64. 

  65. # @celery.task(name='tasks.background_worker_task', bind=True, acks_late=True)

  66. # async def background_worker_task(self, redis_config, kafka_config, gewe_config):

  67. #     # Initialize services inside the task

  68. #     redis_service = RedisService()

  69. #     await redis_service.init(**redis_config)

  70. 

  71. #     login_keys = []

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

  73. #         login_keys.append(key)

  74. 

  75. #     print(login_keys)

  76. 

  77.     # kafka_service = KafkaService(**kafka_config)

  78.     # await kafka_service.start()

  79. 

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

  81. 

  82.     # # Task logic

  83.     # lock_name = "background_wxchat_thread_lock"

  84.     # lock_identifier = str(time.time())

  85.     # while True:

  86.     #     if await redis_service.acquire_lock(lock_name, timeout=10):

  87.     #         try:

  88.     #             logger.info("分布式锁已成功获取")

  89.     #             # Perform task logic

  90.     #         finally:

  91.     #             await redis_service.release_lock(lock_name, lock_identifier)

  92.     #             break

  93.     #     else:

  94.     #         logger.info("获取分布式锁失败，等待10秒后重试...")

  95.     #         await asyncio.sleep(10)

  96. 

  97. 

  98. 

  99. # @celery_app.task(name='tasks.scheduled_task', bind=True, acks_late=True)

  100. # def scheduled_task(self):

  101. #     print("定时任务执行成功！~~~~~~~~~~~~~~~~~")

  102. #     return "Hello from Celery Beat + RedBeat!"

  103. 

  104. 

  105. # @celery_app.task(name='tasks.scheduled_task_sync_wx', bind=True, acks_late=True)

  106. # def scheduled_task_sync_wx(self,redis_service,kafka_service,gewe_service):

  107. #     print("scheduled_task_sync_wx 定时任务执行成功！")

  108. #     return "Hello from Celery Beat + RedBeat!"

  109. 

  110. # @celery_app.task(name='tasks.scheduled_task_sync_wx_info_1', bind=True, acks_late=True)

  111. # def scheduled_task_sync_wx_info_1(self,redis_config, kafka_config, gewe_config):

  112. #     '''

  113. #     定时获取微信号资料

  114. #     '''

  115. #     loop = asyncio.new_event_loop()

  116. #     asyncio.set_event_loop(loop)

  117. #     async def task():

  118. #         try:

  119. #             redis_service = RedisService()

  120. #             await redis_service.init(**redis_config)

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

  122. #             login_keys = []

  123. #             async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  124. #                 login_keys.append(key)

  125. 

  126. #             print(login_keys)

  127. #             # for k in login_keys:

  128. #             #     r = await redis_service.get_hash(k)

  129. #             #     app_id = r.get("appId")

  130. #             #     token_id = r.get("tokenId")

  131. #             #     wxid = r.get("wxid")

  132. #             #     status = r.get('status')

  133. #             #     if status == '0':

  134. #             #         continue

  135. #             #     ret, msg, profile = await gewe_service.get_profile_async(token_id, app_id)

  136. #             #     if ret != 200:

  137. #             #         logger.warning(f"同步微信号 {wxid} 资料失败: {ret}-{msg}")

  138. #             #         continue

  139. #             #     nickname=profile.get("nickName")

  140. #             #     head_img_url=profile.get("smallHeadImgUrl")

  141. #             #     r.update({"nickName":nickname,"headImgUrl":head_img_url,"modify_at":int(time.time())})

  142. #             #     cleaned_login_info = {k: (v if v is not None else '') for k, v in r.items()}

  143. #             #     await redis_service.set_hash(k, cleaned_login_info)

  144. #             #     logger.info(f"同步微信号 {wxid} 资料 成功")

  145. #                 # redis_service.update_hash_field(k,"nickName",nickname)

  146. #                 # redis_service.update_hash_field(k,"headImgUrl",head_img_url)

  147. #                 # redis_service.update_hash_field(k,"modify_at",int(time.time()))

  148. #         except Exception as e:

  149. #             logger.error(f"任务执行过程中发生异常: {e}")

  150. 

  151. #         print("scheduled_task_sync_wx_info 定时任务执行成功！")

  152. #         return "Hello from Celery Beat + RedBeat!"

  153.     

  154. #     loop.run_until_complete(task())

  155. #     loop.close()

  156. 

  157. 

  158. 

  159. # @celery_app.task(name='tasks.scheduled_task_sync_wx_info', bind=True, acks_late=True)

  160. # def scheduled_task_sync_wx_info(self, redis_config, kafka_config, gewe_config):

  161. #     '''

  162. #     定时获取微信号资料

  163. #     '''

  164. #     async def task():

  165. #         try:

  166. #             redis_service = RedisService()

  167. #             await redis_service.init(**redis_config)

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

  169. 

  170. #             login_keys = []

  171. #             async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  172. #                 login_keys.append(key)

  173. 

  174. #             #print(login_keys)

  175. #             for k in login_keys:

  176. #                 r = await redis_service.get_hash(k)

  177. #                 app_id = r.get("appId")

  178. #                 token_id = r.get("tokenId")

  179. #                 wxid = r.get("wxid")

  180. #                 status = r.get('status')

  181. #                 if status == '0':

  182. #                     logger.warning(f"微信号 {wxid} 已经离线")

  183. #                     continue

  184. #                 ret, msg, profile = await gewe_service.get_profile_async(token_id, app_id)

  185. #                 if ret != 200:

  186. #                     logger.warning(f"同步微信号 {wxid} 资料失败: {ret}-{msg}")

  187. #                     continue

  188. #                 nickname=profile.get("nickName")

  189. #                 head_img_url=profile.get("smallHeadImgUrl")

  190. #                 # print(nickname)

  191. 

  192. #                 nickname=profile.get("nickName")

  193. #                 head_img_url=profile.get("smallHeadImgUrl")

  194. #                 r.update({"nickName":nickname,"headImgUrl":head_img_url,"modify_at":int(time.time())})

  195. #                 cleaned_login_info = {k: (v if v is not None else '') for k, v in r.items()}

  196. #                 await redis_service.set_hash(k, cleaned_login_info)

  197. #                 logger.info(f"定时同步微信号{wxid}-昵称{nickname} 资料成功")

  198. 

  199. #         except Exception as e:

  200. #             logger.error(f"任务执行过程中发生异常: {e}")

  201. 

  202. #     loop = asyncio.get_event_loop()

  203. #     if loop.is_closed():

  204. #         loop = asyncio.new_event_loop()

  205. #         asyncio.set_event_loop(loop)

  206. 

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

  208. 

  209. 

  210. @celery_app.task(name='tasks.scheduled_task_sync_wx_info', bind=True, acks_late=True)

  211. def scheduled_task_sync_wx_info(self, redis_config, kafka_config, gewe_config):

  212.     '''

  213.     定时获取微信号资料

  214.     '''

  215.     async def process_key(redis_service, gewe_service, semaphore, key):

  216.         async with semaphore:  # 使用 Semaphore 控制并发

  217.             try:

  218.                 r = await redis_service.get_hash(key)

  219.                 app_id = r.get("appId")

  220.                 token_id = r.get("tokenId")

  221.                 wxid = r.get("wxid")

  222.                 status = r.get('status')

  223.                 if status == '0':

  224.                     logger.warning(f"微信号 {wxid} 已经离线")

  225.                     return

  226.                 ret, msg, profile = await gewe_service.get_profile_async(token_id, app_id)

  227.                 if ret != 200:

  228.                     logger.warning(f"同步微信号 {wxid} 资料失败: {ret}-{msg}")

  229.                     return

  230.                 nickname = profile.get("nickName")

  231.                 head_img_url = profile.get("smallHeadImgUrl")

  232.                 r.update({"nickName": nickname, "headImgUrl": head_img_url, "modify_at": int(time.time())})

  233.                 cleaned_login_info = {k: (v if v is not None else '') for k, v in r.items()}

  234.                 await redis_service.set_hash(key, cleaned_login_info)

  235.                 logger.info(f"定时同步微信号{wxid}-昵称{nickname} 资料成功")

  236.             except Exception as e:

  237.                 logger.error(f"处理键 {key} 时发生异常: {e}")

  238. 

  239.     async def task():

  240.         try:

  241.             redis_service = RedisService()

  242.             await redis_service.init(**redis_config)

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

  244.             login_keys = []

  245.             async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  246.                 login_keys.append(key)

  247. 

  248.             # 设置 Semaphore，限制并发数为 10

  249.             semaphore = asyncio.Semaphore(10)

  250.             # 使用 asyncio.gather 并发处理所有键

  251.             await asyncio.gather(*[process_key(redis_service, gewe_service, semaphore, key) for key in login_keys])

  252.         except Exception as e:

  253.             logger.error(f"任务执行过程中发生异常: {e}")

  254. 

  255.     loop = asyncio.get_event_loop()

  256.     if loop.is_closed():

  257.         loop = asyncio.new_event_loop()

  258.         asyncio.set_event_loop(loop)

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

  260. 

  261. 

  262. 

  263. 

  264. # @celery_app.task(name='tasks.scheduled_task_add_contacts_from_chatrooms', bind=True, acks_late=True)

  265. # def scheduled_task_add_contacts_from_chatrooms(self, redis_config, kafka_config, gewe_config):

  266. 

  267. #     '''

  268. #     关于群加好友的请求规则: 每个智能体一次最多30人,间隔2小时做1次,即最多90人/天。

  269. #     加好友规则:每天处理次数、间隔时间(分钟)、每次加好友人数这3个参数都可以设置。目前默认只是上面的设置。

  270. #     '''

  271. #     async def task():

  272. #         try:

  273. #             now = datetime.now()

  274. #             if now.hour < 8:

  275. #                 logger.info(f"定时群成员定时添好友任务不启动,当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，早于8点")

  276. #                 return

  277. 

  278. #             logger.info('定时群成员定时添好友任务开始')

  279. #             redis_service = RedisService()

  280. #             await redis_service.init(**redis_config)

  281. 

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

  283. 

  284. #             KAFKA_BOOTSTRAP_SERVERS=kafka_config['bootstrap_servers']

  285. #             KAFKA_TOPIC=kafka_config['topic']

  286. #             KAFKA_GROUP_ID=kafka_config['group_id']

  287.             

  288. #             kafka_service= KafkaService(KAFKA_BOOTSTRAP_SERVERS, KAFKA_TOPIC, KAFKA_TOPIC,KAFKA_GROUP_ID)

  289. #             await kafka_service.start_producer()

  290. 

  291. #             global_config=await gewe_service.get_global_config_from_cache_async()

  292. #             scheduled_task_add_contacts_from_chatrooms_config=global_config.get('scheduledTaskAddContactsFromChatrooms',{})

  293.             

  294. #             oneday_add_contacts_total=90

  295. #             once_add_contacts_total=30

  296. #             #oneday_times=3

  297. 

  298. #             if scheduled_task_add_contacts_from_chatrooms_config:

  299. #                 oneday_add_contacts_total=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayAddContactsTotal',90)

  300. #                 once_add_contacts_total=scheduled_task_add_contacts_from_chatrooms_config.get('onceAddContactsTotal',30)

  301. #                 #oneday_times=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayTimes',3)

  302. 

  303. #             # cache_task_run_time_logs= await gewe_service.get_task_run_time_async('scheduled_task_add_contacts_from_chatrooms')

  304. #             # if cache_task_run_time_logs:

  305. #             #     sorted_tasks = sorted(cache_task_run_time_logs, key=lambda x: x.get("runTime"), reverse=True)

  306. #             #     last_run_time=sorted_tasks[0].get("runTime")

  307. 

  308. #             #     if last_run_time > 1e12:  # 毫秒级时间戳

  309. #             #         last_run_time = last_run_time / 1000  # 转换为秒

  310.                 

  311. #             #     # 将时间戳转换为 datetime 对象

  312. #             #     last_run_time = datetime.fromtimestamp(last_run_time)

  313.                 

  314. #             #     # 获取当前时间

  315. #             #     current_time = datetime.now()

  316.                 

  317. #             #     # 计算时间差

  318. #             #     time_difference = current_time - last_run_time

  319.                 

  320. #             #     # 判断是否相差2小时

  321. #             #     if time_difference < timedelta(hours=2):

  322. #             #         logger.info(f"上次定时群成员定时添好友任务在2小时内，不再执行")

  323. #             #         return 

  324. 

  325. #             # time_difference_seconds = today_seconds_remaining()

  326. #             # cache_task_run_time_logs.append({"runTime":int(time.time())})

  327. #             # await gewe_service.save_task_run_time_async('scheduled_task_add_contacts_from_chatrooms',cache_task_run_time_logs,time_difference_seconds)

  328.             

  329. #             login_keys = []

  330. #             async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  331. #                 login_keys.append(key)

  332. 

  333. #             wixd_add_contacts_from_chatrooms_times = {}

  334. 

  335. #             for k in login_keys:

  336. #                 r = await redis_service.get_hash(k)

  337. #                 app_id = r.get("appId")

  338. #                 token_id = r.get("tokenId")

  339. #                 wxid = r.get("wxid")

  340. #                 status = r.get('status')

  341. #                 if status == '0':

  342. #                     logger.warning(f"微信号 {wxid} 已经离线，群成员不能定时添加")

  343. #                     continue

  344. 

  345. #                 config=await gewe_service.get_wxchat_config_from_cache_async(wxid)

  346. #                 validated_config = AgentConfig.model_validate(config)

  347. #                 if not validated_config.agentEnabled:

  348. #                     logger.warning(f"微信号 {wxid} 取消了托管，群成员不能定时添加")

  349. #                     continue

  350. 

  351. 

  352. #                 # 判断是否过于频繁

  353. #                 is_wx_expection = await gewe_service.get_wx_expection_async(wxid,"addGroupMemberAsFriend")

  354. #                 if is_wx_expection:

  355. #                     logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常, {is_wx_expection}，本次群好友邀请任务未开始,跳过任务。")   

  356. #                     continue

  357.                 

  358. 

  359. #                 cache_task_run_time_wxid_logs= await gewe_service.get_task_run_time_by_wxid_async(wxid,'scheduled_task_add_contacts_from_chatrooms')

  360. #                 if cache_task_run_time_wxid_logs:

  361. #                     sorted_tasks = sorted(cache_task_run_time_wxid_logs, key=lambda x: x.get("runTime"), reverse=True)

  362. #                     last_run_time=sorted_tasks[0].get("runTime")

  363. 

  364. #                     if last_run_time > 1e12:  # 毫秒级时间戳

  365. #                         last_run_time = last_run_time / 1000  # 转换为秒

  366.                     

  367. #                     # 将时间戳转换为 datetime 对象

  368. #                     last_run_time = datetime.fromtimestamp(last_run_time)

  369.                     

  370. #                     # 获取当前时间

  371. #                     current_time = datetime.now()

  372.                     

  373. #                     # 计算时间差

  374. #                     time_difference = current_time - last_run_time

  375.                     

  376. #                     # 判断是否相差2小时

  377. #                     if time_difference < timedelta(hours=2):

  378. #                         logger.info(f"{wxid}上次定时群成员定时添好友任务在2小时内，不再执行")

  379. #                         continue 

  380. 

  381. 

  382. #                 cache_task_run_time_wxid_logs.append({"runTime":int(time.time())})

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

  384.                 

  385. 

  386. 

  387. #                 c = await gewe_service.get_wxchat_config_from_cache_async(wxid)

  388. #                 contacts = await gewe_service.get_contacts_brief_from_cache_async(wxid)

  389.             

  390. #                 contact_wxids = [c.get('userName') for c in contacts]

  391. #                 chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])

  392. #                 logger.info(f'{wxid} 定时群成员定时添好友任务开始')

  393. #                 wixd_add_contacts_from_chatrooms_times[wxid] = 0

  394. #                 for chatroom_id in chatrooms:

  395. #                     chatroom = await gewe_service.get_group_info_from_cache_async(wxid, chatroom_id)

  396. #                     chatroom_member=await gewe_service.get_group_members_from_cache_async(wxid, chatroom_id)

  397.                     

  398. #                     chatroom_nickname = chatroom.get('nickName')

  399. #                     chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)

  400. 

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

  402. 

  403. #                     admin_wxids = chatroom_member.get('adminWxid')

  404. #                     if admin_wxids is None:

  405. #                         admin_wxids = []  # 如果 admin_wxids 是 None，将其初始化为空列表

  406.                     

  407. #                     logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxids}')

  408. #                     contact_wxids_set = set(contact_wxids)

  409. #                     # for admin_wxid in admin_wxids:

  410. #                     #     contact_wxids_set.add(admin_wxid)

  411. #                     if admin_wxids:

  412. #                         contact_wxids_set.update(set(admin_wxids))

  413. #                     if chatroom_owner_wxid is not None:

  414. #                         contact_wxids_set.add(chatroom_owner_wxid)

  415.                     

  416. #                     contact_wxids_set.add(wxid)

  417. 

  418. #                     # unavailable_wixds=await gewe_service.check_wixd_group_add_contacts_history_async(wxid,chatroom_id)

  419. #                     # contact_wxids_set.update(set(unavailable_wixds))

  420. 

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

  422. 

  423. 

  424. #                     unavailable_wixds=await gewe_service.check_wixd_group_add_contacts_history_async(wxid,chatroom_id)

  425. #                     if unavailable_wixds:

  426. #                         contact_wxids_set.update(set(unavailable_wixds))

  427.                         

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

  429. #                     if chatroom_member_list is None:

  430. #                         chatroom_member_list = []  # 如果 memberList 是 None，将其初始化为空列表

  431. #                     elif not isinstance(chatroom_member_list, list):

  432. #                         chatroom_member_list = list(chatroom_member_list)  # 如果 memberList 不是列表，将其转换为列表                    

  433. 

  434. 

  435. #                     remaining_chatroot_members = [x for x in chatroom_member_list if x.get('wxid') not in contact_wxids_set]

  436.                 

  437. #                     nickname = next((member['nickName'] for member in chatroom_member_list if member['wxid'] == wxid), None)

  438. 

  439. #                     if not remaining_chatroot_members:

  440. #                         logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里没有好友可以邀请')

  441. #                         # 任务状态推送到kafka 

  442. #                         k_message=wx_add_contacts_from_chatroom_task_status_message(wxid,chatroom_id,2)

  443. #                         await kafka_service.send_message_async(k_message)

  444. #                         continue

  445. 

  446. #                     logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里还可以邀请的好友有：{[x.get("nickName") for x in remaining_chatroot_members]}')

  447. #                     for m in remaining_chatroot_members:

  448. #                         # 判断本次任务是否已经邀请了30个好友

  449. #                         if wixd_add_contacts_from_chatrooms_times[wxid] == once_add_contacts_total:

  450. #                             logger.info(f"{wxid} 本次任务已经邀请了{once_add_contacts_total}人,不再邀请")

  451. #                             continue 

  452. #                         # 判断当天群成员是否已经加了90个好友

  453. #                         is_add_group_times = await gewe_service.is_group_add_contacts_history_one_day_async(wxid,oneday_add_contacts_total)

  454. #                         if is_add_group_times:

  455. #                             logger.info(f"当天 {wxid} 所有群的成员已经加了{oneday_add_contacts_total}个好友，不再添加")

  456. #                             continue

  457.                         

  458. #                         # 判断是否过于频繁

  459. #                         is_wx_expection = await gewe_service.get_wx_expection_async(wxid,"addGroupMemberAsFriend")

  460. #                         if is_wx_expection:

  461. #                             logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}")   

  462. #                             continue

  463. 

  464. #                         contact_wxid= m.get('wxid')

  465. #                         member_nickname=m.get("nickName")

  466. #                         group_add_contacts_history = await gewe_service.get_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid)

  467.                         

  468. #                         if group_add_contacts_history:

  469. #                             sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)

  470.                             

  471. #                             # 已经邀请过两次，不再邀请

  472. #                             if len(sorted_history)==2:

  473. #                                 logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')

  474. #                                 continue

  475. 

  476. #                             # 当天邀请过，不再邀请

  477. #                             if len(sorted_history) > 0:

  478. #                                 last_add_time = sorted_history[0].addTime

  479. #                                 def is_add_time_more_than_one_day(addTime: int) -> bool:

  480. #                                     """

  481. #                                     判断 addTime 是否与当前时间相隔大于 3600 × 24 秒

  482. #                                     :param addTime: Unix 时间戳

  483. #                                     :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False

  484. #                                     """

  485. #                                     # 获取当前时间的时间戳

  486. #                                     current_time = time.time()

  487.                                     

  488. #                                     # 计算时间戳差值

  489. #                                     time_difference = abs(current_time - addTime)

  490.                                     

  491. #                                     # 检查是否大于 3600 × 24 秒

  492. #                                     return time_difference > 3600 * 24

  493.                             

  494. #                                 is_more_than_one_day= is_add_time_more_than_one_day(last_add_time)

  495. 

  496. #                                 if not is_more_than_one_day:

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

  498. #                                     continue

  499. 

  500. #                         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}')

  501. #                         if ret!=200:

  502. #                             logger.warning(f'群好友邀请失败原因：{ret} {data}')

  503. #                             if msg in '操作过于频繁，请稍后再试。':

  504. #                                 await gewe_service.save_wx_expection_async(wxid,"addGroupMemberAsFriend",msg,today_seconds_remaining())

  505. #                                 logger.warning(f'{nickname}-{wxid} 操作过于频繁，本次群好友邀请任务未完成跳过。')

  506. #                                 continue 

  507. 

  508. #                         history=AddGroupContactsHistory.model_validate({

  509. #                             "chatroomId":chatroom_id,

  510. #                             "wxid":wxid,

  511. #                             "contactWixd":contact_wxid,

  512. #                             "addTime":int(time.time())

  513. #                         })

  514. #                         await gewe_service.save_group_add_contacts_history_async(wxid,chatroom_id,contact_wxid,history)

  515. #                         wixd_add_contacts_from_chatrooms_times[wxid]+=1

  516. #                         logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg}')

  517. #                         # 推送到kafka

  518. 

  519. #                         k_message = wx_add_contacts_from_chatroom_message(history.wxid,history.chatroomId,history.contactWixd,history.addTime)

  520. #                         await kafka_service.send_message_async(k_message)

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

  522. #                         await asyncio.sleep(random.uniform(30,60)) 

  523. 

  524. #                     # 任务状态推送到kafka 

  525. #                     task_status=await gewe_service.wx_add_contacts_from_chatroom_task_status_async(wxid,chatroom_id)

  526. #                     k_message=wx_add_contacts_from_chatroom_task_status_message(wxid,chatroom_id,task_status)

  527. #                     await kafka_service.send_message_async(k_message)

  528. #                     # 下一个群

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

  530. #         except Exception as e:

  531. #             # 获取当前的堆栈跟踪

  532. #             tb = sys.exc_info()[2]

  533. #             # 为异常附加堆栈跟踪

  534. #             e = e.with_traceback(tb)

  535. #             # 输出详细的错误信息

  536. #             logger.error(f"任务执行过程中发生异常: {e}\n异常类型: {type(e).__name__}\n异常信息: {str(e)}\n堆栈跟踪: {traceback.format_exc()}")

  537. #         finally:

  538. #             await kafka_service.stop_producer()

  539.     

  540. #     loop = asyncio.get_event_loop()

  541. #     if loop.is_closed():

  542. #         loop = asyncio.new_event_loop()

  543. #         asyncio.set_event_loop(loop)

  544. 

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

  546. 

  547. 

  548. @celery_app.task(name='tasks.scheduled_task_add_contacts_from_chatrooms', bind=True, acks_late=True)

  549. def scheduled_task_add_contacts_from_chatrooms(self, redis_config, kafka_config, gewe_config):

  550. 

  551.     '''

  552.     关于群加好友的请求规则: 每个智能体一次最多30人,间隔2小时做1次,即最多90人/天。

  553.     加好友规则:每天处理次数、间隔时间(分钟)、每次加好友人数这3个参数都可以设置。目前默认只是上面的设置。

  554.     '''

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

  556.         async with semaphore:  # 使用 Semaphore 控制并发

  557.             r = await redis_service.get_hash(k)

  558.             app_id = r.get("appId")

  559.             token_id = r.get("tokenId")

  560.             wxid = r.get("wxid")

  561.             status = r.get('status')

  562. 

  563.             if status == '0':

  564.                 logger.warning(f"微信号 {wxid} 已经离线，群成员不能定时添加")

  565.                 return

  566. 

  567.             config = await gewe_service.get_wxchat_config_from_cache_async(wxid)

  568.             validated_config = AgentConfig.model_validate(config)

  569.             if not validated_config.agentEnabled:

  570.                 logger.warning(f"微信号 {wxid} 取消了托管，群成员不能定时添加")

  571.                 return

  572. 

  573.             # 判断是否过于频繁

  574.             is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")

  575.             if is_wx_expection:

  576.                 logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常, {is_wx_expection}，本次群好友邀请任务未开始,跳过任务。")

  577.                 return

  578. 

  579.             cache_task_run_time_wxid_logs = await gewe_service.get_task_run_time_by_wxid_async(wxid, 'scheduled_task_add_contacts_from_chatrooms')

  580.             if cache_task_run_time_wxid_logs:

  581.                 sorted_tasks = sorted(cache_task_run_time_wxid_logs, key=lambda x: x.get("runTime"), reverse=True)

  582.                 last_run_time = sorted_tasks[0].get("runTime")

  583.                 if last_run_time > 1e12:  # 毫秒级时间戳

  584.                     last_run_time = last_run_time / 1000  # 转换为秒

  585. 

  586.                 # 将时间戳转换为 datetime 对象

  587.                 last_run_time = datetime.fromtimestamp(last_run_time)

  588. 

  589.                 # 获取当前时间

  590.                 current_time = datetime.now()

  591. 

  592.                 # 计算时间差

  593.                 time_difference = current_time - last_run_time

  594. 

  595.                 # 判断是否相差2小时

  596.                 if time_difference < timedelta(hours=2):

  597.                     logger.info(f"{wxid}上次定时群成员定时添好友任务在2小时内，不再执行")

  598.                     return

  599. 

  600.             cache_task_run_time_wxid_logs.append({"runTime": int(time.time())})

  601.             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)

  602.             c:dict = await gewe_service.get_wxchat_config_from_cache_async(wxid)

  603.             contacts = await gewe_service.get_contacts_brief_from_cache_async(wxid)

  604. 

  605.             contact_wxids = [c.get('userName') for c in contacts]

  606.             chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])

  607.             logger.info(f'{wxid} 定时群成员定时添好友任务开始')

  608.             wixd_add_contacts_from_chatrooms_times = {wxid: 0}

  609.             for chatroom_id in chatrooms:

  610.                 chatroom = await gewe_service.get_group_info_from_cache_async(wxid, chatroom_id)

  611.                 chatroom_member = await gewe_service.get_group_members_from_cache_async(wxid, chatroom_id)

  612. 

  613.                 chatroom_nickname = chatroom.get('nickName')

  614.                 chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)

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

  616.                 admin_wxids = chatroom_member.get('adminWxid')

  617.                 if admin_wxids is None:

  618.                     admin_wxids = []  # 如果 admin_wxids 是 None，将其初始化为空列表

  619. 

  620.                 logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxids}')

  621.                 contact_wxids_set = set(contact_wxids)

  622.                 if admin_wxids:

  623.                     contact_wxids_set.update(set(admin_wxids))

  624.                 if chatroom_owner_wxid is not None:

  625.                     contact_wxids_set.add(chatroom_owner_wxid)

  626. 

  627.                 contact_wxids_set.add(wxid)

  628.                 unavailable_wixds = await gewe_service.check_wixd_group_add_contacts_history_async(wxid, chatroom_id)

  629.                 if unavailable_wixds:

  630.                     contact_wxids_set.update(set(unavailable_wixds))

  631. 

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

  633.                 if chatroom_member_list is None:

  634.                     chatroom_member_list = []  # 如果 memberList 是 None，将其初始化为空列表

  635.                 elif not isinstance(chatroom_member_list, list):

  636.                     chatroom_member_list = list(chatroom_member_list)  # 如果 memberList 不是列表，将其转换为列表

  637. 

  638.                 remaining_chatroot_members = [x for x in chatroom_member_list if x.get('wxid') not in contact_wxids_set]

  639. 

  640.                 nickname = next((member['nickName'] for member in chatroom_member_list if member['wxid'] == wxid), None)

  641.                 if not remaining_chatroot_members:

  642.                     logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里没有好友可以邀请')

  643.                     # 任务状态推送到kafka

  644.                     k_message = wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, 2)

  645.                     await kafka_service.send_message_async(k_message)

  646.                     continue 

  647.                 

  648.                 logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里还可以邀请的好友有：{[x.get("nickName") for x in remaining_chatroot_members]}')

  649.                 for m in remaining_chatroot_members:

  650.                     # 判断本次任务是否已经邀请了30个好友

  651.                     if wixd_add_contacts_from_chatrooms_times[wxid] == once_add_contacts_total:

  652.                         logger.info(f"{wxid} 本次任务已经邀请了{once_add_contacts_total}人,不再邀请")

  653.                         return

  654. 

  655.                     # 判断当天群成员是否已经加了90个好友

  656.                     is_add_group_times = await gewe_service.is_group_add_contacts_history_one_day_async(wxid, oneday_add_contacts_total)

  657.                     if is_add_group_times:

  658.                         logger.info(f"当天 {wxid} 所有群的成员已经加了{oneday_add_contacts_total}个好友，不再添加")

  659.                         return

  660. 

  661.                     # 判断是否过于频繁

  662.                     is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")

  663.                     if is_wx_expection:

  664.                         logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}")

  665.                         return

  666.                     

  667.                     contact_wxid = m.get('wxid')

  668.                     member_nickname = m.get("nickName")

  669.                     group_add_contacts_history = await gewe_service.get_group_add_contacts_history_async(wxid, chatroom_id, contact_wxid)

  670. 

  671.                     if group_add_contacts_history:

  672.                         sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)

  673. 

  674.                         # 已经邀请过两次，不再邀请

  675.                         if len(sorted_history) == 2:

  676.                             logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')

  677.                             continue

  678.                         

  679.                         # 当天邀请过，不再邀请

  680.                         if len(sorted_history) > 0:

  681.                             last_add_time = sorted_history[0].addTime

  682. 

  683.                             def is_add_time_more_than_one_day(addTime: int) -> bool:

  684.                                 """

  685.                                 判断 addTime 是否与当前时间相隔大于 3600 × 24 秒

  686.                                 :param addTime: Unix 时间戳

  687.                                 :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False

  688.                                 """

  689.                                 # 获取当前时间的时间戳

  690.                                 current_time = time.time()

  691. 

  692.                                 # 计算时间戳差值

  693.                                 time_difference = abs(current_time - addTime)

  694. 

  695.                                 # 检查是否大于 3600 × 24 秒

  696.                                 return time_difference > 3600 * 24

  697. 

  698.                             is_more_than_one_day = is_add_time_more_than_one_day(last_add_time)

  699.                             if not is_more_than_one_day:

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

  701.                                 continue 

  702. 

  703.                     ret, msg, data = await gewe_service.add_group_member_as_friend_async(token_id, app_id, chatroom_id, m.get('wxid'),

  704.                                                                                         f'我是群聊"{chatroom_nickname}"群的{nickname}')

  705.                     if ret != 200:

  706.                         logger.warning(f'{nickname}-{wxid} 在 {chatroom_nickname}-{chatroom_id} 群好友邀请失败原因：{ret} {msg} {data}')

  707.                         if '操作过于频繁' in data.get('msg'):

  708.                             await gewe_service.save_wx_expection_async(wxid, "addGroupMemberAsFriend", msg, today_seconds_remaining())

  709.                             logger.warning(f'{nickname}-{wxid} 在 {chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 操作过于频繁，本次群好友邀请任务未完成跳过。当天不再处理该号群好友邀请任务')

  710.                             return 

  711.                         

  712.                     history = AddGroupContactsHistory.model_validate({

  713.                         "chatroomId": chatroom_id,

  714.                         "wxid": wxid,

  715.                         "contactWixd": contact_wxid,

  716.                         "addTime": int(time.time())

  717.                     })

  718.                     await gewe_service.save_group_add_contacts_history_async(wxid, chatroom_id, contact_wxid, history)

  719.                     wixd_add_contacts_from_chatrooms_times[wxid] += 1

  720.                     logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg}')

  721.                     # 推送到kafka

  722.                     k_message = wx_add_contacts_from_chatroom_message(history.wxid, history.chatroomId, history.contactWixd, history.addTime)

  723.                     await kafka_service.send_message_async(k_message)

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

  725.                     # await asyncio.sleep(random.uniform(30, 60))

  726.                     await asyncio.sleep(random.uniform(270,300))

  727.                 # 任务状态推送到kafka

  728.                 task_status = await gewe_service.wx_add_contacts_from_chatroom_task_status_async(wxid, chatroom_id)

  729.                 k_message=wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, task_status)

  730.                 await kafka_service.send_message_async(k_message)

  731.                 # 下一个群

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

  733. 

  734.     async def task():

  735.         try:

  736.             now = datetime.now()

  737.             # if 10> now.hour < 8:

  738.             #     logger.info(f"定时群成员定时添好友任务不启动,当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，早于8点")

  739.             #     return

  740. 

  741.             if now.hour < 8 or now.hour > 22:

  742.                 logger.info(f"定时群成员定时添好友任务不启动, 当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，不在8点到22点之间")

  743.                 return

  744. 

  745.             logger.info('定时群成员定时添好友任务开始')

  746.             redis_service = RedisService()

  747.             await redis_service.init(**redis_config)

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

  749.             KAFKA_BOOTSTRAP_SERVERS = kafka_config['bootstrap_servers']

  750.             KAFKA_TOPIC = kafka_config['topic']

  751.             KAFKA_GROUP_ID = kafka_config['group_id']

  752. 

  753.             kafka_service = KafkaService(KAFKA_BOOTSTRAP_SERVERS, KAFKA_TOPIC, KAFKA_TOPIC, KAFKA_GROUP_ID)

  754.             await kafka_service.start_producer()

  755.             global_config = await gewe_service.get_global_config_from_cache_async()

  756.             scheduled_task_add_contacts_from_chatrooms_config = global_config.get('scheduledTaskAddContactsFromChatrooms', {})

  757. 

  758.             oneday_add_contacts_total = 90

  759.             once_add_contacts_total = 30

  760.             # oneday_times=3

  761.             if scheduled_task_add_contacts_from_chatrooms_config:

  762.                 oneday_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('oneDayAddContactsTotal', 90)

  763.                 once_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('onceAddContactsTotal', 30)

  764.                 # oneday_times=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayTimes',3)

  765. 

  766.             login_keys = []

  767.             async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  768.                 login_keys.append(key)

  769. 

  770.             # 设置 Semaphore，限制并发任务数量

  771.             semaphore = asyncio.Semaphore(10)  # 例如，限制同时运行的任务数量为 10

  772.             # 使用 asyncio.gather 并发处理每个 login_key

  773.             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])

  774. 

  775.         except Exception as e:

  776.             # 获取当前的堆栈跟踪

  777.             tb = sys.exc_info()[2]

  778.             # 为异常附加堆栈跟踪

  779.             e = e.with_traceback(tb)

  780.             # 输出详细的错误信息

  781.             logger.error(f"任务执行过程中发生异常: {e}\n异常类型: {type(e).__name__}\n异常信息: {str(e)}\n堆栈跟踪: {traceback.format_exc()}")

  782.         finally:

  783.             await kafka_service.stop_producer()

  784. 

  785.     loop = asyncio.get_event_loop()

  786.     if loop.is_closed():

  787.         loop = asyncio.new_event_loop()

  788.         asyncio.set_event_loop(loop)

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

  790. 

  791. 

  792. 

  793. # @celery_app.task(name='tasks.scheduled_task_add_contacts_from_chatrooms', bind=True, acks_late=True)

  794. # def scheduled_task_add_contacts_from_chatrooms(self, redis_config, kafka_config, gewe_config):

  795. 

  796. #     '''

  797. #     关于群加好友的请求规则: 每个智能体一次最多30人,间隔2小时做1次,即最多90人/天。

  798. #     加好友规则:每天处理次数、间隔时间(分钟)、每次加好友人数这3个参数都可以设置。目前默认只是上面的设置。

  799. #     '''

  800. #     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):

  801. #         r = await redis_service.get_hash(k)

  802. #         app_id = r.get("appId")

  803. #         token_id = r.get("tokenId")

  804. #         wxid = r.get("wxid")

  805. #         status = r.get('status')

  806. 

  807. #         if status == '0':

  808. #             logger.warning(f"微信号 {wxid} 已经离线，群成员不能定时添加")

  809. #             return

  810. 

  811. #         config = await gewe_service.get_wxchat_config_from_cache_async(wxid)

  812. #         validated_config = AgentConfig.model_validate(config)

  813. #         if not validated_config.agentEnabled:

  814. #             logger.warning(f"微信号 {wxid} 取消了托管，群成员不能定时添加")

  815. #             return

  816. 

  817. #         # 判断是否过于频繁

  818. #         is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")

  819. #         if is_wx_expection:

  820. #             logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常, {is_wx_expection}，本次群好友邀请任务未开始,跳过任务。")

  821. #             return

  822. 

  823. #         cache_task_run_time_wxid_logs = await gewe_service.get_task_run_time_by_wxid_async(wxid, 'scheduled_task_add_contacts_from_chatrooms')

  824. #         if cache_task_run_time_wxid_logs:

  825. #             sorted_tasks = sorted(cache_task_run_time_wxid_logs, key=lambda x: x.get("runTime"), reverse=True)

  826. #             last_run_time = sorted_tasks[0].get("runTime")

  827. #             if last_run_time > 1e12:  # 毫秒级时间戳

  828. #                 last_run_time = last_run_time / 1000  # 转换为秒

  829. 

  830. #             # 将时间戳转换为 datetime 对象

  831. #             last_run_time = datetime.fromtimestamp(last_run_time)

  832. 

  833. #             # 获取当前时间

  834. #             current_time = datetime.now()

  835. 

  836. #             # 计算时间差

  837. #             time_difference = current_time - last_run_time

  838. 

  839. #             # 判断是否相差2小时

  840. #             if time_difference < timedelta(hours=2):

  841. #                 logger.info(f"{wxid}上次定时群成员定时添好友任务在2小时内，不再执行")

  842. #                 return

  843. 

  844. #         cache_task_run_time_wxid_logs.append({"runTime": int(time.time())})

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

  846. #         c:dict = await gewe_service.get_wxchat_config_from_cache_async(wxid)

  847. #         contacts = await gewe_service.get_contacts_brief_from_cache_async(wxid)

  848. 

  849. #         contact_wxids = [c.get('userName') for c in contacts]

  850. #         chatrooms = c.get('addContactsFromChatroomIdWhiteList', [])

  851. #         logger.info(f'{wxid} 定时群成员定时添好友任务开始')

  852. #         wixd_add_contacts_from_chatrooms_times = {wxid: 0}

  853. #         for chatroom_id in chatrooms:

  854. #             chatroom = await gewe_service.get_group_info_from_cache_async(wxid, chatroom_id)

  855. #             chatroom_member = await gewe_service.get_group_members_from_cache_async(wxid, chatroom_id)

  856. 

  857. #             chatroom_nickname = chatroom.get('nickName')

  858. #             chatroom_owner_wxid = chatroom_member.get('chatroomOwner', None)

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

  860. #             admin_wxids = chatroom_member.get('adminWxid')

  861. #             if admin_wxids is None:

  862. #                 admin_wxids = []  # 如果 admin_wxids 是 None，将其初始化为空列表

  863. 

  864. #             logger.info(f'{chatroom_nickname} 的群主是 {chatroom_owner_wxid}，管理员是{admin_wxids}')

  865. #             contact_wxids_set = set(contact_wxids)

  866. #             if admin_wxids:

  867. #                 contact_wxids_set.update(set(admin_wxids))

  868. #             if chatroom_owner_wxid is not None:

  869. #                 contact_wxids_set.add(chatroom_owner_wxid)

  870. 

  871. #             contact_wxids_set.add(wxid)

  872. #             unavailable_wixds = await gewe_service.check_wixd_group_add_contacts_history_async(wxid, chatroom_id)

  873. #             if unavailable_wixds:

  874. #                 contact_wxids_set.update(set(unavailable_wixds))

  875. 

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

  877. #             if chatroom_member_list is None:

  878. #                 chatroom_member_list = []  # 如果 memberList 是 None，将其初始化为空列表

  879. #             elif not isinstance(chatroom_member_list, list):

  880. #                 chatroom_member_list = list(chatroom_member_list)  # 如果 memberList 不是列表，将其转换为列表

  881. 

  882. #             remaining_chatroot_members = [x for x in chatroom_member_list if x.get('wxid') not in contact_wxids_set]

  883. 

  884. #             nickname = next((member['nickName'] for member in chatroom_member_list if member['wxid'] == wxid), None)

  885. #             if not remaining_chatroot_members:

  886. #                 logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里没有好友可以邀请')

  887. #                 # 任务状态推送到kafka

  888. #                 k_message = wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, 2)

  889. #                 await kafka_service.send_message_async(k_message)

  890. #                 return 

  891.             

  892. #             logger.info(f'{nickname}-{wxid} 在 {chatroom_nickname} 群里还可以邀请的好友有：{[x.get("nickName") for x in remaining_chatroot_members]}')

  893. #             for m in remaining_chatroot_members:

  894. #                 # 判断本次任务是否已经邀请了30个好友

  895. #                 if wixd_add_contacts_from_chatrooms_times[wxid] == once_add_contacts_total:

  896. #                     logger.info(f"{wxid} 本次任务已经邀请了{once_add_contacts_total}人,不再邀请")

  897. #                     return

  898. 

  899. #                 # 判断当天群成员是否已经加了90个好友

  900. #                 is_add_group_times = await gewe_service.is_group_add_contacts_history_one_day_async(wxid, oneday_add_contacts_total)

  901. #                 if is_add_group_times:

  902. #                     logger.info(f"当天 {wxid} 所有群的成员已经加了{oneday_add_contacts_total}个好友，不再添加")

  903. #                     return

  904. 

  905. #                 # 判断是否过于频繁

  906. #                 is_wx_expection = await gewe_service.get_wx_expection_async(wxid, "addGroupMemberAsFriend")

  907. #                 if is_wx_expection:

  908. #                     logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}")

  909. #                     return

  910.                 

  911. #                 contact_wxid = m.get('wxid')

  912. #                 member_nickname = m.get("nickName")

  913. #                 group_add_contacts_history = await gewe_service.get_group_add_contacts_history_async(wxid, chatroom_id, contact_wxid)

  914. 

  915. #                 if group_add_contacts_history:

  916. #                     sorted_history = sorted(group_add_contacts_history, key=lambda x: x.addTime, reverse=True)

  917. 

  918. #                     # 已经邀请过两次，不再邀请

  919. #                     if len(sorted_history) == 2:

  920. #                         logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经邀请过2次，不再邀请')

  921. #                         return

  922.                     

  923. #                     # 当天邀请过，不再邀请

  924. #                     if len(sorted_history) > 0:

  925. #                         last_add_time = sorted_history[0].addTime

  926. 

  927. #                         def is_add_time_more_than_one_day(addTime: int) -> bool:

  928. #                             """

  929. #                             判断 addTime 是否与当前时间相隔大于 3600 × 24 秒

  930. #                             :param addTime: Unix 时间戳

  931. #                             :return: 如果 addTime 与当前时间相隔大于 3600 × 24 秒，返回 True；否则返回 False

  932. #                             """

  933. #                             # 获取当前时间的时间戳

  934. #                             current_time = time.time()

  935. 

  936. #                             # 计算时间戳差值

  937. #                             time_difference = abs(current_time - addTime)

  938. 

  939. #                             # 检查是否大于 3600 × 24 秒

  940. #                             return time_difference > 3600 * 24

  941. 

  942. #                         is_more_than_one_day = is_add_time_more_than_one_day(last_add_time)

  943. #                         if not is_more_than_one_day:

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

  945. #                             return 

  946. 

  947. #                 ret, msg, data = await gewe_service.add_group_member_as_friend_async(token_id, app_id, chatroom_id, m.get('wxid'),

  948. #                                                                                     f'我是群聊"{chatroom_nickname}"群的{nickname}')

  949. #                 if ret != 200:

  950. #                     logger.warning(f'群好友邀请失败原因：{ret} {msg} {data}')

  951. #                     if '操作过于频繁' in data.get('msg'):

  952. #                         await gewe_service.save_wx_expection_async(wxid, "addGroupMemberAsFriend", msg, today_seconds_remaining())

  953. #                         logger.warning(f'{nickname}-{wxid} 操作过于频繁，本次群好友邀请任务未完成跳过。')

  954. #                         return 

  955.                     

  956. #                 history = AddGroupContactsHistory.model_validate({

  957. #                     "chatroomId": chatroom_id,

  958. #                     "wxid": wxid,

  959. #                     "contactWixd": contact_wxid,

  960. #                     "addTime": int(time.time())

  961. #                 })

  962. #                 await gewe_service.save_group_add_contacts_history_async(wxid, chatroom_id, contact_wxid, history)

  963. #                 wixd_add_contacts_from_chatrooms_times[wxid] += 1

  964. #                 logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg}')

  965. #                 # 推送到kafka

  966. #                 k_message = wx_add_contacts_from_chatroom_message(history.wxid, history.chatroomId, history.contactWixd, history.addTime)

  967. #                 await kafka_service.send_message_async(k_message)

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

  969. #                 await asyncio.sleep(random.uniform(30, 60))

  970. #             # 任务状态推送到kafka

  971. #             task_status = await gewe_service.wx_add_contacts_from_chatroom_task_status_async(wxid, chatroom_id)

  972. #             wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, task_status)

  973. #             await kafka_service.send_message_async(k_message)

  974. #             # 下一个群

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

  976. 

  977. #     async def task():

  978. #         try:

  979. #             now = datetime.now()

  980. #             if now.hour < 8:

  981. #                 logger.info(f"定时群成员定时添好友任务不启动,当前时间为 {now.strftime('%Y-%m-%d %H:%M:%S')}，早于8点")

  982. #                 return

  983. 

  984. #             logger.info('定时群成员定时添好友任务开始')

  985. #             redis_service = RedisService()

  986. #             await redis_service.init(**redis_config)

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

  988. #             KAFKA_BOOTSTRAP_SERVERS = kafka_config['bootstrap_servers']

  989. #             KAFKA_TOPIC = kafka_config['topic']

  990. #             KAFKA_GROUP_ID = kafka_config['group_id']

  991. 

  992. #             kafka_service = KafkaService(KAFKA_BOOTSTRAP_SERVERS, KAFKA_TOPIC, KAFKA_TOPIC, KAFKA_GROUP_ID)

  993. #             await kafka_service.start_producer()

  994. #             global_config = await gewe_service.get_global_config_from_cache_async()

  995. #             scheduled_task_add_contacts_from_chatrooms_config = global_config.get('scheduledTaskAddContactsFromChatrooms', {})

  996. 

  997. #             oneday_add_contacts_total = 90

  998. #             once_add_contacts_total = 30

  999. #             # oneday_times=3

  1000. #             if scheduled_task_add_contacts_from_chatrooms_config:

  1001. #                 oneday_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('oneDayAddContactsTotal', 90)

  1002. #                 once_add_contacts_total = scheduled_task_add_contacts_from_chatrooms_config.get('onceAddContactsTotal', 30)

  1003. #                 # oneday_times=scheduled_task_add_contacts_from_chatrooms_config.get('oneDayTimes',3)

  1004. 

  1005. #             login_keys = []

  1006. #             async for key in redis_service.client.scan_iter(match='__AI_OPS_WX__:LOGININFO:*'):

  1007. #                 login_keys.append(key)

  1008. 

  1009. #             # 使用 asyncio.gather 并发处理每个 login_key

  1010. #             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])

  1011. 

  1012. #         except Exception as e:

  1013. #             # 获取当前的堆栈跟踪

  1014. #             tb = sys.exc_info()[2]

  1015. #             # 为异常附加堆栈跟踪

  1016. #             e = e.with_traceback(tb)

  1017. #             # 输出详细的错误信息

  1018. #             logger.error(f"任务执行过程中发生异常: {e}\n异常类型: {type(e).__name__}\n异常信息: {str(e)}\n堆栈跟踪: {traceback.format_exc()}")

  1019. #         finally:

  1020. #             await kafka_service.stop_producer()

  1021. 

  1022. #     loop = asyncio.get_event_loop()

  1023. #     if loop.is_closed():

  1024. #         loop = asyncio.new_event_loop()

  1025. #         asyncio.set_event_loop(loop)

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

  1027. 

  1028. 

  1029. 

  1030. 

  1031. 

  1032. 

  1033. REDIS_KEY_PATTERN = "friend_add_limit:{date}"

  1034. REDIS_LAST_RUN_KEY = "last_run_time:add_friends_task"

  1035. 

  1036. @celery_app.task(name='tasks.add_friends_task', bind=True, acks_late=True)

  1037. def add_friends_task(self,redis_config):

  1038.     """

  1039.     限制每天最多 15 个，每 2 小时最多 8 个

  1040.     """

  1041.     async def task():

  1042.         redis_service = RedisService()

  1043.         await redis_service.init(**redis_config)

  1044.         today_str = datetime.now().strftime("%Y%m%d")

  1045.         redis_key = REDIS_KEY_PATTERN.format(date=today_str)

  1046. 

  1047.         # 获取当前总添加数量

  1048.         total_added = await redis_service.get_hash_field(redis_key, "total") or 0

  1049.         last_2h_added =await redis_service.get_hash_field(redis_key, "last_2h") or 0

  1050. 

  1051.         total_added = int(total_added)

  1052.         last_2h_added = int(last_2h_added)

  1053. 

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

  1055. 

  1056.         # 判断是否超过限制

  1057.         if total_added >= 15:

  1058.             logger.warning("今日好友添加已达上限！")

  1059.             return

  1060. 

  1061.         if last_2h_added >= 8:

  1062.             logger.warning("过去2小时添加已达上限！")

  1063.             return

  1064. 

  1065.         # 计算本次要添加的好友数量 (控制每天 5-15 个)

  1066.         max_add = min(15 - total_added, 8 - last_2h_added)

  1067.         if max_add <= 0:

  1068.             return

  1069. 

  1070.         num_to_add = min(max_add, 1)  # 每次最多加 1 个

  1071.         logger.info(f"本次添加 {num_to_add} 位好友")

  1072. 

  1073.         # TODO: 调用好友添加逻辑 (接口 or 业务逻辑)

  1074.         # success = add_friends(num_to_add)

  1075. 

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

  1077. 

  1078.         # 更新 Redis 计数

  1079.         if success > 0:

  1080.             await redis_service.increment_hash_field(redis_key, "total", success)

  1081.             await redis_service.increment_hash_field(redis_key, "last_2h", success)

  1082. 

  1083.             # 设置 Redis 过期时间 (每日记录存 1 天, 2 小时记录存 2 小时)

  1084.             await redis_service.expire(redis_key, 86400)  # 24小时

  1085.             await redis_service.expire_field(redis_key, "last_2h", 7200)  # 2小时

  1086. 

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

  1088. 

  1089.         # 生成一个新的随机时间（5-15 分钟之间）

  1090.         # next_interval = random.randint(10, 20)

  1091. 

  1092.         # # 计算新的执行时间

  1093.         # next_run_time = datetime.datetime.now() + timedelta(seconds=next_interval)

  1094. 

  1095.         # # 重新注册 RedBeat 任务，确保下次执行时间不同

  1096.         # redbeat_entry = RedBeatSchedulerEntry(

  1097.         #     name="redbeat:add_friends_task",

  1098.         #     task="tasks.add_friends_task",

  1099.         #     schedule=celery.schedules.schedule(timedelta(seconds=next_interval)),

  1100.         #     args=[redis_config],

  1101.         #     app=celery_app

  1102.         # )

  1103. 

  1104.         # # 设置任务的下次执行时间

  1105.         # redbeat_entry.last_run_at = next_run_time

  1106.         # redbeat_entry.save()

  1107. 

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

  1109. 

  1110.     loop = asyncio.get_event_loop()

  1111.     

  1112.     if loop.is_closed():

  1113.         loop = asyncio.new_event_loop()

  1114.         asyncio.set_event_loop(loop)

  1115. 

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

  1117. 

  1118. def today_seconds_remaining()->int:

  1119.     current_time = datetime.now()

  1120. 

  1121.     # 计算当天的结束时间（23:59:59）

  1122.     end_of_day = datetime(current_time.year, current_time.month, current_time.day, 23, 59, 59)

  1123. 

  1124.     # 计算时间差

  1125.     time_difference = end_of_day - current_time

  1126. 

  1127.     # 将时间差转换为秒数

  1128.     time_difference_seconds = int(time_difference.total_seconds())

  1129. 

  1130.     return time_difference_seconds

  1131. 

  1132. 

  1133. @celery_app.task(name='tasks.random_scheduled_task', bind=True, acks_late=True)

  1134. def random_scheduled_task(self,):

  1135.     print(f"Task executed at {datetime.now()}")

  1136.     # 随机生成下次执行时间（例如：10-60秒内的随机时间）

  1137.     next_run_in = random.randint(10, 60)

  1138.     print(f"Next execution will be in {next_run_in} seconds")

  1139.     

  1140.     # 设置下次执行时间

  1141.     entry = RedBeatSchedulerEntry(

  1142.         name='random-task',

  1143.         task='tasks.random_scheduled_task',

  1144.         schedule=timedelta(seconds=next_run_in),

  1145.         app=celery_app

  1146.     )

  1147.     entry.save()

  1148.     return f"Scheduled next run in {next_run_in} seconds"