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

static check & format

Browse Source
This commit is contained in:
Fengqing Liu
2025-10-25 14:59:24 +11:00
parent cd75e0818c
commit c046deb967
9 changed files with 66 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CLAUDE.md
View File

@@ -300,7 +300,7 @@ The project does not include a test suite. Manual testing workflow:
1. Run with `-vvv` for maximum verbosity (levels: -v, -vv, -vvv, -vvvv)
2. Use `--dump` to generate debug data dumps
3. Check timestamped log files in `./logs/` directory (always created as `TDM.TIMESTAMP.log`)
3. Check log files in `./logs/` directory
4. Use `--debug-ws` for websocket debug logging
5. Use `--debug-gql` for GraphQL debug logging
6. Monitor web GUI console output and browser developer tools

7
add_language_names.py
View File

@@ -4,8 +4,10 @@
import json
from pathlib import Path
LANG_PATH = Path(__file__).parent / "lang"
def add_language_names():
"""Add language_name field to each translation file based on filename."""
for filepath in LANG_PATH.glob("*.json"):
@@ -15,7 +17,7 @@ def add_language_names():
print(f"Processing {filepath.name}...")
# Read the JSON file
with open(filepath, 'r', encoding='utf-8') as f:
with open(filepath, encoding="utf-8") as f:
data = json.load(f)
# Add language_name at the beginning
@@ -23,11 +25,12 @@ def add_language_names():
updated_data.update(data)
# Write back to file with proper formatting
with open(filepath, 'w', encoding='utf-8') as f:
with open(filepath, "w", encoding="utf-8") as f:
json.dump(updated_data, f, ensure_ascii=False, indent=4)
print(f" ✓ Added language_name: {language_name}")
if __name__ == "__main__":
add_language_names()
print("\n✓ All translation files updated!")

22
add_more_translations.py
View File

