first commit

.dockerignore

@@ -0,0 +1,5 @@
+.git
+.venv
+modules/__pycache__
+cache/*
+.env

.github/workflows/publish.yml

@@ -0,0 +1,33 @@
+name: Build and publish container image to GitHub Container Registry
+
+on: push
+
+jobs:
+  ghcr:
+    if: startsWith(github.ref, 'refs/tags')
+    runs-on: ubuntu-20.04
+
+    steps:
+      - name: Checkout git repository
+        uses: actions/checkout@v2
+
+      - name: Get version from tag
+        id: vars
+        run: echo ::set-output name=tag::${GITHUB_REF#refs/*/}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v1
+
+      - name: Login to GitHub Container Registry
+        uses: docker/login-action@v1
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Build and push
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          push: true
+          tags: ghcr.io/${{ github.repository_owner }}/steam-update-notifier:${{ steps.vars.outputs.tag }}

.gitignore

@@ -0,0 +1,3 @@
+.venv
+__pycache__
+.env

Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.9.2-slim
+
+WORKDIR /app
+COPY ./ .
+RUN pip3 install --upgrade pip && pip3 install -r requirements.txt
+
+ENTRYPOINT [ "python3" ]
+CMD [ "app.py" ]

README.md

@@ -0,0 +1,7 @@
+# Steam Update Notifier
+
+A Python script that will let you know via Discord as soon as a new version of your favorite game on Steam is released.
+
+## TODO
+
+* Support multiple apps, multiple branches

app.py

@@ -0,0 +1,93 @@
+import logging
+import os
+import time
+from os.path import dirname, join
+
+from dotenv import load_dotenv
+from gevent import monkey
+
+monkey.patch_all()
+
+from modules import notifier, witness  # noqa: E402
+
+# dotenv
+dotenv_path = join(dirname(__file__), ".env")
+load_dotenv(dotenv_path)
+
+# Load environment variables
+APP_ID = int(os.getenv("APP_ID"))
+WATCHED_BRANCH = os.getenv("WATCHED_BRANCH")
+
+DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")
+DISCORD_USER_ID = os.getenv("DISCORD_USER_ID")
+
+IGNORE_FIRST_NOTIFICATION = os.getenv("IGNORE_FIRST_NOTIFICATION").lower() == "true"
+CHECK_INTERVAL_SEC = int(os.getenv("CHECK_INTERVAL_SEC"))
+PRODUCT_INFO_CACHE = "./cache/product_info.json"
+UPDATED_TIME_CACHE = "./cache/updated_time.txt"
+
+# Logger
+log_format = "%(asctime)s %(filename)s:%(name)s:%(lineno)d [%(levelname)s] %(message)s"
+logging.basicConfig(
+    level=logging.INFO,
+    format=log_format,
+)
+logger = logging.getLogger(__name__)
+
+
+def one_shot():
+    logger.info("Log in to Steam")
+    is_logged_in = witness.Login()
+    if not is_logged_in:
+        logger.error("Failed to log in to Steam")
+        return False
+
+    logger.info("Check if updated")
+    is_updated = witness.Watch(
+        APP_ID,
+        PRODUCT_INFO_CACHE,
+        UPDATED_TIME_CACHE,
+        WATCHED_BRANCH,
+        IGNORE_FIRST_NOTIFICATION,
+    )
+    logger.info("Result: {}".format(is_updated))
+
+    if is_updated is not None and is_updated["updated"]:
+        logger.info("Fire notification")
+        notifier.Fire(
+            DISCORD_WEBHOOK_URL,
+            DISCORD_USER_ID,
+            is_updated["app_name"],
+            is_updated["app_id"],
+            is_updated["branch"],
+            is_updated["timeupdated"]["str"],
+        )
+    return True
+
+
+def main():
+    try:
+        while True:
+            logger.info("Loop start")
+            result = one_shot()
+            if result:
+                logger.info(
+                    "Loop successfully completed. Will sleep {} seconds".format(
+                        CHECK_INTERVAL_SEC
+                    )
+                )
+            else:
+                logger.info(
+                    "Loop failed. Will sleep {} seconds and retry".format(
+                        CHECK_INTERVAL_SEC
+                    )
+                )
+            time.sleep(CHECK_INTERVAL_SEC)
+    except Exception as e:
+        logger.error(e)
+    finally:
+        witness.Logout()
+
+
+if __name__ == "__main__":
+    main()

cache/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

docker-compose.yml

@@ -0,0 +1,13 @@
+version: "3"
+
+services:
+  notifier:
+    image: ghcr.io/kurokobo/steam-update-notifier:${STEAM_UPDATE_NOTIFIER_TAG:?err}
+    restart: always
+    environment:
+      - APP_ID=${APP_ID:?err}
+      - WATCHED_BRANCH=${WATCHED_BRANCH:?err}
+      - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL:?err}
+      - DISCORD_USER_ID=${DISCORD_USER_ID:?err}
+      - IGNORE_FIRST_NOTIFICATION=${IGNORE_FIRST_NOTIFICATION:?err}
+      - CHECK_INTERVAL_SEC=${CHECK_INTERVAL_SEC:?err}

modules/notifier.py

