Merge pull request #2938 from InfinityPacer/feature/push

app/modules/telegram/telegram.py

@@ -206,14 +206,15 @@ class Telegram:
         """
         向Telegram发送报文
         """
-
         if image:
             res = RequestUtils(proxies=settings.PROXY).get_res(image)
             if res is None:
                 raise Exception("获取图片失败")
             if res.content:
                 # 使用随机标识构建图片文件的完整路径，并写入图片内容到文件
-                image_file = Path(settings.TEMP_PATH) / str(uuid.uuid4())
+                image_file = Path(settings.TEMP_PATH) / "telegram" / str(uuid.uuid4())
+                if not image_file.parent.exists():
+                    image_file.parent.mkdir(parents=True, exist_ok=True)
                 image_file.write_bytes(res.content)
                 photo = InputFile(image_file)
                 # 发送图片到Telegram
@@ -223,8 +224,7 @@ class Telegram:
                                            parse_mode="Markdown")
                 if ret is None:
                     raise Exception("发送图片消息失败")
-                if ret:
+                return True
-                    return True
         # 按4096分段循环发送消息
         ret = None
         if len(caption) > 4095:

app/modules/wechat/wechat.py

@@ -10,6 +10,7 @@ from app.log import logger
 from app.utils.common import retry
 from app.utils.http import RequestUtils
 from app.utils.string import StringUtils
+from app.utils.url import UrlUtils
 
 lock = threading.Lock()
 
@@ -31,14 +32,16 @@ class WeChat:
     _proxy = None
 
     # 企业微信发送消息URL
-    _send_msg_url = "/cgi-bin/message/send?access_token=%s"
+    _send_msg_url = "/cgi-bin/message/send?access_token={access_token}"
     # 企业微信获取TokenURL
-    _token_url = "/cgi-bin/gettoken?corpid=%s&corpsecret=%s"
+    _token_url = "/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}"
-    # 企业微信创新菜单URL
+    # 企业微信创建菜单URL
-    _create_menu_url = "/cgi-bin/menu/create?access_token=%s&agentid=%s"
+    _create_menu_url = "/cgi-bin/menu/create?access_token={access_token}&agentid={agentid}"
+    # 企业微信删除菜单URL
+    _delete_menu_url = "/cgi-bin/menu/delete?access_token={access_token}&agentid={agentid}"
 
-    def __init__(self, WECHAT_CORPID: str = None, WECHAT_APP_SECRET: str = None, WECHAT_APP_ID: str = None,
+    def __init__(self, WECHAT_CORPID: str = None, WECHAT_APP_SECRET: str = None,
-                 WECHAT_PROXY: str = None, **kwargs):
+                 WECHAT_APP_ID: str = None, WECHAT_PROXY: str = None, **kwargs):
         """
         初始化
         """
@@ -51,10 +54,10 @@ class WeChat:
         self._proxy = WECHAT_PROXY or "https://qyapi.weixin.qq.com"
 
         if self._proxy:
-            self._proxy = self._proxy.rstrip("/")
+            self._send_msg_url = UrlUtils.adapt_request_url(self._proxy, self._send_msg_url)
-            self._send_msg_url = f"{self._proxy}/{self._send_msg_url}"
+            self._token_url = UrlUtils.adapt_request_url(self._proxy, self._token_url)
-            self._token_url = f"{self._proxy}/{self._token_url}"
+            self._create_menu_url = UrlUtils.adapt_request_url(self._proxy, self._create_menu_url)
-            self._create_menu_url = f"{self._proxy}/{self._create_menu_url}"
+            self._delete_menu_url = UrlUtils.adapt_request_url(self._proxy, self._delete_menu_url)
 
         if self._corpid and self._appsecret and self._appid:
             self.__get_access_token()
@@ -63,7 +66,7 @@ class WeChat:
         """
         获取状态
         """
-        return True if self.__get_access_token else False
+        return True if self.__get_access_token() else False
 
     @retry(Exception, logger=logger)
     def __get_access_token(self, force=False):
