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

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import Any, Callable, List, TYPE_CHECKING, Tuple, TypeVar 

6import logging 

7 

8# -- third party -- 

9# -- own -- 

10from server.endpoint import Client 

11from utils.events import EventHub 

12from wire import msg 

13 

14# -- typing -- 

15if TYPE_CHECKING: 

16 from server.core import Core # noqa: F401 

17 

18 

19# -- code -- 

20log = logging.getLogger('Admin') 

21T = TypeVar('T', bound=msg.Message) 

22STOP = EventHub.STOP_PROPAGATION 

23 

24 

25def _need_admin(f: Callable[[Admin, Client, T], Any]) -> Callable[[Admin, Tuple[Client, T]], EventHub.StopPropagation]: 

26 def wrapper(self: Admin, ev: Tuple[Client, T]) -> EventHub.StopPropagation: 

27 core = self.core 

28 u, m = ev 

29 if core.auth.uid_of(u) not in self.admins: 

30 return STOP 

31 

32 f(self, u, m) 

33 u.write(msg.SystemMsg(msg='成功的执行了管理命令')) 

34 return STOP 

35 

36 return wrapper 

37 

38 

39class Admin(object): 

40 def __init__(self, core: Core): 

41 self.core = core 

42 

43 D = core.events.client_command 

44 D[msg.AdminStacktrace] += self._stacktrace 

45 D[msg.AdminClearZombies] += self._clearzombies 

46 D[msg.AdminMigrate] += self._migrate 

47 D[msg.AdminKick] += self._kick 

48 D[msg.AdminKillGame] += self._kill_game 

49 D[msg.AdminAdd] += self._add 

50 D[msg.AdminRemove] += self._remove 

51 D[msg.AdminAddBigbrother] += self._add_bigbrother 

52 D[msg.AdminRemoveBigbrother] += self._remove_bigbrother 

53 

54 self.admins: List[int] = [2, 109, 351, 3044, 6573, 6584, 9783] 

55 

56 def __repr__(self) -> str: 

57 return self.__class__.__name__ 

58 

59 @_need_admin 

60 def _kick(self, c: Client, m: msg.AdminKick) -> None: 

61 core = self.core 

62 u = core.lobby.get(m.uid) 

63 u and u.terminate() 

64 

65 @_need_admin 

66 def _clearzombies(self, c: Client, m: msg.AdminClearZombies) -> None: 

67 core = self.core 

68 users = core.lobby.all_users() 

69 for u in users: 

70 if u.is_dead(): 70 ↛ 71line 70 didn't jump to line 71, because the condition on line 70 was never true

71 log.info('Clear zombie: %r', u) 

72 core.events.client_dropped.emit(u) 

73 

74 @_need_admin 

75 def _migrate(self, c: Client, m: msg.AdminMigrate) -> None: 

76 core = self.core 

77 

78 @core.runner.spawn 

79 def sysmsg() -> None: 

80 while True: 

81 users = core.lobby.all_users() 

82 users.write(msg.SystemMsg(msg='游戏已经更新，当前的游戏结束后将会被自动踢出，请更新后重新游戏')) 

83 core.runner.sleep(300) 

84 

85 @core.runner.spawn 

86 def kick() -> None: 

87 core.runner.sleep(30) 

88 while True: 

89 users = core.lobby.all_users() 

90 for u in users: 

91 if core.lobby.state_of(u) in ('lobby', 'room', 'ready', 'connected'): 

92 u.terminate() 

93 

94 core.runner.sleep(1) 

95 

96 @_need_admin 

97 def _stacktrace(self, c: Client, m: msg.AdminStacktrace) -> None: 

98 core = self.core 

99 g = core.game.current(c) 

100 if not g: 100 ↛ 103line 100 didn't jump to line 103, because the condition on line 100 was never false

101 return 

102 

103 log.info('>>>>> GAME STACKTRACE <<<<<') 

104 

105 def logtraceback(f: Any) -> None: 

106 import traceback 

107 log.info('----- %r -----\n%s', gr, ''.join(traceback.format_stack(f))) 

108 

109 logtraceback(g) 

110 

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

112 gr = u.get_greenlet() 

113 if gr and gr.gr_frame: 

114 logtraceback(gr.gr_frame) 

115 

116 log.info('===========================') 

117 

118 @_need_admin 

119 def _kill_game(self, c: Client, m: msg.AdminKillGame) -> None: 

120 core = self.core 

121 g = core.room.get(m.gid) 

122 if not g: return 122 ↛ exitline 122 didn't return from function '_kill_game', because the return on line 122 wasn't executed

123 

124 users = core.room.online_users_of(g) 

125 for u in users: 

126 core.room.exit_game(u) 

127 

128 @_need_admin 

129 def _add(self, c: Client, m: msg.AdminAdd) -> None: 

130 self.admins.append(m.uid) 

131 

132 @_need_admin 

133 def _remove(self, c: Client, m: msg.AdminRemove) -> None: 

134 try: 

135 self.admins.remove(m.uid) 

136 except Exception: 

137 pass 

138 

139 @_need_admin 

140 def _add_bigbrother(self, c: Client, m: msg.AdminAddBigbrother) -> None: 

141 core = self.core 

142 core.observe.add_bigbrother(m.uid) 

143 

144 @_need_admin 

145 def _remove_bigbrother(self, c: Client, m: msg.AdminRemoveBigbrother) -> None: 

146 core = self.core 

147 core.observe.remove_bigbrother(m.uid)