|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from devplacepy.config import (
|
|
PRESENCE_ONLINE_LIMIT,
|
|
PRESENCE_ONLINE_MARGIN_SECONDS,
|
|
PRESENCE_TIMEOUT_SECONDS,
|
|
PRESENCE_TRACK_LIMIT,
|
|
PRESENCE_WRITE_SECONDS,
|
|
)
|
|
from devplacepy.database import get_online_users, set_last_seen
|
|
|
|
_last_write: dict[str, float] = {}
|
|
|
|
|
|
def recording_allowed(user_uid: str) -> bool:
|
|
from devplacepy.database import consent_granted
|
|
|
|
return bool(user_uid) and consent_granted("user", user_uid, "activity_recording")
|
|
|
|
|
|
def touch(user_uid: str) -> None:
|
|
if not user_uid:
|
|
return
|
|
now = time.monotonic()
|
|
if now - _last_write.get(user_uid, 0.0) < PRESENCE_WRITE_SECONDS:
|
|
return
|
|
_last_write[user_uid] = now
|
|
if not recording_allowed(user_uid):
|
|
return
|
|
set_last_seen(user_uid, datetime.now(timezone.utc).isoformat())
|
|
|
|
|
|
def seconds_since(last_seen: Optional[str]) -> Optional[float]:
|
|
if not last_seen:
|
|
return None
|
|
try:
|
|
seen = datetime.fromisoformat(last_seen)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
if seen.tzinfo is None:
|
|
seen = seen.replace(tzinfo=timezone.utc)
|
|
return (datetime.now(timezone.utc) - seen).total_seconds()
|
|
|
|
|
|
def stays_online(elapsed: Optional[float], was_online: bool) -> bool:
|
|
if elapsed is None:
|
|
return False
|
|
grace = PRESENCE_TIMEOUT_SECONDS + PRESENCE_ONLINE_MARGIN_SECONDS
|
|
return elapsed < (grace if was_online else PRESENCE_TIMEOUT_SECONDS)
|
|
|
|
|
|
def is_online(user: Optional[dict]) -> bool:
|
|
if not user:
|
|
return False
|
|
return stays_online(seconds_since(user.get("last_seen")), was_online=False)
|
|
|
|
|
|
def _cutoff_iso(seconds: int) -> str:
|
|
return (datetime.now(timezone.utc) - timedelta(seconds=seconds)).isoformat()
|
|
|
|
|
|
def online_cutoff_iso() -> str:
|
|
return _cutoff_iso(PRESENCE_TIMEOUT_SECONDS)
|
|
|
|
|
|
def sort_by_username(rows: list) -> list:
|
|
return sorted(rows, key=lambda row: (row.get("username") or "").lower())
|
|
|
|
|
|
def online_users(limit: int = PRESENCE_ONLINE_LIMIT) -> list:
|
|
return sort_by_username(get_online_users(online_cutoff_iso(), limit))
|
|
|
|
|
|
def online_candidates(limit: int = PRESENCE_TRACK_LIMIT) -> list:
|
|
return sort_by_username(
|
|
get_online_users(
|
|
_cutoff_iso(PRESENCE_TIMEOUT_SECONDS + PRESENCE_ONLINE_MARGIN_SECONDS), limit
|
|
)
|
|
)
|