@@ -82,13 +85,13 @@ class WeChat:
             if not self._corpid or not self._appsecret:
                 return None
             try:
-                token_url = self._token_url % (self._corpid, self._appsecret)
+                token_url = self._token_url.format(corpid=self._corpid, corpsecret=self._appsecret)
                 res = RequestUtils().get_res(token_url)
                 if res:
                     ret_json = res.json()
-                    if ret_json.get('errcode') == 0:
+                    if ret_json.get("errcode") == 0:
-                        self._access_token = ret_json.get('access_token')
+                        self._access_token = ret_json.get("access_token")
-                        self._expires_in = ret_json.get('expires_in')
+                        self._expires_in = ret_json.get("expires_in")
                         self._access_token_time = datetime.now()
                 elif res is not None:
                     logger.error(f"获取微信access_token失败，错误码：{res.status_code}，错误原因：{res.reason}")
@@ -100,8 +103,64 @@ class WeChat:
                 return None
         return self._access_token
 
+    @staticmethod
+    def __split_content(content: str, max_bytes: int = 2048) -> List[str]:
+        """
+        将内容分块为不超过 max_bytes 字节的块
+        :param content: 待拆分的内容
+        :param max_bytes: 最大字节数
+        :return: 分块后的内容列表
+        """
+        content_chunks = []
+        current_chunk = bytearray()
+
+        for line in content.splitlines():
+            encoded_line = (line + "\n").encode("utf-8")
+            line_length = len(encoded_line)
+
+            if line_length > max_bytes:
+                # 在处理长行之前，先将 current_chunk 添加到 content_chunks
+                if current_chunk:
+                    content_chunks.append(current_chunk.decode("utf-8", errors="replace").strip())
+                    current_chunk = bytearray()
+
+                # 处理长行，拆分为多个不超过 max_bytes 的块
+                start = 0
+                while start < line_length:
+                    end = start + max_bytes  # 不再需要为 "..." 预留空间
+                    if end >= line_length:
+                        end = line_length
+                    else:
+                        # 调整以避免拆分多字节字符
+                        while end > start and (encoded_line[end] & 0xC0) == 0x80:
+                            end -= 1
+                        if end == start:
+                            # 单个字符超过了 max_bytes，强制包含整个字符
+                            end = start + 1
+                            while end < line_length and (encoded_line[end] & 0xC0) == 0x80:
+                                end += 1
+                    truncated_line = encoded_line[start:end].decode("utf-8", errors="replace")
+                    content_chunks.append(truncated_line.strip())
+                    start = end
+                continue  # 继续处理下一行
+
+            # 检查添加当前行后是否会超过 max_bytes
+            if len(current_chunk) + line_length > max_bytes:
+                # 将 current_chunk 添加到 content_chunks
+                content_chunks.append(current_chunk.decode("utf-8", errors="replace").strip())
+                current_chunk = bytearray()
+
+            # 将当前行添加到 current_chunk
+            current_chunk += encoded_line
+
+        # 处理剩余的 current_chunk
+        if current_chunk:
+            content_chunks.append(current_chunk.decode("utf-8", errors="replace").strip())
+
+        return content_chunks
+
     def __send_message(self, title: str, text: str = None,
-                       userid: str = None, link: str = None) -> Optional[bool]:
+                       userid: str = None, link: str = None) -> bool:
         """
         发送文本消息
         :param title: 消息标题
@@ -110,62 +169,41 @@ class WeChat:
         :param link: 跳转链接
         :return: 发送状态，错误信息
         """
-        message_url = self._send_msg_url % self.__get_access_token()
+        if not title:
+            logger.error("消息标题不能为空")
+            return False
         if text:
-            content = "%s\n%s" % (title, text.replace("\n\n", "\n"))
+            formatted_text = text.replace("\n\n", "\n")
+            content = f"{title}\n{formatted_text}"
         else:
             content = title
-
         if link:
             content = f"{content}\n点击查看：{link}"
