MRP-2 T13: Low-priority micro-fixes — 8 DRY deduplication patterns

13.1: _get_apps_without_join(table) replaces 4 identical LEFT JOIN queries
13.2: _get_stale_count(table, days) replaces 2 identical stale-count queries
13.3: dataclasses.replace() replaces manual Profile reconstruction (2x)
13.5: _load_json_directory() unifies shared/locale file loading in i18n.py
13.7: _open_readonly_db() in BaseExternalParser for Lutris/itch parsers
13.9: _wire_and_start_track() deduplicates 4x track signal wiring
13.10: _hydrate_row() unifies SmartCollection deserialization (2x)
13.11: _post_search() deduplicates HLTB retry POST request

Skipped 13.4 (mkdir one-liner), 13.6 (minimal ROI), 13.8 (already done in T09).

Author: HeikesFootSlave

src/core/db/enrichment_queries.py

@@ -25,6 +25,41 @@ class EnrichmentQueryMixin:
     Requires ConnectionBase attributes: conn.
     """
 
+    # ── Shared query helpers ─────────────────────────────────────────────
+
+    def _get_apps_without_join(self, table: str) -> list[tuple[int, str]]:
+        """Returns game-type apps with no entry in the given table.
+
+        Args:
+            table: Target table name to LEFT JOIN against.
+
+        Returns:
+            List of (app_id, name) tuples.
+        """
+        cursor = self.conn.execute(
+            f"SELECT g.app_id, g.name FROM games g"
+            f" LEFT JOIN {table} t ON g.app_id = t.app_id"
+            f" WHERE t.app_id IS NULL AND g.app_type IN ('game', '')"
+        )
+        return [(row[0], row[1]) for row in cursor.fetchall()]
+
+    def _get_stale_count(self, table: str, max_age_days: int) -> int:
+        """Counts entries with cache older than max_age_days.
+
+        Args:
+            table: Table name with a last_updated column.
+            max_age_days: Maximum cache age in days.
+
+        Returns:
+            Number of stale entries.
+        """
+        cutoff = int(time.time()) - (max_age_days * 86400)
+        cursor = self.conn.execute(
+            f"SELECT COUNT(*) FROM {table} WHERE last_updated < ?",
+            (cutoff,),
+        )
+        return cursor.fetchone()[0]
+
     # ── Metadata enrichment ──────────────────────────────────────────────
 
     def upsert_game_metadata(self, app_id: int, **fields: Any) -> None:
@@ -130,17 +165,8 @@ def get_apps_missing_metadata(self) -> list[tuple[int, str]]:
         return [(row[0], row[1]) for row in cursor.fetchall()]
 
     def get_apps_without_hltb(self) -> list[tuple[int, str]]:
-        """Returns game-type apps that have no HLTB data.
-
-        Returns:
-            List of (app_id, name) tuples.
-        """
-        cursor = self.conn.execute("""
-            SELECT g.app_id, g.name FROM games g
-            LEFT JOIN hltb_data h ON g.app_id = h.app_id
-            WHERE h.app_id IS NULL AND g.app_type IN ('game', '')
-            """)
-        return [(row[0], row[1]) for row in cursor.fetchall()]
+        """Returns game-type apps that have no HLTB data."""
+        return self._get_apps_without_join("hltb_data")
 
     # ── HLTB ID cache ────────────────────────────────────────────────────
 
@@ -254,17 +280,8 @@ def upsert_protondb(
         )
 
     def get_apps_without_protondb(self) -> list[tuple[int, str]]:
-        """Returns game-type apps that have no ProtonDB rating.
-
-        Returns:
-            List of (app_id, name) tuples.
-        """
-        cursor = self.conn.execute("""
-            SELECT g.app_id, g.name FROM games g
-            LEFT JOIN protondb_ratings p ON g.app_id = p.app_id
-            WHERE p.app_id IS NULL AND g.app_type IN ('game', '')
-            """)
-        return [(row[0], row[1]) for row in cursor.fetchall()]
+        """Returns game-type apps that have no ProtonDB rating."""
+        return self._get_apps_without_join("protondb_ratings")
 
     def get_apps_without_pegi(self) -> list[tuple[int, str]]:
         """Returns game-type apps that have no PEGI age rating.
@@ -355,64 +372,22 @@ def upsert_achievements(self, app_id: int, achievements: list[dict]) -> None:
         )
 
     def get_apps_without_achievements(self) -> list[tuple[int, str]]:
