78 lines
3.0 KiB
Python
78 lines
3.0 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
import logging
|
|
from star_resonance_tracer.proto.stru_place_holder_item_pb2 import PlaceHolderItem
|
|
from star_resonance_tracer.sniffer import Sniffer
|
|
|
|
from inventory_wars.models import User, ItemShare, Event
|
|
from inventory_wars.const import ChitChatNtf, HypertextVariant, decode_placeholder
|
|
from star_resonance_tracer.proto.serv_chit_chat_ntf_pb2 import ChitChatNtf as ChitChatNtfPb
|
|
from star_resonance_tracer.proto.enum_chit_chat_channel_type_pb2 import ChitChatChannelType
|
|
from star_resonance_tracer.proto.enum_chit_chat_msg_type_pb2 import ChitChatMsgType
|
|
|
|
from inventory_wars.sniffer import start_sniffing
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Game:
|
|
def __init__(self, session: Session, *, item_id: int, listening_channels: list[ChitChatChannelType] | None = None):
|
|
self.session = session
|
|
self.listening_channels = listening_channels or []
|
|
self.event = Event(item_id=item_id)
|
|
|
|
def start(self) -> None:
|
|
self.event.timestamp = datetime.now()
|
|
|
|
self.session.add(self.event)
|
|
self.session.commit()
|
|
|
|
logger.info(f"Started {self.event.id} at {self.event.timestamp} for item {self.event.item_id}")
|
|
|
|
sniffer = Sniffer()
|
|
sniffer.set_service_type(
|
|
ChitChatNtf.ServiceId.value,
|
|
ChitChatNtf.Method.NotifyNewestChitChatMsgs.value,
|
|
ChitChatNtfPb.NotifyNewestChitChatMsgs
|
|
)
|
|
sniffer.on_service(ChitChatNtfPb.NotifyNewestChitChatMsgs, self.on_chit_chat_msg)
|
|
|
|
start_sniffing(sniffer)
|
|
|
|
def on_chit_chat_msg(self, event: ChitChatNtfPb.NotifyNewestChitChatMsgs) -> None:
|
|
req = event.vRequest
|
|
if req.channelType not in self.listening_channels:
|
|
return
|
|
|
|
if req.chatMsg.msgInfo.msgType is not ChitChatMsgType.ChatMsgHypertext:
|
|
return
|
|
|
|
hypertext = req.chatMsg.msgInfo.chatHypertext
|
|
if hypertext.configId != HypertextVariant.ITEM_SHARING.value:
|
|
return
|
|
|
|
for placeholder in hypertext.hypertextContents:
|
|
placeholder_content = decode_placeholder(placeholder)
|
|
match placeholder_content:
|
|
case PlaceHolderItem() as item:
|
|
if item.configId != self.event.item_id:
|
|
continue
|
|
|
|
user = User(id=req.chatMsg.sendCharInfo.charID, username=req.chatMsg.sendCharInfo.name)
|
|
item_share = ItemShare(
|
|
timestamp=datetime.fromtimestamp(req.chatMsg.timestamp),
|
|
user_id=user.id,
|
|
event_id=self.event.id,
|
|
count=item.ItemDetail.count,
|
|
raw=item.SerializeToString()
|
|
)
|
|
|
|
self.session.merge(user)
|
|
self.session.add(item_share)
|
|
self.session.commit()
|
|
|
|
logger.info(f"{user.username} guessed {self.event.item_id} "
|
|
f"with {item_share.count} at {item_share.timestamp}")
|