WebHost: add a tracker api endpoint (#1052)

An endpoint from the tracker page.
This commit is contained in:
Aaron Wagener
2025-08-23 01:33:46 -05:00
committed by GitHub
parent bead81b64b
commit 88a4a589a0
3 changed files with 442 additions and 2 deletions

View File

@@ -11,5 +11,5 @@ api_endpoints = Blueprint('api', __name__, url_prefix="/api")
def get_players(seed: Seed) -> List[Tuple[str, str]]:
return [(slot.player_name, slot.game) for slot in seed.slots.order_by(Slot.player_id)]
from . import datapackage, generate, room, user # trigger registration
# trigger endpoint registration
from . import datapackage, generate, room, tracker, user

230
WebHostLib/api/tracker.py Normal file
View File

@@ -0,0 +1,230 @@
from datetime import datetime, timezone
from typing import Any, TypedDict
from uuid import UUID
from flask import abort
from NetUtils import ClientStatus, Hint, NetworkItem, SlotType
from WebHostLib import cache
from WebHostLib.api import api_endpoints
from WebHostLib.models import Room
from WebHostLib.tracker import TrackerData
@api_endpoints.route("/tracker/<suuid:tracker>")
@cache.memoize(timeout=60)
def tracker_data(tracker: UUID) -> dict[str, Any]:
"""
Outputs json data to <root_path>/api/tracker/<id of current session tracker>.
:param tracker: UUID of current session tracker.
:return: Tracking data for all players in the room. Typing and docstrings describe the format of each value.
"""
room: Room | None = Room.get(tracker=tracker)
if not room:
abort(404)
tracker_data = TrackerData(room)
all_players: dict[int, list[int]] = tracker_data.get_all_players()
class PlayerAlias(TypedDict):
player: int
name: str | None
player_aliases: list[dict[str, int | list[PlayerAlias]]] = []
"""Slot aliases of all players."""
for team, players in all_players.items():
team_player_aliases: list[PlayerAlias] = []
team_aliases = {"team": team, "players": team_player_aliases}
player_aliases.append(team_aliases)
for player in players:
team_player_aliases.append({"player": player, "alias": tracker_data.get_player_alias(team, player)})
class PlayerItemsReceived(TypedDict):
player: int
items: list[NetworkItem]
player_items_received: list[dict[str, int | list[PlayerItemsReceived]]] = []
"""Items received by each player."""
for team, players in all_players.items():
player_received_items: list[PlayerItemsReceived] = []
team_items_received = {"team": team, "players": player_received_items}
player_items_received.append(team_items_received)
for player in players:
player_received_items.append(
{"player": player, "items": tracker_data.get_player_received_items(team, player)})
class PlayerChecksDone(TypedDict):
player: int
locations: list[int]
player_checks_done: list[dict[str, int | list[PlayerChecksDone]]] = []
"""ID of all locations checked by each player."""
for team, players in all_players.items():
per_player_checks: list[PlayerChecksDone] = []
team_checks_done = {"team": team, "players": per_player_checks}
player_checks_done.append(team_checks_done)
for player in players:
per_player_checks.append(
{"player": player, "locations": sorted(tracker_data.get_player_checked_locations(team, player))})
total_checks_done: list[dict[str, int]] = [
{"team": team, "checks_done": checks_done}
for team, checks_done in tracker_data.get_team_locations_checked_count().items()
]
"""Total number of locations checked for the entire multiworld per team."""
class PlayerHints(TypedDict):
player: int
hints: list[Hint]
hints: list[dict[str, int | list[PlayerHints]]] = []
"""Hints that all players have used or received."""
for team, players in tracker_data.get_all_slots().items():
per_player_hints: list[PlayerHints] = []
team_hints = {"team": team, "players": per_player_hints}
hints.append(team_hints)
for player in players:
player_hints = sorted(tracker_data.get_player_hints(team, player))
per_player_hints.append({"player": player, "hints": player_hints})
slot_info = tracker_data.get_slot_info(team, player)
# this assumes groups are always after players
if slot_info.type != SlotType.group:
continue
for member in slot_info.group_members:
team_hints[member]["hints"] += player_hints
class PlayerTimer(TypedDict):
player: int
time: datetime | None
activity_timers: list[dict[str, int | list[PlayerTimer]]] = []
"""Time of last activity per player. Returned as RFC 1123 format and null if no connection has been made."""
for team, players in all_players.items():
player_timers: list[PlayerTimer] = []
team_timers = {"team": team, "players": player_timers}
activity_timers.append(team_timers)
for player in players:
player_timers.append({"player": player, "time": None})
client_activity_timers: tuple[tuple[int, int], float] = tracker_data._multisave.get("client_activity_timers", ())
for (team, player), timestamp in client_activity_timers:
# use index since we can rely on order
activity_timers[team]["player_timers"][player - 1]["time"] = datetime.fromtimestamp(timestamp, timezone.utc)
connection_timers: list[dict[str, int | list[PlayerTimer]]] = []
"""Time of last connection per player. Returned as RFC 1123 format and null if no connection has been made."""
for team, players in all_players.items():
player_timers: list[PlayerTimer] = []
team_connection_timers = {"team": team, "players": player_timers}
connection_timers.append(team_connection_timers)
for player in players:
player_timers.append({"player": player, "time": None})
client_connection_timers: tuple[tuple[int, int], float] = tracker_data._multisave.get(
"client_connection_timers", ())
for (team, player), timestamp in client_connection_timers:
connection_timers[team]["players"][player - 1]["time"] = datetime.fromtimestamp(timestamp, timezone.utc)
class PlayerStatus(TypedDict):
player: int
status: ClientStatus
player_status: list[dict[str, int | list[PlayerStatus]]] = []
"""The current client status for each player."""
for team, players in all_players.items():
player_statuses: list[PlayerStatus] = []
team_status = {"team": team, "players": player_statuses}
player_status.append(team_status)
for player in players:
player_statuses.append({"player": player, "status": tracker_data.get_player_client_status(team, player)})
return {
**get_static_tracker_data(room),
"aliases": player_aliases,
"player_items_received": player_items_received,
"player_checks_done": player_checks_done,
"total_checks_done": total_checks_done,
"hints": hints,
"activity_timers": activity_timers,
"connection_timers": connection_timers,
"player_status": player_status,
"datapackage": tracker_data._multidata["datapackage"],
}
@cache.memoize()
def get_static_tracker_data(room: Room) -> dict[str, Any]:
"""
Builds and caches the static data for this active session tracker, so that it doesn't need to be recalculated.
"""
tracker_data = TrackerData(room)
all_players: dict[int, list[int]] = tracker_data.get_all_players()
class PlayerGroups(TypedDict):
slot: int
name: str
members: list[int]
groups: list[dict[str, int | list[PlayerGroups]]] = []
"""The Slot ID of groups and the IDs of the group's members."""
for team, players in tracker_data.get_all_slots().items():
groups_in_team: list[PlayerGroups] = []
team_groups = {"team": team, "groups": groups_in_team}
groups.append(team_groups)
for player in players:
slot_info = tracker_data.get_slot_info(team, player)
if slot_info.type != SlotType.group or not slot_info.group_members:
continue
groups_in_team.append(
{
"slot": player,
"name": slot_info.name,
"members": list(slot_info.group_members),
})
class PlayerName(TypedDict):
player: int
name: str
player_names: list[dict[str, str | list[PlayerName]]] = []
"""Slot names of all players."""
for team, players in all_players.items():
per_team_player_names: list[PlayerName] = []
team_names = {"team": team, "players": per_team_player_names}
player_names.append(team_names)
for player in players:
per_team_player_names.append({"player": player, "name": tracker_data.get_player_name(team, player)})
class PlayerGame(TypedDict):
player: int
game: str
games: list[dict[str, int | list[PlayerGame]]] = []
"""The game each player is playing."""
for team, players in all_players.items():
player_games: list[PlayerGame] = []
team_games = {"team": team, "players": player_games}
games.append(team_games)
for player in players:
player_games.append({"player": player, "game": tracker_data.get_player_game(team, player)})
class PlayerSlotData(TypedDict):
player: int
slot_data: dict[str, Any]
slot_data: list[dict[str, int | list[PlayerSlotData]]] = []
"""Slot data for each player."""
for team, players in all_players.items():
player_slot_data: list[PlayerSlotData] = []
team_slot_data = {"team": team, "players": player_slot_data}
slot_data.append(team_slot_data)
for player in players:
player_slot_data.append({"player": player, "slot_data": tracker_data.get_slot_data(team, player)})
return {
"groups": groups,
"slot_data": slot_data,
}

View File

@@ -16,6 +16,8 @@ Current endpoints:
- [`/status/<suuid:seed>`](#status)
- Room API
- [`/room_status/<suuid:room_id>`](#roomstatus)
- Tracker API
- [`/tracker/<suuid:tracker>`](#tracker)
- User API
- [`/get_rooms`](#getrooms)
- [`/get_seeds`](#getseeds)
@@ -244,6 +246,214 @@ Example:
}
```
## Tracker Endpoints
Endpoints to fetch information regarding players of an active WebHost room with the supplied tracker_ID. The tracker ID
can either be viewed while on a room tracker page, or from the [room's endpoint](#room-endpoints).
### `/tracker/<suuid:tracker>`
<a name=tracker></a>
Will provide a dict of tracker data with the following keys:
- item_link groups and their players (`groups`)
- Each player's slot_data (`slot_data`)
- Each player's current alias (`aliases`)
- Will return the name if there is none
- A list of items each player has received as a NetworkItem (`player_items_received`)
- A list of checks done by each player as a list of the location id's (`player_checks_done`)
- The total number of checks done by all players (`total_checks_done`)
- Hints that players have used or received (`hints`)
- The time of last activity of each player in RFC 1123 format (`activity_timers`)
- The time of last active connection of each player in RFC 1123 format (`connection_timers`)
- The current client status of each player (`player_status`)
- The datapackage hash for each player (`datapackage`)
- This hash can then be sent to the datapackage API to receive the appropriate datapackage as necessary
Example:
```json
{
"groups": [
{
"team": 0,
"groups": [
{
"slot": 5,
"name": "testGroup",
"members": [
1,
2
]
},
{
"slot": 6,
"name": "myCoolLink",
"members": [
3,
4
]
}
]
}
],
"slot_data": [
{
"team": 0,
"players": [
{
"player": 1,
"slot_data": {
"example_option": 1,
"other_option": 3
}
},
{
"player": 2,
"slot_data": {
"example_option": 1,
"other_option": 2
}
}
]
}
],
"aliases": [
{
"team": 0,
"players": [
{
"player": 1,
"alias": "Incompetence"
},
{
"player": 2,
"alias": "Slot_Name_2"
}
]
}
],
"player_items_received": [
{
"team": 0,
"players": [
{
"player": 1,
"items": [
[1, 1, 1, 0],
[2, 2, 2, 1]
]
},
{
"player": 2,
"items": [
[1, 1, 1, 2],
[2, 2, 2, 0]
]
}
]
}
],
"player_checks_done": [
{
"team": 0,
"players": [
{
"player": 1,
"locations": [
1,
2
]
},
{
"player": 2,
"locations": [
1,
2
]
}
]
}
],
"total_checks_done": [
{
"team": 0,
"checks_done": 4
}
],
"hints": [
{
"team": 0,
"players": [
{
"player": 1,
"hints": [
[1, 2, 4, 6, 0, "", 4, 0]
]
},
{
"player": 2,
"hints": []
}
]
}
],
"activity_timers": [
{
"team": 0,
"players": [
{
"player": 1,
"time": "Fri, 18 Apr 2025 20:35:45 GMT"
},
{
"player": 2,
"time": "Fri, 18 Apr 2025 20:42:46 GMT"
}
]
}
],
"connection_timers": [
{
"team": 0,
"players": [
{
"player": 1,
"time": "Fri, 18 Apr 2025 20:38:25 GMT"
},
{
"player": 2,
"time": "Fri, 18 Apr 2025 21:03:00 GMT"
}
]
}
],
"player_status": [
{
"team": 0,
"players": [
{
"player": 1,
"status": 0
},
{
"player": 2,
"status": 0
}
]
}
],
"datapackage": {
"Archipelago": {
"checksum": "ac9141e9ad0318df2fa27da5f20c50a842afeecb",
"version": 0
},
"The Messenger": {
"checksum": "6991cbcda7316b65bcb072667f3ee4c4cae71c0b",
"version": 0
}
}
}
```
## User Endpoints
User endpoints can get room and seed details from the current session tokens (cookies)