Redo channel init
This commit is contained in:
1 parent
268cf439e7
commit
4d2a884fbe
2 files changed
+41
-42
No files matched your search
+40
-38
@@ -7,7 +7,7 @@ import logging
|
|||||||
from base64 import b64encode
|
from base64 import b64encode
|
||||||
from functools import cached_property
|
from functools import cached_property
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from typing import Any, Optional, TYPE_CHECKING
|
from typing import Any, Optional, SupportsInt, TYPE_CHECKING
|
||||||
|
|
||||||
from utils import Game, invalidate_cache
|
from utils import Game, invalidate_cache
|
||||||
from exceptions import MinerException, RequestException
|
from exceptions import MinerException, RequestException
|
||||||
@@ -49,11 +49,20 @@ class Stream:
|
|||||||
|
|
||||||
|
|
||||||
class Channel:
|
class Channel:
|
||||||
def __init__(self, twitch: Twitch, data: JsonType, *, priority: bool = False):
|
def __init__(
|
||||||
self._init_base(twitch)
|
self,
|
||||||
self.id: int = int(data["id"])
|
twitch: Twitch,
|
||||||
self._login: str = data["name"]
|
*,
|
||||||
self._display_name: Optional[str] = data.get("displayName")
|
id: SupportsInt,
|
||||||
|
login: str,
|
||||||
|
display_name: Optional[str] = None,
|
||||||
|
priority: bool = False,
|
||||||
|
):
|
||||||
|
self._twitch: Twitch = twitch
|
||||||
|
self._gui_channels: ChannelList = twitch.gui.channels
|
||||||
|
self.id: int = int(id)
|
||||||
|
self._login: str = login
|
||||||
|
self._display_name: Optional[str] = display_name
|
||||||
self._spade_url: Optional[str] = None
|
self._spade_url: Optional[str] = None
|
||||||
self.points: Optional[int] = None
|
self.points: Optional[int] = None
|
||||||
self._stream: Optional[Stream] = None
|
self._stream: Optional[Stream] = None
|
||||||
@@ -64,48 +73,31 @@ class Channel:
|
|||||||
# • not cleaned up unless they're streaming a game we haven't selected
|
# • not cleaned up unless they're streaming a game we haven't selected
|
||||||
self.priority: bool = priority
|
self.priority: bool = priority
|
||||||
|
|
||||||
def _init_base(self, twitch: Twitch):
|
@classmethod
|
||||||
self._twitch: Twitch = twitch
|
def from_acl(cls, twitch: Twitch, data: JsonType) -> Channel:
|
||||||
self._gui_channels: ChannelList = twitch.gui.channels
|
return cls(
|
||||||
|
twitch,
|
||||||
@property
|
id=data["id"],
|
||||||
def name(self) -> str:
|
login=data["name"],
|
||||||
if self._display_name is not None:
|
display_name=data.get("displayName"),
|
||||||
return self._display_name
|
priority=True,
|
||||||
return self._login
|
)
|
||||||
|
|
||||||
@property
|
|
||||||
def url(self) -> str:
|
|
||||||
return f"{BASE_URL}/{self._login}"
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_directory(cls, twitch: Twitch, data: JsonType, *, priority: bool = False) -> Channel:
|
def from_directory(cls, twitch: Twitch, data: JsonType) -> Channel:
|
||||||
self = super().__new__(cls)
|
|
||||||
self._init_base(twitch)
|
|
||||||
channel = data["broadcaster"]
|
channel = data["broadcaster"]
|
||||||
self.id = int(channel["id"])
|
self = cls(
|
||||||
self._login = channel["login"]
|
twitch, id=channel["id"], login=channel["login"], display_name=channel["displayName"]
|
||||||
self._display_name = channel["displayName"]
|
)
|
||||||
self._spade_url = None
|
|
||||||
self.points = None
|
|
||||||
self._stream = Stream.from_directory(self, data)
|
self._stream = Stream.from_directory(self, data)
|
||||||
self._pending_stream_up = None
|
|
||||||
self.priority = priority
|
|
||||||
return self
|
return self
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def from_name(
|
async def from_name(
|
||||||
cls, twitch: Twitch, channel_login: str, *, priority: bool = False
|
cls, twitch: Twitch, channel_login: str, *, priority: bool = False
|
||||||
) -> Channel:
|
) -> Channel:
|
||||||
self = super().__new__(cls)
|
self = cls(twitch, id=0, login=channel_login, priority=priority)
|
||||||
self._init_base(twitch)
|
# id and display name to be filled/overwritten by get_stream
|
||||||
# id and display name to be filled by get_stream
|
|
||||||
self._login = channel_login
|
|
||||||
self._spade_url = None
|
|
||||||
self.points = None
|
|
||||||
self._stream = None
|
|
||||||
self._pending_stream_up = None
|
|
||||||
self.priority = priority
|
|
||||||
stream = await self.get_stream()
|
stream = await self.get_stream()
|
||||||
if stream is not None:
|
if stream is not None:
|
||||||
self._stream = stream
|
self._stream = stream
|
||||||
@@ -126,6 +118,16 @@ class Channel:
|
|||||||
def __hash__(self) -> int:
|
def __hash__(self) -> int:
|
||||||
return hash((self.__class__.__name__, self.id))
|
return hash((self.__class__.__name__, self.id))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def name(self) -> str:
|
||||||
|
if self._display_name is not None:
|
||||||
|
return self._display_name
|
||||||
|
return self._login
|
||||||
|
|
||||||
|
@property
|
||||||
|
def url(self) -> str:
|
||||||
|
return f"{BASE_URL}/{self._login}"
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def iid(self) -> str:
|
def iid(self) -> str:
|
||||||
"""
|
"""
|
||||||
|
|||||||
+1
-4
@@ -194,10 +194,7 @@ class DropsCampaign:
|
|||||||
self.ends_at: datetime = timestamp(data["endAt"])
|
self.ends_at: datetime = timestamp(data["endAt"])
|
||||||
allowed = data["allow"]
|
allowed = data["allow"]
|
||||||
self.allowed_channels: List[Channel] = (
|
self.allowed_channels: List[Channel] = (
|
||||||
[
|
[Channel.from_acl(twitch, channel_data) for channel_data in allowed["channels"]]
|
||||||
Channel(twitch, channel_data, priority=True)
|
|
||||||
for channel_data in allowed["channels"]
|
|
||||||
]
|
|
||||||
if allowed["channels"] and allowed.get("isEnabled", True) else []
|
if allowed["channels"] and allowed.get("isEnabled", True) else []
|
||||||
)
|
)
|
||||||
self.timed_drops: Dict[str, TimedDrop] = {
|
self.timed_drops: Dict[str, TimedDrop] = {
|
||||||
|
|||||||
Reference in new issue
Block a user