1# -*- coding: utf-8 -*- 

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from collections import defaultdict 

6from typing import Dict, Optional, Set, TYPE_CHECKING, Tuple 

7import logging 

8 

9# -- third party -- 

10from mypy_extensions import TypedDict 

11 

12# -- own -- 

13from server.base import Game 

14from server.endpoint import Client 

15from server.utils import command 

16from utils.events import EventHub 

17import wire 

18 

19# -- typing -- 

20if TYPE_CHECKING: 

21 from server.core import Core # noqa: F401 

22 

23 

24# -- code -- 

25log = logging.getLogger('Invite') 

26 

27 

28class InviteAssocOnGame(TypedDict): 

29 invited: Set[int] 

30 banned: Dict[int, Set[int]] 

31 

32 

33def A(self: Invite, g: Game) -> InviteAssocOnGame: 

34 return g._[self] 

35 

36 

37class Invite(object): 

38 def __init__(self, core: Core): 

39 self.core = core 

40 

41 core.events.game_created += self.handle_game_created 

42 core.events.game_left += self.handle_game_left 

43 core.events.game_successive_create += self.handle_game_successive_create 

44 

45 D = core.events.client_command 

46 D[wire.Invite] += self._invite 

47 D[wire.Kick] += self._kick 

48 

49 D[wire.JoinRoom].subscribe(self._room_join_invite_limit, -3) 

50 

51 def __repr__(self) -> str: 

52 return self.__class__.__name__ 

53 

54 def handle_game_created(self, g: Game) -> Game: 

55 assoc: InviteAssocOnGame = { 

56 'invited': set(), 

57 'banned': defaultdict(set), 

58 } 

59 g._[self] = assoc 

60 return g 

61 

62 def handle_game_successive_create(self, ev: Tuple[Game, Game]) -> Tuple[Game, Game]: 

63 old, g = ev 

64 g._[self] = old._[self] 

65 return ev 

66 

67 def handle_game_left(self, ev: Tuple[Game, Client]) -> Tuple[Game, Client]: 

68 g, c = ev 

69 core = self.core 

70 

71 for bl in A(self, g)['banned'].values(): 

72 bl.discard(core.auth.uid_of(c)) 

73 

74 return ev 

75 

76 # ----- Commands ----- 

77 @command('*') 

78 def _room_join_invite_limit(self, u: Client, ev: wire.JoinRoom) -> Optional[EventHub.StopPropagation]: 

79 core = self.core 

80 g = core.room.get(ev.gid) 

81 if not g: 81 ↛ 82line 81 didn't jump to line 82, because the condition on line 81 was never true

82 return None 

83 

84 flags = core.room.flags_of(g) 

85 uid = core.auth.uid_of(u) 

86 

87 banned = A(self, g)['banned'] 

88 invited = A(self, g)['invited'] 

89 

90 # banned 

91 if len(banned[uid]) >= max(g.n_persons // 2, 1): 

92 u.write(wire.Error('banned')) 

93 return EventHub.STOP_PROPAGATION 

94 

95 # invite 

96 if flags.get('invite') and uid not in invited: 

97 u.write(wire.Error('not_invited')) 

98 return EventHub.STOP_PROPAGATION 

99 

100 return None 

101 

102 @command('room', 'ready') 

103 def _invite(self, u: Client, ev: wire.Invite) -> None: 

104 core = self.core 

105 ouid = ev.uid 

106 

107 other = core.lobby.get(ouid) 

108 if not (other and core.lobby.state_of(other) in ('lobby', 'ob')): 108 ↛ 109line 108 didn't jump to line 109, because the condition on line 108 was never true

109 return 

110 

111 g = core.game.current(u) 

112 assert g 

113 

114 A(self, g)['invited'].add(ouid) 

115 

116 other.write(wire.InviteRequest( 

117 uid=core.auth.uid_of(u), 

118 name=core.auth.name_of(u), 

119 gid=core.room.gid_of(g), 

120 type=g.__class__.__name__, 

121 )) 

122 

123 @command('room', 'ready') 

124 def _kick(self, u: Client, ev: wire.Kick) -> None: 

125 core = self.core 

126 ouid = ev.uid 

127 other = core.lobby.get(ouid) 

128 if not other: 128 ↛ 129line 128 didn't jump to line 129, because the condition on line 128 was never true

129 return 

130 

131 g = core.game.current(u) 

132 g2 = core.game.current(other) 

133 assert g 

134 assert g2 

135 

136 if g is not g2: 136 ↛ 137line 136 didn't jump to line 137, because the condition on line 136 was never true

137 return 

138 

139 if core.lobby.state_of(other) not in ('room', 'ready'): 139 ↛ 140line 139 didn't jump to line 140, because the condition on line 139 was never true

140 return 

141 

142 bl = A(self, g)['banned'][ouid] 

143 bl.add(core.auth.uid_of(u)) 

144 

145 for u in core.room.online_users_of(g): 

146 u.write(wire.KickRequest( 

147 uid=core.auth.uid_of(u), 

148 victim=core.auth.uid_of(other), 

149 votes=len(bl), 

150 )) 

151 

152 if len(bl) >= len(core.room.users_of(g)) // 2: 

153 A(self, g)['invited'].discard(ouid) 

154 core.room.exit_game(other) 

155 

156 # ----- Methods ----- 

157 def add_invited(self, g: Game, u: Client) -> None: 

158 core = self.core 

159 A(self, g)['invited'].add(core.auth.uid_of(u))