utils/authenticator.py

@@ -11,7 +11,6 @@ from pathlib import Path
 import threading
 import time
 from typing import Optional, Tuple
-from datetime import datetime
 
 from request import Request
 
@@ -71,20 +70,19 @@ class Authenticator:
             )  # 指定服务商的访问令牌和失效时间戳
             if time.time() > expired_timestamp:
                 match servicer:
-                    # 获取深圳快瞳访问凭证cd C:\Python\.venv\Scripts
+                    # 刷新深圳快瞳访问凭证cd C:\Python\.venv\Scripts
                     case "inspirvision":
                         token, expired_timestamp = (
-                            self._get_inspirvision_certification()
+                            self._refresh_inspirvision_certification()
                         )
-                    # 获取合力亿捷访问凭证
+                    # 刷新合力亿捷访问凭证
                     case "hollycrm":
-                        token, expired_timestamp = self._get_hollycrm_certification()
+                        token, expired_timestamp = (
-                    # 获取飞书访问凭证
+                            self._refresh_hollycrm_certification()
+                        )
+                    # 刷新飞书访问凭证
                     case "feishu":
-                        token, expired_timestamp = self._get_feishu_certification()
+                        token, expired_timestamp = self._refresh_feishu_certification()
-                    # 获取 Cloudreve 访问凭证
-                    case "cloudreve":
-                        token, expired_timestamp = self._get_cloudreve_certification()
                     case _:
                         raise RuntimeError(f"未知服务商：{servicer}")
 
@@ -105,9 +103,9 @@ class Authenticator:
                     )
             return token
 
-    def _get_inspirvision_certification(self) -> Tuple[str, float]:
+    def _refresh_inspirvision_certification(self) -> Tuple[str, float]:
         """
-        获取深圳快瞳访问凭证
+        刷新深圳快瞳访问凭证
         :return: 深圳快瞳访问令牌和失效时间戳
         """
         response = self.request.get(
@@ -115,15 +113,15 @@ class Authenticator:
         )
         # 若非响应成功则抛出异常
         if not (response["status"] == 200 and response["code"] == 0):
-            raise RuntimeError("获取深圳快瞳访问凭证发生异常")
+            raise RuntimeError("刷新深圳快瞳访问凭证发生异常")
         return (
             response["data"]["access_token"],
             time.time() + response["data"]["expires_in"],
         )  # https://inspirvision.cn/spa/documentCenter/ocr-finance/mark?id=3 深圳快瞳获取访问凭证接口，其中 access_token 为访问令牌，expires_in 为有效剩余时间（单位为秒）
 
-    def _get_hollycrm_certification(self) -> Tuple[str, float]:
+    def _refresh_hollycrm_certification(self) -> Tuple[str, float]:
         """
-        获取合力亿捷访问凭证
+        刷新合力亿捷访问凭证
         :return: 合力亿捷访问令牌和失效时间戳
         """
         # 企业访问标识
@@ -144,15 +142,15 @@ class Authenticator:
         )
         # 若非响应成功则抛出异常
         if not response["success"]:
-            raise RuntimeError("获取合力亿捷访问凭证发生异常")
+            raise RuntimeError("刷新合力亿捷访问凭证发生异常")
         return (
             response["data"],
             time.time() + 1 * 3600,  # 访问令牌有效期为1小时
         )
 
-    def _get_feishu_certification(self) -> Tuple[str, float]:
+    def _refresh_feishu_certification(self) -> Tuple[str, float]:
         """
-        获取飞书访问凭证
+        刷新飞书访问凭证
         :return: 飞书访问令牌和失效时间戳
         """
         response = self.request.post(
@@ -164,30 +162,8 @@ class Authenticator:
         )
         # 若非响应成功则抛出异常
         if not response["code"] == 0:
-            raise RuntimeError("获取飞书访问凭证发生异常")
+            raise RuntimeError("刷新飞书访问凭证发生异常")
         return (
             response["tenant_access_token"],
             time.time() + response["expire"],
         )  # https://open.feishu.cn/document/server-docs/api-call-guide/calling-process/get-access-token 飞书获取访问凭证接口
