ige.BookingMngr.Booking.answer() - Code Metrics - ospaceteam/outerspace - Measure and Improve Code Quality continuously with Scrutinizer

ige.BookingMngr.Booking.answer() A
last analyzed 2021-09-11 21:46 UTC

↳ Parent: ige.BookingMngr

Complexity

Conditions

Size

Total Lines	9
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	2
dl	0
loc	9
rs	9.95
c	0
b	0
f	0

#
#  Copyright 2001 - 2016 Ludek Smid [http://www.ospace.net/]
#
#  This file is part of Outer Space.
#
#  Outer Space is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 2 of the License, or
#  (at your option) any later version.
#
#  Outer Space is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with Outer Space; if not, write to the Free Software
#  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
#

import hashlib
import os
import random
import time

import data
import ige
import log

from ige.ClientMngr import Session
from ige.ospace import Const
from ige.ospace import Utils

from ige import GameException, BookingMngrException
from ige.Transaction import Transaction
from ige.IDataHolder import IDataHolder
from ige.ospace.GalaxyGenerator import GalaxyGenerator

class Booking(object):
    def __init__(self, gal_type):
        self.players = set([])
        self.gal_type = gal_type
        self.last_creation = None
        self.capacity = None
        self.owner = None
        self.owner_nick = None
        self.pw_salt = hashlib.sha256(str(random.random())).hexdigest()
        self.pw_hash = None

    def toggle_booking(self, player):
        try:
            self.players.remove(player)
        except KeyError:
            self.players.add(player)
        return self.is_filled()

    def discard_booking(self, player):
        self.players.discard(player)

    def is_filled(self):
        return len(self.players) == self.capacity

    def set_password(self, password):
        self.pw_hash = hashlib.sha256(password + self.pw_salt).hexdigest()

    def check_password(self, password):
        pw_hash = hashlib.sha256(password + self.pw_salt).hexdigest()
        return self.pw_hash == pw_hash

    def answer(self, player):
        answer = IDataHolder()
        answer.bookings = len(self.players)
        answer.last_creation = self.last_creation
        answer.is_booked = player in self.players
        answer.owner_nick = self.owner_nick
        answer.gal_type = self.gal_type
        answer.capacity = self.capacity
        return answer

