Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 142 additions & 0 deletions src/prism/flashlight.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
import functools
import logging
from json import JSONDecodeError
from typing import Any

import requests
from requests.exceptions import RequestException

from prism.errors import APIError
from prism.player import Tags, TagSeverity
from prism.ratelimiting import RateLimiter
from prism.requests import make_prism_requests_session
from prism.retry import ExecutionError, execute_with_retry

logger = logging.getLogger(__name__)

FLASHLIGHT_API_URL = "https://flashlight.prismoverlay.com"


class FlashlightTagsProvider:
def __init__(
self,
*,
retry_limit: int,
initial_timeout: float,
) -> None:
self._retry_limit = retry_limit
self._initial_timeout = initial_timeout
self._session = make_prism_requests_session()
self._limiter = RateLimiter(limit=120, window=60)

@property
def seconds_until_unblocked(self) -> float:
"""Return the number of seconds until we are unblocked"""
return self._limiter.block_duration_seconds

def _make_tags_request(
self,
*,
url: str,
user_id: str,
last_try: bool,
urchin_api_key: str | None,
) -> requests.Response: # pragma: nocover
headers = {"X-User-Id": user_id}
if urchin_api_key:
headers["X-Urchin-Api-Key"] = urchin_api_key

try:
# Uphold our prescribed rate-limits
with self._limiter:
response = self._session.get(url, headers=headers, timeout=10)
except RequestException as e:
raise ExecutionError(
"Request to flashlight failed due to an unknown error"
) from e

if response.status_code == 429 or response.status_code == 503 and not last_try:
raise ExecutionError(
"Request to flashlight failed due to intermittent error, retrying"
)

return response

def get_tags(
self,
uuid: str,
*,
urchin_api_key: str | None,
user_id: str,
) -> Tags: # pragma: nocover
"""Get the tags for the given player (are they a sniper/cheater)"""
url = f"{FLASHLIGHT_API_URL}/v1/tags/{uuid}"

try:
response = execute_with_retry(
functools.partial(
self._make_tags_request,
url=url,
user_id=user_id,
urchin_api_key=urchin_api_key,
),
retry_limit=self._retry_limit,
initial_timeout=self._initial_timeout,
)
except ExecutionError as e:
raise APIError(f"Request to flashlight failed for {uuid=}.") from e

if not response:
raise APIError(
f"Request to flashlight failed with status code "
f"{response.status_code} when getting tags for player {uuid}. "
f"Response: {response.text}"
)

try:
response_json = response.json()
except JSONDecodeError as e:
raise APIError(
"Failed parsing the response from flashlight. "
f"Raw content: {response.text}"
) from e

return parse_flashlight_tags(response_json)


def validate_tag_severity(tag_severity: object) -> TagSeverity | None:
"""Validate that the string is a valid TagSeverity"""
if not isinstance(tag_severity, str):
return None

if tag_severity == "none":
return "none"
if tag_severity == "medium":
return "medium"
if tag_severity == "high":
return "high"

return None


def parse_flashlight_tags(response_json: Any) -> Tags:
"""Parse the flashlight tags from the response JSON"""
tags_data = response_json.get("tags", {})
if not isinstance(tags_data, dict):
raise APIError(f"Invalid tags data {tags_data=} {type(tags_data)=}")

cheating_severity = validate_tag_severity(tags_data.get("cheating", None))
if cheating_severity is None:
raise APIError(
f"Invalid cheating tag severity " f"{response_json.get('cheating', None)=}"
)
sniper_severity = validate_tag_severity(tags_data.get("sniping", None))
if sniper_severity is None:
raise APIError(
f"Invalid sniping tag severity " f"{response_json.get('sniping', None)=}"
)

return Tags(
cheating=cheating_severity,
sniping=sniper_severity,
)
3 changes: 3 additions & 0 deletions src/prism/overlay/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ def main() -> None: # pragma: nocover
nick_database = NickDatabase.from_disk([], default_database=default_database)

