Initial commit

Basic plugin with support for logging into your steam account, the retrieval/creation of your Steam Web API key and synchronization of your library and game time.

authentication.py

@@ -0,0 +1,91 @@
+import requests
+from bs4 import BeautifulSoup
+import constants
+from galaxy.api.types import (
+    Authentication
+)
+
+import backend
+
+
+
+def convertCookieListToDict(lst):
+    res_dict = {}
+    for i in range(0, len(lst)):
+        if lst[i]["domain"] == "steamcommunity.com":
+            res_dict[lst[i]["name"]] = lst[i]["value"]
+    return res_dict
+
+
+def add_credential_to_url(str, stored_credentials):
+    return str+"&key="+stored_credentials.get(constants.STORED_CREDENTIAL_WEBPAPIKEY_KEY)
+
+# validate steam api key by calling GetPlayerSummaries for the user's steamid 
+def validate_credentials(stored_credentials):
+    player_summaries_response = backend.get_player_summaries(stored_credentials)
+
+    steamid = player_summaries_response.get("response").get("players")[0].get("steamid")   
+    name = player_summaries_response.get("response").get("players")[0].get("personaname") 
+
+    return Authentication(steamid, name)
+
+def get_steamid_and_web_api_key(plugin, cookies):
+    #get captured steamcomunity cookies for use in the api key form requests
+    cookies_dict = convertCookieListToDict(cookies)
+    keyText = ""
+
+    # get steamID from cookievalue
+    cookie_value = cookies_dict["steamLoginSecure"]
+    steamID = cookie_value[: cookie_value.find("%7C%7C")]
+    # get sessionid from cookievalue, in case we need it to submit the api key form
+    sessionid = cookies_dict["sessionid"]
+
+    print(f"steamID: {steamID}")
+    print(f"sessionID: {sessionid}")
+
+    # access the web api form
+    api_form_response = requests.get("https://steamcommunity.com/dev/apikey", cookies=cookies_dict)
+
+    # parse the reponse
+    api_form_soup = BeautifulSoup(api_form_response.text, "html.parser")
+    form_action = api_form_soup.find("form", id="editForm").get("action")
+
+    # find out if the user already has a key, based on the form action
+    # if the action is to revoke, the user already has a key setup and we can use it
+    # if the action is to register, them we register one for them
+    if form_action == constants.ACTION_REVOKE_KEY_URL:  # got the key, extract it
+        print("Already has a key")
+        paragraphText = api_form_soup.find("div", id="bodyContents_ex").p.string
+        keyText = paragraphText[paragraphText.find(": ") + 2 :]
+
+    elif form_action == constants.ACTION_REGISTER_KEY_URL:  # doesn't have a key, create a new one
+        print("Needs a key")
+
+        # build formdata
+        form_data = {
+            'domain': 'Created by GOG Galaxy Steam Integration',
+            'agreeToTerms':'agreed', #valve is going to kill me because of this
+            'sessionid':sessionid,
+            'Submit':'Register'
+        }
+
+        # submit key request form
+        register_response = requests.post(constants.ACTION_REGISTER_KEY_URL, data=form_data, cookies=cookies_dict)
+
+        # parse response and extract the key
+        register_soup = BeautifulSoup(register_response.text, "html.parser")
+        paragraphText = register_soup.find("div", id="bodyContents_ex").p.string
+        keyText = paragraphText[paragraphText.find(": ") + 2 :]
+
+    print(f"keyText: {keyText}")
+
+    credentials = {
+        constants.STORED_CREDENTIAL_STEAMID_KEY: steamID,
+        constants.STORED_CREDENTIAL_WEBPAPIKEY_KEY: keyText
+    }
+
+    plugin.store_credentials(credentials)
+
+    plugin.stored_credentials = credentials
+
+    return validate_credentials(plugin.stored_credentials)

backend.py

