Merge master (commit '8424e1d4304497cbb077ce149687fa196f7c94c6') into…

… gdb

genshin-dev/setup.py

@@ -1,4 +1,5 @@
 """Mock package to install the dev requirements."""
+
 import pathlib
 import typing
 

genshin/__main__.py

@@ -219,7 +219,7 @@ async def starrail_notes(client: genshin.Client, uid: typing.Optional[int]) -> N
         f"{click.style('Echo of War:', bold=True)} {data.remaining_weekly_discounts}/{data.max_weekly_discounts}"
     )
 
-    click.echo(f"\n{click.style('Assignments:', bold=True)} {data.accepted_epedition_num}/{data.total_expedition_num}")
+    click.echo(f"\n{click.style('Assignments:', bold=True)} {data.accepted_expedition_num}/{data.total_expedition_num}")
     for expedition in data.expeditions:
         if expedition.remaining_time > datetime.timedelta(0):
             remaining = f"{expedition.remaining_time} remaining"

genshin/client/components/base.py

@@ -43,6 +43,7 @@ class BaseClient(abc.ABC):
         "authkeys",
         "_hoyolab_id",
         "_accounts",
+        "custom_headers",
     )
 
     USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501
@@ -71,6 +72,9 @@ def __init__(
         game: typing.Optional[types.Game] = None,
         uid: typing.Optional[int] = None,
         hoyolab_id: typing.Optional[int] = None,
+        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,
         debug: bool = False,
     ) -> None:
