Merge branch dev into master (#221)

* Implement detailed Genshin character endpoint (#219)

* Implement detailed Genshin character endpoint.

* Run nox and fix formatting issues.

* Fix type checker issues

* Replace root_validator with validator

* Improve __fill_prop_info

* Rename some fields and models

* Improve artifact set effect impl

---------

Co-authored-by: seriaati <[email protected]>

* Fix error on get_genshin_user

* Add detailed character test

* Add ZZZPropertyType.ANOMALY_MASTERY

* Remove the usage of galias

* Remove custom validation dunder methods

* Remove the use of lang in APIModel

* Fix validation error on FullGenshinUserStats

* Fix validation error on CalculatorConsumable

* Fix TypeError on generic models

* Fix KeyError on GenshinDetailCharacters

* Fix ValidationError on FullHonkaiUserStats

* Fix ValidationError on LineupPreview

* Fix ValidationError on PartialLineupCharacter

* Revert PartialLineupCharacter changes

* Fix ValidationError on PartialLineupCharacter

* Make stored_attendance_refresh_countdown optional

* Allow returning raw data

* Fix returning wrong icons

* Fix _create_icon method

* Fix problem with aliased fields in root validators

* Im dumb

* Fix validation error on DailyReward

* Fix error in getting db char

* Add TheaterDifficulty.IDK

* Fix character icons with img theater

* Remove partial/unknown character test

* Add pyroculi

* Remove messed up character test

* Add missing events in HSR announcements

* Add img field to Announcement

* Rename TheaterDifficulty.IDK to TheaterDifficulty.VISIONARY

* Fix qrcode login

* Change otp code mmt model to v4

* Fix using tuple class to type hint

* Add support for SQLiteCache

* Add clear_cache method to SQLiteCache

* Allow not passing in connection for SQLiteCache

* Fix HSR gacha log route

* Make wish record time timezone aware

* Export StarRailBannerType

* Fix tz offset causing KeyError in Genshin

* Add Natlan tribe reputations

* Fix field not being Aliased

* Add ImgTheater battle stats

* Fix HSR code redeem not working

* Fix ValidationError in TheaterBattleStats

* Disable coverage workflows temporarily

---------

Co-authored-by: Furia <[email protected]>

.github/workflows/checks.yml

@@ -49,44 +49,44 @@ jobs:
           python -m nox -s test --verbose -- --cov-append
           mv .coverage .coverage.${{ matrix.python-version }}
 
-      - name: Upload coverage
-        uses: actions/upload-artifact@v3
-        with:
-          name: coverage
-          path: .coverage.${{ matrix.python-version }}
-          retention-days: 1
-          if-no-files-found: error
-
-  upload-coverage:
-    needs: [test]
-    runs-on: ubuntu-latest
-
-    steps:
-      - name: Checkout repo
-        uses: actions/checkout@v4
-
-      - name: Setup python 3.10
-        uses: actions/setup-python@v5
-        with:
-          python-version: "3.10"
-
-      - name: Download coverage
-        uses: actions/download-artifact@v3
-        with:
-          name: coverage
-
-      - name: Combine coverage
-        run: |
-          pip install -r ./genshin-dev/pytest-requirements.txt
-          coverage combine
-          coverage xml -i
-
-      - name: Upload coverage to codeclimate
-        uses: paambaati/[email protected]
-        env:
-          CC_TEST_REPORTER_ID: cd8c7d84ae5f98d86882d666dce0946fe5aae1e63f442995bd9c6e17869e6513
-        with:
-          coverageLocations: .coverage.xml:coverage.py
+      # - name: Upload coverage
+      #   uses: actions/upload-artifact@v3
+      #   with:
+      #     name: coverage
+      #     path: .coverage.${{ matrix.python-version }}
+      #     retention-days: 1
+      #     if-no-files-found: error
+
+  # upload-coverage:
+  #   needs: [test]
+  #   runs-on: ubuntu-latest
+
+  #   steps:
+  #     - name: Checkout repo
+  #       uses: actions/checkout@v4
+
+  #     - name: Setup python 3.10
+  #       uses: actions/setup-python@v5
+  #       with:
+  #         python-version: "3.10"
+
+  #     - name: Download coverage
+  #       uses: actions/download-artifact@v3
+  #       with:
+  #         name: coverage
+
+  #     - name: Combine coverage
+  #       run: |
+  #         pip install -r ./genshin-dev/pytest-requirements.txt
+  #         coverage combine
+  #         coverage xml -i
+
+  #     - name: Upload coverage to codeclimate
+  #       uses: paambaati/[email protected]
+  #       env:
+  #         CC_TEST_REPORTER_ID: cd8c7d84ae5f98d86882d666dce0946fe5aae1e63f442995bd9c6e17869e6513
+  #       with:
+  #         coverageLocations: .coverage.xml:coverage.py
 
   type-check:
     runs-on: ubuntu-latest

.gitignore

@@ -59,3 +59,7 @@ docs/pdoc
 .mypy_cache/
 .dmypy.json
 dmypy.json
+
+# databases
+*.sqlite3
+*.db

genshin/client/cache.py

@@ -12,8 +12,10 @@
 
 if typing.TYPE_CHECKING:
     import aioredis
+    import aiosqlite
 
-__all__ = ["BaseCache", "Cache", "RedisCache", "StaticCache"]
+
+__all__ = ["BaseCache", "Cache", "RedisCache", "StaticCache", "SQLiteCache"]
 
 MINUTE = 60
 HOUR = MINUTE * 60
@@ -200,3 +202,122 @@ async def set_static(self, key: typing.Any, value: typing.Any) -> None:
             self.serialize_value(value),
             ex=self.static_ttl,
         )
+
+
+class SQLiteCache(BaseCache):
+    """SQLite implementation of the cache."""
+
+    conn: aiosqlite.Connection | None
+    ttl: int
+    static_ttl: int
+
+    def __init__(
+        self,
+        conn: aiosqlite.Connection | None = None,
+        *,
+        ttl: int = HOUR,
+        static_ttl: int = DAY,
+        db_name: str = "genshin_py.db",
+    ) -> None:
+        self.conn = conn
+        self.ttl = ttl
+        self.static_ttl = static_ttl
+        self.db_name = db_name
+
+    async def _clear_cache(self, conn: aiosqlite.Connection) -> None:
+        """Clear timed-out items."""
+        now = time.time()
+
+        await conn.execute("DELETE FROM cache WHERE expiration < ?", (now,))
+        await conn.commit()
+
+    async def initialize(self) -> None:
+        """Initialize the cache."""
+        import aiosqlite
+
+        if self.conn is None:
+            conn = await aiosqlite.connect(self.db_name)
+        else:
+            conn = self.conn
+
+        await conn.execute("CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, value TEXT, expiration INTEGER)")
+        await conn.commit()
+
+        if self.conn is None:
+            await conn.close()
+
+    def serialize_key(self, key: typing.Any) -> str:
+        """Serialize a key by turning it into a string."""
+        return str(key)
+
+    def serialize_value(self, value: typing.Any) -> str:
+        """Serialize a value by turning it into a string."""
+        return json.dumps(value)
+
+    def deserialize_value(self, value: str) -> typing.Any:
+        """Deserialize a value back into data."""
+        return json.loads(value)
+
+    async def get(self, key: typing.Any) -> typing.Optional[typing.Any]:
+        """Get an object with a key."""
+        import aiosqlite
+
+        if self.conn is None:
+            conn = await aiosqlite.connect(self.db_name)
+        else:
+            conn = self.conn
+
+        async with conn.execute(
+            "SELECT value FROM cache WHERE key = ? AND expiration > ?", (self.serialize_key(key), int(time.time()))
+        ) as cursor:
+            value = await cursor.fetchone()
+
+        if self.conn is None:
+            await conn.close()
+
+        if value is None:
+            return None
+
+        return self.deserialize_value(value[0])
+
+    async def set(self, key: typing.Any, value: typing.Any) -> None:
+        """Save an object with a key."""
+        import aiosqlite
+
+        if self.conn is None:
+            conn = await aiosqlite.connect(self.db_name)
+        else:
+            conn = self.conn
+
+        await conn.execute(
+            "INSERT OR REPLACE INTO cache (key, value, expiration) VALUES (?, ?, ?)",
+            (self.serialize_key(key), self.serialize_value(value), int(time.time() + self.ttl)),
+        )
+        await conn.commit()
+        await self._clear_cache(conn)
+
+        if self.conn is None:
+            await conn.close()
+
+    async def get_static(self, key: typing.Any) -> typing.Optional[typing.Any]:
+        """Get a static object with a key."""
+        return await self.get(key)
+
+    async def set_static(self, key: typing.Any, value: typing.Any) -> None:
+        """Save a static object with a key."""
+        import aiosqlite
+
+        if self.conn is None:
+            conn = await aiosqlite.connect(self.db_name)
+        else:
+            conn = self.conn
+
+        await conn.execute(
+            "INSERT OR REPLACE INTO cache (key, value, expiration) VALUES (?, ?, ?)",
+            (self.serialize_key(key), self.serialize_value(value), int(time.time() + self.static_ttl)),
+        )
+        await conn.commit()
+        await self._clear_cache(conn)
+
+        if self.conn is None:
+            await conn.close()

genshin/client/components/auth/client.py

@@ -15,7 +15,6 @@
 from genshin.client import routes
 from genshin.client.components import base
 from genshin.client.manager import managers
-from genshin.client.manager.cookie import fetch_cookie_token_with_game_token, fetch_stoken_with_game_token
 from genshin.models.auth.cookie import (
     AppLoginResult,
     CNWebLoginResult,
@@ -247,35 +246,19 @@ async def login_with_qrcode(self) -> QRLoginResult:
 
         scanned = False
         while True:
-            check_result = await self._check_qrcode(
-                creation_result.app_id, creation_result.device_id, creation_result.ticket
-            )
-            if check_result.status == QRCodeStatus.SCANNED and not scanned:
+            status, cookies = await self._check_qrcode(creation_result.ticket)
+            if status is QRCodeStatus.SCANNED and not scanned:
                 LOGGER_.info("QR code scanned")
                 scanned = True
-            elif check_result.status == QRCodeStatus.CONFIRMED:
+            elif status is QRCodeStatus.CONFIRMED:
                 LOGGER_.info("QR code login confirmed")
                 break
 
-            await asyncio.sleep(2)
-
-        raw_data = check_result.payload.raw
-        assert raw_data is not None
+            await asyncio.sleep(1)
 
-        cookie_token = await fetch_cookie_token_with_game_token(
-            game_token=raw_data.game_token, account_id=raw_data.account_id
-        )
-        stoken = await fetch_stoken_with_game_token(game_token=raw_data.game_token, account_id=int(raw_data.account_id))
-
-        cookies = {
-            "stoken_v2": stoken.token,
-            "ltuid": stoken.aid,
-            "account_id": stoken.aid,
-            "ltmid": stoken.mid,
-            "cookie_token": cookie_token,
-        }
         self.set_cookies(cookies)
-        return QRLoginResult(**cookies)
+        dict_cookies = {key: morsel.value for key, morsel in cookies.items()}
+        return QRLoginResult(**dict_cookies)
 
     @managers.no_multi
     async def create_mmt(self) -> MMT:

genshin/client/components/auth/subclients/app.py

@@ -4,9 +4,8 @@
 """
 
 import json
-import random
 import typing
-from string import ascii_letters, digits
+from http.cookies import SimpleCookie
 
 import aiohttp
 
@@ -15,7 +14,7 @@
 from genshin.client.components import base
 from genshin.models.auth.cookie import AppLoginResult
 from genshin.models.auth.geetest import SessionMMT, SessionMMTResult
-from genshin.models.auth.qrcode import QRCodeCheckResult, QRCodeCreationResult
+from genshin.models.auth.qrcode import QRCodeCreationResult, QRCodeStatus
 from genshin.models.auth.verification import ActionTicket
 from genshin.utility import auth as auth_utility
 from genshin.utility import ds as ds_utility
@@ -180,46 +179,34 @@ async def _verify_email(self, code: str, ticket: ActionTicket) -> None:
 
     async def _create_qrcode(self) -> QRCodeCreationResult:
         """Create a QR code for login."""
-        if self.default_game is None:
-            raise RuntimeError("No default game set.")
-
-        app_id = constants.APP_IDS[self.default_game][self.region]
-        device_id = "".join(random.choices(ascii_letters + digits, k=64))
-
         async with aiohttp.ClientSession() as session:
             async with session.post(
                 routes.CREATE_QRCODE_URL.get_url(),
-                json={"app_id": app_id, "device": device_id},
+                headers=auth_utility.QRCODE_HEADERS,
             ) as r:
                 data = await r.json()
 
         if not data["data"]:
             errors.raise_for_retcode(data)
 
-        url: str = data["data"]["url"]
         return QRCodeCreationResult(
-            app_id=app_id,
-            ticket=url.split("ticket=")[1],
-            device_id=device_id,
-            url=url,
+            ticket=data["data"]["ticket"],
+            url=data["data"]["url"],
         )
 
-    async def _check_qrcode(self, app_id: str, device_id: str, ticket: str) -> QRCodeCheckResult:
+    async def _check_qrcode(self, ticket: str) -> typing.Tuple[QRCodeStatus, SimpleCookie]:
         """Check the status of a QR code login."""
-        payload = {
-            "app_id": app_id,
-            "device": device_id,
-            "ticket": ticket,
-        }
+        payload = {"ticket": ticket}
 
         async with aiohttp.ClientSession() as session:
             async with session.post(
                 routes.CHECK_QRCODE_URL.get_url(),
                 json=payload,
+                headers=auth_utility.QRCODE_HEADERS,
             ) as r:
                 data = await r.json()
 
-        if not data["data"]:
-            errors.raise_for_retcode(data)
+                if not data["data"]:
+                    errors.raise_for_retcode(data)
 
-        return QRCodeCheckResult(**data["data"])
+                return QRCodeStatus(data["data"]["status"]), r.cookies

genshin/client/components/auth/subclients/web.py

@@ -12,7 +12,7 @@
 from genshin.client import routes
 from genshin.client.components import base
 from genshin.models.auth.cookie import CNWebLoginResult, MobileLoginResult, WebLoginResult
-from genshin.models.auth.geetest import SessionMMT, SessionMMTResult
+from genshin.models.auth.geetest import SessionMMT, SessionMMTResult, SessionMMTv4
 from genshin.utility import auth as auth_utility
 from genshin.utility import ds as ds_utility
 
@@ -164,7 +164,7 @@ async def _send_mobile_otp(
         *,
         encrypted: bool = False,
         mmt_result: typing.Optional[SessionMMTResult] = None,
-    ) -> typing.Union[None, SessionMMT]:
+    ) -> typing.Union[None, SessionMMTv4]:
         """Attempt to send OTP to the provided mobile number.
 
         May return aigis headers if captcha is triggered, None otherwise.
@@ -192,7 +192,7 @@ async def _send_mobile_otp(
         if data["retcode"] == -3101:
             # Captcha triggered
             aigis = json.loads(r.headers["x-rpc-aigis"])
-            return SessionMMT(**aigis)
+            return SessionMMTv4(**aigis)
 
         if not data["data"]:
             errors.raise_for_retcode(data)

genshin/client/components/base.py

@@ -84,7 +84,7 @@ def __init__(
         device_id: typing.Optional[str] = None,
         device_fp: typing.Optional[str] = None,
         headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
-        cache: typing.Optional[client_cache.Cache] = None,
+        cache: typing.Optional[client_cache.BaseCache] = None,
         debug: bool = False,
     ) -> None:
         self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)