#106 added domain logic action-state-decision

shvatka/common/log_utils.py

@@ -0,0 +1,6 @@
+from base64 import b64encode
+from typing import Any
+
+
+def obfuscate_sensitive(information: Any) -> str:
+    return b64encode(str(information).encode("utf8")).decode("utf8")

shvatka/core/interfaces/dal/game_play.py

@@ -54,6 +54,11 @@ async def save_key(
     ) -> dto.KeyTime:
         raise NotImplementedError
 
+    async def get_team_typed_keys(
+        self, game: dto.Game, team: dto.Team, level_number: int
+    ) -> list[dto.KeyTime]:
+        raise NotImplementedError
+
     async def level_up(self, team: dto.Team, level: dto.Level, game: dto.Game) -> None:
         raise NotImplementedError
 

shvatka/core/models/dto/scn/__init__.py

@@ -25,6 +25,6 @@
     PhotoHint,
     ContactHint,
 )
-from .level import LevelScenario, SHKey, BonusKey, HintsList
+from .level import LevelScenario, SHKey, BonusKey, HintsList, Conditions
 from .parsed_zip import ParsedZip
 from .time_hint import TimeHint

shvatka/core/models/dto/scn/action/__init__.py

@@ -0,0 +1,11 @@
+from .interface import WinCondition, Action, State, Decision, DecisionType, StateHolder
+from .decisions import NotImplementedActionDecision, Decisions
+from .keys import (
+    KeyDecision,
+    KeyWinCondition,
+    TypedKeyAction,
+    TypedKeysState,
+    BonusKeyDecision,
+    KeyBonusCondition,
+)
+from .state_holder import InMemoryStateHolder

shvatka/core/models/dto/scn/action/decisions.py

@@ -0,0 +1,49 @@
+from abc import abstractmethod
+from dataclasses import dataclass
+from typing import Literal, Sequence, overload
+
+from shvatka.core.models.dto.scn.action.interface import DecisionType, Decision
+
+
+@dataclass
+class NotImplementedActionDecision(Decision):
+    type: Literal[DecisionType.NO_ACTION] = DecisionType.NOT_IMPLEMENTED
+
+
+class Decisions(Sequence[Decision]):
+    def __init__(self, decisions: list[Decision]):
+        self.decisions = decisions
+
+    @overload
+    @abstractmethod
+    def __getitem__(self, index: int) -> Decision:
+        return self.decisions[index]
+
+    @overload
+    @abstractmethod
+    def __getitem__(self, index: slice) -> Sequence[Decision]:
+        return self.decisions[index]
+
+    def __getitem__(self, index):
+        return self.decisions[index]
+
+    def __len__(self):
+        return len(self.decisions)
+
+    def __iter__(self):
+        return iter(self.decisions)
+
+    def get_significant(self) -> "Decisions":
+        return self.get_all_except(DecisionType.NOT_IMPLEMENTED, DecisionType.NO_ACTION)
+
+    def get_implemented(self) -> "Decisions":
+        return self.get_all_except(DecisionType.NOT_IMPLEMENTED)
+
+    def get_all(self, *type_: type) -> "Decisions":
+        return Decisions([d for d in self if isinstance(d, type_)])
+
+    def get_all_except(self, *type_: DecisionType) -> "Decisions":
+        return Decisions([d for d in self if d.type not in type_])
+
+    def get_all_only(self, *type_: DecisionType) -> "Decisions":
+        return Decisions([d for d in self if d.type in type_])

shvatka/core/models/dto/scn/action/interface.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import enum
+import typing
+from typing import Protocol
+
+
+class WinCondition(Protocol):
+    def check(self, action: Action, state_holder: StateHolder) -> Decision:
+        raise NotImplementedError
+
+
+class Action(Protocol):
+    pass
+
+
+class State(Protocol):
+    pass
+
+
+T = typing.TypeVar("T")
+
+
+class StateHolder(Protocol):
+    def get(self, state_class: type[T]) -> T:
+        raise NotImplementedError
+
+
+class Decision(Protocol):
+    type: DecisionType
+
+
+class DecisionType(enum.StrEnum):
+    NOT_IMPLEMENTED = enum.auto()
+    LEVEL_UP = enum.auto()
+    SIGNIFICANT_ACTION = enum.auto()
+    NO_ACTION = enum.auto()
+    BONUS_TIME = enum.auto()

shvatka/core/models/dto/scn/action/keys.py