-        """Returns game-type apps that have no achievement_stats entry.
-
-        Returns:
-            List of (app_id, name) tuples.
-        """
-        cursor = self.conn.execute("""
-            SELECT g.app_id, g.name FROM games g
-            LEFT JOIN achievement_stats a ON g.app_id = a.app_id
-            WHERE a.app_id IS NULL AND g.app_type IN ('game', '')
-            """)
-        return [(row[0], row[1]) for row in cursor.fetchall()]
+        """Returns game-type apps that have no achievement_stats entry."""
+        return self._get_apps_without_join("achievement_stats")
 
     # ── Health check queries ─────────────────────────────────────────────
 
     def get_games_missing_artwork(self) -> list[tuple[int, str]]:
-        """Returns games that have no custom artwork entry.
-
-        Returns:
-            List of (app_id, name) tuples.
-        """
-        cursor = self.conn.execute("""
-            SELECT g.app_id, g.name FROM games g
-            LEFT JOIN custom_artwork ca ON g.app_id = ca.app_id
-            WHERE ca.app_id IS NULL AND g.app_type IN ('game', '')
-        """)
-        return [(row[0], row[1]) for row in cursor.fetchall()]
+        """Returns games that have no custom artwork entry."""
+        return self._get_apps_without_join("custom_artwork")
 
     def get_stale_hltb_count(self, max_age_days: int = 30) -> int:
-        """Counts games with HLTB cache older than max_age_days.
-
-        Args:
-            max_age_days: Maximum cache age in days.
-
-        Returns:
-            Number of games with stale HLTB data.
-        """
-        cutoff = int(time.time()) - (max_age_days * 86400)
-        cursor = self.conn.execute(
-            "SELECT COUNT(*) FROM hltb_data WHERE last_updated < ?",
-            (cutoff,),
-        )
-        return cursor.fetchone()[0]
+        """Counts games with HLTB cache older than max_age_days."""
+        return self._get_stale_count("hltb_data", max_age_days)
 
     def get_stale_protondb_count(self, max_age_days: int = 7) -> int:
-        """Counts games with ProtonDB cache older than max_age_days.
-
-        Args:
-            max_age_days: Maximum cache age in days.
-
-        Returns:
-            Number of games with stale ProtonDB data.
-        """
-        cutoff = int(time.time()) - (max_age_days * 86400)
-        cursor = self.conn.execute(
-            "SELECT COUNT(*) FROM protondb_ratings WHERE last_updated < ?",
-            (cutoff,),
-        )
-        return cursor.fetchone()[0]
+        """Counts games with ProtonDB cache older than max_age_days."""
+        return self._get_stale_count("protondb_ratings", max_age_days)
 
     # ── Import recording ─────────────────────────────────────────────────
 

src/core/profile_manager.py

@@ -14,7 +14,7 @@
 import re
 import shutil
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any
 
@@ -273,20 +273,7 @@ def rename_profile(self, old_name: str, new_name: str) -> bool:
         except FileNotFoundError:
             return False
 
-        # Build renamed profile, preserving original created_at
-        renamed = Profile(
-            name=new_name.strip(),
-            collections=old_profile.collections,
-            autocat_methods=old_profile.autocat_methods,
-            tags_per_game=old_profile.tags_per_game,
-            ignore_common_tags=old_profile.ignore_common_tags,
-            filter_enabled_types=old_profile.filter_enabled_types,
-            filter_enabled_platforms=old_profile.filter_enabled_platforms,
-            filter_active_statuses=old_profile.filter_active_statuses,
-            filter_active_languages=old_profile.filter_active_languages,
-            sort_key=old_profile.sort_key,
-            created_at=old_profile.created_at,
-        )
+        renamed = replace(old_profile, name=new_name.strip())
 
         self.save_profile(renamed)
         # Only delete old if filename actually changed
@@ -343,19 +330,7 @@ def import_profile(self, source_path: Path) -> Profile:
 
         # Stamp import time if no creation time was set
         if profile.created_at == 0.0:
-            profile = Profile(
-                name=profile.name,
-                collections=profile.collections,
-                autocat_methods=profile.autocat_methods,
-                tags_per_game=profile.tags_per_game,
-                ignore_common_tags=profile.ignore_common_tags,
-                filter_enabled_types=profile.filter_enabled_types,
-                filter_enabled_platforms=profile.filter_enabled_platforms,
-                filter_active_statuses=profile.filter_active_statuses,
-                filter_active_languages=profile.filter_active_languages,
-                sort_key=profile.sort_key,
-                created_at=time.time(),
-            )
+            profile = replace(profile, created_at=time.time())
 
         self.save_profile(profile)
         logger.info("Imported profile '%s' from %s", profile.name, source_path)

