have a little magic

bilibili_api/utils/network_httpx.py

@@ -4,16 +4,18 @@
 复写了 .utils.network，使用 httpx
 """
 
-from typing import Any, Union
+from typing import Any, Callable, Dict, Optional, Union
 from urllib.parse import quote
 import json
 import re
 import asyncio
 import atexit
 import uuid
-
+from aiohttp import CookieJar
+from websockets.sync.client import connect
+from inspect import iscoroutinefunction as isAsync
 import httpx
-
+from dataclasses import dataclass
 from ..exceptions import ResponseCodeException, ResponseException, NetworkException
 from .Credential import Credential
 from .. import settings
@@ -210,3 +212,77 @@ def to_form_urlencoded(data: dict) -> str:
         temp.append(f'{k}={quote(str(v)).replace("/", "%2F")}')
 
     return "&".join(temp)
+
+@dataclass
+class Request:
+    url:str
+    method:str
+    path:Optional[Dict[str,Any]] = None
+    query:Optional[Dict[str,Any]] = None
+    body:Optional[Dict[str,Any]] = None
+    form:Optional[Dict[str,Any]] = None
+    cookies:Optional['CookieJar'] = None
+    json:Optional[Dict[str,Any]] = None
+    timeout:Optional[int] = None
+    headers:Optional[Dict[str,Any]] = None
+    def formatURLPath(self,url:str,data:Optional[Dict[str,Any]]= None):
+        if(data != None):
+            backupURL = url
+            for key in data:
+                backupURL = backupURL.replace("{%s}"%(key),str(data.get(key)),1)
+                backupURL = backupURL.replace(":%s"%(key),str(data.get(key)),1)
+            return backupURL
+        return url
+    def covernCookies(self,cookies:Optional[Dict[str,Any]] = None):
+        if cookies != None:
+            # TODO: implement dict to cookiesJar
+            cookiejarInstance = CookieJar()
+            return cookiejarInstance
+    def update(self, **kwargs):
+        "更新参数"
+        self.url = kwargs.get("url", self.url)
+        self.path = kwargs.get("path", self.path)
+        self.method = kwargs.get("method", self.method)
+        overrideHeaders: Optional[Dict[str,Any]] = kwargs.get("headers")
+        if overrideHeaders is not None:
+            self.headers = self.headers or {}
+            for key in overrideHeaders:
+                self.headers[key] = overrideHeaders[key]
+        self.query = kwargs.get("query", self.query)
+        self.json = kwargs.get("json", self.json)
+    def __call__(self,func:Callable):
+        if isAsync(func):
+            async def wapperAsync(*args, **kw):
+                session =  get_session()
+                self.update(kwargs=kw)
+                if(self.method == "ws" or self.method == "websocket"):
+                    with connect(self.url) as websocket:
+                        websocket.send(self.body or self.json or "")
+                        res = websocket.recv()
+                res = await session.request(
+                    method=self.method,
+                    url=self.formatURLPath(self.url,self.path),
+                    json=self.json,
+                    headers=self.headers,
+                    params=self.query,
+                    timeout=self.timeout
+                )
+                return func(response=res,*args,**kw)
+            return wapperAsync
+        else:
+            def wapper(*args, **kw):
+                self.update(kwargs=kw)
+                if(self.method == "ws" or self.method == "websocket"):
+                    with connect(self.url) as websocket:
+                        websocket.send(self.body or self.json or "")
+                        res = websocket.recv()
+                res = httpx.request(
+                    method=self.method,
+                    url=self.formatURLPath(self.url,self.path),
+                    json=self.json,
+                    headers=self.headers,
+                    params=self.query,
+                    timeout=self.timeout
+                )
+                return func(response=res,*args,**kw)
+            return wapper

requirements.txt

@@ -12,4 +12,5 @@ APScheduler~=3.10.1
 rsa~=4.9
 pillow~=9.5.0
 tqdm~=4.65.0
-yarl~=1.9.2
+yarl~=1.9.2
+websockets~=10.0.3