skyeweave/skyeweave/service/atlas.py

import logging
import pathlib
from typing import Annotated, List, Tuple, TypedDict
import requests

# Backport of NotRequired for python 3.10 and older
try:
    from typing import NotRequired
except ImportError:
    from typing_extensions import NotRequired

from ..config import AtlasAPIConfig, PathConfig, ExpressionDefaults, LoggingConfig

LOGGER = logging.getLogger(LoggingConfig.NAME)

class SpritesheetData(TypedDict):
    facesize: Tuple[int, int]
    position: Tuple[int, int]

class ExtendData(TypedDict):
    faceSizeRect: NotRequired[Annotated[List[int], 2]]
    faceSize: NotRequired[int]

def fetch_config(chara_id: int) -> SpritesheetData:
    url = f"https://api.atlasacademy.io/raw/JP/svtScript?charaId={chara_id}"

    LOGGER.debug(f"Loading data for {url}")
    response = requests.get(url, timeout=AtlasAPIConfig.TIMEOUT)
    LOGGER.debug(f"{response.status_code} - {response.text}")
    if not response.ok:
        raise ValueError(f"Failed to fetch data for charaId {chara_id}")

    resp_json = response.json()
    if len(resp_json) == 0:
        raise ValueError(f"No data was returned for get given charaId {chara_id}. Please ensure that this is a valid charaId and not the baseSvtId")
    resp_data = resp_json[0]
    extend_data: ExtendData = resp_data["extendData"]

    if "faceSizeRect" in extend_data:
        facesize: Tuple[int, int] = tuple(extend_data["faceSizeRect"]) # type: ignore
    else:
        facesize = tuple(2 * [ extend_data.get("faceSize", ExpressionDefaults.FACESIZE) ]) # type: ignore

    position: tuple[int, int] = (resp_data["faceX"], resp_data["faceY"])

    returndata: SpritesheetData = {
        "facesize": facesize,
        "position": position
    }

    LOGGER.debug(returndata)
    return returndata

def fetch_mstsvtjson() -> None:
    url = AtlasAPIConfig.MST_SVT_JSON
    filelocation = PathConfig.IMAGES / "mstsvt.json"

    if filelocation.exists():
        LOGGER.info("Found cached asset for mstsvt.json")
        return

    LOGGER.debug(f"Loading data for {url}")
    with open(filelocation, 'wb') as handle:
        response = requests.get(url, stream=True, timeout=AtlasAPIConfig.TIMEOUT)
        status = response.status_code
        LOGGER.debug(f"{response.status_code} - {response.text}")
        if status != 200:
            raise ValueError("Could not fetch mstsvnt.json from atlas - please check your network connection")
        for block in response.iter_content(1024):
            if not block:
                break
            handle.write(block)

def fetch_expression_sheets(tempfolder: pathlib.Path, imageid: int) -> pathlib.Path:
    atlasurl_base = f"https://static.atlasacademy.io/{AtlasAPIConfig.REGION}/CharaFigure/{imageid}"

    savefolder = tempfolder / str(imageid)
    savefolder.mkdir(exist_ok=True, parents=True)

    idx, status = 0, 200

    while status == 200:
        filelocation = savefolder / f"{idx}.png"

        postfix = ""
        if idx == 1:
            postfix = "f"
        elif idx > 1:
            postfix = f"f{idx}"

        if filelocation.exists():
            LOGGER.info(f"Found cached asset for {imageid}{postfix}.png")
            idx += 1
            continue

        filename = f"{imageid}{postfix}.png"
        atlasurl = f"{atlasurl_base}/{filename}"

        LOGGER.debug(f"Loading data for {atlasurl}")
        with open(filelocation, 'wb') as handle:
            response = requests.get(atlasurl, stream=True, timeout=AtlasAPIConfig.TIMEOUT)
            status = response.status_code
            LOGGER.debug(f"{response.status_code} - {response.text}")
            if status != 200 and postfix == "":
                LOGGER.warning(f"There was no expression sheet for the given charaId {imageid} - please ensure that this charaId is valid.")
            if status != 200:
                continue
            for block in response.iter_content(1024):
                if not block:
                    break
                handle.write(block)
        LOGGER.info(f"Finished downloading {filename}")
        idx += 1
    p = savefolder / f"{idx}.png"
    p.unlink(missing_ok=True)

    return savefolder


def fetch_data(servantid: int) -> List[int]:
    atlasurl = f"https://api.atlasacademy.io/nice/{AtlasAPIConfig.REGION}/servant/{servantid}?lore=false&lang=en"

    LOGGER.debug(f"Loading data for {atlasurl}")
    response = requests.get(atlasurl, timeout=AtlasAPIConfig.TIMEOUT)
    LOGGER.debug(f"{response.status_code}")
    if not response.ok:
        LOGGER.debug(f"{response.status_code} - {response.text}")
        raise ValueError(f"{response.status_code} - {response.text}")

    responsedata = response.json()
    svtname = responsedata["name"]
    charascripts: List[dict[str, int]] = responsedata["charaScripts"]
    chara_ids: List[int] = [chara["id"] for chara in charascripts]

    LOGGER.debug(chara_ids)
    LOGGER.info(f"{svtname} ({servantid}) - {len(chara_ids)} charaIds")
    return chara_ids