Implement easier raw and tmpl handling

.github/workflows/checks.yml

@@ -8,12 +8,12 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
-    - name: Setup Python 3.11
-      uses: actions/setup-python@v4
+    - name: Setup Python 3.12
+      uses: actions/setup-python@v5
       with:
-        python-version: "3.11"
+        python-version: "3.12"
 
     - name: install nox
       run: python -m pip install nox
@@ -25,14 +25,14 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11"]
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
       fail-fast: false
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
     - name: Setup Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v4
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
 
@@ -53,12 +53,12 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
-    - name: Setup Python 3.11
-      uses: actions/setup-python@v4
+    - name: Setup Python 3.12
+      uses: actions/setup-python@v5
       with:
-        python-version: "3.11"
+        python-version: "3.12"
 
     - name: install nox
       run: python -m pip install nox
@@ -70,12 +70,12 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
-    - name: Setup Python 3.11
-      uses: actions/setup-python@v4
+    - name: Setup Python 3.12
+      uses: actions/setup-python@v5
       with:
-        python-version: "3.11"
+        python-version: "3.12"
 
     - name: install nox
       run: python -m pip install nox
@@ -87,7 +87,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
     - name: Run prettier
       run: npx prettier --check *.md

arkprts/__init__.py

@@ -1,7 +1,8 @@
 """Arknights python wrapper."""
 
-from . import errors, models
+from . import models
 from .assets import *
 from .auth import *
 from .client import *
+from .errors import *
 from .network import *

arkprts/__main__.py

@@ -16,7 +16,7 @@
 parser.add_argument("--server", type=str, default="en", help="Server to use")
 parser.add_argument("--guest", action="store_true", help="Whether to use a guest account.")
 