-
-    def _get_cloudreve_certification(self) -> Tuple[str, float]:
-        """
-        获取 Cloudreve 访问凭证
-        :return: Cloudreve 访问令牌和失效时间戳
-        """
-        response = self.request.post(
-            url="https://cloudreve.liubiren.cloud/api/v4/session/token",
-            json={
-                "email": "marslbr@qq.com",
-                "password": "Cl198752",
-            },
-        )
-        # 若非响应成功则抛出异常
-        if not response["code"] == 0:
-            raise RuntimeError("获取 Cloudreve 访问凭证发生异常")
-        return (
-            response["data"]["token"]["access_token"],
-            datetime.fromisoformat(
-                response["data"]["token"]["access_expires"]
-            ).timestamp(),
-        )  # https://docs.cloudreve.org/zh/api/auth 通过登录获取 AccessToken，时效时重新登录重新获取

utils/certifications.json

@@ -1,5 +1 @@
-<<<<<<< HEAD
+{"feishu": ["t-g1043kdp74T7OBBWKU32LNRRUIOSIE4Q77JNQTEZ", 1773991516.999697]}
-{"feishu": ["t-g1043mboYRA4SDDKUY5RTRLCHPF3YGJQIYCLGDVP", 1774157093.7511718]}
-=======
-{"feishu": ["t-g1043kdp74T7OBBWKU32LNRRUIOSIE4Q77JNQTEZ", 1773991516.999697]}
->>>>>>> 239b7d7f2837822ce7a85afdcb85bafc4888d6f7

utils/cloudreve.py

@@ -1,69 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-封装 Cloudreve 客户端
-"""
-
-from email.parser import BytesParser
-from email.policy import default
-from email.utils import parsedate_to_datetime
-from imaplib import IMAP4_SSL
-import re
-from time import time
-from typing import Any, Dict, Optional, List
-
-from pydantic.config import JsonEncoder
-
-from authenticator import Authenticator
-from request import Request
-from base64 import b64encode
-from urllib.parse import quote
-
-
-class Cloudreve:
-    """Cloudreve 客户端"""
-
-    def __init__(self):
-        # 实例化认证器
-        self.authenticator = Authenticator()
-        # 实例化请求客户端
-        self.http_client = Request()
-
-    def _create_upload_session(self, uri: str, size: int) -> str:
-        """
-        创建上传会话
-        :param uri: 统一资源标识符
-        :param size: 文件大小
-        :return: 上传会话标识
-        """
-        response = self.http_client.put(
-            url=f"https://cloudreve.liubiren.cloud/api/v4/file/upload",
-            headers={
-                "Authorization": f"Bearer {self.authenticator.get_token(servicer="cloudreve")}",
-                "Content-Type": "application/json; charset=utf-8",
-            },
-            json={
-                "uri": quote(string=uri, safe=":/?&="),  # 编码统一资源标识
-                "size": size,
-            },
-        )
-        # 若非响应成功则抛出异常
-        if not response["code"] == 0:
-            raise RuntimeError("创建上传会话发生异常")
-        return response["data"]["session_id"]
-
-    def _upload_file_chunk(self, session_id: str, index: int, file_chunk: bytes) -> str:
-        """
-        上传文件块
-        :param session_id: 上传会话标识
-        :param index: 文件块索引
-        :return: 上传文件标识
-        """
-        response = self.http_client.post(
-            url=f"https://cloudreve.liubiren.cloud/api/v4/file/upload/{session_id}/{index}",
-            headers={
-                "Authorization": f"Bearer {self.authenticator.get_token(servicer="cloudreve")}",
-                "Content-Type": "application/octet-stream",
-                "Content-Length": len(file_chunk),
-            },
-            json=open(file_path, "rb").read(),
-        )

utils/feishu.py

@@ -38,7 +38,7 @@ class Feishu:
         """
         使用示例：
         feishu = Feishu()
-        feishu.get_mail_verification_code(folder="邮箱验证码", regular_expression=r"【普康健康】您的验证码是：(\\d+)")
+        print(feishu.get_mail_verification_code(folder="邮箱验证码", regular_expression=r"【普康健康】您的验证码是：(\\d+)"))
         输出：123456
         """
         if not folder:
@@ -209,20 +209,15 @@ class Feishu:
 
         return records
 
-    def _convert_to_cloudreve_direct_link(self, material_token: str) -> str:
+    def download_material(self, file_token: str, stream_enabled: bool = False) -> str:
         """