@@ -0,0 +1,82 @@
+import constants
+import requests
+import json
+import logging
+from typing import Dict
+
+from galaxy.api.errors import (
+    InvalidCredentials
+)
+from galaxy.api.types import (
+    GameTime
+)
+
+def get_method_params(method_name, stored_credentials):
+    params = {
+        "format": "json",
+        "key": stored_credentials.get(constants.STORED_CREDENTIAL_WEBPAPIKEY_KEY)
+        }
+
+    if method_name == "GetPlayerSummaries":
+        params["steamids"] = stored_credentials.get(constants.STORED_CREDENTIAL_STEAMID_KEY)
+    elif method_name == "GetOwnedGames":
+        params["steamid"] = stored_credentials.get(constants.STORED_CREDENTIAL_STEAMID_KEY)
+    return params
+
+
+def get_player_summaries(stored_credentials):
+    # get params
+    payload = get_method_params("GetPlayerSummaries", stored_credentials)
+
+    # call method
+    get_players_summaries_response = requests.get(constants.STEAM_API_GET_PAYER_SUMMARIES_URL, params=payload)
+
+    if get_players_summaries_response.status_code in constants.STEAM_API_OK_HTTP_CODES:
+        return get_players_summaries_response.json()
+    elif get_players_summaries_response.status_code in constants.STEAM_API_NOK_HTTP_CODES:
+        raise InvalidCredentials()
+
+def get_steam_owned_games(stored_credentials, appinfo=False, appids_filter=[]):
+    logging.debug("get_steam_owned_games")
+    # get params
+    payload = get_method_params("GetOwnedGames", stored_credentials)
+
+    if appinfo:
+        payload["include_appinfo"] = "true" 
+
+    # TODO: filter by ids appids_filter, https://developer.valvesoftware.com/wiki/Steam_Web_API#GetOwnedGames_.28v0001.29
+    # JSON: "appids_filter: [ 440, 500, 550 ]" )
+    if appids_filter:
+        integer_appids_filter = list(map(int, appids_filter))
+        payload["input_json"] = "{ \"steamid\": "+ stored_credentials.get("steamid") + ", \"appids_filter\": "+json.dumps(integer_appids_filter)+"}"
+        logging.debug("appids_filter")
+
+
+
+    owned_games_response = requests.get(constants.STEAM_API_GET_OWNED_GAMES_URL, params=payload)
+
+    if owned_games_response.status_code in constants.STEAM_API_OK_HTTP_CODES:
+        return owned_games_response.json()
+    elif owned_games_response.status_code in constants.STEAM_API_NOK_HTTP_CODES:
+        raise InvalidCredentials()
+
+async def get_steam_game_times(stored_credentials, game_ids) -> Dict[str, GameTime]:
+    steam_owned_game_list = get_steam_owned_games(stored_credentials, appids_filter=game_ids).get("response").get("games")
+
+    context = {}
+
+    for game in steam_owned_game_list:
+        game_id = str(game.get("appid"))
+        context[game_id] = GameTime(
+            game_id,
+            game.get("playtime_forever") if game.get("playtime_forever") > 0 else None,
+            game.get("rtime_last_played") if game.get("rtime_last_played") > constants.STEAM_API_BEGINNING_OF_TIME else None
+            )
+
+        logging.debug(f"Caching time for game_id: {game_id}")
+        logging.debug("game_id: "+str(context[game_id].game_id))
+        logging.debug("time_played: "+str(context[game_id].time_played))
+        logging.debug("last_played_time: "+str(context[game_id].last_played_time))
+
+
+    return context

constants.py

@@ -0,0 +1,38 @@
+from galaxy.api.types import (
+   Cookie
+)
+
+ACTION_REVOKE_KEY_URL = "https://steamcommunity.com/dev/revokekey"
+ACTION_REGISTER_KEY_URL = "https://steamcommunity.com/dev/registerkey"
+
+
+STEAM_LOGIN_WINDOW_PARAMS = {
+    "window_title": "Login to platform",
+    "window_width": 800,
+    "window_height": 650,
+    "start_uri": "https://steamcommunity.com/login/home/?goto=%2Fdev%2Fapikey",
+    "end_uri_regex": r"^https://steamcommunity.com/dev/apikey",
+}
+
+STEAM_LOGIN_REJECT_COOKIES_COOKIE = [
+    Cookie(
+        "cookieSettings",
+        "%7B%22version%22%3A1%2C%22preference_state%22%3A2%2C%22content_customization%22%3Anull%2C%22valve_analytics%22%3Anull%2C%22third_party_analytics%22%3Anull%2C%22third_party_content%22%3Anull%2C%22utm_enabled%22%3Atrue%7D",
+        "steamcommunity.com",
+    )
+]
+
+STORED_CREDENTIAL_WEBPAPIKEY_KEY = "webapikey"
+STORED_CREDENTIAL_STEAMID_KEY = "steamid"
+
+
+STEAM_API_BASE_URL = "http://api.steampowered.com"
+
+STEAM_API_GET_OWNED_GAMES_URL = f"{STEAM_API_BASE_URL}/IPlayerService/GetOwnedGames/v0001/"
+STEAM_API_GET_PAYER_SUMMARIES_URL = f"{STEAM_API_BASE_URL}/ISteamUser/GetPlayerSummaries/v0002/"
+
+STEAM_API_OK_HTTP_CODES = [200]
+STEAM_API_NOK_HTTP_CODES = [401, 403]
+
+
+STEAM_API_BEGINNING_OF_TIME = 86400