# Import late so we can patch ssl certs in requests
from prism.flashlight import FlashlightTagsProvider
from prism.mojang import MojangAccountProvider
from prism.overlay.antisniper_api import (
StrangePlayerProvider,
Expand All @@ -82,6 +83,7 @@ def main() -> None: # pragma: nocover
retry_limit=5, initial_timeout=2, get_time_ns=time.time_ns
)
winstreak_provider = PlaceholderWinstreakProvider()
tags_provider = FlashlightTagsProvider(retry_limit=5, initial_timeout=2)

controller = OverlayController(
state=OverlayState(),
Expand All @@ -90,6 +92,7 @@ def main() -> None: # pragma: nocover
account_provider=account_provider,
player_provider=player_provider,
winstreak_provider=winstreak_provider,
tags_provider=tags_provider,
)

if options.test_ssl:
Expand Down
16 changes: 16 additions & 0 deletions src/prism/overlay/behaviour.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,22 @@ def get_and_cache_player(
completed_queue.put(username)
logger.debug(f"Updated missing winstreak for {username}")

# Get tags
if isinstance(player, KnownPlayer):
tags = controller.get_tags(player.uuid)
if tags is ERROR_DURING_PROCESSING:
logger.error(f"Error getting tags for {username}")
else:
for alias in player.aliases:
controller.player_cache.update_cached_player(
alias,
functools.partial(KnownPlayer.set_tags, tags=tags),
)

# Tell the main thread that we got the tags
completed_queue.put(username)
logger.debug(f"Set tags for {username}")


def update_settings(new_settings: SettingsDict, controller: OverlayController) -> None:
"""
Expand Down
30 changes: 29 additions & 1 deletion src/prism/overlay/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from typing import TYPE_CHECKING, Protocol

from prism.errors import APIError, APIKeyError, APIThrottleError, PlayerNotFoundError
from prism.player import MISSING_WINSTREAKS, KnownPlayer, Winstreaks
from prism.player import MISSING_WINSTREAKS, KnownPlayer, Tags, Winstreaks
from prism.ssl_errors import MissingLocalIssuerSSLError

if TYPE_CHECKING: # pragma: no cover
Expand Down Expand Up @@ -50,6 +50,19 @@ def get_estimated_winstreaks_for_uuid(
def seconds_until_unblocked(self) -> float: ...


class TagsProvider(Protocol):
def get_tags(
self,
uuid: str,
*,
user_id: str,
urchin_api_key: str | None,
) -> Tags: ...

@property
def seconds_until_unblocked(self) -> float: ...


class OverlayController:
def __init__(
self,
Expand All @@ -59,6 +72,7 @@ def __init__(
account_provider: AccountProvider,
player_provider: PlayerProvider,
winstreak_provider: WinstreakProvider,
tags_provider: TagsProvider,
) -> None:
from prism.overlay.player_cache import PlayerCache

Expand All @@ -81,6 +95,7 @@ def __init__(
self._account_provider = account_provider
self._player_provider = player_provider
self._winstreak_provider = winstreak_provider
self._tags_provider = tags_provider

def get_uuid(self, username: str) -> str | None | ProcessingError:
try:
Expand Down Expand Up @@ -174,3 +189,16 @@ def get_estimated_winstreaks(self, uuid: str) -> tuple[Winstreaks, bool]:
self.antisniper_api_key_throttled = False
self.missing_local_issuer_certificate = False
return winstreaks, accurate

def get_tags(self, uuid: str) -> Tags | ProcessingError:
try:
tags = self._tags_provider.get_tags(
uuid=uuid,
user_id=self.settings.user_id,
urchin_api_key=None, # TODO: Implement urchin api key setting
)
except APIError as e:
logger.error(f"Error getting tags for {uuid=}", exc_info=e)
return ERROR_DURING_PROCESSING
else:
return tags
71 changes: 68 additions & 3 deletions src/prism/overlay/output/cell_renderer.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,20 @@
from collections.abc import Sequence
from dataclasses import dataclass, replace
from functools import lru_cache
from typing import assert_never
from typing import Literal, assert_never

from prism.overlay.output.cells import CellValue, ColorSection, ColumnName
from prism.overlay.output.color import GUIColor, MinecraftColor
from prism.overlay.output.config import RatingConfig, RatingConfigCollection
from prism.player import KnownPlayer, NickedPlayer, PendingPlayer, Player, UnknownPlayer
from prism.player import (
KnownPlayer,
NickedPlayer,
PendingPlayer,
Player,
Tags,
TagSeverity,
UnknownPlayer,
)
from prism.utils import format_seconds_short, truncate_float

GUI_COLORS = (
Expand Down Expand Up @@ -38,6 +46,8 @@ class RenderedStats:
wins: CellValue
sessiontime: CellValue

tags: CellValue


def truncate_float_or_int(value: float | int, decimals: int) -> str:
"""Truncate the decimals of the float, or keep the int"""
Expand Down Expand Up @@ -367,6 +377,59 @@ def render_stars(
return replace(levels_rating, color_sections=color_sections)


def tag_severity_to_color(severity: Literal["medium", "high"]) -> str:
if severity == "medium":
return GUI_COLORS[3]
elif severity == "high":
return GUI_COLORS[4]


def render_tags(tags: Tags | None) -> CellValue:
if tags is None:
# Pending
return CellValue.monochrome(text="-", gui_color=GUI_COLORS[0])

def add_tag(
text: str,
color_sections: tuple[ColorSection, ...],
tag_char: str,
severity: TagSeverity,
) -> tuple[str, tuple[ColorSection, ...]]:
if severity == "none":
return text, color_sections

if text != "":
# There is already a tag, add space separator
text += " "
color_sections += (ColorSection(GUI_COLORS[0], 1),)

return (
text + tag_char,
color_sections + (ColorSection(tag_severity_to_color(severity), 1),),
)

color_sections: tuple[ColorSection, ...] = ()
text = ""

text, color_sections = add_tag(text, color_sections, "C", tags.cheating)
text, color_sections = add_tag(text, color_sections, "S", tags.sniping)

if len(color_sections) == 0:
# The painter needs at least one color
return CellValue.monochrome(text="", gui_color=GUI_COLORS[0])

# Build hover text with detailed tag information
hover_parts: list[str] = []
if tags.cheating != "none":
hover_parts.append(f"C: Cheating ({tags.cheating})")
if tags.sniping != "none":
hover_parts.append(f"S: Sniping ({tags.sniping})")

hover_text = "\n".join(hover_parts) if hover_parts else None

return CellValue(text=text, color_sections=color_sections, hover=hover_text)


@lru_cache(maxsize=100)
def render_stats(
player: Player,
Expand Down Expand Up @@ -480,6 +543,7 @@ def render_stats(
rating_configs.sessiontime.rate_by_level,
rating_configs.sessiontime.sort_ascending,
)
tags_cell = render_tags(player.tags)
else:
if isinstance(player, NickedPlayer):
text = "nick"
Expand All @@ -496,7 +560,7 @@ def render_stats(
cell = CellValue.monochrome(text, gui_color=gui_color)
stars_cell = index_cell = fkdr_cell = kdr_cell = bblr_cell = wlr_cell = cell
winstreak_cell = kills_cell = finals_cell = beds_cell = wins_cell = cell
sessiontime_cell = cell
sessiontime_cell = tags_cell = cell

username_cell = CellValue.monochrome(username_str, gui_color=GUIColor.WHITE)

Expand All @@ -514,6 +578,7 @@ def render_stats(
beds=beds_cell,
wins=wins_cell,
sessiontime=sessiontime_cell,
tags=tags_cell,
)


Expand Down
Loading