调整

app/endpoints/pipeline_endpoint.py

@@ -85,7 +85,7 @@ async def handle_self_cmd_async(request: Request,wxid,msg):
     to_wxid=msg_data["ToUserName"]["string"]
     from_wxid=msg_data["FromUserName"]["string"]
     msg_content=msg_data["Content"]["string"]
-    if from_wxid == to_wxid:
+    if from_wxid == to_wxid and wxid == to_wxid:
         commands = {
             '启用托管': True,
             '关闭托管': False
@@ -98,7 +98,7 @@ async def handle_self_cmd_async(request: Request,wxid,msg):
                 
                 await request.app.state.gewe_service.save_wxchat_config_async(wxid, agent_config.model_dump())
                 logger.info(f'{wxid} {"启动" if commands[msg_content] else "关闭"}托管')
-
+        return {"message": "收到微信回调消息"}
 
    
 
@@ -374,7 +374,7 @@ async def handle_text_async(request: Request,token_id,app_id, wxid,msg_data,from
         return 
     
     # 判断是否转人工处理功能
-    is_human_handle_msg= await request.app.state.gewe_service.is_human_handle_msg_async(to_wxid)
+    is_human_handle_msg= await request.app.state.gewe_service.is_human_handle_msg_with_contact_wxid_async(wxid,from_wxid)
     if is_human_handle_msg:
         logger.warning(f'微信号 {wxid} 发送到微信号{to_wxid} 暂时工人接管30分钟中')
         return
@@ -383,12 +383,12 @@ async def handle_text_async(request: Request,token_id,app_id, wxid,msg_data,from
 
     msg_content=msg_data["Content"]["string"]
     if wxid == from_wxid:  #手动发送消息
-        logger.info("Active message sending detected")
+        logger.info(f"{wxid} 手动发送消息")
         
         await request.app.state.gewe_service.save_contacts_brief_to_cache_async(token_id,app_id,wxid,[to_wxid])
         callback_to_user=msg_data["ToUserName"]["string"]
         # 转人工处理功能
-        await request.app.state.gewe_service.set_human_handle_msg_async(to_wxid,60*30)
+        await request.app.state.gewe_service.set_human_handle_msg_with_contact_wxid_async(wxid,to_wxid,60*30)
 
         # 推送到kafka
         input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}]
@@ -402,7 +402,7 @@ async def handle_text_async(request: Request,token_id,app_id, wxid,msg_data,from
         keywords = ["预约", "报价", "购买", "价钱","价格", "多少钱", "下单"]
         found_keywords = [keyword for keyword in keywords if keyword in msg_content]
         if found_keywords:
-            await request.app.state.gewe_service.set_human_handle_msg_with_contact_wxid_async(wxid,callback_to_user,60*30)
+            await request.app.state.gewe_service.set_human_handle_msg_with_contact_wxid_key_word_async(wxid,callback_to_user,60*30)
             logger.info(f"{wxid} 收到 {callback_to_user} 私聊消息匹配到关键词：{', '.join(found_keywords)}")
 
             # 回复好友
@@ -437,15 +437,11 @@ async def handle_text_async(request: Request,token_id,app_id, wxid,msg_data,from
             await asyncio.sleep(random.uniform(1.5,3))
             assistant_msg_content=f'收到，{assistant_msg_content}'
             await request.app.state.gewe_service.post_text_async(kf_token_id, kf_appid, wxid, assistant_msg_content)
-
-
-
-
             return
         
 
         # 是否在转人工处理
-        is_human_handle_msg_with_contact_wxid= await request.app.state.gewe_service.is_human_handle_msg_with_contact_wxid_async(wxid,callback_to_user)
+        is_human_handle_msg_with_contact_wxid= await request.app.state.gewe_service.is_human_handle_msg_with_contact_wxid_key_word_async(wxid,callback_to_user)
         if is_human_handle_msg_with_contact_wxid:
             logger.warning(f'微信号 {wxid} 与 {callback_to_user} 有关键字匹配,暂时工人接管30分钟中,请查看长服AI商机提醒助理')
 

run.py

@@ -35,13 +35,13 @@ def start_celery_beat():
 if __name__ == "__main__":
     # 启动 FastAPI、Celery Worker 和 Celery Beat
     fastapi_process = start_fastapi()
-    celery_worker_process = start_celery_worker()
-    celery_beat_process = start_celery_beat()
+    #celery_worker_process = start_celery_worker()
+    #celery_beat_process = start_celery_beat()
  
     # 等待子进程完成
     fastapi_process.wait()
-    celery_worker_process.wait()
-    celery_beat_process.wait()
+    #celery_worker_process.wait()
+    #celery_beat_process.wait()
 
 
 # def signal_handler(sig, frame):

services/gewe_service.py

@@ -1747,8 +1747,17 @@ class GeWeService:
     
     async def set_human_handle_msg_with_contact_wxid_async(self,wxid,contact_wxid,expire_time=60*30)->bool:
         hash_key = f"__AI_OPS_WX__:HUMAN_HANDLE_MSG_WTIH_CONTACT:{wxid}:{contact_wxid}"
-        cache=  await self.redis_service.set_hash(hash_key,{"data": json.dumps(int(time.time()), ensure_ascii=False)}, expire_time)
+        await self.redis_service.set_hash(hash_key,{"data": json.dumps(int(time.time()), ensure_ascii=False)}, expire_time)
+
+    async def is_human_handle_msg_with_contact_wxid_key_word_async(self,wxid,contact_wxid)->bool:
+        hash_key = f"__AI_OPS_WX__:HUMAN_HANDLE_MSG_WTIH_CONTACT_KEY_WORD:{wxid}:{contact_wxid}"
+        cache=  await self.redis_service.get_hash_field(hash_key,"data")
         return True if cache else False
+    
+    async def set_human_handle_msg_with_contact_wxid_key_word_async(self,wxid,contact_wxid,expire_time=60*30)->bool:
+        hash_key = f"__AI_OPS_WX__:HUMAN_HANDLE_MSG_WTIH_CONTACT_KEY_WORD:{wxid}:{contact_wxid}"
+        await self.redis_service.set_hash(hash_key,{"data": json.dumps(int(time.time()), ensure_ascii=False)}, expire_time)
+