[0.8.0] refactor app_state using dataclass

Author: Radekyspec

app_state/__init__.py

@@ -1,7 +1,8 @@
+from dataclasses import dataclass, field
 from functools import partial
 from json import dumps, loads
 from queue import Queue
-from typing import Optional
+from typing import Optional, Any, List
 
 from keyring import get_password
 from obsws_python import ReqClient
@@ -10,74 +11,91 @@
 
 import constant
 from constant import *
-from models.classes import ThreadSafeDict
+from .app_state_base import StateBase
 
 dumps = partial(dumps, ensure_ascii=False,
                 separators=(",", ":"))
-_APP_CONFIG_INITIALIZATION = {
-    "proxy_mode": ProxyMode.NONE,
-    "custom_proxy_url": "",
-    "custom_tray_icon": "",
-    "custom_tray_hint": "",
-    "custom_font": "",
-    "prefer_proto": PreferProto.RTMP,
-}
-_OBS_SETTINGS_INITIALIZATION = {
-    "ip_addr": "localhost",
-    "port": "4455",
-    "password": "",
-    "auto_live": False,
-    "auto_connect": False
-}
-_SCAN_STATUS_INITIALIZATION = {
-    "scanned": False, "qr_key": None, "qr_url": None,
-    "expired": False, "is_new": False,
-    "cred_loaded": False,
-    "timeout": False, "wait_for_confirm": False,
-    "area_updated": False, "room_updated": False,
-    "const_updated": False, "announce_updated": False
-}
-_STREAM_STATUS_INITIALIZATION = {
-    "live_status": False,
-    "required_face": False,
-    "identified_face": False,
-    "face_url": None,
-    "stream_addr": None,
-    "stream_key": None
-}
-_ROOM_INFO_INITIALIZATION = {
-    "cover_audit_reason": "",
-    "cover_url": "",
-    "cover_status": CoverStatus.AUDIT_IN_PROGRESS,
-    "cover_data": None,
-    "room_id": "",
-    "title": "",
-    "parent_area": "",
-    "area": "",
-    "announcement": "",
-    "recent_areas": [],
-    "recent_title": [],
-}
+
+
+@dataclass
+class AppSettings(StateBase):
+    proxy_mode: ProxyMode = ProxyMode.NONE
+    custom_proxy_url: str = ""
+    custom_tray_icon: str = ""
+    custom_tray_hint: str = ""
+    custom_font: str = ""
+    prefer_proto: PreferProto = PreferProto.RTMP
+
+
+@dataclass
+class ObsSettings(StateBase):
+    ip_addr: str = "localhost"
+    port: str = "4455"
+    password: str = ""
+    auto_live: bool = False
+    auto_connect: bool = False
+
+
+@dataclass
+class ScanStatus(StateBase):
+    scanned: bool = False
+    qr_key: Optional[str] = None
+    qr_url: Optional[str] = None
+    expired: bool = False
+    is_new: bool = False
+    cred_loaded: bool = False
+    timeout: bool = False
+    wait_for_confirm: bool = False
+    area_updated: bool = False
+    room_updated: bool = False
+    const_updated: bool = False
+    announce_updated: bool = False
+
+
+@dataclass
+class StreamStatus(StateBase):
+    live_status: bool = False
+    required_face: bool = False
+    identified_face: bool = False
+    face_url: Optional[str] = None
+    stream_addr: Optional[str] = None
+    stream_key: Optional[str] = None
+
+
+@dataclass
+class RoomInfo(StateBase):
+    cover_audit_reason: str = ""
+    cover_url: str = ""
+    cover_status: CoverStatus = CoverStatus.AUDIT_IN_PROGRESS
+    cover_data: Any = None
+    room_id: str = ""
+    title: str = ""
+    parent_area: str = ""
+    area: str = ""
+    announcement: str = ""
+    recent_areas: List[str] = field(default_factory=list)
+    recent_title: List[str] = field(default_factory=list)
+
 
 # Queue to communicate with OBS in a separate thread
 obs_req_queue = Queue()
 
 # Scan status flags for login
-scan_status = ThreadSafeDict.new(_SCAN_STATUS_INITIALIZATION)
+scan_status = ScanStatus()
 
 # Stream status stores fetched RTMP info and verification state
-stream_status = ThreadSafeDict.new(_STREAM_STATUS_INITIALIZATION)
+stream_status = StreamStatus()
 
-app_settings = ThreadSafeDict.new(_APP_CONFIG_INITIALIZATION)
+app_settings = AppSettings()
 
 if (app := get_password(KEYRING_SERVICE_NAME,
                         KEYRING_APP_SETTINGS)) is not None:
     app_settings.update(loads(app))
 
 # Managed by models.workers.credential_manager
-room_info = ThreadSafeDict({})
-obs_settings = ThreadSafeDict({})
-usernames = ThreadSafeDict({})
+room_info = RoomInfo()
+obs_settings = ObsSettings()
+usernames = {}
 # A cache of cookie indices
 cookie_indices = []
 
