Module server.ladder_service.violation_service

Classes

class Violation (count: int = 1, time: Optional[datetime.datetime] = None)

Violation(count: int = 1, time: Optional[datetime.datetime] = None)

Expand source code 
@dataclass
class Violation:
    count: int
    time: datetime

    def __init__(self, count: int = 1, time: Optional[datetime] = None):
        self.count = count
        self.time = time or datetime_now()

    def register(self):
        self.count += 1
        self.time = datetime_now()

    def get_ban_expiration(self) -> datetime:
        if self.count < 2:
            # No ban, expires as soon as it's registered
            return self.time
        elif self.count == 2:
            return self.time + timedelta(minutes=10)
        else:
            return self.time + timedelta(minutes=30)

    def get_remaining(self, now: Optional[datetime] = None) -> timedelta:
        return self.get_ban_expiration() - (now or datetime_now())

    def is_expired(self, now: Optional[datetime] = None) -> bool:
        """
        Whether the violation history should be reset. This is different from
        the ban expiration time which should be checked by calling
        `get_ban_expiration`.
        """
        now = now or datetime_now()
        # TODO: Config?
        return self.time + timedelta(hours=1) <= now

    def to_dict(self) -> dict:
        return {
            "count": self.count,
            "time": self.time.isoformat()
        }

Class variables

var count : int
var time : datetime.datetime

Methods

def get_ban_expiration(self) ‑> datetime.datetime
def get_remaining(self, now: Optional[datetime.datetime] = None) ‑> datetime.timedelta
def is_expired(self, now: Optional[datetime.datetime] = None) ‑> bool

Whether the violation history should be reset. This is different from the ban expiration time which should be checked by calling get_ban_expiration.

def register(self)
def to_dict(self) ‑> dict
class ViolationService

Track who is banned from searching and for how long. Apply progressive discipline for repeated violations.

A violation could be anything, but it is usually any time a player fails to connect to a game.

Expand source code 
@with_logger
class ViolationService(Service):
    """
    Track who is banned from searching and for how long. Apply progressive
    discipline for repeated violations.

    A violation could be anything, but it is usually any time a player fails
    to connect to a game.
    """

    def __init__(self):
        # We store a reference to the original `Player` object for logging only
        self._violations: dict[int, tuple[Player, Violation]] = {}

    async def initialize(self):
        self._cleanup_task = at_interval(5, func=self.clear_expired)

    def clear_expired(self):
        now = datetime_now()
        for player, violation in list(self._violations.values()):
            if violation.is_expired(now):
                self._clear_violation(player)

    def register_violations(self, players: list[Player]):
        now = datetime_now()
        for player in players:
            violation = self.get_violation(player)
            if violation is None or violation.is_expired(now):
                violation = Violation(time=now)
                self.set_violation(player, violation)
            else:
                violation.register()

            player.write_message({
                "command": "search_violation",
                **violation.to_dict()
            })
            extra_text = ""
            if violation.count > 1:
                delta_text = humanize.precisedelta(
                    violation.get_ban_expiration() - now
                )
                extra_text = f" You can queue again in {delta_text}"
            player.write_message({
                "command": "notice",
                "style": "info",
                "text": (
                    f"You have caused a matchmaking connection failure {violation.count} time(s). "
                    "Multiple failures result in temporary time-outs from matchmaker. "
                    "Please seek support on the forums or discord for persistent issues." +
                    extra_text
                )
            })

    def get_violations(self, players: list[Player]) -> dict[Player, Violation]:
        now = datetime_now()
        result = {}
        for player in players:
            violation = self.get_violation(player)
            if not violation:
                continue
            elif violation.get_ban_expiration() > now:
                result[player] = violation
            elif violation.is_expired(now):
                self._clear_violation(player)

        return result

    def get_violation(self, player: Player) -> Optional[Violation]:
        _, violation = self._violations.get(player.id, (None, None))
        return violation

    def set_violation(self, player: Player, violation: Violation):
        self._violations[player.id] = (player, violation)

    def _clear_violation(self, player: Player):
        violation = self.get_violation(player)
        self._logger.debug(
            "Cleared violation for player %s: %s",
            player.login,
            violation
        )
        del self._violations[player.id]

Ancestors

Methods

def clear_expired(self)
def get_violation(self, player: Player) ‑> Optional[Violation]
def get_violations(self, players: list[Player]) ‑> dict[Player, Violation]
def register_violations(self, players: list[Player])
def set_violation(self, player: Player, violation: Violation)

Inherited members