@@ -0,0 +1,37 @@
+import logging
+
+from discord_webhook import DiscordEmbed, DiscordWebhook
+
+logger = logging.getLogger(__name__)
+
+
+def create_embed_message(user_id, app_name, app_id, branch, latest):
+    desc = (
+        "<@!{}>\n".format(user_id)
+        + "{} seems to be updated on {} branch.\n".format(app_name, branch)
+        + "Now it's time to dive into the RAM and find new offsets."
+    )
+    embed = DiscordEmbed(
+        title="🚨🚨🚨 {} IS UPDATED 🚨🚨🚨".format(app_name.upper()),
+        description=desc,
+        color="03b2f8",
+    )
+    embed.add_embed_field(name="Updated Time", value=latest)
+    embed.add_embed_field(
+        name="App Info", value="{} ({}) @ {}".format(app_name, app_id, branch)
+    )
+    embed.set_footer(text="Notified by Steam Update Notifier")
+    embed.set_timestamp()
+    return embed
+
+
+def Fire(webhook_url, user_id, app_name, app_id, branch, message):
+    logger.info("Prepare webhook")
+    webhook = DiscordWebhook(url=webhook_url)
+
+    logger.info("Construct embed message")
+    embed = create_embed_message(user_id, app_name, app_id, branch, message)
+
+    logger.info("Post embed message")
+    webhook.add_embed(embed)
+    webhook.execute()

modules/witness.py

@@ -0,0 +1,141 @@
+import datetime
+import json
+import logging
+import os
+
+from steam.client import SteamClient
+from steam.enums import EResult
+
+logger = logging.getLogger(__name__)
+client = SteamClient()
+
+
+def save_dict_as_json(dict, path):
+    with open(path, mode="wt", encoding="utf-8") as file:
+        json.dump(dict, file, ensure_ascii=False, indent=2)
+
+
+def get_updated_time(app_id, info_cache, branch):
+    try:
+        logger.info("Request product information for app_id: {}".format(app_id))
+        product_info = client.get_product_info(apps=[app_id])
+
+        logger.info("Save product information to {}".format(info_cache))
+        save_dict_as_json(product_info, info_cache)
+
+        logger.info("Obtain last updated time of branch: {}".format(branch))
+        updated_time = int(
+            product_info["apps"][app_id]["depots"]["branches"][branch]["timeupdated"]
+        )
+
+        logger.info(
+            "The last updated time is: {} ({})".format(
+                datetime.datetime.utcfromtimestamp(updated_time), updated_time
+            )
+        )
+
+        return {
+            "app_name": product_info["apps"][app_id]["common"]["name"],
+            "updated_time": updated_time,
+        }
+    except Exception as e:
+        logger.info("Failed to gather information: {}".format(e))
+        return None
+
+
+def check_if_updated(current, time_cache, ignore_first):
+    if os.path.exists(time_cache):
+        logger.info("Read cached updated time")
+        with open(time_cache) as file:
+            latest = int(file.read())
+    elif ignore_first:
+        logger.info("No cache exists. To ignore first notification, fake cached time")
+        latest = current
+    else:
+        logger.info("No cache exists. Will be compare to zero")
+        latest = 0
+
+    logger.info(
+        "Cached updated time : {} ({})".format(
+            datetime.datetime.utcfromtimestamp(int(latest)), latest
+        )
+    )
+    logger.info(
+        "Current updated time: {} ({})".format(
+            datetime.datetime.utcfromtimestamp(int(current)), current
+        )
+    )
+
+    logger.info("Save current updated time to cache file: {}".format(time_cache))
+    with open(time_cache, mode="w") as file:
+        file.write(str(current))
+
+    if int(current) > latest:
+        logger.info("Updated")
+        return True
+    else:
+        logger.info("Not updated")
+        return False
+
+
+def Login():
+    logged_on = client.logged_on
+    logger.info("Is logged on: {}".format(logged_on))
+
+    if logged_on:
+        logger.info("Already logged in")
+        return True
+
+    logger.info("Try log in to Steam")
+    if client.relogin_available:
+        logger.info("Invoke relogin")
+        login = client.relogin()
+    else:
+        logger.info("Invoke anonymous login")
+        login = client.anonymous_login()
+    logger.info("Result: {}".format(login))
+
+    logged_on = client.logged_on
+    logger.info("Is logged on: {}".format(logged_on))
+
+    if login == EResult.OK or client.logged_on:
+        logger.info("Successful logged in")
+        return True
+    else:
+        logger.error("Failed to log in to Steam")
+        return False
+
+
+def Watch(app_id, info_cache, time_cache, branch, ignore_first):
+    logger.info("Check the last updated time of app_id: {}".format(app_id))
+    updated_time = get_updated_time(
+        app_id,
+        info_cache,
+        branch,
+    )
+    if updated_time is None:
+        logger.error("Failed to obtain last updated time from Steam")
+        return None
+
+    logger.info("Check if the product is updated ")
+    is_updated = check_if_updated(
+        updated_time["updated_time"], time_cache, ignore_first
+    )
+    return {
+        "timestamp": datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
+        "app_id": app_id,
+        "app_name": updated_time["app_name"],
+        "updated": is_updated,
+        "branch": branch,
+        "timeupdated": {
+            "epoc": updated_time["updated_time"],
+            "str": datetime.datetime.utcfromtimestamp(
+                updated_time["updated_time"]
+            ).strftime("%Y/%m/%d %H:%M:%S"),
+        },
+    }
+
+
+def Logout():
+    logger.info("Invoke logged out")
+    client.logout()

requirements.txt

@@ -0,0 +1,7 @@
+wheel==0.36.2
+python-dotenv==0.17.0
+gevent==21.1.2
+gevent-eventemitter==2.1
+steam==1.2.0
+discord-webhook==0.13.0
+google-api-python-client==2.1.0

sample.env

@@ -0,0 +1,10 @@
+STEAM_UPDATE_NOTIFIER_TAG=
+
+APP_ID=
+WATCHED_BRANCH=
+
+DISCORD_WEBHOOK_URL=
+DISCORD_USER_ID=
+
+IGNORE_FIRST_NOTIFICATION=True
+CHECK_INTERVAL_SEC=300