1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
import fortnitepy
import typing
import time
import datetime
from fortnite_client import *
from telegram_bot import *
from persistence import *
from formatter import *
__accept_keyword__ = "accept"
__decline_keyword__ = "decline"
class ClientInitObserverImpl(ClientInitObserver):
async def on_event(self, fortnite_client: FortniteClient) -> None:
print('----------------')
print('FortniteClient ready as:')
print(fortnite_client.user.display_name)
print(fortnite_client.user.id)
print('----------------')
class FriendPresenceObserverImpl(FriendPresenceObserver):
__telegram_bot: TelegramBot
__presence_repository: PresenceRepository
def __init__(self, telegram_bot: TelegramBot, presence_repository: PresenceRepository):
self.__telegram_bot = telegram_bot
self.__presence_repository = presence_repository
async def update(self, display_name: str, playing: bool, party_size: int) -> None:
if playing:
last_presence = self.__presence_repository.get_last_user_presence(display_name)
diff = time.time() - last_presence
if diff > 60 * 60: # 60 minutes
await self.__notify_friend_playing(display_name, party_size)
self.__presence_repository.set_last_user_presence(display_name, time.time())
async def __notify_friend_playing(self, display_name: str, party_size: int):
await self.__telegram_bot.send_message_to_all(format_friend_online(display_name, party_size))
class NewFriendObserverImpl(NewFriendObserver):
__telegram_bot: TelegramBot
__stats_repository: StatsRepository
def __init__(self, telegram_bot: TelegramBot, stats_repository: StatsRepository):
self.__telegram_bot = telegram_bot
self.__stats_repository = stats_repository
async def on_new_friend_request(self, request_id: str, display_name: str, fortnite_client: FortniteClient) -> None:
__new_friend_reply_markup = telebot.types.InlineKeyboardMarkup()
__new_friend_reply_markup.add(telebot.types.InlineKeyboardButton('Accept', callback_data="{}:{}:{}".format(__accept_keyword__, request_id, display_name)))
__new_friend_reply_markup.add(telebot.types.InlineKeyboardButton('Decline', callback_data="{}:{}:{}".format(__decline_keyword__, request_id, display_name)))
self.__telegram_bot.register_callback_query(
NewFriendCallbackQueryHandler(self.__telegram_bot, fortnite_client, self),
lambda call: call.data.startswith(__accept_keyword__) or call.data.startswith(__decline_keyword__))
await self.__telegram_bot.send_message_to_all(format_new_friend_request(display_name), reply_markup=__new_friend_reply_markup)
async def on_friend_added(self, fornite_friend: fortnitepy.friend.Friend) -> None:
# Try fetch stats
try:
await self.__try_get_stats(fornite_friend)
except:
print(traceback.format_exc())
await self.__on_unavailable_public_stats(fornite_friend.display_name)
friend = User.from_fortnite_friend(fornite_friend)
await self.__stats_repository.put_stats(friend, datetime.datetime.now())
await self.__telegram_bot.send_message_to_all(format_new_friend(friend.display_name))
async def __try_get_stats(self, friend: fortnitepy.friend.Friend):
await friend.fetch_br_stats()
async def __on_unavailable_public_stats(self, display_name) -> None:
await self.__telegram_bot.send_message_to_all(format_friend_disabled_public_stats(display_name))
class NewFriendCallbackQueryHandler(CallbackQueryHandler):
__telegram_bot: TelegramBot
__fortnite_client: FortniteClient
__new_friend_observer: NewFriendObserver
def __init__(self, telegram_bot: TelegramBot, fortnite_client: FortniteClient, new_friend_observer: NewFriendObserver):
self.__telegram_bot = telegram_bot
self.__fortnite_client = fortnite_client
self.__new_friend_observer = new_friend_observer
async def handle(self, call: telebot.types.CallbackQuery):
if self.__fortnite_client.is_initialized():
answered: bool = False
if call.data.startswith(__accept_keyword__) or call.data.startswith(__decline_keyword__):
splitted = call.data.split(":")
await self.__handle_request(splitted[0], splitted[1], splitted[2])
await self.__telegram_bot.answer_callback_query(callback_query_id=call.id)
async def __handle_request(self, action: str, request_id: str, display_name: str):
pending_friends = self.__fortnite_client.pending_friends
for pending_friend in pending_friends:
if isinstance(pending_friend, fortnitepy.friend.IncomingPendingFriend):
incoming_request = typing.cast(fortnitepy.friend.IncomingPendingFriend, pending_friend)
if incoming_request._id == request_id:
if action == __accept_keyword__:
await incoming_request.accept()
accepted_friend = incoming_request.client.get_friend(incoming_request._id)
await self.__new_friend_observer.on_friend_added(accepted_friend)
elif action == __decline_keyword__:
await incoming_request.decline()
await self.__telegram_bot.send_message_to_all(format_friend_declined(display_name))
|
