# Coverage for server/parts/archive.py : 100%

## 32 statements 32 run 0 missing 4 excluded 0 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

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

2 from __future__ import annotations

3

4# -- stdlib --

5 from typing import Any, Dict, TYPE_CHECKING

6 import base64

7 import json

8 import logging

9 import time

10 import zlib

11

12# -- third party --

13 import arrow

14

15# -- own --

16 from server.base import Game, HumanPlayer

17 import settings

18

19# -- typing --

20 if TYPE_CHECKING:

21 from server.core import Core # noqa: F401

22

23

24# -- code --

25 log = logging.getLogger('Archive')

26

27

28 class Archive(object):

29 def __init__(self, core: Core):

30 self.core = core

31 core.events.game_ended += self.handle_game_ended

32

33 def __repr__(self) -> str:

34 return self.__class__.__name__

35

36 def handle_game_ended(self, g: Game) -> Game:

37 core = self.core

38

39 meta = self._meta(g)

40 archive = self._archive(g)

41

42 core.backend.query('''

43 mutation ArchiveGame($game: GameInput!, $archive: String!) {

44 game {

45 archive(game: $meta, archive: $archive) {

46 gid

47 }

48 }

49 }

50 ''', game=meta, archive=archive)

51

52 return g

53

54 # ----- Methods -----

55

56 def _meta(self, g: Game) -> Dict[str, Any]:

57 core = self.core

58 start = core.room.start_time_of(g)

59

60 flags = dict(core.room.flags_of(g))

61 flags['crashed'] = core.game.is_crashed(g)

62 flags['aborted'] = core.game.is_aborted(g)

63

64 return {

65 'gid': core.room.gid_of(g),

66 'name': core.room.name_of(g),

67 'type': g.__class__.__name__,

68 'flags': flags,

69 'players': [core.auth.uid_of(u) for u in core.room.users_of(g)],

70 'winners': [core.auth.uid_of(p.client) if isinstance(p, HumanPlayer) else 0 for p in core.game.winners_of(g)],

71 'startedAt': arrow.get(start).to('Asia/Shanghai').isoformat(),

72 'duration': int(time.time() - start),

73 }

74

75 def _archive(self, g: Game) -> str:

76 core = self.core

77 data = {

78 'version': settings.VERSION,

79 'gid': core.room.gid_of(g),

80 'class': g.__class__.__name__,

81 'params': core.game.params_of(g),

82 'items': core.item.item_skus_of(g),

83 'rngseed': core.game.rngseed_of(g),

84 'players': [core.auth.uid_of(u) for u in core.room.users_of(g)],

85 'data': core.game.get_gamedata_archive(g),

86 }

87

88 return base64.b64encode(

89 zlib.compress(json.dumps(data).encode('utf-8'))

90 ).decode('utf-8')