根据 yml 文件自动生成函数 #307

Drelf2018 · 2023-05-27T10:12:30Z

Drelf2018
May 27, 2023
Collaborator

其实 #288 这个我前段时间研究了下，但是不知道有啥用

from typing import Callable, Dict, List, Optional, Tuple, TypeVar

from yaml import FullLoader, load

from bilibili_api.utils.network_httpx import request

T = TypeVar("T")


def dfs(
    root: Optional[T] = None,
    getchildren: Callable[[T], List[T]] = None,
):
    """
    ### 深度优先搜索
    
    root: 起点 不为空时自动启动一次
    
    getchildren: 获取子节点函数
    """

    def warpper(preorder: Callable):
        def run(root: T):
            dict_path = []
            nodes: List[Optional[Tuple[str, T]]] = [("root", root)]
            while nodes:
                node = nodes.pop(0)
                if node is None:
                    dict_path.pop(-1)
                else:
                    path, info = node
                    dict_path.append(path)
                    juction = "API" + '"]["'.join(dict_path)[6:] + '"]'
                    if preorder(juction, info) is True:
                        nodes = [*getchildren(info), None, *nodes]
                    else:
                        dict_path.pop(-1)
        if root is not None:
            run(root)
        return run
    return warpper


with open("user.yml", "r", encoding="utf-8") as fp:
    API: dict = load(fp, FullLoader)


function_format = '''

async def {function}({variable_params}) -> {result}:
    """
    {comment}
    """

    api = {api}
    params = {constant_params}
    return await request("{method}", api["url"], params=params)
'''


def write_function(s: str, path: str = __file__):
    with open(path, "a", encoding="utf-8") as fp:
        fp.write(s)


@dfs(API, lambda node: node.items())
def makeFunction(path: str, node: dict):
    name = node.get("function")
    if name is None:
        return True
    params: Dict[str, dict] = node["params"]

    variable = []
    constant = []

    for param, info in params.items():
        if info.get("variable", False):
            hint = param
            if info.get("type") is not None:
                hint += f": {info['type']}"
            if info.get("value") is not None:
                v = info["value"]
                if isinstance(v, str):
                    v = '"' + v + '"'
                hint += f" = {v}"
            variable.append(hint)
            constant.append(f'"{param}": {param}')
        else:
            v = info["value"]
            if isinstance(v, str):
                v = '"' + v + '"'
            constant.append(f'"{param}": {v}')

    node["variable_params"] = ", ".join(variable)
    node["constant_params"] = "{\n\t\t" + ",\n\t\t".join(constant) + "\n\t}"

    node["api"] = path
    write_function(function_format.format_map(node))

同目录下有 user.yml 文件

info:
  info:
    function: get_user_info
    url: "https://api.bilibili.com/x/space/wbi/acc/info"
    method: "GET"
    verify: false
    params:
      self:
        variable: true
      mid: 
        type: int
        describe: uid
        variable: true
      token: 
        type: str
        value: ""
        describe: 密钥
      platform:
        type: str
        value: "web"
        describe: 平台
      web_location:
        type: int
        value: 1550101
    result: dict
    comment: "用户基本信息"

最后结果

async def get_user_info(self, mid: int) -> dict:
    """
    用户基本信息
    """

    api = API["info"]["info"]
    params = {
		"self": self,
		"mid": mid,
		"token": "",
		"platform": "web",
		"web_location": 1550101
	}
    return await request("GET", api["url"], params=params)

z0z0r4 · 2023-05-27T10:20:24Z

z0z0r4
May 27, 2023
Collaborator

意思是，以后添加新封装接口直接生成然后加注释？好东西qeq

1 reply

z0z0r4 May 27, 2023
Collaborator

我的打算是，改成

async def get_user_info(self, mid: int) -> dict:
    """
    用户基本信息
    """

    api = API["info"]["info"]
    return await request("GET", api, mid=mid)