GitHub/TwitchDropsMiner
1
0
Fork 0
mirror of https://github.com/rangermix/TwitchDropsMiner.git synced 2026-06-06 04:19:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove redundant stuff

Browse Source
This commit is contained in:
Fengqing Liu
2025-10-19 17:08:18 +11:00
parent e8144e9591
commit d466f46d6f
39 changed files with 372 additions and 573 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/models/benefit.py
View File

@@ -10,6 +10,7 @@ if TYPE_CHECKING:
class BenefitType(Enum):
"""Type of drop benefit (reward)."""
UNKNOWN = "UNKNOWN"
BADGE = "BADGE"
EMOTE = "EMOTE"
@@ -21,6 +22,7 @@ class BenefitType(Enum):
class Benefit:
"""Represents a reward/benefit from a completed drop."""
__slots__ = ("id", "name", "type", "image_url")
def __init__(self, data: JsonType):

20
src/models/campaign.py
View File

@@ -41,7 +41,8 @@ class DropsCampaign:
allowed: JsonType = data["allow"]
self.allowed_channels: list[Channel] = (
[Channel.from_acl(twitch, channel_data) 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] = {
drop_data["id"]: TimedDrop(self, drop_data, claimed_benefits)
@@ -139,13 +140,15 @@ class DropsCampaign:
self.eligible # account is eligible
and self.active # campaign is active (and valid)
and (
channel is None or ( # channel isn't specified,
channel is None
or ( # channel isn't specified,
# or there's no ACL, or the channel is in the ACL
(not self.allowed_channels or channel in self.allowed_channels)
# and the channel is live and playing the campaign's game
and (
ignore_channel_status
or channel.game is not None and channel.game == self.game
or channel.game is not None
and channel.game == self.game
)
)
)
@@ -163,13 +166,10 @@ class DropsCampaign:
)
)
def can_earn(
self, channel: Channel | None = None, ignore_channel_status: bool = False
) -> bool:
def can_earn(self, channel: Channel | None = None, ignore_channel_status: bool = False) -> bool:
# True if any of the containing drops can be earned
return (
self._base_can_earn(channel, ignore_channel_status)
and any(drop._base_can_earn() for drop in self.drops)
return self._base_can_earn(channel, ignore_channel_status) and any(
drop._base_can_earn() for drop in self.drops
)
def can_earn_within(self, stamp: datetime) -> bool:
@@ -193,7 +193,7 @@ class DropsCampaign:
# Executes if any drop's extra_current_minutes reach MAX_ESTIMATED_MINUTES
# TODO: Figure out a better way to handle this case
logger.warning(
f"At least one of the drops in campaign \"{self.name}({self.game.name})\" "
f'At least one of the drops in campaign "{self.name}({self.game.name})" '
"has reached the maximum extra minutes limit!"
)
self._twitch.change_state(State.CHANNEL_SWITCH)

25
src/models/channel.py
View File

@@ -60,7 +60,7 @@ class Stream:
"muted": False,
"player": "site",
"user_id": self.channel._twitch._auth_state.user_id,
}
},
}
]
return {"data": (b64encode(json_minify(payload).encode("utf8"))).decode("utf8")}
@@ -107,7 +107,7 @@ class Stream:
token_value = token_data["value"]
token_signature = token_data["signature"]
# using the token, query Twitch for a list of all available stream qualities
available_qualities: str = ''
available_qualities: str = ""
try:
async with self.channel._twitch.request(
"GET",
@@ -128,7 +128,7 @@ class Stream:
if isinstance(available_json, list):
available_json = available_json[0]
if "error" in available_json:
logger.error(f"Stream URL get error: \"{available_json['error']}\"")
logger.error(f'Stream URL get error: "{available_json["error"]}"')
self.channel.set_offline()
return None
# pick the last URL from the list, usually with the lowest quality stream
@@ -141,8 +141,15 @@ class Stream:
class Channel:
__slots__ = (
"_twitch", "_gui_channels", "id", "_login", "_display_name", "_spade_url",
"_stream", "_pending_stream_up", "acl_based"
"_twitch",
"_gui_channels",
"id",
"_login",
"_display_name",
"_spade_url",
"_stream",
"_pending_stream_up",
"acl_based",
)
def __init__(
@@ -289,9 +296,7 @@ class Channel:
For mobile view, spade_url is available immediately from the page, skipping step #2.
"""
SETTINGS_PATTERN: str = (
r'src="(https://[\w.]+/config/settings\.[0-9a-f]{32}\.js)"'
)
SETTINGS_PATTERN: str = r'src="(https://[\w.]+/config/settings\.[0-9a-f]{32}\.js)"'
SPADE_PATTERN: str = (
r'"spade_?url": ?"(https://video-edge-[.\w\-/]+\.ts(?:\?allow_stream=true)?)"'
)
@@ -441,7 +446,7 @@ class Channel:
# the response may contain some invalid JSON with duplicate double quotes
# in the value strings: we need to get rid of them by removing the "url" key entirely
# if no JSON can be found within the response, this is a NOOP
available_chunks = re.sub(r'"url": ?".+}",', '', available_chunks)
available_chunks = re.sub(r'"url": ?".+}",', "", available_chunks)
# try to decode the suspected JSON
try:
available_json: JsonType = json.loads(available_chunks)
@@ -453,7 +458,7 @@ class Channel:
if isinstance(available_json, list):
available_json = available_json[0]
if "error" in available_json:
logger.error(f"Send watch error: \"{available_json['error']}\"")
logger.error(f'Send watch error: "{available_json["error"]}"')
return False
# the list contains ~10-13 chunks of the stream at 2s intervals,
# pick the last chunk URL available. Ensure it's not the end-of-stream tag,

27
src/models/drop.py
View File

@@ -22,12 +22,12 @@ if TYPE_CHECKING:
logger = logging.getLogger("TwitchDrops")
DIMS_PATTERN = re.compile(r'-\d+x\d+(?=\.(?:jpg|png|gif)$)', re.I)
DIMS_PATTERN = re.compile(r"-\d+x\d+(?=\.(?:jpg|png|gif)$)", re.I)
def remove_dimensions(url: str) -> str:
"""Remove dimension suffix from Twitch image URLs (e.g., -285x380.jpg)."""
return DIMS_PATTERN.sub('', url)
return DIMS_PATTERN.sub("", url)
class BaseDrop:
@@ -71,7 +71,7 @@ class BaseDrop:
elif self.can_earn():
additional = ", can_earn=True"
else:
additional = ''
additional = ""
return f"Drop({self.rewards_text()}{additional})"
@property
@@ -107,11 +107,9 @@ class BaseDrop:
and self.starts_at < stamp
)
def can_earn(
self, channel: Channel | None = None, ignore_channel_status: bool = False
) -> bool:
return (
self._base_can_earn() and self.campaign._base_can_earn(channel, ignore_channel_status)
def can_earn(self, channel: Channel | None = None, ignore_channel_status: bool = False) -> bool:
return self._base_can_earn() and self.campaign._base_can_earn(
channel, ignore_channel_status
)
@property
@@ -152,7 +150,7 @@ class BaseDrop:
# two different claim texts, becase a new line after the game name
# looks ugly in the output window - replace it with a space
self._twitch.print(
_("status", "claimed_drop").format(drop=claim_text.replace('\n', ' '))
_("status", "claimed_drop").format(drop=claim_text.replace("\n", " "))
)
self._twitch.gui.tray.notify(claim_text, _("gui", "tray", "notification_title"))
else:
@@ -183,9 +181,9 @@ class BaseDrop:
elif "claimDropRewards" in data:
if not data["claimDropRewards"]:
return False
elif (
data["claimDropRewards"]["status"]
in ("ELIGIBLE_FOR_ALL", "DROP_INSTANCE_ALREADY_CLAIMED")
elif data["claimDropRewards"]["status"] in (
"ELIGIBLE_FOR_ALL",
"DROP_INSTANCE_ALREADY_CLAIMED",
):
return True
return False
@@ -211,11 +209,11 @@ class TimedDrop(BaseDrop):
elif self.can_earn():
additional = ", can_earn=True"
else:
additional = ''
additional = ""
if 0 < self.current_minutes < self.required_minutes:
minutes = f", {self.current_minutes}/{self.required_minutes}"
else:
minutes = ''
minutes = ""
return f"Drop({self.rewards_text()}{minutes}{additional})"
@property
@@ -257,6 +255,7 @@ class TimedDrop(BaseDrop):
@property
def availability(self) -> float:
import math
now = datetime.now(timezone.utc)
if self.required_minutes > 0 and self.total_remaining_minutes > 0 and now < self.ends_at:
return ((self.ends_at - now).total_seconds() / 60) / self.total_remaining_minutes

6
src/models/game.py
View File

@@ -40,9 +40,9 @@ class Game:
Converts the game name into a slug, useable for the GQL API.
"""
# remove specific characters
slug_text = re.sub(r'\'', '', self.name.lower())
slug_text = re.sub(r"\'", "", self.name.lower())
# remove non alpha-numeric characters
slug_text = re.sub(r'\W+', '-', slug_text)
slug_text = re.sub(r"\W+", "-", slug_text)
# strip and collapse dashes
slug_text = re.sub(r'-{2,}', '-', slug_text.strip('-'))
slug_text = re.sub(r"-{2,}", "-", slug_text.strip("-"))
return slug_text