-subparsers: argparse._SubParsersAction[argparse.ArgumentParser] = parser.add_subparsers(dest="command", required=True)
+subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]" = parser.add_subparsers(dest="command", required=True)
 
 parser_search: argparse.ArgumentParser = subparsers.add_parser("search", description="Get user info.")
 parser_search.add_argument(
@@ -29,7 +29,7 @@
 
 parser_api: argparse.ArgumentParser = subparsers.add_parser("api", description="Make a request towards the API.")
 parser_api.add_argument("endpoint", type=str, nargs="?", help="Endpoint path, not full url")
-parser_api.add_argument("payload", type=str, nargs="?", default="{}", help="JSON payload")
+parser_api.add_argument("payload", type=str, nargs="?", default=r"{}", help="JSON payload")
 
 
 async def search(client: arkprts.Client, nickname: typing.Optional[str] = None) -> None:
@@ -70,14 +70,11 @@ async def search(client: arkprts.Client, nickname: typing.Optional[str] = None)
 
 async def api(client: arkprts.Client, endpoint: str, payload: typing.Optional[str] = None) -> None:
     """Make a request."""
-    client.assets.loaded = True
     try:
-        data = await client.request(endpoint, json=payload and json.loads(payload))
+        data = await client.auth.auth_request(endpoint, json=payload and json.loads(payload), handle_errors=False)
         json.dump(data, sys.stdout, indent=4, ensure_ascii=False)
-    except arkprts.errors.GameServerError as e:
-        json.dump(e.data, sys.stdout, indent=4, ensure_ascii=False)
     finally:
-        await client.network.close()
+        await client.auth.network.close()
 
     sys.stdout.write("\n")
 

arkprts/assets/bundle.py

@@ -343,7 +343,8 @@ def __init__(
         except ImportError as e:
             raise ImportError("Cannot use BundleAssets without arkprts[assets]") from e
         try:
-            subprocess.run(["flatc", "--version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)  # noqa: S603
+            cmd = ["flatc", "--version"]
+            subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)  # noqa: S603
         except OSError as e:
             raise ImportError("Cannot use BundleAssets without a flatc executable") from e
 

arkprts/assets/git.py

@@ -80,7 +80,7 @@ async def download_github_tarball(
 
     async with aiohttp.ClientSession(auto_decompress=False) as session, session.get(url) as response:
         response.raise_for_status()
-        with destination.open("wb") as file:  # noqa: ASYNC101  # would need another dependency
+        with destination.open("wb") as file:  # noqa: ASYNC230
             async for chunk in response.content.iter_any():
                 file.write(chunk)
 
@@ -124,8 +124,11 @@ async def download_repository(
     try:
         commit = await get_github_repository_commit(repository, branch=branch)
     except aiohttp.ClientResponseError:
-        LOGGER.warning("Failed to get %s commit, skipping download", repository, exc_info=True)
-        return
+        if commit_file.exists():
+            LOGGER.warning("Failed to get %s commit, skipping download", repository, exc_info=True)
+            return
+
+        commit = "null"
 
     if not force and commit_file.exists() and commit_file.read_text() == commit:
         LOGGER.debug("%s is up to date [%s]", repository, commit)

arkprts/auth.py

@@ -180,15 +180,15 @@ async def auth_request(
         """Send an authenticated request to the arkights game server."""
 
 
-class Auth(abc.ABC, CoreAuth):
+class Auth(CoreAuth):
     """Authentication client for single sessions."""
 
     server: netn.ArknightsServer
     """Arknights server."""
     network: netn.NetworkSession
     """Network session."""
     device_ids: tuple[str, str, str]
-    """Device ids."""
+    """Random device ids."""
     session: AuthSession
     """Authentication session."""
 
@@ -206,6 +206,25 @@ def __init__(
         self.session = AuthSession(self.server, "", "")
         self.device_ids = create_random_device_ids()
 
+    @classmethod
+    def create(
+        cls,
+        server: netn.ArknightsServer | None = None,
+        *,
+        network: netn.NetworkSession | None = None,
+    ) -> Auth:
+        """Find and create an appropriate auth client."""
+        if server in ("en", "jp", "kr"):
+            return YostarAuth(server, network=network)
+        if server == "cn":
+            return HypergryphAuth(server, network=network)
+        if server == "bili":
+            return BilibiliAuth(server, network=network)
+        if server == "tw":
+            return LongchengAuth(server, network=network)
+
+        return cls(server, network=network)
+
     @property
     def uid(self) -> str:
         """Arknights user UID."""
@@ -216,6 +235,11 @@ def secret(self) -> str:
         """Arknights session token."""
         return self.session.secret
 
+    @property
+    def seqnum(self) -> int:
+        """Last sent seqnum."""
+        return self.session.seqnum
+
     async def request(
         self,
         domain: netn.ArknightsDomain,
@@ -329,21 +353,23 @@ async def from_token(
         network: netn.NetworkSession | None = None,
     ) -> Auth:
         """Create a client from a token."""
-        if server in ("en", "jp", "kr"):
-            auth = YostarAuth(server, network=network)
-            await auth.login_with_token(channel_uid, token)
-        elif server == "cn":
-            auth = HypergryphAuth(server, network=network)
-            await auth.login_with_token(channel_uid, token)
-        elif server == "bili":
-            auth = BilibiliAuth(server, network=network)
-            await auth.login_with_token(channel_uid, token)
-        elif server == "tw":
-            auth = LongchengAuth(server, network=network)
-            await auth.login_with_token(channel_uid, token)
-        else:
-            raise ValueError(f"Cannot create a generic auth client for server {server!r}")
+        auth = cls.create(server, network=network)
+        await auth.login_with_token(channel_uid, token)
+        return auth
 
+    @classmethod
+    async def from_session(
+        cls,
+        server: netn.ArknightsServer,
+        *,
+        uid: str,
+        secret: str,
+        seqnum: str | int,
+        network: netn.NetworkSession,
+    ) -> Auth:
+        """Create an auth from an already ongoing session."""
+        auth = cls.create(server, network=network)
+        auth.session = AuthSession(server, uid=uid, secret=secret, seqnum=int(seqnum))
         return auth
 
 
@@ -491,14 +517,14 @@ async def _get_hypergryph_access_token(self, username: str, password: str) -> st
             "platform": 1,
         }
         data["sign"] = generate_u8_sign(data)
-        data = await self.network.request("as", "user/login", json=data)
+        data = await self.request("as", "user/login", json=data)
         return data["token"]
 
     async def _get_hypergryph_uid(self, token: str) -> str:
         """Get a channel uid from a hypergryph access token."""
         data = {"token": token}
         data["sign"] = generate_u8_sign(data)
-        data = await self.network.request("as", "user/auth", json=data)
+        data = await self.request("as", "user/auth", json=data)
         return data["uid"]
 
     async def login_with_token(self, channel_uid: str, access_token: str) -> None:

arkprts/errors.py

@@ -28,10 +28,10 @@ class ArkPrtsError(BaseArkprtsError):
 
     def __init__(self, data: typing.Mapping[str, typing.Any]) -> None:
         self.data = data
-        super().__init__(f"[{data.get('result')}] {self.message} {data}")
+        super().__init__(f"[{data.get('result')}] {self.message} {json.dumps(data)}")
 
 
-class GameServerError(BaseArkprtsError):
+class GameServerError(ArkPrtsError):
     """Game server error."""
 
     data: typing.Mapping[str, typing.Any]
@@ -49,7 +49,7 @@ def __init__(self, data: typing.Mapping[str, typing.Any]) -> None:
         self.msg = data.get("msg", "")
         self.info = json.loads(data.get("info", "{}"))
 
-        super().__init__(str(data))
+        BaseArkprtsError.__init__(self, json.dumps(data))
 
 
 class GeetestError(ArkPrtsError):
@@ -65,7 +65,7 @@ def __init__(self, data: typing.Mapping[str, typing.Any]) -> None:
         super().__init__(data)
 
 
-class InvalidStatusError(BaseArkprtsError):
+class InvalidStatusError(ArkPrtsError):
     """Raised when a response has an invalid status code."""
 
     status: int
@@ -75,7 +75,7 @@ def __init__(self, status: int, data: typing.Mapping[str, typing.Any]) -> None:
         self.status = status
         self.data = data
 
-        super().__init__(f"[{status}] {data}")
+        BaseArkprtsError.__init__(self, f"[{status}] {json.dumps(data)}")
 
 
 class InvalidContentTypeError(BaseArkprtsError):

arkprts/models/base.py

@@ -62,11 +62,15 @@ def __init__(self, client: CoreClient | None = None, **kwargs: typing.Any) -> No
         if client:
             _set_recursively(self, "client", client)
 
-    @pydantic.model_validator(mode="before")  # pyright: ignore
+    @pydantic.model_validator(mode="before")
+    @classmethod
     def _fix_amiya(cls, value: typing.Any, info: pydantic.ValidationInfo) -> typing.Any:
         """Flatten Amiya to only keep her selected form if applicable."""
         if value and value.get("tmpl"):
-            # tmplId present in battle replays
+            value["variations"] = {
+                tmplid: cls(value["client"], **{**value, **tmpl}) for tmplid, tmpl in value["tmpl"].items()
+            }
+            # tmplId present in battle replays, sometimes the tmpl for amiya guard is not actually present
             current_tmpl = value["currentTmpl"] if "currentTmpl" in value else value["tmplId"]
             current = value["tmpl"].get(current_tmpl, next(iter(value["tmpl"].values())))
             value.update(current)

arkprts/models/battle.py

@@ -68,6 +68,8 @@ class Character(base.BaseModel):
 
     tmpl: typing.Mapping[str, base.DDict] = pydantic.Field(default_factory=base.DDict, repr=False)
     """Alternative operator class data. Only for Amiya."""
+    variations: typing.Mapping[str, "Character"] = pydantic.Field(default_factory=dict, repr=False)
+    """All representations of amiya."""
 
 
 class Signature(base.BaseModel):

arkprts/models/data.py

@@ -145,6 +145,8 @@ class SquadSlot(base.BaseModel):
     """Currently equipped module ID."""
     tmpl: typing.Mapping[str, base.DDict] = pydantic.Field(default_factory=base.DDict, repr=False)
     """Alternative operator class data. Only for Amiya."""
+    variations: typing.Mapping[str, "SquadSlot"] = pydantic.Field(default_factory=dict, repr=False)
+    """All representations of amiya."""
 
 
 class Squads(base.BaseModel):
@@ -224,6 +226,8 @@ class Character(base.BaseModel):
     """Whether the operator is marked as favorite."""
     tmpl: typing.Mapping[str, base.DDict] = pydantic.Field(default_factory=base.DDict, repr=False)
     """Alternative operator class data. Only for Amiya."""
+    variations: typing.Mapping[str, "Character"] = pydantic.Field(default_factory=dict, repr=False)
+    """All representations of amiya."""
 
     @property
     def static(self) -> base.DDict:
@@ -287,6 +291,8 @@ class AssistChar(base.BaseModel):
     """Currently equipped module."""
     tmpl: typing.Mapping[str, base.DDict] = pydantic.Field(default_factory=base.DDict, repr=False)
     """Alternative operator class data. Only for Amiya."""
+    variations: typing.Mapping[str, "AssistChar"] = pydantic.Field(default_factory=dict, repr=False)
+    """All representations of amiya."""
 
 
 class Social(base.BaseModel):

arkprts/models/social.py

@@ -81,6 +81,8 @@ class AssistChar(base.BaseModel):
     """Equipped modules. Module ID to module info."""
     tmpl: typing.Mapping[str, base.DDict] = pydantic.Field(default_factory=base.DDict, repr=False)
     """Alternative operator class data. Only for Amiya."""
+    variations: typing.Mapping[str, "AssistChar"] = pydantic.Field(default_factory=dict, repr=False)
+    """All representations of amiya."""
 
     @property
     def static(self) -> base.DDict: