merge dev into rewrite

css_browserhook.py

@@ -4,6 +4,7 @@
 import css_inject
 
 MAX_QUEUE_SIZE = 500
+ON_WEBSOCKET_CONNECT = asyncio.Event()
 
 class BrowserTabHook:
     def __init__(self, browserHook, sessionId : str, targetInfo : dict):
@@ -427,6 +428,8 @@ async def health_check(self):
                 Log("Connected to Steam Browser")
                 await self.send_command("Target.setDiscoverTargets", {"discover": True}, None, False)
 
+                ON_WEBSOCKET_CONNECT.set()
+
                 async for message in self.websocket:
                     data = message.json()
                     for x in self.ws_response:
@@ -436,6 +439,8 @@ async def health_check(self):
             except Exception as e:
                 Result(False, f"[Health Check] {str(e)}")
 
+            ON_WEBSOCKET_CONNECT.clear()
+
             try:
                 await self.close_websocket()
             except:

css_inject.py

@@ -6,32 +6,7 @@
 from typing import List
 from css_utils import Result, Log, store_read, get_theme_path
 from css_browserhook import BrowserTabHook as CssTab, inject, remove
-
-CLASS_MAPPINGS = {}
-
-def initialize_class_mappings():
-    css_translations_path = os.path.join(get_theme_path(), "css_translations.json")
-
-    if not os.path.exists(css_translations_path):
-        Log("Failed to get css translations from local file")
-        return
-
-    try:
-        with open(css_translations_path, "r", encoding="utf-8") as fp:
-            data : dict = json.load(fp)
-    except Exception as e:
-        Log(f"Failed to load css translations from local file: {str(e)}")
-        return
-
-    CLASS_MAPPINGS.clear()
-
-    # Data is in the format of { "uid": ["ver1", "ver2", "ver3"]}
-    for uid in data:
-        latest_value = data[uid][-1]
-        for y in data[uid][:-1]:
-            CLASS_MAPPINGS[y] = latest_value
-
-    Log(f"Loaded {len(CLASS_MAPPINGS)} css translations from local file")
+from css_mappings import CLASS_MAPPINGS
 
 ALL_INJECTS = []
 
@@ -55,7 +30,7 @@ def __init__(self, cssPath : str, tabs : List[str], theme):
 
     async def load(self) -> Result:
         try:
-            with open(self.cssPath, "r") as fp:
+            with open(self.cssPath, "r", encoding="utf-8") as fp:
                 self.css = fp.read()
 
             split_css = re.split(r"(\.[_a-zA-Z]+[_a-zA-Z0-9-]*)", self.css)

css_loader.py

@@ -36,11 +36,10 @@ async def load(self, inject_now : bool = True):
         for x in self.themes:
             await self._set_theme_score(x)
 
-        Log(self.scores)
         self.themes.sort(key=lambda d: self.scores[d.name])
 
         for x in self.themes:
-            Log(f"Loading theme {x.name}")
+            #Log(f"Loading theme {x.name}")
             await x.load(inject_now)
 
         await self._cache_lists()