@@ -125,24 +143,24 @@ def create_session() -> Session:
     return session
 
 
-def app_settings_default():
-    app_settings.update(_APP_CONFIG_INITIALIZATION)
+def app_settings_default() -> None:
+    app_settings.reset()
 
 
-def scan_settings_default():
-    scan_status.update(_SCAN_STATUS_INITIALIZATION)
+def scan_settings_default() -> None:
+    scan_status.reset()
     scan_status["const_updated"] = True
 
 
-def room_info_default():
-    _ROOM_INFO_INITIALIZATION["recent_areas"].clear()
-    _ROOM_INFO_INITIALIZATION["recent_title"].clear()
-    room_info.update(_ROOM_INFO_INITIALIZATION)
+def room_info_default() -> None:
+    room_info.recent_areas.clear()
+    room_info.recent_title.clear()
+    room_info.reset()
 
 
-def stream_status_default():
-    stream_status.update(_STREAM_STATUS_INITIALIZATION)
+def stream_status_default() -> None:
+    stream_status.reset()
 
 
-def obs_settings_default():
-    obs_settings.update(_OBS_SETTINGS_INITIALIZATION)
+def obs_settings_default() -> None:
+    obs_settings.reset()

app_state/app_state_base.py

@@ -0,0 +1,103 @@
+from dataclasses import dataclass, field, fields, MISSING
+from typing import Any, Mapping, Iterator, Tuple
+from copy import deepcopy
+
+from PySide6.QtCore import QMutex, QMutexLocker
+
+
+@dataclass
+class StateBase:
+    _lock: QMutex = field(default_factory=QMutex, init=False, repr=False)
+
+    # obj["field"]
+    def __getitem__(self, key: str) -> Any:
+        with QMutexLocker(self._lock):
+            if not hasattr(self, key):
+                raise KeyError(key)
+            return getattr(self, key)
+
+    # obj["field"] = value
+    def __setitem__(self, key: str, value: Any) -> None:
+        with QMutexLocker(self._lock):
+            if not hasattr(self, key):
+                raise KeyError(key)
+            setattr(self, key, value)
+
+    # obj.get("field", default)
+    def get(self, key: str, default: Any = None) -> Any:
+        with QMutexLocker(self._lock):
+            if not hasattr(self, key):
+                return default
+            return getattr(self, key)
+
+    # obj.update({...})
+    def update(self, mapping: Mapping[str, Any] | None = None,
+               **kwargs: Any) -> None:
+        with QMutexLocker(self._lock):
+            if mapping:
+                for k, v in mapping.items():
+                    if hasattr(self, k):
+                        setattr(self, k, v)
+            for k, v in kwargs.items():
+                if hasattr(self, k):
+                    setattr(self, k, v)
+
+    def as_dict(self) -> dict[str, Any]:
+        with QMutexLocker(self._lock):
+            return {
+                f.name: getattr(self, f.name)
+                for f in fields(self)
+                if not f.name.startswith("_")
+            }
+
+    @classmethod
+    def default_dict(cls) -> dict[str, Any]:
+        result: dict[str, Any] = {}
+        for f in fields(cls):
+            # 跳过内部字段（比如 _lock）
+            if f.name.startswith("_"):
+                continue
+
+            if f.default_factory is not MISSING:
+                value = f.default_factory()
+            elif f.default is not MISSING:
+                # 防止用到可变对象，做一层 deepcopy 稳妥
+                value = deepcopy(f.default)
+            else:
+                value = None
+
+            result[f.name] = value
+        return result
+
+    def reset(self) -> None:
+        defaults = type(self).default_dict()
+        with QMutexLocker(self._lock):
+            for name, value in defaults.items():
+                setattr(self, name, value)
+
+    @property
+    def internal(self) -> dict[str, Any]:
+        return self.as_dict()
+
+    def items(self) -> Iterator[Tuple[str, Any]]:
+        return iter(self.as_dict().items())
+
+    def values(self) -> Iterator[Any]:
+        return iter(self.as_dict().values())
+
+    def keys(self) -> Iterator[str]:
+        return iter(self.as_dict().keys())
+
+    def __iter__(self) -> Iterator[str]:
+        return self.keys()
+
+    def __contains__(self, key: str) -> bool:
+        with QMutexLocker(self._lock):
+            return hasattr(self, key)
+
+    def __len__(self) -> int:
+        with QMutexLocker(self._lock):
+            return len([f for f in fields(self) if not f.name.startswith("_")])
+
+    def __bool__(self) -> bool:
+        return bool(len(self))

models/classes/__init__.py

@@ -3,4 +3,3 @@
 from .focus_placeholder_line_edit import FocusPlaceholderLineEdit
 from .completion_combo import CompletionComboBox
 from .single_instance_window import SingleInstanceWindow
-from .thread_safe_dict import ThreadSafeDict