@@ -90,6 +94,10 @@ def __init__(
         self.uid = uid
         self.hoyolab_id = hoyolab_id
 
+        self.custom_headers = dict(headers or {})
+        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
+        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})
+
     def __repr__(self) -> str:
         kwargs = dict(
             lang=self.lang,
@@ -338,6 +346,7 @@ async def request(
 
         headers = dict(headers or {})
         headers["User-Agent"] = self.USER_AGENT
+        headers.update(self.custom_headers)
 
         if method is None:
             method = "POST" if data else "GET"
@@ -384,6 +393,7 @@ async def request_webstatic(
 
         headers = dict(headers or {})
         headers["User-Agent"] = self.USER_AGENT
+        headers.update(self.custom_headers)
 
         await self._request_hook("GET", url, headers=headers, **kwargs)
 
@@ -593,7 +603,6 @@ def region_specific(region: types.Region) -> typing.Callable[[AsyncCallableT], A
     def decorator(func: AsyncCallableT) -> AsyncCallableT:
         @functools.wraps(func)
         async def wrapper(self: typing.Any, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
-
             if not hasattr(self, "region"):
                 raise TypeError("Cannot use @region_specific on a plain function.")
             if region != self.region:

genshin/client/components/chronicle/starrail.py

@@ -138,17 +138,16 @@ async def get_starrail_pure_fiction(
         """Get starrail pure fiction runs."""
         payload = dict(schedule_type=2 if previous else 1, need_all="true")
         data = await self._request_starrail_record("challenge_story", uid, lang=lang, payload=payload)
-
-        # In "groups", it contains time data from both the current season and previous season.
-        data = dict(data)
-        time = data["groups"][0]
-        if previous is True and len(data["groups"]) > 1:
-            time = data["groups"][1]
-
-        # Extract the time data.
-        data["schedule_id"] = time["schedule_id"]
-        data["begin_time"] = time["begin_time"]
-        data["end_time"] = time["end_time"]
-        data["name_mi18n"] = time["name_mi18n"]
-
         return models.StarRailPureFiction(**data)
+
+    async def get_starrail_apc_shadow(
+        self,
+        uid: typing.Optional[int] = None,
+        *,
+        previous: bool = False,
+        lang: typing.Optional[str] = None,
+    ) -> models.StarRailAPCShadow:
+        """Get starrail apocalyptic shadow runs."""
+        payload = dict(schedule_type=2 if previous else 1, need_all="true")
+        data = await self._request_starrail_record("challenge_boss", uid, lang=lang, payload=payload)
+        return models.StarRailAPCShadow(**data)

genshin/client/components/daily.py

@@ -17,8 +17,6 @@
 
 __all__ = ["DailyRewardClient"]
 
-CN_TIMEZONE = datetime.timezone(datetime.timedelta(hours=8))
-
 
 class DailyRewardClient(base.BaseClient):
     """Daily reward component."""
@@ -105,7 +103,7 @@ async def get_monthly_rewards(
             game=game,
             static_cache=cache.cache_key(
                 "rewards",
-                month=datetime.datetime.now(CN_TIMEZONE).month,
+                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                 region=self.region,
                 game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                 lang=lang or self.lang,

genshin/client/components/diary.py

@@ -8,13 +8,12 @@
 from genshin.client import cache, routes
 from genshin.client.components import base
 from genshin.client.manager import managers
+from genshin.constants import CN_TIMEZONE
 from genshin.models.genshin import diary as models
 from genshin.utility import deprecation
 
 __all__ = ["DiaryClient"]
 
-CN_TIMEZONE = datetime.timezone(datetime.timedelta(hours=8))
-
 
 class DiaryCallback(typing.Protocol):
     """Callback which requires a diary page."""

genshin/client/manager/managers.py

@@ -65,6 +65,7 @@ class BaseCookieManager(abc.ABC):
     """A cookie manager for making requests."""
 
     _proxy: typing.Optional[yarl.URL] = None
+    _socks_proxy: typing.Optional[str] = None
 
     @classmethod
     def from_cookies(cls, cookies: typing.Optional[AnyCookieOrHeader] = None) -> BaseCookieManager:
@@ -115,15 +116,28 @@ def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
             return
 
         proxy = yarl.URL(proxy)
-        if str(proxy.scheme) not in ("https", "http", "ws", "wss"):
+
+        if proxy.scheme in {"socks4", "socks5"}:
+            self._socks_proxy = str(proxy)
+            return
+
+        if proxy.scheme not in {"https", "http", "ws", "wss"}:
             raise ValueError("Proxy URL must have a valid scheme.")
 
         self._proxy = proxy
 
     def create_session(self, **kwargs: typing.Any) -> aiohttp.ClientSession:
         """Create a client session."""
+        if self._socks_proxy is not None:
+            import aiohttp_socks
+
+            connector = aiohttp_socks.ProxyConnector.from_url(self._socks_proxy)
+        else:
+            connector = None
+
         return aiohttp.ClientSession(
             cookie_jar=aiohttp.DummyCookieJar(),
+            connector=connector,
             **kwargs,
         )
 

genshin/constants.py

@@ -1,5 +1,7 @@
 """Constants hardcoded for optimizations."""
 
+import datetime
+
 from . import types
 
 __all__ = ["APP_IDS", "APP_KEYS", "DS_SALT", "GEETEST_RETCODES", "LANGS"]
@@ -83,3 +85,5 @@
     types.Game.ZZZ: "nap_game_record",
 }
 """Keys used to submit geetest result."""
+
+CN_TIMEZONE = datetime.timezone(datetime.timedelta(hours=8))

genshin/models/genshin/chronicle/abyss.py

@@ -9,6 +9,7 @@
     except ImportError:
         import pydantic
 
+from genshin.constants import CN_TIMEZONE
 from genshin.models.genshin import character
 from genshin.models.model import Aliased, APIModel
 
@@ -113,6 +114,10 @@ def __nest_ranks(cls, values: typing.Dict[str, typing.Any]) -> typing.Dict[str,
         values.setdefault("ranks", {}).update(values)
         return values
 
+    @pydantic.validator("start_time", "end_time", pre=True)
+    def __parse_timezones(cls, value: str) -> datetime.datetime:
+        return datetime.datetime.fromtimestamp(int(value), tz=CN_TIMEZONE)
+
 
 class SpiralAbyssPair(APIModel):
     """Pair of both current and previous spiral abyss.

genshin/models/genshin/daily.py

@@ -3,6 +3,7 @@
 import datetime
 import typing
 
+from genshin.constants import CN_TIMEZONE
 from genshin.models.model import Aliased, APIModel, Unique
 
 __all__ = ["ClaimedDailyReward", "DailyReward", "DailyRewardInfo"]
@@ -16,8 +17,7 @@ class DailyRewardInfo(typing.NamedTuple):
 
     @property
     def missed_rewards(self) -> int:
-        cn_timezone = datetime.timezone(datetime.timedelta(hours=8))
-        now = datetime.datetime.now(cn_timezone)
+        now = datetime.datetime.now(CN_TIMEZONE)
         return now.day - self.claimed_rewards