Lawnmower/defs/ci.py

import datetime
from configparser import RawConfigParser
from os.path import exists
from os import remove, sep
from typing import Optional
from httpx import get, AsyncClient

""" Init httpx client """
# 使用自定义 UA
config = RawConfigParser()
config.read("config.ini")
ghp: str = ""
ghp = config.get("basic", "ghp", fallback=ghp)
headers = {
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.132 Safari/537.36",
    "Authorization": f"token {ghp}",
}
client = AsyncClient(timeout=10.0, headers=headers, follow_redirects=True)


class Artifact:
    def __init__(self, data: dict):
        self.id: int = data.get("id", 0)
        self.name: str = "Grasscutter.zip"
        self.url: str = data.get("archive_download_url", "")
        size: int = data.get("archive_size", 0)
        self.size: str = f"{size / 1024 / 1024:.2f} MB"
        self.expired: bool = data.get("expired", False)
        created_at: str = data.get("created_at", "")
        # 2022-04-27T06:31:46Z
        try:
            self.created_at: datetime.datetime = datetime.datetime.strptime(
                created_at, "%Y-%m-%dT%H:%M:%SZ"
            ) + datetime.timedelta(hours=8)
        except ValueError:
            self.created_at: datetime.datetime = datetime.datetime.utcnow() + datetime.timedelta(hours=8)
        self.created_time: str = self.created_at.strftime("%Y-%m-%d %H:%M:%S")
        self.hash: str = get_hash()

    async def download(self, path: str = f"data{sep}Grasscutter.zip"):
        if exists(path):
            remove(path)

        content = await client.get(self.url)
        if content.status_code == 404:
            raise FileNotFoundError
        content = content.content
        with open(path, 'wb') as f:
            f.write(content)


async def get_artifact() -> Optional[Artifact]:
    req = await client.get("https://api.github.com/repos/Grasscutters/Grasscutter/actions/artifacts?per_page=1")
    if req.status_code == 200:
        data = req.json()
        if data.get("artifacts", []):
            return Artifact(data["artifacts"][0])
    return None


def get_hash() -> str:
    req = get("https://api.github.com/repos/Grasscutters/Grasscutter/actions/runs?per_page=1&status=success")
    if req.status_code == 200:
        data = req.json()
        if data.get("workflow_runs", []):
            return data["workflow_runs"][0]["head_commit"]["id"][:6] + "  " + \
                   data["workflow_runs"][0]["head_commit"]["message"][:50]
    return "Unknown."