src/integrations/external_games/base_parser.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import logging
+import sqlite3
 from abc import ABC, abstractmethod
 from pathlib import Path
 
@@ -70,3 +71,20 @@ def _find_config_file(self) -> Path | None:
                 logger.debug("Found %s config: %s", self.platform_name(), path)
                 return path
         return None
+
+    def _open_readonly_db(self, db_path: Path) -> sqlite3.Connection | None:
+        """Opens a SQLite database in read-only mode with Row factory.
+
+        Args:
+            db_path: Path to the SQLite database file.
+
+        Returns:
+            Connection with Row factory set, or None on error.
+        """
+        try:
+            conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+            conn.row_factory = sqlite3.Row
+            return conn
+        except sqlite3.Error as e:
+            logger.warning("Failed to open %s database: %s", self.platform_name(), e)
+            return None

src/integrations/external_games/itch_parser.py

@@ -91,11 +91,8 @@ def read_games(self) -> list[ExternalGame]:
         if not db_path.exists():
             return []
 
-        try:
-            conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
-            conn.row_factory = sqlite3.Row
-        except sqlite3.Error as e:
-            logger.warning("Failed to open itch.io database: %s", e)
+        conn = self._open_readonly_db(db_path)
+        if not conn:
             return []
 
         games: list[ExternalGame] = []

src/integrations/external_games/lutris_parser.py

@@ -84,11 +84,8 @@ def read_games(self) -> list[ExternalGame]:
         if not db_path:
             return []
 
-        try:
-            conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
-            conn.row_factory = sqlite3.Row
-        except sqlite3.Error as e:
-            logger.warning("Failed to open Lutris database: %s", e)
+        conn = self._open_readonly_db(db_path)
+        if not conn:
             return []
 
         games: list[ExternalGame] = []

src/integrations/hltb_api.py

@@ -267,6 +267,27 @@ def search_game(self, name: str, app_id: int = 0) -> HLTBResult | None:
             return to_result(match)
         return None
 
+    def _post_search(self, payload: dict) -> requests.Response:
+        """Sends a search POST request to the HLTB API.
+
+        Args:
+            payload: JSON payload for the search.
+
+        Returns:
+            Response object from the API.
+        """
+        return self._session.post(
+            f"{_HLTB_BASE}/api/{self._api_path}",
+            json=payload,
+            headers={
+                "Content-Type": "application/json",
+                "Origin": _HLTB_BASE,
+                "Referer": f"{_HLTB_BASE}/",
+                "x-auth-token": self._auth_token,
+            },
+            timeout=15,
+        )
+
     def _search_and_find(self, search_name: str) -> tuple[dict | None, int]:
         """Performs an HLTB API search and returns the best match with distance.
 
@@ -306,38 +327,15 @@ def _search_and_find(self, search_name: str) -> tuple[dict | None, int]:
             "useCache": True,
         }
 
-        search_url = f"{_HLTB_BASE}/api/{self._api_path}"
-
         try:
-            resp = self._session.post(
-                search_url,
-                json=payload,
-                headers={
-                    "Content-Type": "application/json",
-                    "Origin": _HLTB_BASE,
-                    "Referer": f"{_HLTB_BASE}/",
-                    "x-auth-token": self._auth_token,
-                },
-                timeout=15,
-            )
+            resp = self._post_search(payload)
             # If 404 or 403, invalidate cache and retry once
             if resp.status_code in (403, 404):
                 logger.info("HLTB endpoint returned %d, refreshing...", resp.status_code)
                 self._cache_time = 0.0
                 if not self._ensure_api_ready():
                     return None, 0
-                search_url = f"{_HLTB_BASE}/api/{self._api_path}"
-                resp = self._session.post(
-                    search_url,
-                    json=payload,
-                    headers={
-                        "Content-Type": "application/json",
-                        "Origin": _HLTB_BASE,
-                        "Referer": f"{_HLTB_BASE}/",
-                        "x-auth-token": self._auth_token,
-                    },
-                    timeout=15,
-                )
+                resp = self._post_search(payload)
 
             resp.raise_for_status()
             data = resp.json()