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

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from collections import defaultdict 

6from typing import Dict, List, TYPE_CHECKING, Tuple 

7import logging 

8 

9# -- third party -- 

10from mypy_extensions import TypedDict 

11 

12# -- own -- 

13from game.base import GameItem 

14from server.base import Game 

15from server.endpoint import Client 

16from server.utils import command 

17from utils.misc import BusinessException 

18import wire 

19 

20# -- typing -- 

21if TYPE_CHECKING: 

22 from server.core import Core # noqa: F401 

23 

24 

25# -- code -- 

26log = logging.getLogger('server.parts.item') 

27 

28 

29class ItemAssocOnGame(TypedDict): 

30 items: Dict[int, List[GameItem]] 

31 

32 

33def A(self: Item, g: Game) -> ItemAssocOnGame: 

34 return g._[self] 

35 

36 

37class Item(object): 

38 def __init__(self, core: Core): 

39 self.core = core 

40 core.events.game_created += self.handle_game_created 

41 core.events.game_started += self.handle_game_started 

42 _ = core.events.client_command 

43 _[wire.UseItem] += self._use_item 

44 

45 def __repr__(self) -> str: 

46 return self.__class__.__name__ 

47 

48 def handle_game_started(self, g: Game) -> Game: 

49 core = self.core 

50 final: Dict[int, List[GameItem]] = {} 

51 

52 for uid, l in A(self, g)['items'].items(): 

53 consumed = [] 

54 for i in l: 

55 try: 

56 rst = core.backend.query(''' 

57 mutation($id: Int!, $sku: String, $r: String) { 

58 item { 

59 remove(player: $id, sku: $sku, reason: $r) 

60 } 

61 } 

62 ''', id=uid, sku=i.sku, reason="Use in game %s" % core.room.gid_of(g)) 

63 

64 if rst['item']['remove']: 64 ↛ 54line 64 didn't jump to line 54, because the condition on line 64 was never false

65 consumed.append(i) 

66 except Exception: 

67 log.exception('Error consuming item') 

68 

69 final[uid] = consumed 

70 

71 A(self, g)['items'] = final 

72 

73 return g 

74 

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

76 assoc: ItemAssocOnGame = { 

77 'items': defaultdict(list), 

78 } 

79 g._[self] = assoc 

80 return g 

81 

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

83 g, u = ev 

84 core = self.core 

85 if not core.room.is_started(g) and not g.ended: 

86 A(self, g)['items'].pop(core.auth.uid_of(u), None) 

87 

88 return ev 

89 

90 # ----- Command ----- 

91 @command('room') 

92 def _use_item(self, u: Client, ev: wire.UseItem) -> None: 

93 core = self.core 

94 g = core.game.current(u) 

95 assert g 

96 

97 try: 

98 uid = core.auth.uid_of(u) 

99 i = GameItem.from_sku(ev.sku) 

100 i.should_usable(core, g, u) 

101 

102 have_item = core.backend.query(''' 

103 query($uid: Int!, $sku: String!) { 

104 player(id: $id) { 

105 haveItem(sku: $sku) 

106 } 

107 } 

108 ''', uid=uid, sku=ev.sku)['player']['haveItem'] 

109 

110 if not have_item: 

111 from utils.misc import exceptions 

112 raise exceptions.ItemNotFound 

113 

114 A(self, g)['items'][uid].append(i) 

115 u.write(wire.Info('use_item_success')) 

116 except BusinessException as e: 

117 uid = core.auth.uid_of(u) 

118 log.error('User %s failed to use item %s: %s', uid, ev.sku, e.name) 

119 u.write(wire.Error(e.snake_case)) 

120 

121 # ----- Methods ------ 

122 # ----- Public Methods ----- 

123 def items_of(self, g: Game) -> Dict[int, List[GameItem]]: 

124 return A(self, g)['items'] 

125 

126 def item_skus_of(self, g: Game) -> Dict[int, List[str]]: 

127 return {k: [i.sku for i in v] for k, v in self.items_of(g).items()}