Update for CC pyrolysis

.github/workflows/checks.yml

@@ -36,8 +36,8 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
 
-    - name: install flatbuffers
-      run: sudo apt-get install -y flatbuffers-compiler
+    # - name: install flatbuffers
+    #   run: sudo apt-get install -y flatbuffers-compiler
 
     - name: install nox
       run: python -m pip install nox

README.md

@@ -122,7 +122,7 @@ Programmatically getting auth tokens from a user on your website.
 @route("/code")
 def code(request):
     auth = arkprts.YostarAuth(request.query["server"], network=...)
-    await auth.get_token_from_email_code(request.query["email"])
+    await auth.send_email_code(request.query["email"])
 
     return "Code sent!"
 

arkprts/assets/base.py

@@ -12,6 +12,7 @@
 import json
 import logging
 import pathlib
+import subprocess
 import typing
 
 from arkprts import network as netn
@@ -24,6 +25,20 @@
 LOGGER = logging.getLogger("arkprts.assets")
 
 
+def has_unitypy_flatc() -> bool:
+    try:
+        importlib.util.find_spec("UnityPy")
+    except ImportError:
+        return False
+
+    try:
+        subprocess.run("flatc --version", check=False)  # noqa: S603
+    except FileNotFoundError:
+        return False
+
+    return True
+
+
 class Assets(abc.ABC):
     """Game assets client."""
 
@@ -56,17 +71,14 @@ def create(
         default_server: netn.ArknightsServer | None = None,
     ) -> Assets:
         """Create a new assets file based on what libraries are available."""
-        try:
-            importlib.util.find_spec("UnityPy")
-        except ImportError:
-            from . import git
-
-            LOGGER.debug("Creating GitAssets due to UnityPy being unavailable")
-            return git.GitAssets(path, default_server=default_server or "en")
-        else:
+        if has_unitypy_flatc():
             from . import bundle
 
             return bundle.BundleAssets(path, default_server=default_server, network=network)
+        else:  # noqa: RET505
+            from . import git
+
+            return git.GitAssets(path, default_server=default_server or "en")
 
     @abc.abstractmethod
     async def update_assets(self) -> None:

arkprts/assets/bundle.py

@@ -145,8 +145,8 @@ async def update_fbs_schema(*, force: bool = False) -> None:
             continue
 
         UPDATED_FBS[server] = True
-        directory = resolve_fbs_schema_directory(server).parent
-        await git.update_repository("MooncellWiki/OpenArknightsFBS", directory, branch=branch, force=force)
+        directory = resolve_fbs_schema_directory(server).parent  # pyright: ignore[reportArgumentType]
+        await git.download_repository("MooncellWiki/OpenArknightsFBS", directory, branch=branch, force=force)
 
 
 def recursively_collapse_keys(obj: typing.Any) -> typing.Any:

arkprts/assets/git.py

@@ -72,6 +72,7 @@ async def download_github_tarball(
 ) -> pathlib.Path:
     """Download a tarball from github."""
     destination = pathlib.Path(destination or tempfile.mktemp(f"{repository.split('/')[-1]}.tar.gz"))
+    destination.parent.mkdir(parents=True, exist_ok=True)
     if branch == "HEAD":
         url = f"https://api.github.com/repos/{repository}/tarball"
     else:
@@ -138,10 +139,8 @@ async def download_repository(
     )
 
     LOGGER.debug("Decompressing %s", repository)
-    tarball_commit = decompress_tarball(tarball_path, destination, allow=allow)
-    LOGGER.debug("Decompressed %s %s", repository, tarball_commit)
-    if tarball_commit not in commit:
-        raise RuntimeError(f"Tarball commit {tarball_commit} does not match github commit {commit}")
+    decompress_tarball(tarball_path, destination, allow=allow)
+    LOGGER.debug("Decompressed %s %s", repository, commit)
 
     # sometimes the contents are identical, we still want to update the mtime
     commit_file.write_text(commit)
@@ -165,6 +164,7 @@ async def update_git_repository(repository: str, directory: PathLike, *, branch:
             f"https://github.com/{repository}.git",
             cwd=directory.parent,
         )
+        (directory.parent / repository.split("/", 1)[1]).rename(directory)
         await proc.wait()
     else:
         LOGGER.info("Updating %s in %s", repository, directory)

arkprts/auth.py

@@ -126,8 +126,8 @@ def generate_u8_sign(data: typing.Mapping[str, object]) -> str:
     """u8 auth sign."""
     query = urllib.parse.urlencode(sorted(data.items()))
 
-    hama_code = hmac.new(b"91240f70c09a08a6bc72af1a5c8d4670", query.encode(), "sha1")
-    return hama_code.hexdigest().lower()
+    code = hmac.new(b"91240f70c09a08a6bc72af1a5c8d4670", query.encode(), "sha1")
+    return code.hexdigest().lower()
 
 
 @dataclasses.dataclass()
