cheeseburger/PyT1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: clean up code formatting

Browse Source
This commit is contained in:
WOBBLEFANG THE THIRD
2026-01-23 12:20:37 +01:00
parent 710713224b
commit 3e18988aac
4 changed files with 30 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
main.py
View File

@@ -1,9 +1,9 @@
import logging
import os
import time
import logging
import requests
from logging.handlers import RotatingFileHandler
import requests
# =========================
# Configuration
@@ -16,7 +16,6 @@ DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"]
CHECK_INTERVAL_SECONDS = 30
POST_DETECTION_COOLDOWN_SECONDS = 600
# =========================
# Logging Setup
# =========================
@@ -24,17 +23,10 @@ POST_DETECTION_COOLDOWN_SECONDS = 600
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"
)
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 = RotatingFileHandler("formd_monitor.log", maxBytes=5 * 1024 * 1024, backupCount=2)
file_handler.setFormatter(log_format)
logger.addHandler(file_handler)
@@ -55,14 +47,10 @@ def send_discord_notification(product_name, variant_name, checkout_link):
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})"
)
}
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)
@@ -70,16 +58,11 @@ def send_discord_notification(product_name, variant_name, checkout_link):
if response.status_code == 204:
logger.info("Discord notification sent successfully.")
else:
logger.error(
"Discord notification failed. "
f"HTTP status code: {response.status_code}"
)
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
)
logger.error("Failed to send Discord notification due to an exception.", exc_info=exception)
# =========================
@@ -92,31 +75,18 @@ def check_stock_continuously():
variant is available for purchase.
"""
logger.info(
f"Stock monitor started. Watching variant ID: {TARGET_VARIANT_ID}"
)
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"
}
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
)
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}"
)
logger.error(f"HTTP request failed. Status code: {response.status_code}")
time.sleep(CHECK_INTERVAL_SECONDS)
continue
@@ -124,18 +94,10 @@ def check_stock_continuously():
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
)
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."
)
logger.error(f"Variant ID {TARGET_VARIANT_ID} was not found in the product data.")
time.sleep(CHECK_INTERVAL_SECONDS)
continue
@@ -143,44 +105,26 @@ def check_stock_continuously():
is_in_stock = target_variant.get("available", False)
if is_in_stock:
logger.warning(
f"Stock available! Variant detected: {variant_name}"
)
logger.warning(f"Stock available! Variant detected: {variant_name}")
checkout_link = (
f"https://formdt1.com/cart/{TARGET_VARIANT_ID}:1"
)
checkout_link = (f"https://formdt1.com/cart/{TARGET_VARIANT_ID}:1")
send_discord_notification(
product_name,
variant_name,
checkout_link
)
send_discord_notification(product_name, variant_name, checkout_link)
logger.info(
"Cooldown started to prevent duplicate alerts."
)
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}"
)
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."
)
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."
)
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."
)
logger.exception("An unexpected error occurred during the stock check.")
time.sleep(CHECK_INTERVAL_SECONDS)