112 lines
4.1 KiB
Python
112 lines
4.1 KiB
Python
import logging
|
|
from datetime import datetime, UTC
|
|
from typing import TYPE_CHECKING, Protocol
|
|
|
|
from star_resonance_tracer.proto.stru_notify_newest_chit_chat_msgs_request_pb2 import NotifyNewestChitChatMsgsRequest
|
|
from star_resonance_tracer.proto.stru_place_holder_item_pb2 import PlaceHolderItem
|
|
|
|
from inventory_wars.const import decode_placeholder
|
|
from inventory_wars.models import Event, User, ItemShare
|
|
from inventory_wars.scoring import Scoring
|
|
|
|
if TYPE_CHECKING:
|
|
from inventory_wars.game import Game
|
|
|
|
__all__ = (
|
|
"State",
|
|
"GameOngoing",
|
|
"GameIdle"
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class State(Protocol):
|
|
def start(self, item_id: int, scoring: type[Scoring]) -> None:
|
|
raise NotImplementedError
|
|
|
|
def end(self) -> None:
|
|
raise NotImplementedError
|
|
|
|
def handle_msg(self, msg: NotifyNewestChitChatMsgsRequest) -> None:
|
|
raise NotImplementedError
|
|
|
|
|
|
class GameIdle(State):
|
|
def __init__(self, game: Game):
|
|
self.game = game
|
|
|
|
def start(self, item_id: int, scoring: Scoring) -> None:
|
|
event = Event(item_id=item_id)
|
|
|
|
self.game.session.add(event)
|
|
self.game.session.commit()
|
|
|
|
self.game.set_state(GameOngoing(self.game, scoring, event))
|
|
|
|
logger.info(f"Started {event.id} at {event.start_at} for item {event.item_id}.")
|
|
|
|
def handle_msg(self, msg: NotifyNewestChitChatMsgsRequest) -> None:
|
|
hypertext = msg.chatMsg.msgInfo.chatHypertext
|
|
for placeholder in hypertext.hypertextContents:
|
|
placeholder_content = decode_placeholder(placeholder)
|
|
match placeholder_content:
|
|
case PlaceHolderItem() as item:
|
|
logger.info(item)
|
|
|
|
|
|
class GameOngoing(State):
|
|
def __init__(self, game: Game, scoring: Scoring, event: Event):
|
|
self.game = game
|
|
self.scoring = scoring
|
|
self.event = event
|
|
|
|
def end(self):
|
|
self.event.end_at = datetime.now(UTC)
|
|
end = self.scoring.calculate_end_score()
|
|
if end:
|
|
end.user.score += end.score
|
|
self.game.session.commit()
|
|
|
|
self.game.set_state(GameIdle(self.game))
|
|
|
|
logger.info(f"Ended {self.event.id} at {self.event.end_at}.")
|
|
if end:
|
|
logger.info(f"{end.user.username} with {end.item_id} at {end.count} earned {end.score} scores.")
|
|
|
|
def handle_msg(self, msg: NotifyNewestChitChatMsgsRequest) -> None:
|
|
hypertext = msg.chatMsg.msgInfo.chatHypertext
|
|
for placeholder in hypertext.hypertextContents:
|
|
placeholder_content = decode_placeholder(placeholder)
|
|
match placeholder_content:
|
|
case PlaceHolderItem() as item:
|
|
user_id = msg.chatMsg.sendCharInfo.charID
|
|
user = self.game.session.get(User, user_id)
|
|
if user is None:
|
|
user = User(id=user_id, username=msg.chatMsg.sendCharInfo.name, score=0)
|
|
self.game.session.add(user)
|
|
|
|
item_share = ItemShare(
|
|
timestamp=datetime.fromtimestamp(msg.chatMsg.timestamp, UTC),
|
|
user=user,
|
|
event_id=self.event.id if item.configId == self.event.item_id else None,
|
|
item_id=item.configId,
|
|
count=item.ItemDetail.count,
|
|
raw=item.SerializeToString()
|
|
)
|
|
|
|
if item_share.event_id:
|
|
item_share.score = self.scoring.calculate_score(item_share)
|
|
if item_share.score:
|
|
logger.info(user.score)
|
|
logger.info(item_share.score)
|
|
user.score += item_share.score
|
|
|
|
self.game.session.add(item_share)
|
|
self.game.session.commit()
|
|
|
|
logger.info(f"{user.username} guessed {item_share.item_id} "
|
|
f"with {item_share.count} at {item_share.timestamp}.")
|
|
if item_share.score:
|
|
logger.info(f"{user.username} earned {item_share.score} scores.")
|