-
         if not userid:
             userid = "@all"
-
+        # 分块处理逻辑
-        # Check if content exceeds 2048 bytes and split if necessary
+        content_chunks = self.__split_content(content)
-        if len(content.encode('utf-8')) > 2048:
+        # 逐块发送消息
-            content_chunks = []
+        for chunk in content_chunks:
-            current_chunk = ""
-            for line in content.splitlines():
-                if len(current_chunk.encode('utf-8')) + len(line.encode('utf-8')) > 2048:
-                    content_chunks.append(current_chunk.strip())
-                    current_chunk = ""
-                current_chunk += line + "\n"
-            if current_chunk:
-                content_chunks.append(current_chunk.strip())
-
-            # Send each chunk as a separate message
-            result = True
-            for chunk in content_chunks:
-                req_json = {
-                    "touser": userid,
-                    "msgtype": "text",
-                    "agentid": self._appid,
-                    "text": {
-                        "content": chunk
-                    },
-                    "safe": 0,
-                    "enable_id_trans": 0,
-                    "enable_duplicate_check": 0
-                }
-                result = self.__post_request(message_url, req_json)
-                if not result:
-                    return False
-        else:
             req_json = {
                 "touser": userid,
                 "msgtype": "text",
                 "agentid": self._appid,
                 "text": {
-                    "content": content
+                    "content": chunk
                 },
                 "safe": 0,
                 "enable_id_trans": 0,
                 "enable_duplicate_check": 0
             }
-            result = self.__post_request(message_url, req_json)
+            try:
-
+                # 如果是超长消息，有一个发送失败就全部失败
-        return result
+                if not self.__post_request(self._send_msg_url, req_json):
+                    return False
+            except Exception as e:
+                logger.error(f"发送消息块失败：{e}")
+                return False
+        return True
 
     def __send_image_message(self, title: str, text: str, image_url: str,
                              userid: str = None, link: str = None) -> Optional[bool]:
