119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""GUI本地状态存储。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from loguru import logger
|
|
|
|
from douyu.config import Account, ProxyConfig
|
|
from douyu.email_verifier import get_email_config_for_account
|
|
|
|
|
|
DEFAULT_STATE_PATH = Path("data/gui_state.json")
|
|
|
|
|
|
def account_to_dict(account: Account) -> dict[str, Any]:
|
|
"""把账号对象转换成可保存的字典。"""
|
|
return {
|
|
"username": account.username,
|
|
"password": account.password,
|
|
"email": account.email,
|
|
"email_password": account.email_password,
|
|
"email_imap_server": account.email_imap_server,
|
|
"email_imap_port": account.email_imap_port,
|
|
}
|
|
|
|
|
|
def account_from_dict(data: dict[str, Any]) -> Account | None:
|
|
"""从本地状态恢复账号对象。"""
|
|
username = str(data.get("username", "")).strip()
|
|
password = str(data.get("password", "")).strip()
|
|
email = str(data.get("email", "")).strip()
|
|
email_password = str(data.get("email_password", "")).strip()
|
|
|
|
if not all([username, password, email, email_password]):
|
|
return None
|
|
|
|
email_config = get_email_config_for_account(email)
|
|
imap_server = str(data.get("email_imap_server") or email_config["server"]).strip()
|
|
try:
|
|
imap_port = int(data.get("email_imap_port") or email_config["port"])
|
|
except (TypeError, ValueError):
|
|
imap_port = int(email_config["port"])
|
|
|
|
return Account(
|
|
username=username,
|
|
password=password,
|
|
email=email,
|
|
email_password=email_password,
|
|
email_imap_server=imap_server,
|
|
email_imap_port=imap_port,
|
|
)
|
|
|
|
|
|
class GuiStateStore:
|
|
"""负责保存和读取GUI状态。"""
|
|
|
|
def __init__(self, path: str | Path = DEFAULT_STATE_PATH):
|
|
self.path = Path(path)
|
|
|
|
def load(self) -> dict[str, Any]:
|
|
"""读取本地状态文件。"""
|
|
if not self.path.exists():
|
|
return {}
|
|
|
|
try:
|
|
with self.path.open("r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return data if isinstance(data, dict) else {}
|
|
except Exception as exc:
|
|
logger.warning(f"读取GUI状态失败，将使用默认值: {exc}")
|
|
return {}
|
|
|
|
def save(self, state: dict[str, Any]) -> None:
|
|
"""原子写入本地状态文件。"""
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = self.path.with_suffix(f"{self.path.suffix}.tmp")
|
|
|
|
with tmp_path.open("w", encoding="utf-8") as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
tmp_path.replace(self.path)
|
|
|
|
@staticmethod
|
|
def accounts_from_state(state: dict[str, Any]) -> list[Account]:
|
|
"""从状态字典恢复账号列表。"""
|
|
accounts = []
|
|
raw_accounts = state.get("accounts", [])
|
|
if not isinstance(raw_accounts, list):
|
|
return accounts
|
|
|
|
for raw_account in raw_accounts:
|
|
if not isinstance(raw_account, dict):
|
|
continue
|
|
account = account_from_dict(raw_account)
|
|
if account:
|
|
accounts.append(account)
|
|
|
|
return accounts
|
|
|
|
@staticmethod
|
|
def proxy_from_state(state: dict[str, Any]) -> ProxyConfig:
|
|
"""从状态字典恢复代理配置。"""
|
|
proxy = state.get("proxy", {})
|
|
if not isinstance(proxy, dict):
|
|
return ProxyConfig()
|
|
|
|
return ProxyConfig(
|
|
enabled=bool(proxy.get("enabled", False)),
|
|
api_url=str(proxy.get("api_url", "")).strip(),
|
|
http=str(proxy.get("http", "")).strip(),
|
|
https=str(proxy.get("https", "")).strip(),
|
|
whitelist_enabled=bool(proxy.get("whitelist_enabled", False)),
|
|
whitelist_uid=str(proxy.get("whitelist_uid", "")).strip(),
|
|
whitelist_ukey=str(proxy.get("whitelist_ukey", "")).strip(),
|
|
)
|