@@ -315,6 +315,10 @@ async def _get_secret(
         LOGGER.info("Logged in with UID %s", uid)
         return secret
 
+    @abc.abstractmethod
+    async def login_with_token(self, channel_uid: str, token: str, /) -> None:
+        """Login with a channel uid and token."""
+
     @classmethod
     async def from_token(
         cls,
@@ -371,7 +375,7 @@ async def _get_access_token(self, channel_uid: str, yostar_token: str) -> str:
         data = await self.request_passport("user/login", json=body)
         return data["accessToken"]
 
-    async def _request_yostar_auth(self, email: str) -> None:
+    async def send_email_code(self, email: str) -> None:
         """Request to log in with a yostar account."""
         LOGGER.debug("Sending code to %s.", email)
         body = {"platform": "android", "account": email, "authlang": "en"}
@@ -430,7 +434,7 @@ async def get_token_from_email_code(
             email = input("Enter email:")
 
         if not code:
-            await self._request_yostar_auth(email)
+            await self.send_email_code(email)
             if not stdin:
                 return "", ""
 

arkprts/automation.py

@@ -21,46 +21,6 @@
 __all__ = ["AutomationClient"]
 
 
-def recursively_update_dict(
-    target: typing.MutableMapping[typing.Any, typing.Any],
-    source: typing.Mapping[typing.Any, typing.Any],
-) -> None:
-    """Recursively update a dictionary.
-
-    This is used to update the player data.
-    """
-    for key, value in source.items():
-        if isinstance(value, dict):
-            recursively_update_dict(target[key], typing.cast("dict[object, object]", value))
-        elif isinstance(value, list):
-            for i, v in enumerate(typing.cast("list[object]", value)):
-                if isinstance(v, dict):
-                    recursively_update_dict(target[key][i], typing.cast("dict[object, object]", v))
-        else:
-            target[key] = value
-
-
-def recursively_delete_dict(
-    target: typing.MutableMapping[typing.Any, typing.Any],
-    source: typing.Mapping[typing.Any, typing.Any],
-) -> None:
-    """Recursively delete items from a dictionary.
-
-    This is used to update the player data.
-    """
-    for key, value in source.items():
-        if isinstance(value, dict):
-            recursively_delete_dict(target[key], typing.cast("dict[object, object]", value))
-        elif isinstance(value, list):
-            for i, v in enumerate(typing.cast("list[object]", value)):
-                if isinstance(v, dict):
-                    recursively_delete_dict(target[key][i], typing.cast("dict[object, object]", v))
-                else:
-                    target[key].remove(v)
-        else:
-            del target[key]
-
-
 # https://github.com/Rhine-Department-0xf/Rhine-DFramwork/blob/main/client/encryption.py
 
 
@@ -164,8 +124,7 @@ def update_player_data_delta(
 
         Does not implement deleted as it is often overwritten with "modified".
         """
-        recursively_delete_dict(self.data, delta["deleted"])
-        recursively_update_dict(self.data, delta["modified"])
+        self.data.update(delta["modified"])
 
     async def request(self, endpoint: str, json: typing.Mapping[str, object] = {}, **kwargs: typing.Any) -> typing.Any:
         """Send an authenticated request to the arknights game server."""

arkprts/errors.py

@@ -28,7 +28,7 @@ class ArkPrtsError(BaseArkprtsError):
 
     def __init__(self, data: typing.Mapping[str, typing.Any]) -> None:
         self.data = data
-        super().__init__(f"[{data['result']}] {self.message} {data}")
+        super().__init__(f"[{data.get('result')}] {self.message} {data}")
 
 
 class GameServerError(BaseArkprtsError):

arkprts/models/data.py

@@ -296,8 +296,10 @@ class Social(base.BaseModel):
     """Support operators."""
     yesterday_reward: base.DDict = pydantic.Field(alias="yesterdayReward")
     """IDK. Clue exchange data."""
-    y_crisis_ss: typing.Union[str, typing.Any] = pydantic.Field(alias="yCrisisSs", repr=False)
+    y_crisis_ss: str = pydantic.Field(alias="yCrisisSs", repr=False)
     """IDK. Crisis refers to contingency contract. Always empty string."""
+    y_crisis_v2_ss: str = pydantic.Field(alias="yCrisisV2Ss", repr=False)
+    """IDK. Crisis refers to contingency contract. Empty string when cc is not happening."""
     medal_board: base.DDict = pydantic.Field(default_factory=base.DDict, alias="medalBoard")
     """Medal board."""
 

arkprts/models/social.py

@@ -72,7 +72,9 @@ class AssistChar(base.BaseModel):
     level: int
     """Operator level."""
     crisis_record: typing.Mapping[str, int] = pydantic.Field(alias="crisisRecord")
-    """Maximum risk this operator was used as a support in. -1 if never used."""
+    """Maximum risk this operator was used as a support in. Only for operations (before Pinch-Out). -1 if never used."""
+    crisis_v2_record: typing.Mapping[str, int] = pydantic.Field(alias="crisisV2Record")
+    """Maximum risk this operator was used as a support in. Only for battleplans (after Pinch-Out). -1 if never used."""
     current_equip: typing.Optional[str] = pydantic.Field(alias="currentEquip")
     """ID of the currently equipped module."""
     equip: typing.Mapping[str, UniEquip]

dev-requirements/pytest.txt

@@ -1,5 +1,5 @@
 pytest
-pytest-asyncio==0.21.1
+pytest-asyncio
 pytest-dotenv
 pytest-cov
 coverage[toml]

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "arkprts"
 requires-python = ">=3.9"
-version = "0.3.7"
+version = "0.3.8"
 dynamic = [
     "dependencies",
     "description",

setup.py

@@ -5,7 +5,7 @@
 
 setup(
     name="arkprts",
-    version="0.3.7",
+    version="0.3.8",
     description="Arknights python wrapper.",
     url="https://github.com/thesadru/arkprts",
     packages=find_packages(exclude=["tests", "tests.*"]),

tests/conftest.py

@@ -12,7 +12,10 @@ def event_loop() -> typing.Iterator[asyncio.AbstractEventLoop]:
     with warnings.catch_warnings():
         warnings.simplefilter("ignore")
 
-        loop = asyncio.get_event_loop()
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
 
     yield loop
     loop.close()