class BookingMngr(object):
    def __init__(self, clientMngr, gameMngr, db):
        self.clientMngr = clientMngr
        self.gameMngr = gameMngr
        self.db = db
        self.db.nextID = ige.Const.BID_FREESTART
        self.offerings = GalaxyGenerator().templates
        self.init_bookings()

    def _create_booking(self, gal_type):
        book = Booking(gal_type)
        if self.offerings[gal_type].scenario == Const.SCENARIO_SINGLE:
            # hardcoded 1, as some galaxies has rebels in them
            book.capacity = 1
        else:
            book.capacity = self.offerings[gal_type].players
        return book

    def init_bookings(self):
        for gal_type in self.offerings:
            bookings = self._get_type_bookings(gal_type)
            if not bookings:
                book = self._create_booking(gal_type)
                self.db.create(book)
                bookings.append(book)

        # cleanup of those not used anymore
        for bookID in self.db.keys():
            gal_type = self.db[bookID].gal_type
            if gal_type not in self.offerings:
                del self.db[bookID]

    def shutdown(self):
        log.message('Shutdown')
        self.db.shutdown()

    def checkpoint(self):
        self.db.checkpoint()

    def clear(self):
        self.db.clear()

    def backup(self, basename):
        self.db.backup(basename)

    def upgrade(self):
        return

    def _get_type_bookings(self, gal_type):
        bookings = []
        for bookID in self.db.keys():
            book = self.db[bookID]
            if book.gal_type == gal_type:
                bookings.append(book)
        return bookings

    def _get_challenges(self, template):
        challenges = []
        if template.scenario == Const.SCENARIO_SINGLE and template.players > 1:
            challenges.append(Const.T_AIPLAYER)
        if set([Const.SR_TL1A, Const.SR_TL1B]).intersection(template.resources):
            challenges.append(Const.T_AIRENPLAYER)
        if set([Const.DISEASE_MUTANT]).intersection(template.diseases):
            challenges.append(Const.T_AIMUTPLAYER)
        if set([Const.SR_TL3A, Const.SR_TL3B, Const.SR_TL3C]).intersection(template.resources):
            challenges.append(Const.T_AIPIRPLAYER)
        if set([Const.SR_TL5A, Const.SR_TL5B, Const.SR_TL5C]).intersection(template.resources):
            challenges.append(Const.T_AIEDENPLAYER)
        return challenges

    def _create_answer(self, gal_type):
        template = self.offerings[gal_type]
        answer = IDataHolder()
        answer.scenario = template.scenario
        answer.minPlanets = template.minPlanets
        answer.maxPlanets = template.maxPlanets
        answer.radius = template.radius
        answer.players = template.players
        answer.resources = template.resources.keys()
        answer.challenges = self._get_challenges(template)

        if not template.startR[0] or not template.startR[1]:
            # that means grouping is used to maximize distance between players
            # most likely brawl scenario
            answer.playerGroup = 1
        else:
            answer.playerGroup = template.playerGroup
        return answer

    def _is_valid_offer(self, sid, gal_type):
        hideSingle = self.gameMngr.singleGamesLimit(sid)
        template = self.offerings[gal_type]
        if gal_type == "Test":
            return False
        if hideSingle and template.scenario == Const.SCENARIO_SINGLE:
            return False
        return True

    def get_booking_offers(self, sid):
        answers = {}
        for gal_type in self.offerings:
            if not self._is_valid_offer(sid, gal_type):
                continue
            answers[gal_type] = self._create_answer(gal_type)
        return answers, None

    def _get_booking_answers(self, sid):
        login = self.clientMngr.getSession(sid).login
        answers = {}
        for bookID in self.db.keys():
            book = self.db[bookID]
            if not self._is_valid_offer(sid, book.gal_type):
                continue
            answers[bookID] = book.answer(login)
        return answers

    def get_booking_answers(self, sid):
        return self._get_booking_answers(sid), None

    def toggle_booking(self, sid, bookID, password):
        session = self.clientMngr.getSession(sid)
        book = self.db[bookID]
        template = self.offerings[book.gal_type]
        if book.owner:
            # private booking
            if book.owner == session.login:
                raise BookingMngrException('Owners cannot toggle their own private bookings.')
            if not book.check_password(password):
                raise BookingMngrException('Incorrect password.')
        if not self._is_valid_offer(sid, book.gal_type):
            # limit reached, no toggling allowed
            raise BookingMngrException('Cannot toggle, limit of single player galaxies reached?')
        log.debug("Player '{0}' toggled booking of '{1}/{2}'".format(session.login, book.gal_type, book.owner))
        triggered = False
        if book.toggle_booking(session.login):
            self.trigger_galaxy_start(bookID)
            triggered = True
        answers = self._get_booking_answers(sid)
        answers[ige.Const.BID_TRIGGERED] = triggered
        return answers, None

    def _private_bookings_limit(self, login):
        no_books = 0
        for bookID in self.db.keys():
            book = self.db[bookID]
            if book.owner == login:
                no_books += 1
        return no_books >= Const.BOOKING_PRIVATE_LIMIT

    def create_private_booking(self, sid, bookID, password):
        session = self.clientMngr.getSession(sid)
        if not password:
            raise BookingMngrException('Password is needed for private bookings.')
        sourceBook = self.db[bookID]
        scenario = self.offerings[sourceBook.gal_type].scenario
        if scenario == Const.SCENARIO_SINGLE:
            raise BookingMngrException('Single scenarios cannot be privately booked.')
        if self._private_bookings_limit(session.login):
            raise BookingMngrException('Limit of private bookings reached.')
        book = self._create_booking(sourceBook.gal_type)
        book.owner = session.login
        book.owner_nick = session.nick
        book.players.add(session.login)
        book.set_password(password)
        self.db.create(book)
        return self.get_booking_answers(sid)

    def delete_private_booking(self, sid, bookID):
        login = self.clientMngr.getSession(sid).login
        book = self.db[bookID]
        if not book.owner == login:
            raise BookingMngrException('Only the owner can delete private booking.')
        del self.db[bookID]
        return self.get_booking_answers(sid)

    def trigger_galaxy_start(self, bookID):
        book = self.db[bookID]
        log.debug("Triggering new subscribed galaxy '{0}'".format(book.gal_type))
        # triggers galaxy start regardless booking numbers

        # TODO: I don't know what I am doing, this seems to be a terrible hack
        # but right now, I don't see a way without faking admin access
        universe = self.gameMngr.db[ige.Const.OID_UNIVERSE]
        tran = Transaction(self.gameMngr, ige.Const.OID_ADMIN)
        session = Session(None)
        session.cid = ige.Const.OID_ADMIN
        tran.session = session
        name = self._find_galaxy_name(book.gal_type, list(book.players))
        new_galaxy_ID = self.gameMngr.cmdPool[universe.type].createNewSubscribedGalaxy(tran, universe, name, book.gal_type, list(book.players))

        if book.owner:
            del self.db[bookID]
        else:
            book.players = set([])
            book.last_creation = time.time()

        return new_galaxy_ID, None

    def _find_galaxy_name(self, gal_type, players):
        scenario = self.offerings[gal_type].scenario
        universe = self.gameMngr.db[ige.Const.OID_UNIVERSE]
        names_in_use = set([])
        # fetch list of used galaxy names
        for galaxy_id in universe.galaxies:
            galaxy = self.gameMngr.db[galaxy_id]
            names_in_use.add(galaxy.name)

        if scenario == Const.SCENARIO_SINGLE:
            player_name = players[0]
            name = "{0}-{1}-{2}".format(gal_type, player_name, universe.turn)
            if name not in names_in_use:
                return name
            iterator = 1
            while True:
                name = "{0}-{1}-{2}.{3}".format(gal_type, player_name, universe.turn, iterator)
                if name in names_in_use:
                    iterator += 1
                    continue
                # found new name!
                return name
        elif scenario in [Const.SCENARIO_COOP, Const.SCENARIO_BRAWL]:
            name = "{0}-{1}".format(gal_type, universe.turn)
            if name not in names_in_use:
                return name
            iterator = 1
            while True:
                name = "{0}-{1}.{2}".format(gal_type, universe.turn, iterator)
                if name in names_in_use:
                    iterator += 1
                    continue
                # found new name!
                return name
        # Outer Space has privilege of having traditional galaxy names
        log.debug("Searching for available galaxy name")
        all_names = []
        for name in open(data.GALAXY_NAMES_FILE):
            all_names.append(name.strip())

        for name in all_names:
            if name in names_in_use:
                continue
            return name
        # no name available
        return None