-        转为Cloudreve直链
+        下载素材
-        :param material_token: 素材标识
+        :param file_token: 素材标识
-        :return: 素材直链地址
+        :param stream_enabled: 使用流式传输，默认 False
+        :return: 素材 base64 编码的字符串
         """
-        # 获取 Cloudreve 上传 session_id
-        session_id = self.http_client.
-
-
-
-
         # 构建下载素材的请求地址
-        url = f"https://open.feishu.cn/open-apis/drive/v1/medias/{material_token}/download"  # https://open.feishu.cn/document/server-docs/docs/drive-v1/media/download
+        url = f"https://open.feishu.cn/open-apis/drive/v1/medias/{file_token}/download"  # https://open.feishu.cn/document/server-docs/docs/drive-v1/media/download
 
         headers = self._get_headers()
         # 添加 Content-Type 请求头
@@ -235,18 +230,12 @@ class Feishu:
         response = self.http_client.download(
             url=url,
             headers=headers,
-<<<<<<< HEAD
-            stream_enabled=True,
-        )  # 默认使用流式传输
-        
-=======
             stream_enabled=stream_enabled,
         )
         print(type(response[0]))
         print(type(response[1]))
         print(type(response[2]))
 
->>>>>>> 239b7d7f2837822ce7a85afdcb85bafc4888d6f7
 
 
 a = Feishu()

utils/request.py

@@ -171,7 +171,8 @@ class Caches(SQLite):
 class Request:
     """
     请求客户端，支持：
-    GET、PUT和POST等请求
+    get：GET请求
+    post：POST请求
     download：下载
     """
 
@@ -250,14 +251,6 @@ class Request:
         """
         return self._request(method="GET", parameters=Parameters(**kwargs))
 
-    def put(self, **kwargs) -> Any:
-        """
-        GET请求
-        :param kwargs: 请求参数
-        :return: 响应内容
-        """
-        return self._request(method="PUT", parameters=Parameters(**kwargs))
-
     def post(self, **kwargs) -> Any:
         """
         POST请求
@@ -271,12 +264,12 @@ class Request:
         return self._request(method="POST", parameters=Parameters(**kwargs))
 
     def download(
-        self, stream_enabled: bool = False, chunk_size: int = 1024 * 1024, **kwargs
+        self, stream_enabled: bool = False, chunk_size: int = 1024, **kwargs
     ) -> Any:
         """
         下载
-        :param stream_enabled: 使用流式传输，默认为关闭流式传输
+        :param stream_enabled: 使用流式传输
-        :param chunk_size: 分块大小，若开启流式传输则分块大小默认为 1MB
+        :param chunk_size: 流式传输的分块大小
         :param kwargs: 请求参数
         :return: 响应内容
         """
@@ -284,18 +277,15 @@ class Request:
             method="GET",
             parameters=Parameters(**{"stream_enabled": stream_enabled, **kwargs}),
         )
-        # 若使用流式传输则返回响应内容迭代器
+        # 若使用流式传输则处理流式传输响应
         if stream_enabled:
             return self._process_stream_response(
                 response=response, chunk_size=chunk_size
             )
-
         return response
 
     @restrict(max_tokens=5, refill_rate=5.0)
-    def _request(
+    def _request(self, method: Literal["GET", "POST"], parameters: Parameters) -> Any:
-        self, method: Literal["GET", "PUT", "POST"], parameters: Parameters
-    ) -> Any:
         """
         请求
         :param method: 请求方法
@@ -403,9 +393,11 @@ class Request:
 
     # 处理流式传输响应
     @staticmethod
-    def _process_stream_response(response: Response, chunk_size: int) -> Generator:
+    def _process_stream_response(
+        response: Response, chunk_size: int
+    ) -> Generator[bytes, None, None]:
         """
-        处理流式传输响应
+        处理流式响应
         :param response: 响应对象
         :param chunk_size: 分块大小
         :return: 响应内容迭代器