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)
60 flags = dict(core.room.flags_of(g))
61 flags['crashed'] = core.game.is_crashed(g)
62 flags['aborted'] = core.game.is_aborted(g)
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 }
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 }
88 return base64.b64encode(
89 zlib.compress(json.dumps(data).encode('utf-8'))
90 ).decode('utf-8')