|
# retoor <retoor@molodetz.nl>
|
|
|
|
from .core import TTLCache, bump_cache_version, db, sync_local_cache
|
|
|
|
|
|
def get_users_by_uids(uids):
|
|
if not uids or "users" not in db.tables:
|
|
return {}
|
|
users = db["users"]
|
|
if "uid" not in users.columns:
|
|
return {}
|
|
seen = set()
|
|
unique = [u for u in uids if u not in seen and not seen.add(u)]
|
|
return {u["uid"]: u for u in users.find(users.table.columns.uid.in_(unique))}
|
|
|
|
|
|
_admins_cache = TTLCache(ttl=300, max_size=4)
|
|
# The primary administrator must be an account that can actually authenticate, so scan a
|
|
# few of the earliest admins and skip any that are soft-deleted or deactivated.
|
|
PRIMARY_ADMIN_CANDIDATES = 50
|
|
|
|
|
|
def invalidate_admins_cache() -> None:
|
|
_admins_cache.clear()
|
|
bump_cache_version("admins")
|
|
|
|
|
|
def get_admin_uids():
|
|
sync_local_cache("admins", _admins_cache)
|
|
cached = _admins_cache.get("uids")
|
|
if cached is not None:
|
|
return list(cached)
|
|
if "users" not in db.tables:
|
|
return []
|
|
rows = db.query("SELECT uid FROM users WHERE role = 'Admin'")
|
|
uids = [row["uid"] for row in rows]
|
|
_admins_cache.set("uids", uids)
|
|
return list(uids)
|
|
|
|
|
|
def set_user_timezone(user_uid: str, tz_name: str) -> None:
|
|
if "users" not in db.tables or not user_uid or not tz_name:
|
|
return
|
|
users = db["users"]
|
|
if not users.has_column("timezone"):
|
|
users.create_column_by_example("timezone", "")
|
|
current = users.find_one(uid=user_uid)
|
|
if current and current.get("timezone") == tz_name:
|
|
return
|
|
users.update({"uid": user_uid, "timezone": tz_name}, ["uid"])
|
|
|
|
|
|
def set_last_seen(user_uid: str, iso: str) -> None:
|
|
if "users" not in db.tables or not user_uid or not iso:
|
|
return
|
|
users = db["users"]
|
|
if not users.has_column("last_seen"):
|
|
users.create_column_by_example("last_seen", "")
|
|
users.update({"uid": user_uid, "last_seen": iso}, ["uid"])
|
|
|
|
|
|
def get_online_users(cutoff_iso: str, limit: int = 30) -> list:
|
|
if "users" not in db.tables:
|
|
return []
|
|
users = db["users"]
|
|
if "last_seen" not in users.columns:
|
|
return []
|
|
return list(
|
|
users.find(
|
|
last_seen={">=": cutoff_iso},
|
|
order_by=["username"],
|
|
_limit=limit,
|
|
)
|
|
)
|
|
|
|
|
|
def _can_hold_primary_admin(row, tracks_active):
|
|
if row.get("deleted_at"):
|
|
return False
|
|
if not tracks_active:
|
|
return True
|
|
is_active = row.get("is_active")
|
|
return is_active is None or bool(is_active)
|
|
|
|
|
|
def get_primary_admin_uid():
|
|
sync_local_cache("admins", _admins_cache)
|
|
cached = _admins_cache.get("primary")
|
|
if cached is not None:
|
|
return cached or None
|
|
if "users" not in db.tables:
|
|
return None
|
|
rows = list(
|
|
db.query(
|
|
"SELECT * FROM users WHERE role = 'Admin' "
|
|
"ORDER BY (created_at IS NULL OR created_at = ''), created_at ASC, id ASC "
|
|
"LIMIT :cap",
|
|
cap=PRIMARY_ADMIN_CANDIDATES,
|
|
)
|
|
)
|
|
tracks_active = "is_active" in db["users"].columns
|
|
primary = next(
|
|
(row["uid"] for row in rows if _can_hold_primary_admin(row, tracks_active)),
|
|
None,
|
|
)
|
|
_admins_cache.set("primary", primary or "")
|
|
return primary
|
|
|
|
|
|
def search_users_by_username(q, *, exclude_uid=None, limit=10):
|
|
if not q or "users" not in db.tables:
|
|
return []
|
|
if exclude_uid is not None:
|
|
rows = db.query(
|
|
"SELECT uid, username FROM users WHERE username LIKE :q AND uid != :me LIMIT :limit",
|
|
q=f"%{q}%",
|
|
me=exclude_uid,
|
|
limit=limit,
|
|
)
|
|
else:
|
|
rows = db.query(
|
|
"SELECT uid, username FROM users WHERE username LIKE :q LIMIT :limit",
|
|
q=f"%{q}%",
|
|
limit=limit,
|
|
)
|
|
return [{"uid": r["uid"], "username": r["username"]} for r in rows]
|