250 lines
8.8 KiB
Python
Executable file
250 lines
8.8 KiB
Python
Executable file
#!/usr/bin/env python
|
|
|
|
# -*- coding: utf-8 -*-
|
|
# pylint: disable= I0011,C0103
|
|
|
|
from __future__ import print_function
|
|
from __future__ import division
|
|
from __future__ import unicode_literals
|
|
|
|
# imports
|
|
import sys
|
|
import logging
|
|
import contextlib
|
|
import json
|
|
import getpass
|
|
import argparse
|
|
import socket
|
|
import html5lib
|
|
import requests
|
|
|
|
# python 3
|
|
import http.cookiejar as cookiejar
|
|
from http.client import BadStatusLine
|
|
from urllib.parse import urlencode
|
|
from urllib.request import HTTPCookieProcessor, HTTPError, URLError, build_opener, Request
|
|
|
|
__appname__ = 'gogconnect-notify.py'
|
|
__author__ = 'eNBeWe'
|
|
__version__ = '0.1'
|
|
__url__ = 'https://github.com/eNBeWe/gogconnect-notify'
|
|
|
|
# configure logging
|
|
logFormatter = logging.Formatter("%(asctime)s | %(message)s", datefmt='%H:%M:%S')
|
|
rootLogger = logging.getLogger('ws')
|
|
rootLogger.setLevel(logging.DEBUG)
|
|
consoleHandler = logging.StreamHandler(sys.stdout)
|
|
consoleHandler.setFormatter(logFormatter)
|
|
rootLogger.addHandler(consoleHandler)
|
|
|
|
# filepath constants
|
|
COOKIES_FILENAME = r'gog-cookies.dat'
|
|
|
|
# global web utilities
|
|
global_cookies = cookiejar.LWPCookieJar(COOKIES_FILENAME)
|
|
cookieproc = HTTPCookieProcessor(global_cookies)
|
|
opener = build_opener(cookieproc)
|
|
|
|
# GOG URLs
|
|
GOG_HOME_URL = r'https://www.gog.com'
|
|
GOG_LOGIN_URL = r'https://login.gog.com/login_check'
|
|
GOG_API_URL = r'https://www.gog.com/api/v1/users/'
|
|
|
|
# HTTP request settings
|
|
HTTP_RETRY_COUNT = 3
|
|
HTTP_PERM_ERRORCODES = (404, 403, 503)
|
|
|
|
def request(url, args=None, retries=HTTP_RETRY_COUNT):
|
|
"""Performs web request to url with optional retries, delay, and byte range.
|
|
"""
|
|
_retry = False
|
|
|
|
try:
|
|
if args is not None:
|
|
enc_args = urlencode(args)
|
|
enc_args = enc_args.encode('ascii') # needed for Python 3
|
|
else:
|
|
enc_args = None
|
|
req = Request(url, data=enc_args)
|
|
page = opener.open(req)
|
|
except (HTTPError, URLError, socket.error, BadStatusLine) as e:
|
|
if isinstance(e, HTTPError):
|
|
if e.code in HTTP_PERM_ERRORCODES: # do not retry these HTTP codes
|
|
rootLogger.warn('request failed: %s. will not retry.', e)
|
|
raise
|
|
if retries > 0:
|
|
_retry = True
|
|
else:
|
|
raise
|
|
|
|
if _retry:
|
|
rootLogger.warn('request failed: %s (%d retries left)', e, retries)
|
|
return request(url=url, args=args, retries=retries-1)
|
|
|
|
return contextlib.closing(page)
|
|
|
|
def load_cookies():
|
|
# try to load as default lwp format
|
|
try:
|
|
global_cookies.load()
|
|
return
|
|
except IOError:
|
|
pass
|
|
|
|
rootLogger.error('failed to load cookies, did you login first?')
|
|
raise SystemExit(1)
|
|
|
|
def process_argv(argv):
|
|
p1 = argparse.ArgumentParser(description='%s (%s)' % (__appname__, __url__), add_help=False)
|
|
sp1 = p1.add_subparsers(help='commands', dest='cmd', title='commands')
|
|
|
|
g1 = sp1.add_parser('login',
|
|
help='Login to GOG and save a local copy of your authenticated cookie')
|
|
g1.add_argument('username', action='store', help='GOG username/email', nargs='?', default=None)
|
|
g1.add_argument('password', action='store', help='GOG password', nargs='?', default=None)
|
|
|
|
g1 = sp1.add_parser('check', help='Check for new games in GOG connect')
|
|
|
|
g1 = p1.add_argument_group('other')
|
|
g1.add_argument('-h', '--help', action='help', help='show help message and exit')
|
|
g1.add_argument('-v', '--version', action='version', help='show version number and exit',
|
|
version="%s (version %s)" % (__appname__, __version__))
|
|
|
|
# parse the given argv. raises SystemExit on error
|
|
args = p1.parse_args(argv[1:])
|
|
|
|
return args
|
|
|
|
# --------
|
|
# Commands
|
|
# --------
|
|
def cmd_login(user, passwd):
|
|
"""Attempts to log into GOG and saves the resulting cookiejar to disk.
|
|
"""
|
|
login_data = {'user': user,
|
|
'passwd': passwd,
|
|
'auth_url': None,
|
|
'login_token': None,
|
|
'two_step_url': None,
|
|
'two_step_token': None,
|
|
'two_step_security_code': None,
|
|
'login_success': False,
|
|
}
|
|
|
|
global_cookies.clear() # reset cookiejar
|
|
|
|
# prompt for login/password if needed
|
|
if login_data['user'] is None:
|
|
login_data['user'] = input("Username: ")
|
|
if login_data['passwd'] is None:
|
|
login_data['passwd'] = getpass.getpass()
|
|
|
|
rootLogger.info("attempting gog login as '%s' ...", login_data['user'])
|
|
|
|
# fetch the auth url
|
|
with request(GOG_HOME_URL) as page:
|
|
etree = html5lib.parse(page, namespaceHTMLElements=False)
|
|
for elm in etree.findall('.//script'):
|
|
if elm.text is not None and 'GalaxyAccounts' in elm.text:
|
|
login_data['auth_url'] = elm.text.split("'")[1]
|
|
break
|
|
|
|
# fetch the login token
|
|
with request(login_data['auth_url']) as page:
|
|
etree = html5lib.parse(page, namespaceHTMLElements=False)
|
|
# Bail if we find a request for a reCAPTCHA
|
|
if len(etree.findall('.//div[@class="g-recaptcha"]')) > 0:
|
|
rootLogger.error(
|
|
"cannot continue, gog is asking for a reCAPTCHA :( try again in a few minutes.")
|
|
return
|
|
for elm in etree.findall('.//input'):
|
|
if elm.attrib['id'] == 'login__token':
|
|
login_data['login_token'] = elm.attrib['value']
|
|
break
|
|
|
|
# perform login and capture two-step token if required
|
|
with request(GOG_LOGIN_URL, args={'login[username]': login_data['user'],
|
|
'login[password]': login_data['passwd'],
|
|
'login[login]': '',
|
|
'login[_token]': login_data['login_token']}) as page:
|
|
etree = html5lib.parse(page, namespaceHTMLElements=False)
|
|
if 'two_step' in page.geturl():
|
|
login_data['two_step_url'] = page.geturl()
|
|
for elm in etree.findall('.//input'):
|
|
if elm.attrib['id'] == 'second_step_authentication__token':
|
|
login_data['two_step_token'] = elm.attrib['value']
|
|
break
|
|
elif 'on_login_success' in page.geturl():
|
|
login_data['login_success'] = True
|
|
|
|
# perform two-step if needed
|
|
if login_data['two_step_url'] is not None:
|
|
login_data['two_step_security_code'] = input("enter two-step security code: ")
|
|
|
|
# Send the security code back to GOG
|
|
with request(login_data['two_step_url'],
|
|
args={'second_step_authentication[token][letter_1]': login_data['two_step_security_code'][0],
|
|
'second_step_authentication[token][letter_2]': login_data['two_step_security_code'][1],
|
|
'second_step_authentication[token][letter_3]': login_data['two_step_security_code'][2],
|
|
'second_step_authentication[token][letter_4]': login_data['two_step_security_code'][3],
|
|
'second_step_authentication[send]': "",
|
|
'second_step_authentication[_token]': login_data['two_step_token']
|
|
}) as page:
|
|
if 'on_login_success' in page.geturl():
|
|
login_data['login_success'] = True
|
|
|
|
# save cookies on success
|
|
if login_data['login_success']:
|
|
rootLogger.info('login successful!')
|
|
global_cookies.save()
|
|
else:
|
|
rootLogger.error('login failed, verify your username/password and try again.')
|
|
|
|
|
|
def cmd_check():
|
|
|
|
load_cookies()
|
|
|
|
session = requests.Session()
|
|
session.headers["User-Agent"] = "Mozilla/5.0 (X11; Linux x86_64; rv:46.0) Gecko/20100101 Firefox/46.0"
|
|
|
|
session.cookies = global_cookies
|
|
|
|
user_data = json.loads(session.get("{}/userData.json".format(GOG_HOME_URL)).text)
|
|
|
|
|
|
# Refresh Steam products
|
|
refresh_url = "{}/{}/gogLink/steam/synchronizeUserProfile".format(GOG_API_URL,
|
|
user_data["userId"])
|
|
session.get(refresh_url)
|
|
|
|
steam_products_url = "{}/{}/gogLink/steam/exchangeableProducts".format(GOG_API_URL,
|
|
user_data["userId"])
|
|
steam_products = json.loads(session.get(steam_products_url).text)
|
|
|
|
games_available = False
|
|
if len(steam_products["items"]) > 0:
|
|
for key, value in steam_products["items"].items():
|
|
if value["status"] == "available" or value["status"] == "readyToLink":
|
|
games_available = True
|
|
break
|
|
|
|
if games_available:
|
|
print("New games available!")
|
|
|
|
def main(args):
|
|
if args.cmd == 'login':
|
|
cmd_login(args.username, args.password)
|
|
elif args.cmd == 'check':
|
|
cmd_check()
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main(process_argv(sys.argv))
|
|
except KeyboardInterrupt:
|
|
sys.exit(1)
|
|
except SystemExit:
|
|
raise
|
|
except:
|
|
rootLogger.exception('fatal...')
|
|
sys.exit(1)
|