perf(wecom): add supports for images

libs/wecom_api/api.py

@@ -1,11 +1,14 @@
 from quart import request
 from .WXBizMsgCrypt3 import WXBizMsgCrypt
-
+import base64
+import binascii
 import httpx
 from quart import Quart
 import xml.etree.ElementTree as ET
 from typing import Callable, Dict, Any
 from .wecomevent import WecomEvent
+from pkg.platform.types import events as platform_events, message as platform_message
+import aiofiles
 
 
 class WecomClient():
@@ -42,7 +45,6 @@ class WecomClient():
             else:
                 raise Exception(f"未获取access token: {data}")
 
-
     async def get_users(self):
         if not self.check_access_token_for_contacts():
             self.access_token_for_contacts = await self.get_access_token(self.secret_for_contacts)
@@ -88,6 +90,30 @@ class WecomClient():
                 data = response.json()
                 if data['errcode'] != 0:
                     raise Exception("Failed to send message: "+str(data))
+
+    async def send_image(self,user_id:str,agent_id:int,media_id:str):
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+        url = self.base_url+'/media/upload?access_token='+self.access_token
+        async with httpx.AsyncClient() as client:
+            params = {
+                "touser" : user_id,
+                "toparty" : "",
+                "totag":"",
+                "agentid" : agent_id,
+                "msgtype" : "image",
+                "image" : {
+                    "media_id" : media_id,
+                },
+                "safe":0,
+                "enable_id_trans": 0,
+                "enable_duplicate_check": 0,
+                "duplicate_check_interval": 1800
+            }
+            response = await client.post(url,json=params)
+            data = response.json()
+            if data['errcode'] != 0:
+                raise Exception("Failed to send image: "+str(data))
             
     async def send_private_msg(self,user_id:str, agent_id:int,content:str):
         if not await self.check_access_token():
@@ -188,13 +214,92 @@ class WecomClient():
             "MsgId": int(root.find("MsgId").text) if root.find("MsgId") is not None else None,
             "AgentID": int(root.find("AgentID").text) if root.find("AgentID") is not None else None,
         }
-        return message_data
-
-
-
-
-            
-
-
+        if message_data["MsgType"] == "image":
+            message_data["MediaId"] = root.find("MediaId").text if root.find("MediaId") is not None else None
+            message_data["PicUrl"] = root.find("PicUrl").text if root.find("PicUrl") is not None else None
     
-    
+        return message_data
+    
+    @staticmethod
+    async def get_image_type(image_bytes: bytes) -> str:
+        """
+        通过图片的magic numbers判断图片类型
+        """
+        magic_numbers = {
+            b'\xFF\xD8\xFF': 'jpg',
+            b'\x89\x50\x4E\x47': 'png',
+            b'\x47\x49\x46': 'gif',
+            b'\x42\x4D': 'bmp',
+            b'\x00\x00\x01\x00': 'ico'
+        }
+    
+        for magic, ext in magic_numbers.items():
+            if image_bytes.startswith(magic):
+                return ext
+        return 'jpg'  # 默认返回jpg
+    
+
+    async def upload_to_work(self, image: platform_message.Image):
+        """
+        获取 media_id
+        """
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+
+        url = self.base_url + '/media/upload?access_token=' + self.access_token + '&type=file'
+        file_bytes = None
+        file_name = "uploaded_file.txt"
+
+        # 获取文件的二进制数据
+        if image.path:
+            async with aiofiles.open(image.path, 'rb') as f:
+                file_bytes = await f.read()
+                file_name = image.path.split('/')[-1]
+        elif image.url:
+            file_bytes = await self.download_image_to_bytes(image.url)
+            file_name = image.url.split('/')[-1]
+        elif image.base64:
+            try:
+                base64_data = image.base64
+                if ',' in base64_data:
+                    base64_data = base64_data.split(',', 1)[1]
+                padding = 4 - (len(base64_data) % 4) if len(base64_data) % 4 else 0
+                padded_base64 = base64_data + '=' * padding
+                file_bytes = base64.b64decode(padded_base64)
+            except binascii.Error as e:
+                raise ValueError(f"Invalid base64 string: {str(e)}")
+        else:
+            raise ValueError("image对象出错")
+
+        # 设置 multipart/form-data 格式的文件
+        boundary = "-------------------------acebdf13572468"
+        headers = {
+            'Content-Type': f'multipart/form-data; boundary={boundary}'
+        }
+        body = (
+            f"--{boundary}\r\n"
+            f"Content-Disposition: form-data; name=\"media\"; filename=\"{file_name}\"; filelength={len(file_bytes)}\r\n"
+            f"Content-Type: application/octet-stream\r\n\r\n"
+        ).encode('utf-8') + file_bytes + f"\r\n--{boundary}--\r\n".encode('utf-8')
+
+        # 上传文件
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, headers=headers, content=body)
+            data = response.json()
+            if data.get('errcode', 0) != 0:
+                raise Exception("failed to upload file")
+
+            return data.get('media_id')
+
+
+    async def download_image_to_bytes(self,url:str) -> bytes:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url)
+            response.raise_for_status()
+            return response.content
+
+    #进行media_id的获取
+    async def get_media_id(self, image: platform_message.Image):
+
+        media_id = await self.upload_to_work(image=image)
+        return media_id

libs/wecom_api/wecomevent.py

@@ -35,6 +35,13 @@ class WecomEvent(dict):
             str: 事件类型。
         """
         return self.get("MsgType", "")
+    
+    @property
+    def picurl(self) -> str:
+        """
+        图片链接
+        """
+        return self.get("PicUrl")
 
     @property
     def detail_type(self) -> str: