168 lines
5.2 KiB
Python
168 lines
5.2 KiB
Python
"""Persistent launch history for the app launcher.
|
|
|
|
Remembers how often and how recently each application was launched so
|
|
``AppProvider`` can rank frequently used apps first (frecency, the
|
|
same frequency + recency idea as zoxide). The file lives in the system
|
|
cache directory - losing it only resets the ranking, never breaks the
|
|
launcher.
|
|
|
|
Layout (JSON, written atomically)::
|
|
|
|
{"version": 1, "entries": {"steam.desktop": {"count": 12, "last_used": 1712345678.0}}}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from loguru import logger
|
|
from platformdirs import user_cache_dir
|
|
|
|
from .ranking import frecency, popularity_bonus
|
|
|
|
CACHE_VERSION = 1
|
|
MAX_ENTRIES = 500
|
|
MAX_AGE_DAYS = 365.0
|
|
DEFAULT_HALF_LIFE_DAYS = 30.0
|
|
|
|
|
|
def default_cache_path() -> Path:
|
|
return Path(user_cache_dir("sims")) / "launcher-history.json"
|
|
|
|
|
|
@dataclass
|
|
class _Entry:
|
|
count: int = 0
|
|
last_used: float = 0.0
|
|
|
|
|
|
class LaunchHistory:
|
|
"""Launch counts/recency, keyed by launcher-specific item id."""
|
|
|
|
def __init__(
|
|
self,
|
|
path: Path | str | None = None,
|
|
*,
|
|
half_life_days: float = DEFAULT_HALF_LIFE_DAYS,
|
|
):
|
|
self.path = Path(path) if path is not None else default_cache_path()
|
|
self._half_life_days = half_life_days
|
|
self._entries: dict[str, _Entry] = self._load()
|
|
|
|
def record(self, key: str, *, now: float | None = None) -> None:
|
|
"""Count one launch of *key* and persist the history."""
|
|
if not key:
|
|
return
|
|
now = time.time() if now is None else now
|
|
entry = self._entries.get(key)
|
|
if entry is None:
|
|
entry = self._entries[key] = _Entry()
|
|
entry.count += 1
|
|
entry.last_used = now
|
|
self._save()
|
|
|
|
def bonus(
|
|
self,
|
|
key: str,
|
|
*,
|
|
now: float | None = None,
|
|
max_bonus: float = 150.0,
|
|
) -> float:
|
|
"""Bounded popularity bonus for *key*, 0.0 when unknown."""
|
|
entry = self._entries.get(key)
|
|
if entry is None or entry.count <= 0:
|
|
return 0.0
|
|
now = time.time() if now is None else now
|
|
score = frecency(
|
|
entry.count,
|
|
entry.last_used,
|
|
now,
|
|
half_life_days=self._half_life_days,
|
|
)
|
|
return popularity_bonus(score, max_bonus=max_bonus)
|
|
|
|
# -- persistence ------------------------------------------------------
|
|
|
|
def _load(self) -> dict[str, _Entry]:
|
|
try:
|
|
with open(self.path, "r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except FileNotFoundError:
|
|
return {}
|
|
except (OSError, ValueError) as exc:
|
|
logger.warning("launcher history {} unreadable: {}", self.path, exc)
|
|
return {}
|
|
|
|
raw_entries = data.get("entries") if isinstance(data, dict) else None
|
|
if not isinstance(raw_entries, dict):
|
|
logger.warning("launcher history {}: unexpected format", self.path)
|
|
return {}
|
|
|
|
entries: dict[str, _Entry] = {}
|
|
for key, raw in raw_entries.items():
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
try:
|
|
count = int(raw.get("count", 0))
|
|
last_used = float(raw.get("last_used", 0.0))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if count > 0:
|
|
entries[key] = _Entry(count=count, last_used=last_used)
|
|
|
|
return self._prune(entries, time.time())
|
|
|
|
def _prune(self, entries: dict[str, _Entry], now: float) -> dict[str, _Entry]:
|
|
cutoff = now - MAX_AGE_DAYS * 86_400.0
|
|
kept = {
|
|
key: entry
|
|
for key, entry in entries.items()
|
|
if entry.count > 0 and entry.last_used >= cutoff
|
|
}
|
|
if len(kept) > MAX_ENTRIES:
|
|
top = sorted(
|
|
kept.items(),
|
|
key=lambda item: frecency(
|
|
item[1].count,
|
|
item[1].last_used,
|
|
now,
|
|
half_life_days=self._half_life_days,
|
|
),
|
|
reverse=True,
|
|
)[:MAX_ENTRIES]
|
|
kept = dict(top)
|
|
return kept
|
|
|
|
def _save(self) -> None:
|
|
self._entries = self._prune(self._entries, time.time())
|
|
payload = {
|
|
"version": CACHE_VERSION,
|
|
"entries": {
|
|
key: {"count": entry.count, "last_used": entry.last_used}
|
|
for key, entry in sorted(self._entries.items())
|
|
},
|
|
}
|
|
temp_path: str | None = None
|
|
try:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temp_path = tempfile.mkstemp(
|
|
dir=self.path.parent, prefix=self.path.name, suffix=".tmp"
|
|
)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle)
|
|
os.replace(temp_path, self.path)
|
|
temp_path = None
|
|
except OSError as exc:
|
|
logger.warning("could not write launcher history {}: {}", self.path, exc)
|
|
finally:
|
|
if temp_path is not None:
|
|
try:
|
|
os.unlink(temp_path)
|
|
except OSError:
|
|
pass
|