标签设置

app/endpoints/label_endpoint.py

@@ -32,36 +32,109 @@ class LabelModifyRequest(BaseModel):
 async def add_label(request: Request,  body: LabelAddRequest):
     wxid = body.wxid
     label_name = body.labelName
+    if not wxid:
+        return {"code": 400, "message": "wxid 不能为空"}
+        
+    k,loginfo=await request.app.state.gewe_service.get_login_info_by_wxid_async(wxid)
 
-    return { "labelName":label_name,"labelId":random.randint(1,100)} 
+    if not k:
+        return  {"code":404,"message":f"{wxid} 没有对应的登录信息"}
+    
+    login_status=loginfo.get('status','0')
+    
+    if login_status != '1':
+        return {"code": 401, "message": f"{wxid} 已经离线"}
+    
+    token_id=loginfo.get('tokenId','')
+    app_id=loginfo.get('appId','')
+
+    ret,msg,data=await request.app.state.gewe_service.label_add_async(token_id,app_id,label_name)
+    if ret!=200:
+        return {'code':ret,'message':msg}
+
+    return { "labelName":data.get('labelName'),"labelId":data.get('labelId')} 
 
 
 @label_router.post("/delete", response_model=None)
 async def delete_label(request: Request,  body: LabelDelRequest, ):
-   wxid = body.wxid
-   label_ids= body.labelIds
+    wxid = body.wxid
+    label_ids= body.labelIds
 
+    if not wxid:
+            return {"code": 400, "message": "wxid 不能为空"}
+                
+    k,loginfo=await request.app.state.gewe_service.get_login_info_by_wxid_async(wxid)
+
+    if not k:
+        return  {"code":404,"message":f"{wxid} 没有对应的登录信息"}
+            
+    login_status=loginfo.get('status','0')
+            
+    if login_status != '1':
+            return {"code": 401, "message": f"{wxid} 已经离线"}
+            
+    token_id=loginfo.get('tokenId','')
+    app_id=loginfo.get('appId','')
+    
+    ret,msg,data=await request.app.state.gewe_service.label_delete_async(token_id,app_id,label_ids)
+
+    if ret!=200:
+        return {'code':ret,'message':msg}   
 
-   return {"message":"操作成功"}
+    return {"message":"操作成功"}
 
 @label_router.post("/list", response_model=None)
 async def list_label(request: Request,  body: LabelListRequest, ):
     wxid = body.wxid
+    if not wxid:
+        return {"code": 400, "message": "wxid 不能为空"}
+                
+    k,loginfo=await request.app.state.gewe_service.get_login_info_by_wxid_async(wxid)
+
+    if not k:
+        return  {"code":404,"message":f"{wxid} 没有对应的登录信息"}
+            
+    login_status=loginfo.get('status','0')
+            
+    if login_status != '1':
+            return {"code": 401, "message": f"{wxid} 已经离线"}
+            
+    token_id=loginfo.get('tokenId','')
+    app_id=loginfo.get('appId','')
+
+    ret,msg,data=await request.app.state.gewe_service.label_list_async(token_id,app_id)
+    if ret!=200:
+        return {'code':ret,'message':msg}   
+    
+    return {"labelList":data.get('labelList',[]) if data!=None else []}
 
-    return {
-        "labelList":[{
-        "labelName": "朋友",
-        "labelId": 1
-            },{
-                "labelName": "邻居",
-                "labelId": 2
-            }]
-        }
 
 @label_router.post("/modifymembers", response_model=None)
 async def modifymembers_label(request: Request,  body: LabelModifyRequest, ):
     wxid = body.wxid
     label_ids= body.labelIds
     wxids= body.wxIds
+
+    wxid = body.wxid
+    if not wxid:
+        return {"code": 400, "message": "wxid 不能为空"}
+                
+    k,loginfo=await request.app.state.gewe_service.get_login_info_by_wxid_async(wxid)
+
+    if not k:
+        return  {"code":404,"message":f"{wxid} 没有对应的登录信息"}
+            
+    login_status=loginfo.get('status','0')
+            
+    if login_status != '1':
+            return {"code": 401, "message": f"{wxid} 已经离线"}
+            
+    token_id=loginfo.get('tokenId','')
+    app_id=loginfo.get('appId','')
+
+    ret,msg,data=await request.app.state.gewe_service.label_modify_members_async(token_id,app_id,label_ids,wxids)
+    if ret!=200:
+        return {'code':ret,'message':msg}   
+
     return {"message":"操作成功"}
 

app/endpoints/pipeline_endpoint.py

