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 format_branch(branch: str) -> str:
    for i in ["-", ".", " ", ":"]:
        branch = branch.replace(i, "_")
    return branch


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", []):
            branch = data["workflow_runs"][0]["head_branch"]
            hash_str = data["workflow_runs"][0]["head_commit"]["id"][:7]
            message = data["workflow_runs"][0]["head_commit"]["message"][:200]
            return f"#{format_branch(branch)} #{hash_str} \n\n" \
                   f"{message}"
    return "Unknown."