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
|
#!/usr/bin/python3
import os, time, telebot, asyncio, telebot.async_telebot, nest_asyncio, sys, logging, time
from FortniteStatusNotifier import *
from Formatter import *
from FortniteClient import *
from FortniteEvents import *
from persistence import UserRepository, StatsRepository, PresenceRepository
from TelegramBot import TelegramBot, CommandHandler
from Commands import *
class ClientInitObserver(ClientInit):
async def on_event(self) -> None:
# Accept pending friends
for friend_request in fortniteClient.incoming_pending_friends:
await fortniteClient.event_friend_request(friend_request)
# Record user stats
if len(statsRepository.getStats()) == 0:
await recordStatsCommand.handle(None)
class FortniteStatusObserver(Observer):
async def update(self, fortniteStatus) -> None:
await telegramBot.send_message_to_all(formatFortniteStatus(fortniteStatus))
class FortnitePresenceObserver(PresenceObserver):
async def update(self, display_name: str, playing: bool, party_size: int) -> None:
if playing:
last_presence = prensenceRepository.getLastUserPresence(display_name)
diff = time.time() - last_presence
if diff > 60 * 60: # 60 minutes
await self.__notifyFriendPlaying(display_name, party_size)
prensenceRepository.setLastUserPresence(display_name, time.time())
async def __notifyFriendPlaying(self, display_name: str, party_size: int):
await telegramBot.send_message_to_all(formatFriendOnline(display_name, party_size))
userRepository = UserRepository()
statsRepository = StatsRepository()
prensenceRepository = PresenceRepository()
telegramBot = TelegramBot(userRepository)
fortniteStatusWrapper = FortniteStatusNotifier(FortniteStatusObserver())
fortniteClient = FortniteClient(FortnitePresenceObserver(), ClientInitObserver())
recordStatsCommand = RecordStatsCommand(telegramBot, fortniteClient, statsRepository)
telegramBot.register_command_handler('start', StartCommand(telegramBot, userRepository))
telegramBot.register_command_handler('status', GetStatusCommand(telegramBot))
telegramBot.register_command_handler('friends', GetFriendsCommand(telegramBot, fortniteClient))
telegramBot.register_command_handler('stats', GetStatsCommand(telegramBot, fortniteClient))
telegramBot.register_command_handler('todaystats', GetTodayStatsCommand(telegramBot, fortniteClient, statsRepository))
telegramBot.register_command_handler('recordstats', recordStatsCommand)
async def run_tgbot():
await telegramBot.run()
async def run_fortniteStatusWrapper():
await fortniteStatusWrapper.run()
async def run_fortniteClient():
fortniteClient.run()
async def run_record_stats():
while True:
t = time.localtime()
if t.tm_hour == 5: # only at 05:00
await recordStatsCommand.handle(None)
await asyncio.sleep(60 * 60) # 1 hour
async def run_all():
await asyncio.gather(run_tgbot(), run_fortniteStatusWrapper(), run_fortniteClient(), run_record_stats())
if __name__ == '__main__':
nest_asyncio.apply()
loop = asyncio.get_event_loop()
loop.run_until_complete(run_all())
loop.close()
|