1			#
2			# Copyright 2001 - 2016 Ludek Smid [http://www.ospace.net/]
3			#
4			# This file is part of Outer Space.
5			#
6			# Outer Space is free software; you can redistribute it and/or modify
7			# it under the terms of the GNU General Public License as published by
8			# the Free Software Foundation; either version 2 of the License, or
9			# (at your option) any later version.
10			#
11			# Outer Space is distributed in the hope that it will be useful,
12			# but WITHOUT ANY WARRANTY; without even the implied warranty of
13			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			# GNU General Public License for more details.
15			#
16			# You should have received a copy of the GNU General Public License
17			# along with Outer Space; if not, write to the Free Software
18			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19			#
20
21			import hashlib
22			import os
23			import random
24			import time
25
26			import data
27			import ige
28			import log
29
30			from ige.ClientMngr import Session
31			from ige.ospace import Const
32			from ige.ospace import Utils
33
34			from ige import GameException, BookingMngrException
35			from ige.Transaction import Transaction
36			from ige.IDataHolder import IDataHolder
37			from ige.ospace.GalaxyGenerator import GalaxyGenerator
38
39			class Booking(object):
40			def __init__(self, gal_type):
41			self.players = set([])
42			self.gal_type = gal_type
43			self.last_creation = None
44			self.capacity = None
45			self.owner = None
46			self.owner_nick = None
47			self.pw_salt = hashlib.sha256(str(random.random())).hexdigest()
48			self.pw_hash = None
49
50			def toggle_booking(self, player):
51			try:
52			self.players.remove(player)
53			except KeyError:
54			self.players.add(player)
55			return self.is_filled()
56
57			def discard_booking(self, player):
58			self.players.discard(player)
59
60			def is_filled(self):
61			return len(self.players) == self.capacity
62
63			def set_password(self, password):
64			self.pw_hash = hashlib.sha256(password + self.pw_salt).hexdigest()
65
66			def check_password(self, password):
67			pw_hash = hashlib.sha256(password + self.pw_salt).hexdigest()
68			return self.pw_hash == pw_hash
69
70			def answer(self, player):
71			answer = IDataHolder()
72			answer.bookings = len(self.players)
73			answer.last_creation = self.last_creation
74			answer.is_booked = player in self.players
75			answer.owner_nick = self.owner_nick
76			answer.gal_type = self.gal_type
77			answer.capacity = self.capacity
78			return answer
79
80			class BookingMngr(object):
81			def __init__(self, clientMngr, gameMngr, db):
82			self.clientMngr = clientMngr
83			self.gameMngr = gameMngr
84			self.db = db
85			self.db.nextID = ige.Const.BID_FREESTART
86			self.offerings = GalaxyGenerator().templates
87			self.init_bookings()
88
89			def _create_booking(self, gal_type):
90			book = Booking(gal_type)
91			if self.offerings[gal_type].scenario == Const.SCENARIO_SINGLE:
92			# hardcoded 1, as some galaxies has rebels in them
93			book.capacity = 1
94			else:
95			book.capacity = self.offerings[gal_type].players
96			return book
97
98			def init_bookings(self):
99			for gal_type in self.offerings:
100			bookings = self._get_type_bookings(gal_type)
101			if not bookings:
102			book = self._create_booking(gal_type)
103			self.db.create(book)
104			bookings.append(book)
105
106			# cleanup of those not used anymore
107			for bookID in self.db.keys():
108			gal_type = self.db[bookID].gal_type
109			if gal_type not in self.offerings:
110			del self.db[bookID]
111
112			def shutdown(self):
113			log.message('Shutdown')
114			self.db.shutdown()
115
116			def checkpoint(self):
117			self.db.checkpoint()
118
119			def clear(self):
120			self.db.clear()
121
122			def backup(self, basename):
123			self.db.backup(basename)
124
125			def upgrade(self):
126			return
127
128			def _get_type_bookings(self, gal_type):
129			bookings = []
130			for bookID in self.db.keys():
131			book = self.db[bookID]
132			if book.gal_type == gal_type:
133			bookings.append(book)
134			return bookings
135
136			def _get_challenges(self, template):
137			challenges = []
138			if template.scenario == Const.SCENARIO_SINGLE and template.players > 1:
139			challenges.append(Const.T_AIPLAYER)
140			if set([Const.SR_TL1A, Const.SR_TL1B]).intersection(template.resources):
141			challenges.append(Const.T_AIRENPLAYER)
142			if set([Const.DISEASE_MUTANT]).intersection(template.diseases):
143			challenges.append(Const.T_AIMUTPLAYER)
144			if set([Const.SR_TL3A, Const.SR_TL3B, Const.SR_TL3C]).intersection(template.resources):
145			challenges.append(Const.T_AIPIRPLAYER)
146			if set([Const.SR_TL5A, Const.SR_TL5B, Const.SR_TL5C]).intersection(template.resources):
147			challenges.append(Const.T_AIEDENPLAYER)
148			return challenges
149
150			def _create_answer(self, gal_type):
151			template = self.offerings[gal_type]
152			answer = IDataHolder()
153			answer.scenario = template.scenario
154			answer.minPlanets = template.minPlanets
155			answer.maxPlanets = template.maxPlanets
156			answer.radius = template.radius
157			answer.players = template.players
158			answer.resources = template.resources.keys()
159			answer.challenges = self._get_challenges(template)
160
161			if not template.startR[0] or not template.startR[1]:
162			# that means grouping is used to maximize distance between players
163			# most likely brawl scenario
164			answer.playerGroup = 1
165			else:
166			answer.playerGroup = template.playerGroup
167			return answer
168
169			def _is_valid_offer(self, sid, gal_type):
170			hideSingle = self.gameMngr.singleGamesLimit(sid)
171			template = self.offerings[gal_type]
172			if gal_type == "Test":
173			return False
174			if hideSingle and template.scenario == Const.SCENARIO_SINGLE:
175			return False
176			return True
177
178			def get_booking_offers(self, sid):
179			answers = {}
180			for gal_type in self.offerings:
181			if not self._is_valid_offer(sid, gal_type):
182			continue
183			answers[gal_type] = self._create_answer(gal_type)
184			return answers, None
185
186			def _get_booking_answers(self, sid):
187			login = self.clientMngr.getSession(sid).login
188			answers = {}
189			for bookID in self.db.keys():
190			book = self.db[bookID]
191			if not self._is_valid_offer(sid, book.gal_type):
192			continue
193			answers[bookID] = book.answer(login)
194			return answers
195
196			def get_booking_answers(self, sid):
197			return self._get_booking_answers(sid), None
198
199			def toggle_booking(self, sid, bookID, password):
200			session = self.clientMngr.getSession(sid)
201			book = self.db[bookID]
202			template = self.offerings[book.gal_type]
203			if book.owner:
204			# private booking
205			if book.owner == session.login:
206			raise BookingMngrException('Owners cannot toggle their own private bookings.')
207			if not book.check_password(password):
208			raise BookingMngrException('Incorrect password.')
209			if not self._is_valid_offer(sid, book.gal_type):
210			# limit reached, no toggling allowed
211			raise BookingMngrException('Cannot toggle, limit of single player galaxies reached?')
212			log.debug("Player '{0}' toggled booking of '{1}/{2}'".format(session.login, book.gal_type, book.owner))
213			triggered = False
214			if book.toggle_booking(session.login):
215			self.trigger_galaxy_start(bookID)
216			triggered = True
217			answers = self._get_booking_answers(sid)
218			answers[ige.Const.BID_TRIGGERED] = triggered
219			return answers, None
220
221			def _private_bookings_limit(self, login):
222			no_books = 0
223			for bookID in self.db.keys():
224			book = self.db[bookID]
225			if book.owner == login:
226			no_books += 1
227			return no_books >= Const.BOOKING_PRIVATE_LIMIT
228
229			def create_private_booking(self, sid, bookID, password):
230			session = self.clientMngr.getSession(sid)
231			if not password:
232			raise BookingMngrException('Password is needed for private bookings.')
233			sourceBook = self.db[bookID]
234			scenario = self.offerings[sourceBook.gal_type].scenario
235			if scenario == Const.SCENARIO_SINGLE:
236			raise BookingMngrException('Single scenarios cannot be privately booked.')
237			if self._private_bookings_limit(session.login):
238			raise BookingMngrException('Limit of private bookings reached.')
239			book = self._create_booking(sourceBook.gal_type)
240			book.owner = session.login
241			book.owner_nick = session.nick
242			book.players.add(session.login)
243			book.set_password(password)
244			self.db.create(book)
245			return self.get_booking_answers(sid)
246
247			def delete_private_booking(self, sid, bookID):
248			login = self.clientMngr.getSession(sid).login
249			book = self.db[bookID]
250			if not book.owner == login:
251			raise BookingMngrException('Only the owner can delete private booking.')
252			del self.db[bookID]
253			return self.get_booking_answers(sid)
254
255			def trigger_galaxy_start(self, bookID):
256			book = self.db[bookID]
257			log.debug("Triggering new subscribed galaxy '{0}'".format(book.gal_type))
258			# triggers galaxy start regardless booking numbers
259
260			# TODO: I don't know what I am doing, this seems to be a terrible hack
261			# but right now, I don't see a way without faking admin access
262			universe = self.gameMngr.db[ige.Const.OID_UNIVERSE]
263			tran = Transaction(self.gameMngr, ige.Const.OID_ADMIN)
264			session = Session(None)
265			session.cid = ige.Const.OID_ADMIN
266			tran.session = session
267			name = self._find_galaxy_name(book.gal_type, list(book.players))
268			new_galaxy_ID = self.gameMngr.cmdPool[universe.type].createNewSubscribedGalaxy(tran, universe, name, book.gal_type, list(book.players))
269
270			if book.owner:
271			del self.db[bookID]
272			else:
273			book.players = set([])
274			book.last_creation = time.time()
275
276			return new_galaxy_ID, None
277
278			def _find_galaxy_name(self, gal_type, players):
279			scenario = self.offerings[gal_type].scenario
280			universe = self.gameMngr.db[ige.Const.OID_UNIVERSE]
281			names_in_use = set([])
282			# fetch list of used galaxy names
283			for galaxy_id in universe.galaxies:
284			galaxy = self.gameMngr.db[galaxy_id]
285			names_in_use.add(galaxy.name)
286
287			if scenario == Const.SCENARIO_SINGLE:
288			player_name = players[0]
289			name = "{0}-{1}-{2}".format(gal_type, player_name, universe.turn)
290			if name not in names_in_use:
291			return name
292			iterator = 1
293			while True:
294			name = "{0}-{1}-{2}.{3}".format(gal_type, player_name, universe.turn, iterator)
295			if name in names_in_use:
296			iterator += 1
297			continue
298			# found new name!
299			return name
300			elif scenario in [Const.SCENARIO_COOP, Const.SCENARIO_BRAWL]:
301			name = "{0}-{1}".format(gal_type, universe.turn)
302			if name not in names_in_use:
303			return name
304			iterator = 1
305			while True:
306			name = "{0}-{1}.{2}".format(gal_type, universe.turn, iterator)
307			if name in names_in_use:
308			iterator += 1
309			continue
310			# found new name!
311			return name
312			# Outer Space has privilege of having traditional galaxy names
313			log.debug("Searching for available galaxy name")
314			all_names = []
315			for name in open(data.GALAXY_NAMES_FILE):
316			all_names.append(name.strip())
317
318			for name in all_names:
319			if name in names_in_use:
320			continue
321			return name
322			# no name available
323			return None
324
325

ospaceteam / outerspace

ige.BookingMngr.Booking.answer() A last analyzed 2021-09-11 21:46 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ige.BookingMngr.Booking.answer() A
last analyzed 2021-09-11 21:46 UTC