@@ -660,6 +660,7 @@ async def handle_text_group_async(request: Request,token_id,app_id, wxid,msg_dat
             await request.app.state.gewe_service.post_text_async(token_id,app_id,callback_to_user,reply_content)
             await request.app.state.gewe_service.save_session_messages_to_cache_async(hash_key, {"role": "assistant", "content": reply_content})
             # 回复的对话
+            reply_content=f'{wxid}:\n'+ reply_content
             input_wx_content_dialogue_message=[{"type": "text", "text": reply_content}]
             input_message=dialogue_message(wxid,callback_to_user,input_wx_content_dialogue_message,True)
             await request.app.state.kafka_service.send_message_async(input_message)

celery_app.py

@@ -36,7 +36,7 @@ elif environment == 'test':
     scheduled_task_add_contacts_from_chatrooms_interval = 60*11
 else:
     scheduled_task_sync_wx_info_interval = 6000
-    scheduled_task_add_contacts_from_chatrooms_interval=10
+    scheduled_task_add_contacts_from_chatrooms_interval=6000
     
 
 # 定义定时任务列表 (任务 ID, 任务名称, 执行间隔秒, 任务参数)

services/gewe_service.py

@@ -10,6 +10,9 @@ import time,datetime
 import uuid
 from fastapi import FastAPI, Depends
 from common.singleton import singleton
+
+from aiohttp import ClientError
+from json.decoder import JSONDecodeError
 from common.log import logger
 from model.models import AddGroupContactsHistory
 from services.redis_service import RedisService
@@ -765,6 +768,116 @@ class GeWeService:
                 response_object = await response.json()
         return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
 
+    ###############################     标签模块   ###############################
+    async def label_add_async(self, token_id, app_id, label_name):
+        api_url = f"{self.base_url}/v2/api/label/add"
+        headers = {
+            'X-GEWE-TOKEN': token_id,
+            'Content-Type': 'application/json'
+        }
+        data = {
+            "appId": app_id,
+            "labelName": label_name,
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.post(api_url, headers=headers, json=data) as response:
+                response_object = await response.json()
+        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
+
+    async def label_delete_async(self, token_id, app_id, label_ids: list):
+        api_url = f"{self.base_url}/v2/api/label/delete"
+        label_ids_str = ','.join(map(str, label_ids))
+        headers = {
+            'X-GEWE-TOKEN': token_id,
+            'Content-Type': 'application/json'
+        }
+        data = {
+            "appId": app_id,
+            "labelIds": label_ids_str,
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.post(api_url, headers=headers, json=data) as response:
+                response_object = await response.json()
+        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)   
+
+    # async def label_list_async(self, token_id, app_id):
+    #     api_url = f"{self.base_url}/v2/api/label/list"
+    #     headers = {
+    #         'X-GEWE-TOKEN': token_id,
+    #         'Content-Type': 'application/json'
+    #     }
+    #     data = {
+    #         "appId": app_id,
+    #     }
+    #     async with aiohttp.ClientSession() as session:
+    #         async with session.post(api_url, headers=headers, json=data) as response:
+    #             response_object = await response.json()
+    #     return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)  
+
+    async def label_list_async(self, token_id, app_id):
+        api_url = f"{self.base_url}/v2/api/label/list"
+        headers = {
+            'X-GEWE-TOKEN': token_id,
+            'Content-Type': 'application/json'
+        }
+        data = {
+            "appId": app_id,
+        }
+        
+        try:
+            async with aiohttp.ClientSession() as session:
+                try:
+                    async with session.post(api_url, headers=headers, json=data) as response:
+                        # 检查响应状态码
+                        if response.status != 200:
+                            return response.status, f"HTTP Error: {response.status}", None
+                        
+                        try:
+                            response_object = await response.json()
+                        except JSONDecodeError:
+                            return 501, "Invalid JSON response", None
+                        
+                        # 检查返回的数据结构
+                        if not isinstance(response_object, dict):
+                            return 501, "Invalid response format", None
+                        
+                        # 返回处理后的数据
+                        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)
+                
+                except ClientError as e:
+                    return 501, f"Network error: {str(e)}", None
+        
+        except Exception as e:
+            return None, f"Unexpected error: {str(e)}", None  
+    
+    async def label_modify_members_async(self, token_id, app_id,label_ids: list,wx_ids: list):
+        '''
+        注意
+        由于好友标签信息存储在用户客户端，因此每次在修改时都需要进行全量修改。举例来说，考虑好友A（wxid_asdfaihp123），该好友已经被标记为标签ID为1和2。
+        
+        在添加标签ID为3时，传递的参数如下：labelIds：1,2,3，wxIds：[wxid_asdfaihp123]。这表示要给好友A添加标签ID为3，同时保留已有的标签ID 1和2。
+        
+        而在删除标签ID为1时，传递的参数如下：labelIds：2,3 ，wxIds：[wxid_asdfaihp123]。这表示要将好友A的标签ID 1删除，而保留标签ID 2。
+        '''
+
+        api_url = f"{self.base_url}/v2/api/label/modifyMemberList"
+        label_ids_str = ','.join(map(str, label_ids))
+
+        headers = {
+            'X-GEWE-TOKEN': token_id,
+            'Content-Type': 'application/json'
+        }
+        data = {
+            "appId": app_id,
+            "wxIds": wx_ids,
+            "labelIds": label_ids_str
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.post(api_url, headers=headers, json=data) as response:
+                response_object = await response.json()
+        return response_object.get('ret', None), response_object.get('msg', None), response_object.get('data', None)  
+
+
     ###############################     其他   ###############################
     
     async def save_session_messages_to_cache_async(self, hash_key,item:object)->list:
@@ -1105,7 +1218,6 @@ class GeWeService:
         
         return ""    
 
-
     async def save_login_wx_captch_code_to_cache_async(self,tel,captch_code):
         hash_key = f"__AI_OPS_WX__:WXCAPTCHCODE:{tel}"
         await self.redis_service.set_hash(hash_key,{"data":captch_code},30) 
@@ -1177,9 +1289,7 @@ class GeWeService:
                     if len(today_list) == 200:
                         return True 
         return False
-
-        
-        
+    
     async def enqueue_to_add_contacts_async(self,wxid,scene:int,v3,v4):
         """
         入列待添加好友