Coverage for src/web_signals.py: 100%
217 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-08 21:05 +0000
1"""Cross-thread signalling for the embedded web UI.
3The web UI (``src.web``) runs in a daemon thread alongside the sync
4loop (``src.sync``). Two things need to flow between them:
61. **Force-sync sentinels** — when the user taps "Sync now" on the
7 dashboard, the web thread touches ``$CONFIG_DIR/.force-sync-<svc>``
8 and the sync loop deletes the sentinel + zeroes the countdown on
9 its next iteration.
112. **Last-sync state** — after each per-service sync run, the sync
12 loop writes a small JSON file the dashboard reads on every
13 refresh (last completion time, file counts, error count).
15Files are chosen over a shared module-level singleton so the same
16mechanism keeps working if a future refactor splits sync + web into
17two processes. They live in ``ICLOUD_DOCKER_CONFIG_DIR`` (default
18``/config``) — same place the keyring and session cookies live, so
19they're persisted across container recreations.
20"""
22__author__ = "Mandar Patil (mandarons@pm.me)"
24import json
25import os
26import time
27from typing import Any
29from src import get_logger
31LOGGER = get_logger()
34def _config_dir() -> str:
35 """Resolve the directory force-sync sentinels + state JSON live in.
37 Mirrors the ICLOUD_DOCKER_CONFIG_DIR / DEFAULT_COOKIE_DIRECTORY
38 setup: same logic the keyring redirect uses, so dev hosts without
39 ``/config`` still work via a tempdir. Reads
40 ``DEFAULT_COOKIE_DIRECTORY`` via ``sys.modules`` so the test
41 fixture's monkeypatch is honoured -- a ``from src import
42 DEFAULT_COOKIE_DIRECTORY`` at module top would bind the value once
43 at import time and miss the redirect.
44 """
45 import sys
47 # DEFAULT_COOKIE_DIRECTORY is "<config_dir>/session_data"; strip the
48 # trailing component to recover the config dir.
49 cookie_dir = sys.modules["src"].DEFAULT_COOKIE_DIRECTORY
50 return os.path.dirname(cookie_dir) or "/config"
53_VALID_SERVICES = ("drive", "photos")
56def _sentinel_path(service: str) -> str:
57 return os.path.join(_config_dir(), f".force-sync-{service}")
60def _state_path() -> str:
61 return os.path.join(_config_dir(), ".last-sync-state.json")
64def _reauth_sentinel_path() -> str:
65 return os.path.join(_config_dir(), ".reauth-completed")
68def record_reauth_completed() -> bool:
69 """Signal that a re-auth just succeeded in the web UI.
71 Deliberately not the force-sync sentinel. That one is a button meaning
72 "sync everything now"; borrowing it would queue a full photo
73 re-enumeration -- hours on a large library -- that nobody asked for.
74 This one means only "the wait you are serving is over".
75 """
76 path = _reauth_sentinel_path()
77 try:
78 os.makedirs(os.path.dirname(path), exist_ok=True)
79 with open(path, "w") as f:
80 f.write(str(time.time()))
81 return True
82 except OSError as e:
83 LOGGER.warning(f"web_signals: failed to write {path}: {e!s}")
84 return False
87def consume_reauth_completed() -> bool:
88 """Atomically check + clear the re-auth signal."""
89 path = _reauth_sentinel_path()
90 try:
91 os.unlink(path)
92 return True
93 except FileNotFoundError:
94 return False
95 except OSError as e:
96 LOGGER.warning(f"web_signals: failed to clear {path}: {e!s}")
97 return False
100def request_force_sync(service: str) -> bool:
101 """Touch the sentinel for ``service``.
103 Returns True on success, False on validation/IO failure. Idempotent —
104 re-tapping while a previous request is still queued is a no-op.
105 """
106 if service not in _VALID_SERVICES:
107 return False
108 path = _sentinel_path(service)
109 try:
110 os.makedirs(os.path.dirname(path), exist_ok=True)
111 with open(path, "w") as f:
112 f.write(str(time.time()))
113 return True
114 except OSError as e:
115 LOGGER.warning(f"web_signals: failed to write {path}: {e!s}")
116 return False
119def pending_force_syncs() -> list[str]:
120 """Return services currently queued for an immediate sync.
122 Used by the dashboard to render "Queued ✓" instead of "Sync now".
123 """
124 return [
125 service
126 for service in _VALID_SERVICES
127 if os.path.isfile(_sentinel_path(service))
128 ]
131def consume_force_sync(service: str) -> bool:
132 """Atomically check + delete the sentinel. Sync loop calls this on
133 each iteration; True means "user requested an immediate run."
135 ``os.unlink`` raises ``FileNotFoundError`` if another caller beat us
136 to it — treated as "no request" rather than an error.
137 """
138 if service not in _VALID_SERVICES:
139 return False
140 try:
141 os.unlink(_sentinel_path(service))
142 return True
143 except FileNotFoundError:
144 return False
145 except OSError as e:
146 LOGGER.warning(f"web_signals: failed to consume {service} sentinel: {e!s}")
147 return False
150def record_sync_completion(
151 service: str,
152 *,
153 files_downloaded: int | None = None,
154 files_skipped: int | None = None,
155 files_removed: int | None = None,
156 errors: int | None = None,
157 duration_seconds: float | None = None,
158) -> None:
159 """Persist per-service stats after a sync run completes.
161 All counters are optional — passing ``None`` leaves the previous
162 value alone. Writes atomically (temp + rename) so a partial write
163 can't corrupt the file.
164 """
165 if service not in _VALID_SERVICES:
166 return
167 state = _load_state()
168 entry = state.get(service, {})
169 entry["completed_at"] = time.time()
170 if files_downloaded is not None:
171 entry["files_downloaded"] = int(files_downloaded)
172 if files_skipped is not None:
173 entry["files_skipped"] = int(files_skipped)
174 if files_removed is not None:
175 entry["files_removed"] = int(files_removed)
176 if errors is not None:
177 entry["errors"] = int(errors)
178 if duration_seconds is not None:
179 entry["duration_seconds"] = float(duration_seconds)
180 state[service] = entry
181 _save_state(state)
184def get_sync_state(service: str) -> dict[str, Any]:
185 """Return the persisted last-sync state for ``service``.
187 Empty dict on missing/corrupt file — the dashboard renders absence
188 gracefully.
189 """
190 if service not in _VALID_SERVICES:
191 return {}
192 return _load_state().get(service, {})
195def _load_state() -> dict[str, dict[str, Any]]:
196 path = _state_path()
197 if not os.path.isfile(path):
198 return {}
199 try:
200 with open(path) as f:
201 data = json.load(f)
202 if isinstance(data, dict):
203 return data
204 except (OSError, json.JSONDecodeError) as e:
205 LOGGER.warning(f"web_signals: failed to load {path}: {e!s} — treating as empty")
206 return {}
209def _save_state(state: dict[str, dict[str, Any]]) -> None:
210 path = _state_path()
211 tmp = path + ".tmp"
212 try:
213 os.makedirs(os.path.dirname(path), exist_ok=True)
214 with open(tmp, "w") as f:
215 json.dump(state, f, indent=2)
216 os.rename(tmp, path)
217 except OSError as e:
218 LOGGER.warning(f"web_signals: failed to save {path}: {e!s}")
219 try:
220 os.unlink(tmp)
221 except OSError:
222 pass
225_TRUST_STATE_KEY = "_trust"
226_AUTH_BLOCKED_STATE_KEY = "_auth_blocked"
227_AUTH_METHOD_STATE_KEY = "_auth_method"
230def record_trust_state(
231 *,
232 expires_at_iso: str | None,
233 warned_for_expires_at: str | None = None,
234) -> None:
235 """Persist Apple trust-cookie expiry + whether we've warned for it.
237 Stored under a reserved ``_trust`` key in the same JSON file as
238 per-service sync state. ``warned_for_expires_at`` carries the iso
239 timestamp the most recent threshold-cross warning was fired for, so
240 a cookie refresh (new expires_at) automatically rearms warning
241 eligibility -- compare ``warned_for_expires_at`` against the live
242 ``expires_at_iso`` to decide whether to fire again.
243 """
244 state = _load_state()
245 entry = state.get(_TRUST_STATE_KEY, {})
246 entry["expires_at"] = expires_at_iso
247 entry["last_updated"] = time.time()
248 if warned_for_expires_at is not None:
249 entry["warned_for_expires_at"] = warned_for_expires_at
250 state[_TRUST_STATE_KEY] = entry
251 _save_state(state)
254def get_trust_state() -> dict[str, Any]:
255 """Return persisted trust state. Empty dict if never recorded."""
256 return _load_state().get(_TRUST_STATE_KEY, {})
259def record_auth_blocked(*, blocked: bool, reason: str | None = None) -> None:
260 """Record whether the sync loop is currently unable to authenticate.
262 ``_detect_auth_state`` can only see on-disk signals -- a username in
263 the config and a password in the keyring -- so it reports "ready" for
264 an account that is in fact stuck on a second factor. Only the sync
265 loop knows it is failing, so it publishes that here and the dashboard
266 stops claiming everything is healthy while nothing is syncing.
267 """
268 state = _load_state()
269 state[_AUTH_BLOCKED_STATE_KEY] = {
270 "blocked": bool(blocked),
271 "reason": reason,
272 "last_updated": time.time(),
273 }
274 _save_state(state)
277def get_auth_blocked() -> dict[str, Any]:
278 """Return the recorded auth-blocked state. Empty dict if never recorded."""
279 entry = _load_state().get(_AUTH_BLOCKED_STATE_KEY)
280 return entry if isinstance(entry, dict) else {}
283def record_auth_method(*, username: str, method: str) -> None:
284 """Remember which second factor Apple demands for this Apple ID.
286 Keyed by username because the factor is an account property, not a
287 container property -- an operator who repoints the container at a
288 different Apple ID must not inherit the previous account's mode.
290 ``method`` is "security_key" once Apple has answered with an
291 ``fsaChallenge`` (hardware key or passkey enrolled, which disables
292 6-digit codes entirely), otherwise "code".
293 """
294 state = _load_state()
295 entry = state.get(_AUTH_METHOD_STATE_KEY, {})
296 entry[username] = {"method": method, "last_seen": time.time()}
297 state[_AUTH_METHOD_STATE_KEY] = entry
298 _save_state(state)
301def get_auth_method(username: str) -> str | None:
302 """Return the recorded second factor for ``username``, or None if unknown."""
303 entry = _load_state().get(_AUTH_METHOD_STATE_KEY, {}).get(username)
304 if not isinstance(entry, dict):
305 return None
306 method = entry.get("method")
307 return method if isinstance(method, str) else None
310def format_relative_time(epoch_seconds: float, *, now: float | None = None) -> str:
311 """Human-friendly relative time for dashboard display.
313 "Just now" / "5 min ago" / "2 h ago" / "3 d ago". Avoids
314 "1 hour ago" pluralisation gymnastics by sticking to compact
315 unit suffixes.
316 """
317 if not epoch_seconds:
318 return ""
319 if now is None:
320 now = time.time()
321 delta = max(0, now - epoch_seconds)
322 if delta < 30:
323 return "Just now"
324 if delta < 120:
325 return f"{int(delta)} sec ago"
326 if delta < 3600:
327 return f"{int(delta // 60)} min ago"
328 if delta < 86400:
329 return f"{int(delta // 3600)} h ago"
330 return f"{int(delta // 86400)} d ago"
333_LIBRARY_STATE_KEY = "_libraries"
336def record_library_started(library: str) -> None:
337 """Mark ``library`` as syncing right now.
339 A photo library can take an hour, and until it finishes there was
340 nothing anywhere saying which one was being worked on -- the
341 per-service record is only written once the whole pass completes, so
342 a library that never got its turn was indistinguishable from one
343 that had nothing to do.
344 """
345 state = _load_state()
346 libraries = state.get(_LIBRARY_STATE_KEY, {})
347 entry = libraries.get(library, {})
348 entry["state"] = "syncing"
349 entry["started_at"] = time.time()
350 entry.pop("error", None)
351 libraries[library] = entry
352 state[_LIBRARY_STATE_KEY] = libraries
353 _save_state(state)
356def record_library_finished(
357 library: str,
358 *,
359 ok: bool,
360 error: str | None = None,
361 downloaded: int | None = None,
362 skipped: int | None = None,
363) -> None:
364 """Record how ``library`` ended.
366 ``ok=False`` keeps the previous ``completed_at`` untouched, so the
367 dashboard can still show when the library last succeeded alongside
368 the fact that it is failing now.
369 """
370 state = _load_state()
371 libraries = state.get(_LIBRARY_STATE_KEY, {})
372 entry = libraries.get(library, {})
373 entry["state"] = "ok" if ok else "failed"
374 entry["finished_at"] = time.time()
375 if ok:
376 entry["completed_at"] = entry["finished_at"]
377 entry.pop("error", None)
378 else:
379 entry["error"] = error
380 if downloaded is not None:
381 entry["downloaded"] = downloaded
382 if skipped is not None:
383 entry["skipped"] = skipped
384 libraries[library] = entry
385 state[_LIBRARY_STATE_KEY] = libraries
386 _save_state(state)
389def get_library_states() -> dict[str, Any]:
390 """Return per-library sync state. Empty dict if never recorded."""
391 return _load_state().get(_LIBRARY_STATE_KEY, {})
394_PHOTOS_INDEXING_STATE_KEY = "_photos_indexing"
397def record_photos_indexing(*, waiting: bool) -> None:
398 """Record whether Photos is waiting on Apple to finish indexing.
400 Service-level on purpose: while any zone of the account is being
401 indexed, icloudpy cannot open Photos at all, so this belongs to the
402 service and not to any one library. ``since`` is kept from the first
403 cycle that saw it -- the wait can run for days, and how long it has
404 been going is the only part of it the operator can act on.
406 Clearing is a no-op when nothing was recorded, so the ordinary
407 success path does not rewrite the state file every cycle.
408 """
409 state = _load_state()
410 entry = state.get(_PHOTOS_INDEXING_STATE_KEY)
411 entry = entry if isinstance(entry, dict) else {}
412 if not waiting:
413 if not entry.get("waiting"):
414 return
415 entry = {"waiting": False}
416 elif not entry.get("waiting"):
417 entry = {"waiting": True, "since": time.time()}
418 else:
419 return
420 state[_PHOTOS_INDEXING_STATE_KEY] = entry
421 _save_state(state)
424def get_photos_indexing() -> dict[str, Any]:
425 """Return the recorded Photos indexing wait. Empty dict if never recorded."""
426 entry = _load_state().get(_PHOTOS_INDEXING_STATE_KEY)
427 return entry if isinstance(entry, dict) else {}
430def clear_stale_library_states() -> None:
431 """Demote any library left mid-sync with no process still on it.
433 The state file outlives the container, so a restart during a library
434 would otherwise leave it reading "syncing" forever. Called at
435 sync-loop startup, when nothing can legitimately be in flight, and
436 whenever a photos pass is abandoned part-way through a library.
437 """
438 state = _load_state()
439 libraries = state.get(_LIBRARY_STATE_KEY, {})
440 changed = False
441 for entry in libraries.values():
442 if entry.get("state") == "syncing":
443 entry["state"] = "interrupted"
444 changed = True
445 if changed:
446 state[_LIBRARY_STATE_KEY] = libraries
447 _save_state(state)