@@ -298,6 +297,7 @@ async def _parse_themes(self, themesDir : str, configDir : str = None) -> list[t
 
         possibleThemeDirs = [str(x) for x in listdir(themesDir)]
         fails = []
+        theme_count = 0
 
         for x in possibleThemeDirs:
             themePath = themesDir + "/" + x
@@ -310,7 +310,7 @@ async def _parse_themes(self, themesDir : str, configDir : str = None) -> list[t
             try:
                 theme = None
                 if path.exists(themeDataPath):
-                    with open(themeDataPath, "r") as fp:
+                    with open(themeDataPath, "r", encoding="utf-8") as fp:
                         theme = json.load(fp)
 
                 themeData = Theme(themePath, theme, configPath)
@@ -325,11 +325,13 @@ async def _parse_themes(self, themesDir : str, configDir : str = None) -> list[t
 
                 if (themeData.name not in theme_names):
                     self.themes.append(themeData)
-                    Log(f"Found theme {themeData.name}")
+                    theme_count += 1
 
             except Exception as e:
                 Result(False, f"Failed parsing '{x}': {e}") # Couldn't properly parse everything
                 fails.append((x, str(e)))
+
+        Log(f"Loaded {theme_count} themes, failed to load {len(fails)} themes")
 
         return fails
 

css_mappings.py

@@ -0,0 +1,199 @@
+import os, asyncio, aiohttp, json, time
+from css_utils import Log, get_theme_path, get_steam_version
+from css_settings import setting_beta_mappings
+from css_browserhook import ON_WEBSOCKET_CONNECT
+
+STARTED_FETCHING_TRANSLATIONS = False
+SUCCESSFUL_FETCH_THIS_RUN = False
+CLASS_MAPPINGS = {}
+
+def __get_target_branch() -> str:
+    is_beta = setting_beta_mappings()
+    return "beta" if is_beta else "stable"
+
+def __get_same_branch_versions(data : dict) -> list[str]:
+    target_branch = __get_target_branch()
+    return [x for x in data['versions'] if data['versions'][x] == target_branch][::-1]
+
+def __get_target_steam_version(data : dict) -> str|None:
+    local_steam_version = get_steam_version()
+
+    if local_steam_version and local_steam_version in data['versions'] and data['versions'][local_steam_version] == __get_target_branch():
+        target_steam_version = local_steam_version
+    else:
+        target_steam_version = None
+        prev = "9999999999999"
+        for version in __get_same_branch_versions(data):
+            if local_steam_version == None:
+                target_steam_version = version
+                break
+
+            if int(prev) > int(local_steam_version) and int(version) < int(local_steam_version):
+                target_steam_version = version
+                break
+
+            prev = version
+
+        if target_steam_version not in data['versions']:
+            Log("Cannot find suitable version for translation.")
+            return None
+
+    return target_steam_version
+
+def generate_webpack_id_name_list_from_local_file() -> dict[str, dict]:
+    name_list = {}
+    path = os.path.join(get_theme_path(), "css_translations.json")
+
+    try:
+        with open(path, 'r', encoding="utf-8") as fp:
+            data = json.load(fp)
+    except Exception as e:
+        Log(f"Error while loading translations: {str(e)}.")
+        return name_list 
+
+    target_steam_version = __get_target_steam_version(data)
+    same_branch_versions = __get_same_branch_versions(data)
+    if target_steam_version == None:
+        return name_list
+
+    for _, (module_id, module_data) in enumerate(data['module_mappings'].items()):
+        if target_steam_version in module_data['ids']:
+            new_module_id = module_data['ids'][target_steam_version]
+        else: 
+            prev = "9999999999999"
+            new_module_id = None
+            for _, (steam_version, module_id_of_steam_version) in list(enumerate(module_data['ids'].items()))[::-1]:
+                if target_steam_version not in same_branch_versions:
+                    continue
+
+                if int(prev) > int(target_steam_version) and int(steam_version) < int(target_steam_version):
+                    new_module_id = module_id_of_steam_version
+                    break
+
+                prev = steam_version
+
+        if new_module_id == None:
+            # Assuming module doesn't exist in this steam version
+            continue
+
+        name_list[new_module_id] = {"name": str(module_id) if module_data['name'] is None else module_data['name'], "ignore": module_data['ignore_webpack_keys']}
+
+    return name_list
+
+def generate_translations_from_local_file() -> dict[str, str]:
+    translations = {}
+    timer = time.time()
+    failed_match_version = 0
+
+    path = os.path.join(get_theme_path(), "css_translations.json")
+
+    if not os.path.exists(path):
+        Log("Translations file does not exist.")
+        return translations
+
+    try:
+        with open(path, 'r', encoding="utf-8") as fp:
+            data = json.load(fp)
+    except Exception as e:
+        Log(f"Error while loading translations: {str(e)}.")
+        return translations
+
+    target_steam_version = __get_target_steam_version(data)
+    same_branch_versions = __get_same_branch_versions(data)
+    if target_steam_version == None:
+        return translations
+
+    Log(f"Using steam version {target_steam_version} for translations. Available versions for the {__get_target_branch()} branch: {same_branch_versions}")
+
+    for _, (module_id, module_data) in enumerate(data['module_mappings'].items()):
+        module_name = (str(module_id) if module_data['name'] is None else module_data['name'])
+        for _, (class_name, class_mappings) in enumerate(module_data['classname_mappings'].items()):
+            if target_steam_version in class_mappings:
+                class_target = class_mappings[target_steam_version]
+            else:
+                prev = "9999999999999"
+                class_target = None
+                for _, (class_mapping_name, class_mapping_value) in list(enumerate(class_mappings.items()))[::-1]:
+                    if target_steam_version not in same_branch_versions:
+                        continue
+
+                    if int(prev) > int(target_steam_version) and int(class_mapping_name) < int(target_steam_version):
+                        class_target = class_mapping_value
+                        break
+
+                    prev = class_mapping_name
+
+                if class_target == None:
+                    #Log(f"No suitable version found for mapping {module_id}_{class_name}. Using last")
+                    failed_match_version += 1
+                    class_target = class_mappings[list(class_mappings)[-1]]
+
+            for class_mapping_value in class_mappings.values():
+                if class_mapping_value == class_target:
+                    continue
+
+                translations[class_mapping_value] = class_target
+
+            translations[f"{module_name}_{class_name}"] = class_target
+            translations[f"{module_id}_{class_name}"] = class_target
+
+    Log(f"Loaded {len(translations)} css translations from local file in {time.time() - timer:.1f}s. Failed to match version on {failed_match_version} translations.")
+    return translations
+
+def load_global_translations():
+    CLASS_MAPPINGS.clear()
+
+    try:
+        for _, (k, v) in enumerate(generate_translations_from_local_file().items()):
+            CLASS_MAPPINGS[k] = v
+    except Exception as e:
+        Log(f"Error while loading global translations: {str(e)}")
+
+async def __fetch_class_mappings(css_translations_path : str, loader):
+    global SUCCESSFUL_FETCH_THIS_RUN
+
+    if SUCCESSFUL_FETCH_THIS_RUN:
+        return
+
+    css_translations_url = "https://api.deckthemes.com/mappings.json"
+    Log(f"Fetching CSS mappings from {css_translations_url}")
+
+    try:
+        async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False, use_dns_cache=False), timeout=aiohttp.ClientTimeout(total=30)) as session:
+            async with session.get(css_translations_url) as response:
+                if response.status == 200:
+                    text = await response.text()
+
+                    if len(text.strip()) <= 0:
+                        raise Exception("Empty response")
+
+                    with open(css_translations_path, "w", encoding="utf-8") as fp:
+                        fp.write(text)
+
+                    SUCCESSFUL_FETCH_THIS_RUN = True
+                    Log(f"Fetched css translations from server")
+                    load_global_translations()
+                    asyncio.get_running_loop().create_task(loader.reset(silent=True))
+
+    except Exception as ex:
+        Log(f"Failed to fetch css translations from server [{type(ex).__name__}]: {str(ex)}")
+
+async def __every(__seconds: float, func, *args, **kwargs):
+    global SUCCESSFUL_FETCH_THIS_RUN
+
+    await ON_WEBSOCKET_CONNECT.wait()
+
+    while not SUCCESSFUL_FETCH_THIS_RUN:
+        await func(*args, **kwargs)
+        await asyncio.sleep(__seconds)
+
+async def force_fetch_translations(loader):
+    global SUCCESSFUL_FETCH_THIS_RUN
+
+    SUCCESSFUL_FETCH_THIS_RUN = False
+    css_translations_path = os.path.join(get_theme_path(), "css_translations.json")
+    await __fetch_class_mappings(css_translations_path, loader)
+
+def start_fetch_translations(loader):
+    css_translations_path = os.path.join(get_theme_path(), "css_translations.json")
+    asyncio.get_event_loop().create_task(__every(60, __fetch_class_mappings, css_translations_path, loader))

css_remoteinstall.py

@@ -1,5 +1,6 @@
 import asyncio, json, tempfile, os, aiohttp, zipfile, shutil
-from css_utils import Result, Log, get_theme_path, store_or_file_config
+from css_utils import Result, Log, get_theme_path
+from css_settings import setting_install_dependencies
 from css_theme import CSS_LOADER_VER, Theme
 
 async def run(command : str) -> str:
@@ -57,7 +58,7 @@ async def install(id : str, base_url : str, local_themes : list) -> Result:
 
     tempDir.cleanup()
 
-    if not store_or_file_config("no_deps_install"):
+    if setting_install_dependencies():
         for x in data["dependencies"]:
             if x["name"] in local_themes:
                 continue

css_settings.py

@@ -0,0 +1,29 @@
+import os
+from css_utils import get_theme_path, store_write, store_read, is_steam_beta_active
+
+def store_or_file_config(key : str) -> bool:
+    if os.path.exists(os.path.join(get_theme_path(), key.upper())):
+        return True
+
+    read = store_read(key)
+    return read == "True" or read == "1"
+
+def setting_beta_mappings() -> bool:
+    setting = store_read("beta_translations")
+
+    return ((len(setting.strip()) <= 0 or setting == "-1" or setting == "auto") and is_steam_beta_active()) or (setting == "1" or setting == "true")
+
+def setting_redirect_logs() -> bool:
+    return not store_or_file_config("no_redirect_logs")
+
+def setting_watch_files() -> bool:
+    return store_or_file_config("watch")
+
+def set_setting_watch_files(watch : bool):
+    store_write("watch", "1" if watch else "0")
+
+def setting_run_server() -> bool:
+    return store_or_file_config("server")
+
+def setting_install_dependencies() -> bool:
+    return not store_or_file_config("no_deps_install")