feat: add stock monitoring and Discord notification functionality

2026-01-23 12:10:19 +01:00
parent 4a591d8893
commit acfcfb0a64
3 changed files with 293 additions and 1 deletions
--- a/main.py
+++ b/main.py
@@ -0,0 +1,193 @@
+import os
+import time
+import logging
+import requests
+from logging.handlers import RotatingFileHandler
+
+
+# =========================
+# Configuration
+# =========================
+
+PRODUCT_URL = f"{os.environ["PRODUCT_URL"]}.js"
+TARGET_VARIANT_ID = os.environ["TARGET_VARIANT_ID"]
+DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"]
+
+CHECK_INTERVAL_SECONDS = 30
+POST_DETECTION_COOLDOWN_SECONDS = 600
+
+
+# =========================
+# Logging Setup
+# =========================
+
+logger = logging.getLogger("FormDStockMonitor")
+logger.setLevel(logging.INFO)
+
+log_format = logging.Formatter(
+    "%(asctime)s - %(levelname)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S"
+)
+
+# Write logs to a rotating file so disk space is not exhausted
+file_handler = RotatingFileHandler(
+    "formd_monitor.log",
+    maxBytes=5 * 1024 * 1024,
+    backupCount=2
+)
+file_handler.setFormatter(log_format)
+logger.addHandler(file_handler)
+
+# Also show logs directly in the terminal
+console_handler = logging.StreamHandler()
+console_handler.setFormatter(log_format)
+logger.addHandler(console_handler)
+
+
+# =========================
+# Discord Notification
+# =========================
+
+def send_discord_notification(product_name, variant_name, checkout_link):
+    """
+    Sends a stock alert to Discord using a webhook.
+    """
+
+    logger.critical("Stock detected! Sending Discord notification.")
+
+    payload = {
+        "content": (
+            "🚨 **RESTOCK ALERT** 🚨\n\n"
+            f"**{product_name}**\n"
+            f"Variant: {variant_name}\n\n"
+            f"👉 [ONE-CLICK CHECKOUT]({checkout_link})"
+        )
+    }
+
+    try:
+        response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
+
+        if response.status_code == 204:
+            logger.info("Discord notification sent successfully.")
+        else:
+            logger.error(
+                "Discord notification failed. "
+                f"HTTP status code: {response.status_code}"
+            )
+
+    except Exception as exception:
+        logger.error(
+            "Failed to send Discord notification due to an exception.",
+            exc_info=exception
+        )
+
+
+# =========================
+# Stock Monitoring Logic
+# =========================
+
+def check_stock_continuously():
+    """
+    Continuously checks the product JSON endpoint to see if the target
+    variant is available for purchase.
+    """
+
+    logger.info(
+        f"Stock monitor started. Watching variant ID: {TARGET_VARIANT_ID}"
+    )
+
+    request_headers = {
+        "User-Agent": (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/120.0.0.0 Safari/537.36"
+        ),
+        "Accept": "application/json"
+    }
+
+    while True:
+        try:
+            response = requests.get(
+                PRODUCT_URL,
+                headers=request_headers,
+                timeout=10
+            )
+
+            if response.status_code != 200:
+                logger.error(
+                    f"HTTP request failed. Status code: {response.status_code}"
+                )
+                time.sleep(CHECK_INTERVAL_SECONDS)
+                continue
+
+            product_data = response.json()
+            product_name = product_data.get("title", "FormD T1")
+            variants = product_data.get("variants", [])
+
+            target_variant = next(
+                (
+                    variant for variant in variants
+                    if variant.get("id") == TARGET_VARIANT_ID
+                ),
+                None
+            )
+
+            if not target_variant:
+                logger.error(
+                    f"Variant ID {TARGET_VARIANT_ID} was not found in the product data."
+                )
+                time.sleep(CHECK_INTERVAL_SECONDS)
+                continue
+
+            variant_name = target_variant.get("title", "Unknown Variant")
+            is_in_stock = target_variant.get("available", False)
+
+            if is_in_stock:
+                logger.warning(
+                    f"Stock available! Variant detected: {variant_name}"
+                )
+
+                checkout_link = (
+                    f"https://formdt1.com/cart/{TARGET_VARIANT_ID}:1"
+                )
+
+                send_discord_notification(
+                    product_name,
+                    variant_name,
+                    checkout_link
+                )
+
+                logger.info(
+                    "Cooldown started to prevent duplicate alerts."
+                )
+                time.sleep(POST_DETECTION_COOLDOWN_SECONDS)
+
+            else:
+                logger.info(
+                    f"Out of stock. Checked variant: {variant_name}"
+                )
+
+        except requests.exceptions.ConnectionError:
+            logger.error(
+                "Network connection error. Unable to reach the product server."
+            )
+
+        except requests.exceptions.Timeout:
+            logger.warning(
+                "Request timed out. The server may be slow or blocking requests."
+            )
+
+        except Exception as exception:
+            logger.exception(
+                "An unexpected error occurred during the stock check."
+            )
+
+        time.sleep(CHECK_INTERVAL_SECONDS)
+
+
+# =========================
+# Application Entry Point
+# =========================
+
+if __name__ == "__main__":
+    check_stock_continuously()