manifest.json

@@ -0,0 +1,11 @@
+{
+    "name": "A GOG Galaxy Steam Integration Plugin",
+    "platform": "steam",
+    "guid": "c6cb8cda-9a27-48a7-a50e-f2bccb856d95",
+    "version": "0.1.0",
+    "description": "My attempt at building a gog galaxy steam integration",
+    "author": "novettam",
+    "email": "[email protected]",
+    "url": "https://github.com/Novettam/galaxy-integration-steam",
+    "script": "plugin.py"
+}

plugin.py

@@ -0,0 +1,79 @@
+import sys
+import logging
+from typing import Any
+
+from galaxy.api.plugin import Plugin, create_and_run_plugin
+from galaxy.api.consts import Platform
+from galaxy.api.types import (
+    NextStep,
+    Game,
+    LicenseInfo,
+    LicenseType,
+    GameTime,
+)
+import constants
+import authentication
+import backend
+
+class SteamIntegrationPlugin(Plugin):
+    def __init__(self, reader, writer, token):
+        super().__init__(
+            Platform.Steam,  # choose platform from available list
+            "1.0.2",  # version
+            reader,
+            writer,
+            token,
+        )
+
+    # authenticate user with steam
+    async def authenticate(self, stored_credentials=None):
+        if not stored_credentials:
+            return NextStep(
+                "web_session", constants.STEAM_LOGIN_WINDOW_PARAMS, constants.STEAM_LOGIN_REJECT_COOKIES_COOKIE
+            )  # steamcommunity login page, with redirect to api page
+        self.stored_credentials = stored_credentials
+        # validate stored credentials and return authentication
+        return authentication.validate_credentials(stored_credentials)
+
+    # second authentication step to setup the steam web api key
+    async def pass_login_credentials(self, step, credentials, cookies):
+        return authentication.get_steamid_and_web_api_key(self, cookies)
+
+    # required
+    async def get_owned_games(self):
+        steam_owned_game_list = backend.get_steam_owned_games(self.stored_credentials, True)
+        #logging.log(logging.DEBUG,"get_owned_games")
+        game_list = []
+
+        for game in steam_owned_game_list.get("response").get("games"):
+            game_list.append(
+                Game(
+                    str(game.get("appid")),
+                    game.get("name"),
+                    None,
+                    LicenseInfo(LicenseType.SinglePurchase),
+                )
+            )
+
+        return game_list
+
+    async def prepare_game_times_context(self, game_ids) -> Any:
+        return await backend.get_steam_game_times(self.stored_credentials, game_ids)
+
+    async def get_game_time(self, game_id, context) -> GameTime:
+        game_time = context.get(game_id)
+        logging.debug(f"Updating time for game_id: {game_id}")
+        logging.debug("game_id: "+str(game_time.game_id))
+        logging.debug("time_played: "+str(game_time.time_played))
+        logging.debug("last_played_time: "+str(game_time.last_played_time))
+        return game_time
+        #return GameTime(game_id=game_id, time_played=game_time.get("time_played"), last_played_time=game_time.get("last_played_time"))
+
+
+def main():
+    create_and_run_plugin(SteamIntegrationPlugin, sys.argv)
+
+
+# run plugin event loop
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,3 @@
+beautifulsoup4==4.12.2 ; python_full_version > "3.7.0"
+galaxy-plugin-api==0.69 ; python_full_version > "3.7.0"
+requests==2.29.0 ; python_full_version > "3.7.0"