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

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from collections import defaultdict 

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

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) -> Union[wire.JoinRoom, 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 ev 

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 ev 

101 

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

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

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 

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

114 

115 other.write(wire.InviteRequest( 

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

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

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

119 type=g.__class__.__name__, 

120 )) 

121 

122 return ev 

123 

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

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

126 core = self.core 

127 ouid = ev.uid 

128 other = core.lobby.get(ouid) 

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

130 return 

131 

132 g = core.game.current(u) 

133 g2 = core.game.current(other) 

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

135 return 

136 

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

138 return 

139 

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

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

142 

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

144 u.write(wire.KickRequest( 

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

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

147 votes=len(bl), 

148 )) 

149 

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

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

152 core.room.exit_game(other) 

153 

154 return ev 

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))