@@ -178,7 +216,6 @@ class WeChat:
         :param link: 跳转链接
         :return: 发送状态，错误信息
         """
-        message_url = self._send_msg_url % self.__get_access_token()
         if text:
             text = text.replace("\n\n", "\n")
         if not userid:
@@ -198,7 +235,11 @@ class WeChat:
                 ]
             }
         }
-        return self.__post_request(message_url, req_json)
+        try:
+            return self.__post_request(self._send_msg_url, req_json)
+        except Exception as e:
+            logger.error(f"发送图文消息失败：{e}")
+            return False
 
     def send_msg(self, title: str, text: str = "", image: str = "",
                  userid: str = None, link: str = None) -> Optional[bool]:
@@ -230,7 +271,6 @@ class WeChat:
             logger.error("获取微信access_token失败，请检查参数配置")
             return None
 
-        message_url = self._send_msg_url % self.__get_access_token()
         if not userid:
             userid = "@all"
         articles = []
@@ -256,7 +296,11 @@ class WeChat:
                 "articles": articles
             }
         }
-        return self.__post_request(message_url, req_json)
+        try:
+            return self.__post_request(self._send_msg_url, req_json)
+        except Exception as e:
+            logger.error(f"发送消息失败：{e}")
+            return False
 
     def send_torrents_msg(self, torrents: List[Context],
                           userid: str = "", title: str = "", link: str = None) -> Optional[bool]:
@@ -272,7 +316,6 @@ class WeChat:
             self.__send_message(title=title, userid=userid, link=link)
 
         # 发送列表
-        message_url = self._send_msg_url % self.__get_access_token()
         if not userid:
             userid = "@all"
         articles = []
@@ -289,11 +332,11 @@ class WeChat:
                             f"{StringUtils.str_filesize(torrent.size)} " \
                             f"{torrent.volume_factor} " \
                             f"{torrent.seeders}↑"
-            title = re.sub(r"\s+", " ", title).strip()
+            torrent_title = re.sub(r"\s+", " ", torrent_title).strip()
             articles.append({
                 "title": torrent_title,
-                "description": torrent.description if index == 1 else '',
+                "description": torrent.description if index == 1 else "",
-                "picurl": mediainfo.get_message_image() if index == 1 else '',
+                "picurl": mediainfo.get_message_image() if index == 1 else "",
                 "url": torrent.page_url
             })
             index += 1
@@ -306,35 +349,41 @@ class WeChat:
                 "articles": articles
             }
         }
-        return self.__post_request(message_url, req_json)
+        try:
+            return self.__post_request(self._send_msg_url, req_json)
+        except Exception as e:
+            logger.error(f"发送消息失败：{e}")
+            return False
 
-    def __post_request(self, message_url: str, req_json: dict) -> bool:
+    @retry(Exception, logger=logger)
+    def __post_request(self, url: str, req_json: dict) -> bool:
         """
         向微信发送请求
         """
-        try:
+        url = url.format(access_token=self.__get_access_token())
-            res = RequestUtils(content_type='application/json').post(
+        res = RequestUtils(content_type="application/json").post(
-                message_url,
+            url=url,
-                data=json.dumps(req_json, ensure_ascii=False).encode('utf-8')
+            data=json.dumps(req_json, ensure_ascii=False).encode("utf-8")
-            )
+        )
-            if res and res.status_code == 200:
+        if res is None:
-                ret_json = res.json()
+            error_msg = "发送请求失败，未获取到返回信息"
-                if ret_json.get('errcode') == 0:
+            raise Exception(error_msg)
-                    return True
+        if res.status_code != 200:
-                else:
+            error_msg = f"发送请求失败，错误码：{res.status_code}，错误原因：{res.reason}"
-                    if ret_json.get('errcode') == 42001:
+            raise Exception(error_msg)
-                        self.__get_access_token(force=True)
+
-                    logger.error(f"发送请求失败，错误信息：{ret_json.get('errmsg')}")
+        ret_json = res.json()
-                    return False
+        if ret_json.get("errcode") == 0:
-            elif res is not None:
+            return True
-                logger.error(f"发送请求失败，错误码：{res.status_code}，错误原因：{res.reason}")
+        else:
-                return False
+            if ret_json.get("errcode") == 42001:
+                self.__get_access_token(force=True)
+                error_msg = (f"access_token 已过期，尝试重新获取 access_token,"
+                             f"errcode: {ret_json.get('errcode')}, errmsg: {ret_json.get('errmsg')}")
+                raise Exception(error_msg)
             else:
-                logger.error(f"发送请求失败，未获取到返回信息")
+                logger.error(f"发送请求失败，错误信息：{ret_json.get('errmsg')}")
                 return False
-        except Exception as err:
-            logger.error(f"发送请求失败，错误信息：{str(err)}")
-            return False
 
     def create_menus(self, commands: Dict[str, dict]):
         """
@@ -376,12 +425,12 @@ class WeChat:
         }
         """
         # 请求URL
-        req_url = self._create_menu_url % (self.__get_access_token(), self._appid)
+        req_url = self._create_menu_url.format(access_token="{access_token}", agentid=self._appid)
 
         # 对commands按category分组
         category_dict = {}
         for key, value in commands.items():
-            category: Dict[str, dict] = value.get("category")
+            category: str = value.get("category")
             if category:
                 if not category_dict.get(category):
                     category_dict[category] = {}
@@ -405,6 +454,19 @@ class WeChat:
 
         if buttons:
             # 发送请求
-            self.__post_request(req_url, {
+            try:
-                "button": buttons[:3]
+                self.__post_request(req_url, {
-            })
+                    "button": buttons[:3]
+                })
+            except Exception as e:
+                logger.error(f"创建菜单失败：{e}")
+                return False
+
+    def delete_menus(self):
+        """
+        删除微信菜单
+        """
+        # 请求URL
+        req_url = self._delete_menu_url.format(access_token=self.__get_access_token(), agentid=self._appid)
+        # 发送请求
+        RequestUtils().get(req_url)