import datetime
import json
import math
import re
import time
import arrow
import requests
import random
from ogame import constants
from ogame.errors import BAD_UNIVERSE_NAME, BAD_DEFENSE_ID, NOT_LOGGED, BAD_CREDENTIALS, CANT_PROCESS, BAD_BUILDING_ID, BAD_SHIP_ID, BAD_RESEARCH_ID
from bs4 import BeautifulSoup
from dateutil import tz
def parse_int(text):
return int(text.replace('.', '').replace(',', '').strip())
def for_all_methods(decorator):
def decorate(cls):
for attr in cls.__dict__: # there's propably a better way to do this
if callable(getattr(cls, attr)):
setattr(cls, attr, retry_if_logged_out(decorator(getattr(cls, attr))))
return cls
return decorate
def sandbox_decorator(some_fn):
def wrapper(ogame, *args, **kwargs):
fn_name = some_fn.__name__
local_fns = ['get_datetime_from_time']
if fn_name in local_fns:
return some_fn(ogame, *args, **kwargs)
if fn_name == '__init__' or not ogame.sandbox:
return some_fn(ogame, *args, **kwargs)
if fn_name in ogame.sandbox_obj:
return ogame.sandbox_obj[fn_name]
return None
return wrapper
def retry_if_logged_out(method):
def wrapper(self, *args, **kwargs):
attempt = 0
time_to_sleep = 0
working = False
while not working:
try:
working = True
res = method(self, *args, **kwargs)
except NOT_LOGGED:
time.sleep(time_to_sleep)
attempt += 1
time_to_sleep += 1
if attempt > 5:
raise CANT_PROCESS
working = False
self.login()
return res
return wrapper
def get_nbr(soup, name):
div = soup.find('div', {'class': name})
level = div.find('span', {'class': 'level'})
for tag in level.findAll(True):
tag.extract()
return parse_int(level.text)
def metal_mine_production(level, universe_speed=1):
return int(math.floor(30 * level * 1.1 ** level) * universe_speed)
def get_planet_infos_regex(text):
result = re.search(r'(\w+) \[(\d+):(\d+):(\d+)\]([\d\.]+)km \((\d+)/(\d+)\)([-\d]+).+C (?:bis|to|à) ([-\d]+).+C', text)
if result is not None :
return result #is a plenet
else :
return re.search(r'(\w+) \[(\d+):(\d+):(\d+)\]([\d\.]+)km \((\d+)/(\d+)\)', text) #is a moon
def get_code(name):
if name in constants.Buildings.keys():
return constants.Buildings[name]
if name in constants.Facilities.keys():
return constants.Facilities[name]
if name in constants.Defense.keys():
return constants.Defense[name]
if name in constants.Ships.keys():
return constants.Ships[name]
if name in constants.Research.keys():
return constants.Research[name]
print('Couldn\'t find code for {}'.format(name))
return None
@for_all_methods(sandbox_decorator)
class OGame(object):
def __init__(self, universe, username, password, domain='en.ogame.gameforge.com', auto_bootstrap=True, sandbox=False, sandbox_obj=None):
self.session = requests.session()
self.session.headers.update({'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36'})
self.sandbox = sandbox
self.sandbox_obj = sandbox_obj if sandbox_obj is not None else {}
self.universe = universe
self.domain = domain
self.username = username
self.password = password
self.universe_speed = 1
self.server_url = ''
self.server_tz = 'GMT+1'
if auto_bootstrap:
self.login()
self.universe_speed = self.get_universe_speed()
def login(self):
"""Get the ogame session token."""
payload = {'kid': '',
'language': 'en',
'autologin': 'false',
'credentials[email]': self.username,
'credentials[password]': self.password}
time.sleep(random.uniform(1, 2))
res = self.session.post('https://lobby-api.ogame.gameforge.com/users', data=payload)
php_session_id = None
for c in res.cookies:
if c.name == 'PHPSESSID':
php_session_id = c.value
break
cookie = {'PHPSESSID': php_session_id}
res = self.session.get('https://lobby-api.ogame.gameforge.com/servers').json()
server_num = None
for server in res:
name = server['name'].lower()
if self.universe.lower() == name:
server_num = server['number']
break
res = self.session.get('https://lobby-api.ogame.gameforge.com/users/me/accounts', cookies=cookie)
selected_server_id = None
lang = None
server_accounts = res.json()
for server_account in server_accounts:
if server_account['server']['number'] == server_num:
lang = server_account['server']['language']
selected_server_id = server_account['id']
break
time.sleep(random.uniform(1, 2))
res = self.session.get('https://lobby-api.ogame.gameforge.com/users/me/loginLink?id={}&server[language]={}&server[number]={}'
.format(selected_server_id, lang, str(server_num)), cookies=cookie).json()
selected_server_url = res['url']
b = re.search('https://(.+\.ogame\.gameforge\.com)/game', selected_server_url)
self.server_url = b.group(1)
res = self.session.get(selected_server_url).content
soup = BeautifulSoup(res, 'html.parser')
session_found = soup.find('meta', {'name': 'ogame-session'})
if session_found:
self.ogame_session = session_found.get('content')
else:
raise BAD_CREDENTIALS
def logout(self):
self.session.get(self.get_url('logout'))
def is_logged(self, html=None):
if not html:
html = self.session.get(self.get_url('overview')).content
soup = BeautifulSoup(html, 'html.parser')
session = soup.find('meta', {'name': 'ogame-session'})
return session is not None
def get_page_content(self, page='overview', cp=None):
"""Return the html of a specific page."""
payload = {}
if cp is not None:
payload.update({'cp': cp})
html = self.session.get(self.get_url(page, payload)).content
if not self.is_logged(html):
raise NOT_LOGGED
return html
def fetch_eventbox(self):
res = self.session.get(self.get_url('fetchEventbox')).content.decode('utf8')
try:
obj = json.loads(res)
except ValueError:
raise NOT_LOGGED
return obj
def fetch_resources(self, planet_id):
url = self.get_url('fetchResources', {'cp': planet_id})
res = self.session.get(url).content.decode('utf8')
try:
obj = json.loads(res)
except ValueError:
raise NOT_LOGGED
return obj
def get_resources(self, planet_id):
"""Returns the planet resources stats."""
resources = self.fetch_resources(planet_id)
metal = resources['metal']['resources']['actual']
crystal = resources['crystal']['resources']['actual']
deuterium = resources['deuterium']['resources']['actual']
energy = resources['energy']['resources']['actual']
darkmatter = resources['darkmatter']['resources']['actual']
result = {'metal': metal, 'crystal': crystal, 'deuterium': deuterium,
'energy': energy, 'darkmatter': darkmatter}
return result
def get_universe_speed(self, res=None):
if not res:
res = self.session.get(self.get_url('techtree', {'tab': 2, 'techID': 1})).content
soup = BeautifulSoup(res, 'html.pars