@@ -0,0 +1,136 @@
+import typing
+from dataclasses import dataclass
+from typing import Literal
+
+from shvatka.core.models import enums, dto
+from . import StateHolder
+from .decisions import NotImplementedActionDecision
+from .interface import Action, State, Decision, WinCondition, DecisionType
+from shvatka.core.models.dto.scn import BonusKey
+
+SHKey: typing.TypeAlias = str
+
+
+@dataclass
+class TypedKeyAction(Action):
+    key: SHKey
+
+
+@dataclass
+class TypedKeysState(State):
+    typed_correct: set[SHKey]
+    all_typed: set[SHKey]
+
+    def is_duplicate(self, action: TypedKeyAction) -> bool:
+        return action.key in self.all_typed
+
+
+@dataclass
+class WrongKeyDecision(Decision):
+    duplicate: bool
+    key: str
+    type: Literal[DecisionType.NO_ACTION] = DecisionType.NO_ACTION
+    key_type: typing.Literal[enums.KeyType.wrong] = enums.KeyType.wrong
+
+
+@dataclass
+class KeyDecision(Decision):
+    type: DecisionType
+    key_type: enums.KeyType
+    duplicate: bool
+    key: SHKey
+
+    def is_level_up(self) -> bool:
+        return self.type == DecisionType.LEVEL_UP
+
+    def to_parsed_key(self) -> dto.ParsedKey:
+        return dto.ParsedKey(
+            type_=self.key_type,
+            text=self.key,
+        )
+
+    @property
+    def key_text(self) -> str:
+        return self.key
+
+
+@dataclass
+class KeyWinCondition(WinCondition):
+    keys: set[SHKey]
+
+    def check(self, action: Action, state_holder: StateHolder) -> Decision:
+        if not isinstance(action, TypedKeyAction):
+            return NotImplementedActionDecision()
+        state = state_holder.get(TypedKeysState)
+        type_: DecisionType
+        if not self._is_correct(action):
+            return WrongKeyDecision(duplicate=state.is_duplicate(action), key=action.key)
+        if not state.is_duplicate(action):
+            if self._is_all_typed(action, state):
+                type_ = DecisionType.LEVEL_UP
+            else:
+                type_ = DecisionType.SIGNIFICANT_ACTION
+        else:
+            type_ = DecisionType.NO_ACTION
+        return KeyDecision(
+            type=type_,
+            key_type=enums.KeyType.simple if self._is_correct(action) else enums.KeyType.wrong,
+            duplicate=state.is_duplicate(action),
+            key=action.key,
+        )
+
+    def _is_correct(self, action: TypedKeyAction) -> bool:
+        return action.key in self.keys
+
+    def _is_all_typed(self, action: TypedKeyAction, state: TypedKeysState) -> bool:
+        return self.keys == {*state.typed_correct, action.key}
+
+
+@dataclass
+class BonusKeyDecision(Decision):
+    type: DecisionType
+    key_type: enums.KeyType
+    duplicate: bool
+    key: BonusKey | None
+
+    def to_parsed_key(self) -> dto.ParsedKey:
+        if self.type == DecisionType.BONUS_TIME:
+            return dto.ParsedBonusKey(
+                type_=enums.KeyType.bonus,
+                text=self.key.text,
+                bonus_minutes=self.key.bonus_minutes,
+            )
+        else:
+            return dto.ParsedKey(
+                type_=enums.KeyType.wrong,
+                text=self.key.text,
+            )
+
+    @property
+    def key_text(self) -> str:
+        return self.key.text
+
+
+@dataclass
+class KeyBonusCondition(WinCondition):
+    keys: set[BonusKey]
+
+    def check(self, action: Action, state_holder: StateHolder) -> Decision:
+        if not isinstance(action, TypedKeyAction):
+            return NotImplementedActionDecision()
+        state = state_holder.get(TypedKeysState)
+        bonus = self._get_bonus(action)
+        if bonus is None:
+            return WrongKeyDecision(duplicate=state.is_duplicate(action), key=action.key)
+        return BonusKeyDecision(
+            type=DecisionType.BONUS_TIME,
+            key_type=enums.KeyType.bonus,
+            duplicate=state.is_duplicate(action),
+            key=bonus,
+        )
+
+    def _get_bonus(self, action: TypedKeyAction) -> BonusKey | None:
+        for bonus_key in self.keys:
+            if action.key == bonus_key.text:
+                return bonus_key
+        return None

shvatka/core/models/dto/scn/action/state_holder.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+
+from . import TypedKeysState
+from .interface import StateHolder, T
+from shvatka.core.models.dto import scn
+
+
+@dataclass
+class InMemoryStateHolder(StateHolder):
+    typed_correct: set[scn.SHKey]
+    all_typed: set[scn.SHKey]
+
+    def get(self, state_class: type[T]) -> T:
+        if isinstance(state_class, TypedKeysState):
+            return TypedKeysState(
+                typed_correct=self.typed_correct,
+                all_typed=self.all_typed,
+            )
+        else:
+            raise NotImplementedError(f"unknown state type {type(state_class)}")