91 lines
2.3 KiB
Python
91 lines
2.3 KiB
Python
from dataclasses import dataclass
|
|
import datetime
|
|
from typing import Literal
|
|
|
|
import aiohttp
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class SimpfunServerConfig(BaseModel):
|
|
plugin_simpfun_api_key: str = ""
|
|
plugin_simpfun_base_url: str = "https://api.simpfun.cn"
|
|
plugin_simpfun_instance_id: int = 0
|
|
|
|
|
|
def get_config():
|
|
from nonebot import get_plugin_config
|
|
|
|
return get_plugin_config(SimpfunServerConfig)
|
|
|
|
|
|
class PowerManageResult(BaseModel):
|
|
code: int
|
|
status: bool
|
|
msg: str
|
|
|
|
|
|
class SimpfunServerDetailUtilization(BaseModel):
|
|
memory_bytes: int
|
|
cpu_absolute: float
|
|
disk_bytes: int
|
|
network_rx_bytes: int
|
|
network_tx_bytes: int
|
|
uptime: float
|
|
disk_last_check_time: datetime.datetime
|
|
|
|
|
|
class SimpfunServerDetailData(BaseModel):
|
|
id: int
|
|
name: str
|
|
is_pro: bool
|
|
|
|
status: str
|
|
"运行中的话，是 running"
|
|
|
|
is_suspended: bool
|
|
utilization: SimpfunServerDetailUtilization
|
|
|
|
|
|
class SimpfunServerDetailResp(BaseModel):
|
|
code: int
|
|
data: SimpfunServerDetailData
|
|
|
|
|
|
@dataclass
|
|
class SimpfunServer:
|
|
instance_id: int
|
|
api_key: str
|
|
base_url: str
|
|
|
|
async def power(
|
|
self, action: Literal["start", "stop", "restart", "kill"]
|
|
) -> PowerManageResult:
|
|
url = f"{self.base_url}/api/ins/{self.instance_id}/power"
|
|
|
|
async with aiohttp.ClientSession(
|
|
headers={"Authorization": self.api_key}
|
|
) as session:
|
|
async with session.get(url, params={"action": action}) as resp:
|
|
resp.raise_for_status()
|
|
return PowerManageResult.model_validate_json(await resp.read())
|
|
|
|
async def detail(self) -> SimpfunServerDetailResp:
|
|
url = f"{self.base_url}/api/ins/{self.instance_id}/power"
|
|
|
|
async with aiohttp.ClientSession(
|
|
headers={"Authorization": self.api_key}
|
|
) as session:
|
|
async with session.get(url) as resp:
|
|
resp.raise_for_status()
|
|
return SimpfunServerDetailResp.model_validate_json(await resp.read())
|
|
|
|
@staticmethod
|
|
def new(config: SimpfunServerConfig | None = None):
|
|
if config is None:
|
|
config = get_config()
|
|
return SimpfunServer(
|
|
instance_id=config.plugin_simpfun_instance_id,
|
|
api_key=config.plugin_simpfun_api_key,
|
|
base_url=config.plugin_simpfun_base_url,
|
|
)
|