LangBot/pkg/utils/center/apigroup.py

import abc
import uuid
import json
import logging
import threading

import requests


class APIGroup(metaclass=abc.ABCMeta):
    """API 组抽象类"""
    _basic_info: dict = None
    _runtime_info: dict = None

    prefix = None

    def __init__(self, prefix: str):
        self.prefix = prefix

    def do(
        self,
        method: str,
        path: str,
        data: dict = None,
        params: dict = None,
        headers: dict = {},
        **kwargs
    ):
        """执行一个请求"""
        def thr_wrapper(
            self,
            method: str,
            path: str,
            data: dict = None,
            params: dict = None,
            headers: dict = {},
            **kwargs
        ):
            try:
                url = self.prefix + path
                data = json.dumps(data)
                headers['Content-Type'] = 'application/json'
                
                ret =  requests.request(
                    method,
                    url,
                    data=data,
                    params=params,
                    headers=headers,
                    **kwargs
                )

                logging.debug("data: %s", data)

                logging.debug("ret: %s", ret.json())
            except Exception as e:
                logging.debug("上报数据失败: %s", e)
        
        thr = threading.Thread(target=thr_wrapper, args=(
            self,
            method,
            path,
            data,
            params,
            headers,
        ), kwargs=kwargs)
        thr.start()

            
    def gen_rid(
        self
    ):
        """生成一个请求 ID"""
        return str(uuid.uuid4())

    def basic_info(
        self
    ):
        """获取基本信息"""
        basic_info = APIGroup._basic_info.copy()
        basic_info['rid'] = self.gen_rid()
        return basic_info
    
    def runtime_info(
        self
    ):
        """获取运行时信息"""
        return APIGroup._runtime_info