diff --git a/README.md b/README.md index 0a81fa9..9b2dc09 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ Nothing is ever deleted from the archive. ## MCP -Read-only: `list_watched_bots`, `get_bot_stats`, `get_bot_updates`, `search_messages_tool`, `list_chats`, `get_chat_history` (with deleted messages included), `get_deleted_messages`, `get_message_versions`, `get_media`, `get_callbacks`, `presence_history`, `get_peer_history`, `get_stories`, `get_annotations`, `list_watches`, `list_alerts`. The one write is `set_watch`. There are no send, edit or delete tools and there will not be. +Read-only: `list_watched_bots`, `get_bot_stats`, `get_bot_updates`, `search_messages_tool`, `list_chats`, `get_chat_history` (with deleted messages included), `get_deleted_messages`, `get_message_versions`, `get_media`, `get_callbacks`, `presence_history`, `presence_stats`, `get_peer_history`, `get_stories`, `get_annotations`, `list_watches`, `list_alerts`. The one write is `set_watch`. There are no send, edit or delete tools and there will not be. ## Run diff --git a/backend/migrations/versions/d8f4a1c6e2b7_capture_uptime.py b/backend/migrations/versions/d8f4a1c6e2b7_capture_uptime.py new file mode 100644 index 0000000..fe3daaf --- /dev/null +++ b/backend/migrations/versions/d8f4a1c6e2b7_capture_uptime.py @@ -0,0 +1,50 @@ +"""capture uptime + +Revision ID: d8f4a1c6e2b7 +Revises: d7a3c5e9f2b1 +Create Date: 2026-10-19 03:00:00.000000 + +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "d8f4a1c6e2b7" +down_revision: str | None = "d7a3c5e9f2b1" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_UPTIME = """ +CREATE TABLE capture_uptime ( + account_id integer NOT NULL, + started_at timestamptz NOT NULL, + seen_at timestamptz NOT NULL, + PRIMARY KEY (account_id, started_at) +) +""" + +_BACKFILL = """ +INSERT INTO capture_uptime (account_id, started_at, seen_at) +SELECT account_id, min(ts), max(ts) +FROM ( + SELECT account_id, ts, + count(*) FILTER (WHERE gap > interval '30 minutes') + OVER (PARTITION BY account_id ORDER BY ts) AS stretch + FROM ( + SELECT account_id, ts, + ts - lag(ts) OVER (PARTITION BY account_id ORDER BY ts) AS gap + FROM presence + ) gaps +) stretches +GROUP BY account_id, stretch +""" + + +def upgrade() -> None: + op.execute(_UPTIME) + op.execute(_BACKFILL) + + +def downgrade() -> None: + op.execute("DROP TABLE IF EXISTS capture_uptime") diff --git a/backend/src/api/mcp/server.py b/backend/src/api/mcp/server.py index 0b5cb13..c321fc1 100644 --- a/backend/src/api/mcp/server.py +++ b/backend/src/api/mcp/server.py @@ -20,9 +20,19 @@ from api.mcp.download import ( from api.mcp.format import build_transcript, load_notes, load_photos, resolve_names from dependencies.container import container from utils.files import resolve_mime -from utils.read import annotations, chats, media, peers, presence, social, watches +from utils.read import ( + annotations, + chats, + media, + peers, + presence, + presence_stats, + social, + watches, +) from utils.read.accounts import self_user_id, unknown_account_hint from utils.read.models import DEFAULT_LIMIT, Page +from utils.read.period import DEFAULT_TZ, MAX_DAYS, known_timezone from utils.read.read_receipts import read_info from utils.search.models import SearchFilters from utils.search.repository import search_messages @@ -335,6 +345,37 @@ async def presence_history( ) +@mcp.tool(name="presence_stats") +async def presence_stats_tool( + account_id: int, + peer_id: int, + days: int = 30, + tz: str = DEFAULT_TZ, + limit: int = DEFAULT_LIMIT, + offset: int = 0, +) -> dict[str, Any]: + """Time a peer spent online, from status pushes turned into online sessions. + + `tz` is an IANA zone name; days and hours are counted in it. `week` holds + the average online minutes for each weekday (0 is Monday) and hour, null + where the archive has no data for that hour. `days` gives online seconds + per day, `gaps` the stretches when capture was down, `sessions` the latest + sessions (online stretches with breaks under 5 minutes), paged by + `limit`/`offset`. `hidden` means the peer hides their last seen time. + """ + pool = await _pool() + if not 1 <= days <= MAX_DAYS: + msg = f"days must be between 1 and {MAX_DAYS}" + raise ToolError(msg) + if not await known_timezone(pool, tz): + msg = f"Unknown time zone: {tz}" + raise ToolError(msg) + stats = await presence_stats.presence_stats( + pool, account_id, peer_id, Page(limit=limit, offset=offset), days=days, tz=tz + ) + return stats.model_dump(mode="json") + + @mcp.tool async def get_peer_history(account_id: int, peer_id: int) -> list[dict[str, Any]]: """Get name/username/avatar change history of a contact.""" diff --git a/backend/src/api/realtime.py b/backend/src/api/realtime.py index 5a1001a..293e1e6 100644 --- a/backend/src/api/realtime.py +++ b/backend/src/api/realtime.py @@ -59,16 +59,12 @@ class EventHub: return account_id = event.get("account_id") chat_id = event.get("chat_id") - scoped = event.get("kind") == "presence" + account_wide = event.get("kind") == "presence" targets = [ sub for sub in self._subscribers if sub.account_id == account_id - and ( - sub.chat_id == chat_id - if scoped - else sub.chat_id is None or sub.chat_id == chat_id - ) + and (account_wide or sub.chat_id is None or sub.chat_id == chat_id) ] if not targets: return diff --git a/backend/src/api/routers/analytics.py b/backend/src/api/routers/analytics.py index 7cdd57c..3da748f 100644 --- a/backend/src/api/routers/analytics.py +++ b/backend/src/api/routers/analytics.py @@ -4,8 +4,10 @@ import asyncpg from dishka.integrations.fastapi import DishkaRoute, FromDishka from fastapi import APIRouter, Query +from api.timezone import checked_tz from utils.read import analytics from utils.read.models import ResponseStats, VolumeBucket +from utils.read.period import DEFAULT_TZ, MAX_DAYS router = APIRouter(prefix="/api/analytics", tags=["analytics"], route_class=DishkaRoute) @@ -18,9 +20,12 @@ async def volume( pool: FromDishka[asyncpg.Pool], account_id: AccountId, chat_id: ChatId, - days: Annotated[int, Query()] = 90, + days: Annotated[int, Query(ge=1, le=MAX_DAYS)] = 90, + tz: Annotated[str, Query()] = DEFAULT_TZ, ) -> list[VolumeBucket]: - return await analytics.message_volume(pool, account_id, chat_id, days=days) + return await analytics.message_volume( + pool, account_id, chat_id, days=days, tz=await checked_tz(pool, tz) + ) @router.get("/response-time") diff --git a/backend/src/api/routers/presence.py b/backend/src/api/routers/presence.py index 67d3b8a..51b40bd 100644 --- a/backend/src/api/routers/presence.py +++ b/backend/src/api/routers/presence.py @@ -5,8 +5,17 @@ import asyncpg from dishka.integrations.fastapi import DishkaRoute, FromDishka from fastapi import APIRouter, Query -from utils.read import presence -from utils.read.models import DEFAULT_LIMIT, Page, PresenceHourly, PresenceSample +from api.timezone import checked_tz +from utils.read import presence, presence_stats +from utils.read.models import ( + DEFAULT_LIMIT, + Page, + PresenceHourly, + PresenceSample, + PresenceSession, + PresenceStats, +) +from utils.read.period import DEFAULT_TZ, MAX_DAYS router = APIRouter(prefix="/api/presence", tags=["presence"], route_class=DishkaRoute) @@ -14,6 +23,8 @@ AccountId = Annotated[int, Query()] PeerId = Annotated[int, Query()] DateFrom = Annotated[datetime | None, Query()] DateTo = Annotated[datetime | None, Query()] +Days = Annotated[int, Query(ge=1, le=MAX_DAYS)] +Tz = Annotated[str, Query()] @router.get("") @@ -54,3 +65,43 @@ async def presence_hourly( return await presence.presence_hourly( pool, account_id, peer_id, date_from=date_from, date_to=date_to ) + + +@router.get("/stats") +async def stats( + pool: FromDishka[asyncpg.Pool], + account_id: AccountId, + peer_id: PeerId, + days: Days = 30, + tz: Tz = DEFAULT_TZ, + limit: Annotated[int, Query()] = DEFAULT_LIMIT, + offset: Annotated[int, Query()] = 0, +) -> PresenceStats: + return await presence_stats.presence_stats( + pool, + account_id, + peer_id, + Page(limit=limit, offset=offset), + days=days, + tz=await checked_tz(pool, tz), + ) + + +@router.get("/sessions") +async def sessions( + pool: FromDishka[asyncpg.Pool], + account_id: AccountId, + peer_id: PeerId, + days: Days = 30, + tz: Tz = DEFAULT_TZ, + limit: Annotated[int, Query()] = DEFAULT_LIMIT, + offset: Annotated[int, Query()] = 0, +) -> list[PresenceSession]: + return await presence_stats.presence_sessions( + pool, + account_id, + peer_id, + Page(limit=limit, offset=offset), + days=days, + tz=await checked_tz(pool, tz), + ) diff --git a/backend/src/api/timezone.py b/backend/src/api/timezone.py new file mode 100644 index 0000000..1526663 --- /dev/null +++ b/backend/src/api/timezone.py @@ -0,0 +1,10 @@ +import asyncpg +from fastapi import HTTPException + +from utils.read.period import known_timezone + + +async def checked_tz(pool: asyncpg.Pool, tz: str) -> str: + if not await known_timezone(pool, tz): + raise HTTPException(status_code=422, detail=f"Unknown time zone: {tz}") + return tz diff --git a/backend/src/userbot/modules/uptime/__init__.py b/backend/src/userbot/modules/uptime/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/src/userbot/modules/uptime/repository.py b/backend/src/userbot/modules/uptime/repository.py new file mode 100644 index 0000000..649d41f --- /dev/null +++ b/backend/src/userbot/modules/uptime/repository.py @@ -0,0 +1,16 @@ +import asyncpg + +_HEARTBEAT = """ +WITH extended AS ( + UPDATE capture_uptime SET seen_at = now() + WHERE account_id = $1 AND seen_at >= now() - interval '150 seconds' + RETURNING account_id +) +INSERT INTO capture_uptime (account_id, started_at, seen_at) +SELECT $1, now(), now() +WHERE NOT EXISTS (SELECT 1 FROM extended) +""" + + +async def heartbeat(pool: asyncpg.Pool, account_id: int) -> None: + await pool.execute(_HEARTBEAT, account_id) diff --git a/backend/src/userbot/runner.py b/backend/src/userbot/runner.py index 549bc1c..118ec3c 100644 --- a/backend/src/userbot/runner.py +++ b/backend/src/userbot/runner.py @@ -14,6 +14,7 @@ from userbot.modules.capture import CaptureContext, build_capture_context from userbot.modules.catchup import SavedState, catch_up, read_saved_state from userbot.modules.client import PyroClient from userbot.modules.jobs import JobConsumer +from userbot.modules.uptime import repository as uptime from utils.env import env from utils.jobs import enqueue from utils.listen import Listener @@ -31,6 +32,7 @@ setup_logging() SESSION_FLUSH_INTERVAL = 60 SESSION_RETRY_INTERVAL = 60 +UPTIME_INTERVAL = 60 @dataclass @@ -171,6 +173,15 @@ async def _retry_sessions(registry: AccountRegistry) -> None: logger.exception("[red]Session sync failed[/]") +async def _keep_uptime(registry: AccountRegistry, pool: asyncpg.Pool) -> None: + while True: + for client in registry.clients: + if client.capture is not None and client.is_connected: + with contextlib.suppress(Exception): + await uptime.heartbeat(pool, client.capture.account_id) + await asyncio.sleep(UPTIME_INTERVAL) + + def _change_listener(registry: AccountRegistry, tasks: set[asyncio.Task]) -> Listener: def spawn(coro: Coroutine[Any, Any, None]) -> None: task = asyncio.create_task(coro) @@ -220,6 +231,7 @@ async def runner() -> None: asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop.set) flush = asyncio.create_task(_flush_sessions(registry)) retry = asyncio.create_task(_retry_sessions(registry)) + beat = asyncio.create_task(_keep_uptime(registry, pool)) try: await registry.sync() if not registry.clients: @@ -230,6 +242,7 @@ async def runner() -> None: finally: await _cancel(flush) await _cancel(retry) + await _cancel(beat) await listener.stop() await registry.close() await container.close() diff --git a/backend/src/utils/read/analytics.py b/backend/src/utils/read/analytics.py index b128e6f..9b35a7d 100644 --- a/backend/src/utils/read/analytics.py +++ b/backend/src/utils/read/analytics.py @@ -4,23 +4,31 @@ import asyncpg from utils.read.accounts import self_user_id from utils.read.models import ResponseStats, VolumeBucket +from utils.read.period import period_bounds + +_VOLUME = """ +WITH counts AS ( + SELECT (date AT TIME ZONE $5)::date AS day, count(*) AS total, + count(*) FILTER (WHERE sender_id = $3) AS outgoing + FROM messages + WHERE account_id = $1 AND chat_id = $2 AND date >= $4 + GROUP BY 1 +) +SELECT local_day::date AS bucket, coalesce(total, 0) AS total, + coalesce(outgoing, 0) AS outgoing +FROM generate_series(($4 AT TIME ZONE $5)::date, (now() AT TIME ZONE $5)::date, + interval '1 day') AS local_day +LEFT JOIN counts ON counts.day = local_day::date +ORDER BY 1 +""" async def message_volume( - pool: asyncpg.Pool, account_id: int, chat_id: int, *, days: int = 90 + pool: asyncpg.Pool, account_id: int, chat_id: int, *, days: int, tz: str ) -> list[VolumeBucket]: self_id = await self_user_id(pool, account_id) - rows = await pool.fetch( - "SELECT date_trunc('day', date) AS bucket, count(*) AS total, " - "count(*) FILTER (WHERE sender_id = $3) AS outgoing " - "FROM messages " - "WHERE account_id = $1 AND chat_id = $2 AND date >= $4 " - "GROUP BY bucket ORDER BY bucket", - account_id, - chat_id, - self_id, - datetime.now(UTC) - timedelta(days=days), - ) + since, _ = await period_bounds(pool, days, tz) + rows = await pool.fetch(_VOLUME, account_id, chat_id, self_id, since, tz) return [ VolumeBucket( bucket=row["bucket"], diff --git a/backend/src/utils/read/models.py b/backend/src/utils/read/models.py index b62cbcb..d1806a9 100644 --- a/backend/src/utils/read/models.py +++ b/backend/src/utils/read/models.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import date, datetime from typing import Any from pydantic import BaseModel @@ -362,8 +362,61 @@ class PresenceHourly(BaseModel): last_seen: datetime | None +class PresenceSession(BaseModel): + started: datetime + ended: datetime + seconds: float + online_seconds: float + ongoing: bool + + +class PresenceHour(BaseModel): + weekday: int + hour: int + hours: int + online_minutes: float | None + covered_hours: float + + +class PresenceDay(BaseModel): + day: date + online_seconds: float + covered_seconds: float + + +class CoverageGap(BaseModel): + started: datetime + ended: datetime + + +class PresenceSummary(BaseModel): + online_seconds: float + daily_seconds: float | None + covered_seconds: float + session_count: int + median_session_seconds: float | None + first_online: int | None + last_offline: int | None + last_seen: datetime | None + + +class PresenceStats(BaseModel): + peer_id: int + tz: str + since: datetime + until: datetime + current: PresenceSample | None + online: bool + hidden: bool + summary: PresenceSummary + week: list[PresenceHour] + days: list[PresenceDay] + gaps: list[CoverageGap] + sessions: list[PresenceSession] + + class VolumeBucket(BaseModel): - bucket: datetime + bucket: date total: int outgoing: int incoming: int diff --git a/backend/src/utils/read/period.py b/backend/src/utils/read/period.py new file mode 100644 index 0000000..ee3b91b --- /dev/null +++ b/backend/src/utils/read/period.py @@ -0,0 +1,27 @@ +from datetime import datetime + +import asyncpg + +DEFAULT_TZ = "UTC" +MAX_DAYS = 365 + +_timezones: set[str] = set() + + +async def known_timezone(pool: asyncpg.Pool, tz: str) -> bool: + if not _timezones: + rows = await pool.fetch("SELECT name FROM pg_timezone_names") + _timezones.update(row["name"] for row in rows) + return tz in _timezones + + +async def period_bounds( + pool: asyncpg.Pool, days: int, tz: str +) -> tuple[datetime, datetime]: + row = await pool.fetchrow( + "SELECT (date_trunc('day', now() AT TIME ZONE $1) " + "- make_interval(days => $2 - 1)) AT TIME ZONE $1 AS since, now() AS until", + tz, + days, + ) + return row["since"], row["until"] diff --git a/backend/src/utils/read/presence_stats.py b/backend/src/utils/read/presence_stats.py new file mode 100644 index 0000000..0e3bc07 --- /dev/null +++ b/backend/src/utils/read/presence_stats.py @@ -0,0 +1,289 @@ +import json + +import asyncpg + +from utils.read.models import ( + CoverageGap, + Page, + PresenceDay, + PresenceHour, + PresenceSession, + PresenceStats, + PresenceSummary, +) +from utils.read.period import period_bounds +from utils.read.presence import current_presence + +SECONDS_IN_HOUR = 3600 +SECONDS_IN_DAY = 86_400 +MIN_CELL_COVERAGE = 1800 +MIN_CELL_SHARE = 0.5 +STATS_WORK_MEM = "32MB" +FUZZY_STATUSES = {"recently", "last_week", "last_month", "long_time_ago"} + +_SESSIONS = """ +marks AS ( + SELECT ts, status, + least(next_offline_date, ts + interval '5 minutes', + lead(coalesce(last_online_date, ts)) OVER w, $4) AS ended, + lag(CASE WHEN status = 'online' + THEN least(next_offline_date, ts + interval '5 minutes') + ELSE coalesce(last_online_date, ts) END) OVER w AS previous + FROM presence + WHERE account_id = $1 AND peer_id = $2 + AND ts >= $3::timestamptz - interval '5 minutes' AND ts <= $4 + WINDOW w AS (ORDER BY ts) +), +online AS ( + SELECT greatest(ts, $3) AS started, ended, + count(*) FILTER (WHERE previous IS NULL + OR ts > previous + interval '5 minutes') + OVER (ORDER BY ts) AS visit + FROM marks + WHERE status = 'online' AND ended > greatest(ts, $3) +), +sessions AS ( + SELECT min(started) AS started, max(ended) AS ended, + extract(epoch FROM sum(ended - started))::float8 AS online_seconds + FROM online + GROUP BY visit +) +""" + +_SESSION_FIELDS = """ +started, ended, extract(epoch FROM ended - started)::float8 AS seconds, +online_seconds, ended >= $4 AS ongoing +""" + +_SESSION_PAGE = f""" +WITH {_SESSIONS} +SELECT {_SESSION_FIELDS} +FROM sessions +ORDER BY started DESC +LIMIT $5 OFFSET $6 +""" # noqa: S608 + +_COVERED_HOURS = """ + SELECT local_hour, + extract(epoch FROM sum( + least(ended, (local_hour + interval '1 hour') AT TIME ZONE $5) + - greatest(started, local_hour AT TIME ZONE $5)))::float8 AS seconds + FROM uptime, + generate_series(date_trunc('hour', started AT TIME ZONE $5), + date_trunc('hour', ended AT TIME ZONE $5), + interval '1 hour') AS local_hour + GROUP BY local_hour +""" + +_STATS = f""" +WITH {_SESSIONS}, +uptime AS ( + SELECT greatest(started_at, $3) AS started, + least(seen_at + interval '90 seconds', $4) AS ended + FROM capture_uptime + WHERE account_id = $1 AND started_at < $4 + AND seen_at + interval '90 seconds' > $3 +), +quarters AS ( + SELECT quarter, sum(ended - started - spill) AS inside, sum(spill) AS spill + FROM (SELECT started, ended, quarter, + greatest(ended - quarter - interval '15 minutes', + interval '0') AS spill + FROM (SELECT started, ended, + date_bin('15 minutes', started, timestamptz 'epoch') AS quarter + FROM online) binned) split + GROUP BY quarter +), +online_hours AS ( + SELECT date_trunc('hour', quarter AT TIME ZONE $5) AS local_hour, + extract(epoch FROM sum(share))::float8 AS seconds + FROM (SELECT quarter, inside AS share FROM quarters + UNION ALL + SELECT quarter + interval '15 minutes', spill FROM quarters) shares + GROUP BY 1 +), +covered_hours AS ({_COVERED_HOURS}), +hours AS ( + SELECT local_hour, + extract(epoch FROM + least((local_hour + interval '1 hour') AT TIME ZONE $5, $4) + - greatest(local_hour AT TIME ZONE $5, $3))::float8 AS span, + coalesce(online.seconds, 0) AS online, + least(3600, greatest(coalesce(covered.seconds, 0), + coalesce(online.seconds, 0))) AS covered + FROM generate_series(date_trunc('hour', $3 AT TIME ZONE $5), + date_trunc('hour', $4 AT TIME ZONE $5), + interval '1 hour') AS local_hour + LEFT JOIN online_hours AS online USING (local_hour) + LEFT JOIN covered_hours AS covered USING (local_hour) +), +week AS ( + SELECT extract(isodow FROM local_hour)::int - 1 AS weekday, + extract(hour FROM local_hour)::int AS hour, + count(*) AS hours, sum(span) AS span, + sum(online) AS online, sum(covered) AS covered + FROM hours + GROUP BY 1, 2 +), +days AS ( + SELECT local_hour::date AS day, sum(online) AS online, sum(covered) AS covered + FROM hours + GROUP BY 1 +), +uptime_reach AS ( + SELECT started, + coalesce(max(ended) OVER (ORDER BY started + ROWS BETWEEN UNBOUNDED PRECEDING + AND 1 PRECEDING), $3) AS reached + FROM uptime +), +gaps AS ( + SELECT reached AS started, started AS ended FROM uptime_reach + UNION ALL + SELECT coalesce(max(ended), $3), $4 FROM uptime +), +quiet AS ( + SELECT hour + FROM week + GROUP BY hour + ORDER BY sum(online) / nullif(sum(covered), 0) NULLS LAST, hour + LIMIT 1 +), +activity_days AS ( + SELECT (sessions.started AT TIME ZONE $5) - make_interval(hours => quiet.hour) + AS started, + (sessions.ended AT TIME ZONE $5) - make_interval(hours => quiet.hour) + AS ended, + (($4 AT TIME ZONE $5) - make_interval(hours => quiet.hour))::date AS today + FROM sessions, quiet +), +daily AS ( + SELECT extract(epoch FROM min(started - started::date))::float8 AS first_online, + extract(epoch FROM max(ended - started::date))::float8 AS last_offline + FROM activity_days + WHERE started::date < today + GROUP BY started::date +) +SELECT + (SELECT count(*) FROM sessions) AS session_count, + (SELECT coalesce(sum(online_seconds), 0) FROM sessions) AS online_seconds, + (SELECT sum(covered) FROM hours) AS covered_seconds, + (SELECT percentile_cont(0.5) WITHIN GROUP ( + ORDER BY extract(epoch FROM ended - started)) FROM sessions) AS median_session, + (SELECT max(ended) FROM sessions) AS last_seen, + (SELECT hour FROM quiet) AS day_start, + (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY first_online) + FROM daily) AS first_online, + (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY last_offline) + FROM daily) AS last_offline, + (SELECT json_agg(json_build_object( + 'weekday', weekday, 'hour', hour, 'hours', hours, 'span', span, + 'online', online, 'covered', covered)) + FROM week) AS week, + (SELECT json_agg(json_build_object( + 'day', day, 'online_seconds', online, 'covered_seconds', covered) + ORDER BY day) + FROM days) AS days, + (SELECT json_agg(json_build_object('started', started, 'ended', ended) + ORDER BY started) + FROM gaps WHERE ended - started > interval '5 minutes') AS gaps, + (SELECT json_agg(row_to_json(page) ORDER BY page.started DESC) + FROM (SELECT {_SESSION_FIELDS} FROM sessions + ORDER BY started DESC LIMIT $6 OFFSET $7) page) AS sessions +""" # noqa: S608 + + +def _time_of_day(shifted: float | None, day_start: int | None) -> int | None: + if shifted is None or day_start is None: + return None + return round(shifted + day_start * SECONDS_IN_HOUR) % SECONDS_IN_DAY + + +def _week(rows: list[dict]) -> list[PresenceHour]: + cells = {(row["weekday"], row["hour"]): row for row in rows} + week: list[PresenceHour] = [] + for weekday in range(7): + for hour in range(24): + row = cells.get( + (weekday, hour), {"hours": 0, "span": 0, "online": 0, "covered": 0} + ) + covered = row["covered"] + minutes = ( + min(60.0, row["online"] / covered * 60) + if covered > 0 + and covered >= min(MIN_CELL_COVERAGE, row["span"] * MIN_CELL_SHARE) + else None + ) + week.append( + PresenceHour( + weekday=weekday, + hour=hour, + hours=row["hours"], + online_minutes=minutes, + covered_hours=covered / SECONDS_IN_HOUR, + ) + ) + return week + + +async def presence_stats( # noqa: PLR0913 + pool: asyncpg.Pool, account_id: int, peer_id: int, page: Page, *, days: int, tz: str +) -> PresenceStats: + since, until = await period_bounds(pool, days, tz) + async with pool.acquire() as conn, conn.transaction(): + await conn.execute(f"SET LOCAL work_mem = '{STATS_WORK_MEM}'") + row = await conn.fetchrow( + _STATS, + account_id, + peer_id, + since, + until, + tz, + page.capped_limit, + page.offset, + ) + current = await current_presence(pool, account_id, peer_id) + sessions = [PresenceSession(**item) for item in json.loads(row["sessions"] or "[]")] + covered = row["covered_seconds"] or 0 + online = row["online_seconds"] + return PresenceStats( + peer_id=peer_id, + tz=tz, + since=since, + until=until, + current=current, + online=current is not None + and current.status == "online" + and current.next_offline_date is not None + and current.next_offline_date > until, + hidden=row["session_count"] == 0 + and current is not None + and current.status in FUZZY_STATUSES, + summary=PresenceSummary( + online_seconds=online, + daily_seconds=online / covered * SECONDS_IN_DAY + if covered >= SECONDS_IN_HOUR + else None, + covered_seconds=covered, + session_count=row["session_count"], + median_session_seconds=row["median_session"], + first_online=_time_of_day(row["first_online"], row["day_start"]), + last_offline=_time_of_day(row["last_offline"], row["day_start"]), + last_seen=row["last_seen"] + or (current.last_online_date if current else None), + ), + week=_week(json.loads(row["week"] or "[]")), + days=[PresenceDay(**item) for item in json.loads(row["days"] or "[]")], + gaps=[CoverageGap(**item) for item in json.loads(row["gaps"] or "[]")], + sessions=sessions, + ) + + +async def presence_sessions( # noqa: PLR0913 + pool: asyncpg.Pool, account_id: int, peer_id: int, page: Page, *, days: int, tz: str +) -> list[PresenceSession]: + since, until = await period_bounds(pool, days, tz) + rows = await pool.fetch( + _SESSION_PAGE, account_id, peer_id, since, until, page.capped_limit, page.offset + ) + return [PresenceSession(**dict(row)) for row in rows]