@@ -5,9 +5,9 @@ Adds English text as placeholders where translations are missing.
"""
import json
import os
from pathlib import Path
# Translations to add/update for each language
# Format: {language_code: {key_path: translation}}
TRANSLATIONS = {
@@ -36,10 +36,10 @@ TRANSLATIONS = {
"gui.help.how_to_use": "使用方法",
"gui.help.how_to_use_items": [
"使用您的 Twitch 账号登录（OAuth 设备代码流程）",
"在 <a href=\"https://www.twitch.tv/drops/campaigns\" target=\"_blank\">twitch.tv/drops/campaigns</a> 关联您的账号",
'在 <a href="https://www.twitch.tv/drops/campaigns" target="_blank">twitch.tv/drops/campaigns</a> 关联您的账号',
"矿工将自动发现活动并开始挖掘",
"在设置中配置优先游戏以关注您想要的内容",
"在主界面和库存选项卡中监控进度"
"在主界面和库存选项卡中监控进度",
],
"gui.help.features": "功能",
"gui.help.features_items": [
@@ -47,13 +47,13 @@ TRANSLATIONS = {
"游戏优先级和排除列表",
"同时跟踪最多 199 个频道",
"自动切换频道",
"实时进度跟踪"
"实时进度跟踪",
],
"gui.help.important_notes": "重要提示",
"gui.help.important_notes_items": [
"挖掘时请勿在同一账号上观看流",
"保护好您的 cookies.jar 文件",
"需要关联游戏账号才能掉宝"
"需要关联游戏账号才能掉宝",
],
"gui.help.github_repo": "GitHub 仓库",
"gui.header.language": "语言：",
@@ -88,10 +88,10 @@ DEFAULT_TRANSLATIONS = {
"gui.help.how_to_use": "How to Use",
"gui.help.how_to_use_items": [
"Login using your Twitch account (OAuth device code flow)",
"Link your accounts at <a href=\"https://www.twitch.tv/drops/campaigns\" target=\"_blank\">twitch.tv/drops/campaigns</a>",
'Link your accounts at <a href="https://www.twitch.tv/drops/campaigns" target="_blank">twitch.tv/drops/campaigns</a>',
"The miner will automatically discover campaigns and start mining",
"Configure priority games in Settings to focus on what you want",
"Monitor progress in the Main and Inventory tabs"
"Monitor progress in the Main and Inventory tabs",
],
"gui.help.features": "Features",
"gui.help.features_items": [
@@ -99,13 +99,13 @@ DEFAULT_TRANSLATIONS = {
"Game priority and exclusion lists",
"Tracks up to 199 channels simultaneously",
"Automatic channel switching",
"Real-time progress tracking"
"Real-time progress tracking",
],
"gui.help.important_notes": "Important Notes",
"gui.help.important_notes_items": [
"Do not watch streams on the same account while mining",
"Keep your cookies.jar file secure",
"Requires linked game accounts for drops"
"Requires linked game accounts for drops",
],
"gui.help.github_repo": "GitHub Repository",
"gui.header.language": "Language:",
@@ -139,7 +139,7 @@ def get_nested_value(data, key_path, default=None):
def update_language_file(file_path):
"""Update a language file with missing translations."""
with open(file_path, 'r', encoding='utf-8') as f:
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
language_name = data.get("english_name", "Unknown")
@@ -164,7 +164,7 @@ def update_language_file(file_path):
if updated:
# Write back with proper formatting
with open(file_path, 'w', encoding='utf-8') as f:
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
print(f" ✓ Saved {language_name}")
else:

30
src/core/client.py
View File

@@ -6,7 +6,7 @@ from collections import OrderedDict, abc, deque
from datetime import datetime, timedelta, timezone
from functools import partial
from time import time
from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn
from typing import TYPE_CHECKING, Any, Final, Literal
import aiohttp
@@ -32,7 +32,6 @@ from src.services.message_handlers import MessageHandlerService
from src.services.watch_service import WatchService
from src.utils import (
AwaitableValue,
task_wrapper,
)
from src.websocket import WebsocketPool
@@ -209,9 +208,14 @@ class Twitch:
# Add default topics
self.websocket.add_topics(
[
WebsocketTopic("User", "Drops", auth_state.user_id, self._message_handler_service.process_drops),
WebsocketTopic(
"User", "Notifications", auth_state.user_id, self._message_handler_service.process_notifications
"User", "Drops", auth_state.user_id, self._message_handler_service.process_drops
),
WebsocketTopic(
"User",
"Notifications",
auth_state.user_id,
self._message_handler_service.process_notifications,
),
]
)
@@ -433,12 +437,18 @@ class Twitch:
for channel_id in channels:
to_add_topics.append(
WebsocketTopic(
"Channel", "StreamState", channel_id, self._message_handler_service.process_stream_state
"Channel",
"StreamState",
channel_id,
self._message_handler_service.process_stream_state,
)
)
to_add_topics.append(
WebsocketTopic(
"Channel", "StreamUpdate", channel_id, self._message_handler_service.process_stream_update
"Channel",
"StreamUpdate",
channel_id,
self._message_handler_service.process_stream_update,
)
)
self.websocket.add_topics(to_add_topics)
@@ -512,7 +522,9 @@ class Twitch:
self.exit_manual_mode("No channels available for manual game")
# Auto-select best channel based on priority
else:
for channel in sorted(channels.values(), key=self._channel_service.get_priority):
for channel in sorted(
channels.values(), key=self._channel_service.get_priority
):
if self.can_watch(channel) and self.should_switch(channel):
new_watching = channel
break
@@ -531,7 +543,9 @@ class Twitch:
if self.is_manual_mode() and self._manual_target_game:
status_text = f"🎯 Manual Mode: Watching {watching_channel.name} for {self._manual_target_game.name}"
else:
status_text = _.t["status"]["watching"].format(channel=watching_channel.name)
status_text = _.t["status"]["watching"].format(
channel=watching_channel.name
)
self.gui.status.update(status_text)
self._state_change.clear()
else:

5
src/i18n/translator.py
View File

@@ -6,6 +6,7 @@ from typing import TypedDict, cast
from src.config import DEFAULT_LANG, LANG_PATH
class StatusMessages(TypedDict):
terminated: str
watching: str
@@ -213,8 +214,9 @@ class Translator:
self.t: Translation
# load available languages from JSON files by reading language_name field
for filepath in LANG_PATH.glob("*.json"):
with filepath.open("r", encoding="utf-8") as json_file:
try:
loaded_translation: Translation = json.load(open(filepath, "r"))
loaded_translation: Translation = json.load(json_file)
self._langs[loaded_translation["language_name"]] = loaded_translation
except Exception as e:
# if we can't read the file, skip it
@@ -233,4 +235,5 @@ class Translator:
self.current_language = language
self.t = cast(Translation, self._langs.get(language))
_ = Translator()

11
src/services/channel_service.py
View File

@@ -150,14 +150,21 @@ class ChannelService:
# NOTE: Have to do this here, because "channels" can be any iterable
return
stream_gql_tasks: list[asyncio.Task[list[JsonType]]] = [
# gql_request may return either a single JsonType or a list[JsonType],
# so accept the union in the Task type.
stream_gql_tasks: list[asyncio.Task[JsonType | list[JsonType]]] = [
asyncio.create_task(self._twitch.gql_request(stream_gql_chunk))
for stream_gql_chunk in chunk(stream_gql_ops, 20)
]
try:
for coro in asyncio.as_completed(stream_gql_tasks):
response_list: list[JsonType] = await coro
response = await coro
# Normalize response to a list for uniform processing
if isinstance(response, list):
response_list: list[JsonType] = response
else:
response_list = [response]
for response_json in response_list:
channel_data: JsonType = response_json["data"]["user"]
if channel_data is not None:

4
src/services/inventory_service.py
View File

@@ -185,7 +185,9 @@ class InventoryService:
for i, coro in enumerate(asyncio.as_completed(add_campaign_tasks), start=1):
await coro
status_update(
_.t["gui"]["status"]["adding_campaigns"].format(counter=f"({i}/{len(campaigns)})")
_.t["gui"]["status"]["adding_campaigns"].format(
counter=f"({i}/{len(campaigns)})"
)
)
# this is needed here explicitly, because cache reads from disk don't raise this
from src.config import State

2
src/web/managers/login.py
View File

@@ -105,7 +105,7 @@ class LoginFormManager:
Returns:
Dictionary with status, user_id, and optional oauth_pending data
"""
result = {"status": self._status, "user_id": self._user_id}
result: dict[str, Any] = {"status": self._status, "user_id": self._user_id}
# Include OAuth code if pending
if self._oauth_pending:
result["oauth_pending"] = self._oauth_pending

5
src/web/managers/settings.py
View File

@@ -69,10 +69,13 @@ class SettingsManager:
_.set_language(language)
self._settings.language = language
# Notify clients that translations need to be reloaded
asyncio.create_task(self._broadcaster.emit("language_changed", {"language": language}))
asyncio.create_task(
self._broadcaster.emit("language_changed", {"language": language})
)
except ValueError as e:
# Invalid language, log warning
import logging
logging.warning(f"Invalid language '{language}': {e}")
if "connection_quality" in settings_data:
self._settings.connection_quality = settings_data["connection_quality"]