Coverage for server/parts/archive.py : 100%
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 -*-
2from __future__ import annotations
4# -- stdlib --
5from typing import Any, Dict, TYPE_CHECKING
6import base64
7import json
8import logging
9import time
10import zlib
12# -- third party --
13import arrow
15# -- own --
16from server.base import Game, HumanPlayer
17import settings
19# -- typing --
20if TYPE_CHECKING:
21 from server.core import Core # noqa: F401
24# -- code --
25log = logging.getLogger('Archive')
28class Archive(object):
29 def __init__(self, core: Core):
30 self.core = core
31 core.events.game_ended += self.handle_game_ended
33 def __repr__(self) -> str:
34 return self.__class__.__name__
36 def handle_game_ended(self, g: Game) -> Game:
37 core = self.core
39 meta = self._meta(g)
40 archive = self._archive(g)
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)
52 return g
54 # ----- Methods -----
56 def _meta(self, g: Game) -> Dict[str, Any]:
57 core = self.core
58 start = core.room.start_time_of(g)
59 flags = core.room.flags_of(g)
61 if core.game.is_crashed(g):
62 flags['crashed'] = True
64 if core.game.is_aborted(g):
65 flags['aborted'] = True
67 return {
68 'gid': core.room.gid_of(g),
69 'name': core.room.name_of(g),
70 'type': g.__class__.__name__,
71 'flags': flags,
72 'players': [core.auth.uid_of(u) for u in core.room.users_of(g)],
73 'winners': [core.auth.uid_of(p.client) if isinstance(p, HumanPlayer) else 0 for p in core.game.winners_of(g)],
74 'startedAt': arrow.get(start).to('Asia/Shanghai').isoformat(),
75 'duration': int(time.time() - start),
76 }
78 def _archive(self, g: Game) -> str:
79 core = self.core
80 data = {
81 'version': settings.VERSION,
82 'gid': core.room.gid_of(g),
83 'class': g.__class__.__name__,
84 'params': core.game.params_of(g),
85 'items': core.item.item_skus_of(g),
86 'rngseed': core.game.rngseed_of(g),
87 'players': [core.auth.uid_of(u) for u in core.room.users_of(g)],
88 'data': core.game.get_gamedata_archive(g),
89 }
91 return base64.b64encode(
92 zlib.compress(json.dumps(data).encode('utf-8'))
93 ).decode('utf-8')