J1147
/
AiOpsWechat
Прати 1
Волим 0
Креирај огранак 0
Код Дискусије 0 Захтеви за спајање 0 Издања 0 Вики Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
172 Комити
5 Гране
2.4MB
Дрво: e7a7014801
Гране Ознаке
${ item.name }
Create branch ${ searchTerm }
from 'e7a7014801'
${ noResults }
AiOpsWechat/tasks.py

1189 lines
63KB
Датотека 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_chatroom_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_chatroom_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_chatroom_members]}')

  447. #                     for m in remaining_chatroom_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.             try:  

  558.                 r = await redis_service.get_hash(k)

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

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

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

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

  563. 

  564.                 if status == '0':

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

  566.                     return

  567. 

  568.                 config = await gewe_service.get_wxchat_config_from_cache_async(wxid)

  569.                 validated_config = AgentConfig.model_validate(config)

  570.                 if not validated_config.agentEnabled:

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

  572.                     return

  573. 

  574.                 # 判断是否过于频繁

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

  576.                 if is_wx_expection:

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

  578.                     return

  579. 

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

  581.                 if cache_task_run_time_wxid_logs:

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

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

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

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

  586. 

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

  588.                     last_run_time = datetime.fromtimestamp(last_run_time)

  589. 

  590.                     # 获取当前时间

  591.                     current_time = datetime.now()

  592. 

  593.                     # 计算时间差

  594.                     time_difference = current_time - last_run_time

  595. 

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

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

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

  599.                         return

  600. 

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

  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.                 wixd_add_contacts_from_chatrooms_times = {wxid: 0}

  608. 

  609.                 for chatroom_id in chatrooms:

  610.                     logger.info(f'{wxid} 进行{chatroom_id} 群成员定时添好友')

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

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

  613. 

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

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

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

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

  618.                     if admin_wxids is None:

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

  620. 

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

  622.                     contact_wxids_set = set(contact_wxids)

  623.                     if admin_wxids:

  624.                         contact_wxids_set.update(set(admin_wxids))

  625.                     if chatroom_owner_wxid is not None:

  626.                         contact_wxids_set.add(chatroom_owner_wxid)

  627. 

  628.                     contact_wxids_set.add(wxid)

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

  630.                     if unavailable_wixds:

  631.                         contact_wxids_set.update(set(unavailable_wixds))

  632. 

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

  634.                     if chatroom_member_list is None:

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

  636.                     elif not isinstance(chatroom_member_list, list):

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

  638. 

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

  640. 

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

  642.                     if not remaining_chatroom_members:

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

  644.                         # 任务状态推送到kafka

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

  646.                         await kafka_service.send_message_async(k_message)

  647.                         continue 

  648.                     

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

  650.                     for m in remaining_chatroom_members:

  651.                        

  652. 

  653.                         # 判断群已经退出白名单

  654.                         wx_config=await gewe_service.get_wxchat_config_from_cache_async(wxid)

  655.                         add_contacts_from_chatroomid_list = wx_config.get('addContactsFromChatroomIdWhiteList', [])

  656.                         if chatroom_id not in add_contacts_from_chatroomid_list:

  657.                             logger.info(f"{wxid} 群{chatroom_id} 已经退出白名单，不再邀请")

  658.                             return

  659.                         

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

  661.                         if wixd_add_contacts_from_chatrooms_times[wxid] == once_add_contacts_total:

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

  663.                             

  664.                             # 更新任务运行时间

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

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

  667.                             logger.info(f'{wxid} 定时群成员定时添好友任务，在本次邀请了{once_add_contacts_total}个情况完成，退出，本次共邀请好友 {wixd_add_contacts_from_chatrooms_times[wxid]} 个')   

  668. 

  669.                             return

  670. 

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

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

  673.                         if is_add_group_times:

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

  675.                             # 更新任务运行时间

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

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

  678.                             logger.info(f'{wxid} 定时群成员定时添好友任务，在当天邀请了{once_add_contacts_total}个情况完成，退出，本次共邀请好友 {wixd_add_contacts_from_chatrooms_times[wxid]} 个')   

  679. 

  680.                             return

  681. 

  682.                         # 判断是否过于频繁

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

  684.                         if is_wx_expection:

  685.                             logger.info(f"{wxid} 本次任务接口addGroupMemberAsFriend异常,不再邀请,{is_wx_expection}，本次共邀请好友 {wixd_add_contacts_from_chatrooms_times[wxid]} 个")

  686.                             return

  687.                         

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

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

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

  691. 

  692.                         if group_add_contacts_history:

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

  694. 

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

  696.                             if len(sorted_history) == 2:

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

  698.                                 continue

  699.                             

  700.                             # 24小时邀请过，不再邀请

  701.                             if len(sorted_history) > 0:

  702.                                 last_add_time = sorted_history[0].addTime

  703. 

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

  705.                                     """

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

  707.                                     :param addTime: Unix 时间戳

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

  709.                                     """

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

  711.                                     current_time = time.time()

  712. 

  713.                                     # 计算时间戳差值

  714.                                     time_difference = abs(current_time - addTime)

  715. 

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

  717.                                     return time_difference > 3600 * 24

  718. 

  719.                                 is_more_than_one_day = is_add_time_more_than_one_day(last_add_time)

  720.                                 if not is_more_than_one_day:

  721.                                     logger.info(f'{nickname}-{wxid}在{chatroom_nickname}-{chatroom_id} 群的 {member_nickname}-{contact_wxid} 已经24小时邀请，不再邀请')

  722.                                     continue 

  723. 

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

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

  726.                         

  727.                         history = AddGroupContactsHistory.model_validate({

  728.                             "chatroomId": chatroom_id,

  729.                             "wxid": wxid,

  730.                             "contactWixd": contact_wxid,

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

  732.                         }) 

  733. 

  734.                         if ret != 200:

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

  736.                             

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

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

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

  740.                                 wixd_add_contacts_from_chatrooms_times[wxid] += 1

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

  742.                                 return 

  743.                             

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

  745.                         wixd_add_contacts_from_chatrooms_times[wxid] += 1

  746.                         logger.info(f'{nickname} 向 {chatroom_nickname}-{chatroom_id} 群的  {m.get("nickName")}-{m.get("wxid")} 发送好友邀请  {msg} \n {data} \n 当前已邀请好友数:{wixd_add_contacts_from_chatrooms_times[wxid]}')

  747.                         # 推送到kafka

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

  749.                         await kafka_service.send_message_async(k_message)

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

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

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

  753.                     # 任务状态推送到kafka

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

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

  756.                     await kafka_service.send_message_async(k_message)

  757.                     # 下一个群

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

  759. 

  760.                 # 更新任务运行时间

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

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

  763.         

  764.                 logger.info(f'{wxid} 定时群成员定时添好友任务完成，本次共邀请好友{wixd_add_contacts_from_chatrooms_times[wxid]}个')

  765.             except Exception as e:

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

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

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

  769.                 e = e.with_traceback(tb)

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

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

  772.         

  773.     async def task():

  774.         try:

  775.             now = datetime.now()

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

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

  778.             #     return

  779. 

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

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

  782.                 return

  783. 

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

  785.             redis_service = RedisService()

  786.             await redis_service.init(**redis_config)

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

  788.             KAFKA_BOOTSTRAP_SERVERS = kafka_config['bootstrap_servers']

  789.             KAFKA_TOPIC = kafka_config['topic']

  790.             KAFKA_GROUP_ID = kafka_config['group_id']

  791. 

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

  793.             await kafka_service.start_producer()

  794.             global_config = await gewe_service.get_global_config_from_cache_async()

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

  796. 

  797.             oneday_add_contacts_total = 90

  798.             once_add_contacts_total = 30

  799.             # oneday_times=3

  800.             if scheduled_task_add_contacts_from_chatrooms_config:

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

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

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

  804. 

  805.             login_keys = []

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

  807.                 login_keys.append(key)

  808. 

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

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

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

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

  813. 

  814.         except Exception as e:

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

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

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

  818.             e = e.with_traceback(tb)

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

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

  821.         finally:

  822.             await kafka_service.stop_producer()

  823. 

  824.     loop = asyncio.get_event_loop()

  825.     if loop.is_closed():

  826.         loop = asyncio.new_event_loop()

  827.         asyncio.set_event_loop(loop)

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

  829. 

  830. 

  831. 

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

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

  834. 

  835. #     '''

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

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

  838. #     '''

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

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

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

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

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

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

  845. 

  846. #         if status == '0':

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

  848. #             return

  849. 

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

  851. #         validated_config = AgentConfig.model_validate(config)

  852. #         if not validated_config.agentEnabled:

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

  854. #             return

  855. 

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

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

  858. #         if is_wx_expection:

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

  860. #             return

  861. 

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

  863. #         if cache_task_run_time_wxid_logs:

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

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

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

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

  868. 

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

  870. #             last_run_time = datetime.fromtimestamp(last_run_time)

  871. 

  872. #             # 获取当前时间

  873. #             current_time = datetime.now()

  874. 

  875. #             # 计算时间差

  876. #             time_difference = current_time - last_run_time

  877. 

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

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

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

  881. #                 return

  882. 

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

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

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

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

  887. 

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

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

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

  891. #         wixd_add_contacts_from_chatrooms_times = {wxid: 0}

  892. #         for chatroom_id in chatrooms:

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

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

  895. 

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

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

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

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

  900. #             if admin_wxids is None:

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

  902. 

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

  904. #             contact_wxids_set = set(contact_wxids)

  905. #             if admin_wxids:

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

  907. #             if chatroom_owner_wxid is not None:

  908. #                 contact_wxids_set.add(chatroom_owner_wxid)

  909. 

  910. #             contact_wxids_set.add(wxid)

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

  912. #             if unavailable_wixds:

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

  914. 

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

  916. #             if chatroom_member_list is None:

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

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

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

  920. 

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

  922. 

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

  924. #             if not remaining_chatroom_members:

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

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

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

  928. #                 await kafka_service.send_message_async(k_message)

  929. #                 return 

  930.             

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

  932. #             for m in remaining_chatroom_members:

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

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

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

  936. #                     return

  937. 

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

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

  940. #                 if is_add_group_times:

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

  942. #                     return

  943. 

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

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

  946. #                 if is_wx_expection:

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

  948. #                     return

  949.                 

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

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

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

  953. 

  954. #                 if group_add_contacts_history:

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

  956. 

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

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

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

  960. #                         return

  961.                     

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

  963. #                     if len(sorted_history) > 0:

  964. #                         last_add_time = sorted_history[0].addTime

  965. 

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

  967. #                             """

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

  969. #                             :param addTime: Unix 时间戳

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

  971. #                             """

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

  973. #                             current_time = time.time()

  974. 

  975. #                             # 计算时间戳差值

  976. #                             time_difference = abs(current_time - addTime)

  977. 

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

  979. #                             return time_difference > 3600 * 24

  980. 

  981. #                         is_more_than_one_day = is_add_time_more_than_one_day(last_add_time)

  982. #                         if not is_more_than_one_day:

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

  984. #                             return 

  985. 

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

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

  988. #                 if ret != 200:

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

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

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

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

  993. #                         return 

  994.                     

  995. #                 history = AddGroupContactsHistory.model_validate({

  996. #                     "chatroomId": chatroom_id,

  997. #                     "wxid": wxid,

  998. #                     "contactWixd": contact_wxid,

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

  1000. #                 })

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

  1002. #                 wixd_add_contacts_from_chatrooms_times[wxid] += 1

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

  1004. #                 # 推送到kafka

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

  1006. #                 await kafka_service.send_message_async(k_message)

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

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

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

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

  1011. #             wx_add_contacts_from_chatroom_task_status_message(wxid, chatroom_id, task_status)

  1012. #             await kafka_service.send_message_async(k_message)

  1013. #             # 下一个群

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

  1015. 

  1016. #     async def task():

  1017. #         try:

  1018. #             now = datetime.now()

  1019. #             if now.hour < 8:

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

  1021. #                 return

  1022. 

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

  1024. #             redis_service = RedisService()

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

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

  1027. #             KAFKA_BOOTSTRAP_SERVERS = kafka_config['bootstrap_servers']

  1028. #             KAFKA_TOPIC = kafka_config['topic']

  1029. #             KAFKA_GROUP_ID = kafka_config['group_id']

  1030. 

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

  1032. #             await kafka_service.start_producer()

  1033. #             global_config = await gewe_service.get_global_config_from_cache_async()

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

  1035. 

  1036. #             oneday_add_contacts_total = 90

  1037. #             once_add_contacts_total = 30

  1038. #             # oneday_times=3

  1039. #             if scheduled_task_add_contacts_from_chatrooms_config:

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

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

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

  1043. 

  1044. #             login_keys = []

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

  1046. #                 login_keys.append(key)

  1047. 

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

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

  1050. 

  1051. #         except Exception as e:

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

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

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

  1055. #             e = e.with_traceback(tb)

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

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

  1058. #         finally:

  1059. #             await kafka_service.stop_producer()

  1060. 

  1061. #     loop = asyncio.get_event_loop()

  1062. #     if loop.is_closed():

  1063. #         loop = asyncio.new_event_loop()

  1064. #         asyncio.set_event_loop(loop)

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

  1066. 

  1067. 

  1068. 

  1069. 

  1070. 

  1071. 

  1072. REDIS_KEY_PATTERN = "friend_add_limit:{date}"

  1073. REDIS_LAST_RUN_KEY = "last_run_time:add_friends_task"

  1074. 

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

  1076. def add_friends_task(self,redis_config):

  1077.     """

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

  1079.     """

  1080.     async def task():

  1081.         redis_service = RedisService()

  1082.         await redis_service.init(**redis_config)

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

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

  1085. 

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

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

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

  1089. 

  1090.         total_added = int(total_added)

  1091.         last_2h_added = int(last_2h_added)

  1092. 

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

  1094. 

  1095.         # 判断是否超过限制

  1096.         if total_added >= 15:

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

  1098.             return

  1099. 

  1100.         if last_2h_added >= 8:

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

  1102.             return

  1103. 

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

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

  1106.         if max_add <= 0:

  1107.             return

  1108. 

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

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

  1111. 

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

  1113.         # success = add_friends(num_to_add)

  1114. 

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

  1116. 

  1117.         # 更新 Redis 计数

  1118.         if success > 0:

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

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

  1121. 

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

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

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

  1125. 

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

  1127. 

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

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

  1130. 

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

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

  1133. 

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

  1135.         # redbeat_entry = RedBeatSchedulerEntry(

  1136.         #     name="redbeat:add_friends_task",

  1137.         #     task="tasks.add_friends_task",

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

  1139.         #     args=[redis_config],

  1140.         #     app=celery_app

  1141.         # )

  1142. 

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

  1144.         # redbeat_entry.last_run_at = next_run_time

  1145.         # redbeat_entry.save()

  1146. 

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

  1148. 

  1149.     loop = asyncio.get_event_loop()

  1150.     

  1151.     if loop.is_closed():

  1152.         loop = asyncio.new_event_loop()

  1153.         asyncio.set_event_loop(loop)

  1154. 

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

  1156. 

  1157. def today_seconds_remaining()->int:

  1158.     current_time = datetime.now()

  1159. 

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

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

  1162. 

  1163.     # 计算时间差

  1164.     time_difference = end_of_day - current_time

  1165. 

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

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

  1168. 

  1169.     return time_difference_seconds

  1170. 

  1171. 

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

  1173. def random_scheduled_task(self,):

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

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

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

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

  1178.     

  1179.     # 设置下次执行时间

  1180.     entry = RedBeatSchedulerEntry(

  1181.         name='random-task',

  1182.         task='tasks.random_scheduled_task',

  1183.         schedule=timedelta(seconds=next_run_in),

  1184.         app=celery_app

  1185.     )

  1186.     entry.save()

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