-
-
Notifications
You must be signed in to change notification settings - Fork 9
/
Copy pathmain.py
165 lines (141 loc) · 6.61 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
import os, sys, datetime, traceback, apprise
import signal
import time
from selenium import webdriver
from datetime import datetime
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service
from dotenv import load_dotenv
# Setup environment
load_dotenv()
# Print Welcome Message
print(f"\n{'-' * 75}\nWelcome to Wired Coupon Scraper.\n\nAn automation script to retrieve multiple retailer promotion codes and offers from the Wired website.\nCreated by Prem-ium (https://github.com/Prem-ium)\n\n")
print(f"{'-' * 75}\nWant coupons on the go? Become a Gold Sponsor for exclusive Discord Bot support!\n\nAs a Gold Sponsor, you'll gain:\n- Instant access to coupon notifications anytime, anywhere\n- Be the first to know about new promo codes and offers\n- Priority support for any issues or requests\n\nBecome a Gold Sponsor today and never miss a deal again: [Gold Sponsor page](https://github.com/sponsors/Prem-ium)\n")
# Retrieve CLI Arguments & Environment Variables
if len(sys.argv) == 2:
RETAILERS = sys.argv[1].split(",")
print(f"RETAILERS argument received.\nGathering coupon codes for: {RETAILERS}\n{'-' * 50}")
else:
RETAILERS = os.environ.get("RETAILERS", None)
if RETAILERS is None:
print(f"No arguments or environment variables received for RETAILERS... Defaulting to Walmart.\n{'-' * 50}")
RETAILERS = ["walmart"]
ALLOW_DUPLICATES = True if os.environ.get("ALLOW_DUPLICATES", "False").lower() == "true" else False
if os.environ.get("KEEP_ALIVE", "False").lower() == "true":
from keep_alive import keep_alive
keep_alive()
KEEP_ALIVE = True
else:
KEEP_ALIVE = False
# Set up Apprise, if enabled
APPRISE_ALERTS = os.environ.get("APPRISE_ALERTS", None)
if APPRISE_ALERTS:
APPRISE_ALERTS = APPRISE_ALERTS.split(",")
alerts = apprise.Apprise()
for service in APPRISE_ALERTS:
alerts.add(service)
def getDriver():
"""Initialize and return the WebDriver for Chrome."""
chrome_options = Options()
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')
try:
driver = webdriver.Chrome(
service=Service(ChromeDriverManager().install()),
options=chrome_options)
except Exception as e:
driver = webdriver.Chrome(options=chrome_options)
return driver
def cached_codes_init():
"""Initialize cached codes file, clearing it on the first day of the month."""
if not os.path.isfile("codes.txt"):
open("codes.txt", "a").close()
print("Created new codes.txt file")
else:
with open("codes.txt", "r") as f:
if datetime.now().day == 1 and f.readlines():
print("It's the first day of the month. Clearing the codes.txt file.")
with open("codes.txt", "w"): pass
else:
print("codes.txt file already exists")
def check_cached_codes(code):
"""Check if the code is already cached."""
with open("codes.txt") as f:
lines = f.readlines()
lines = [line.strip() for line in lines]
if f"{str(code)}\n" in lines:
print(f"Code {code} found in cache")
return True
return False
def append_cached_code(code):
"""Append the code to the cache file."""
with open("codes.txt", "a") as f:
f.write(f"{code}\n")
print(f"Appended {code} to codes.txt")
def handle_shutdown_signal(signum, frame):
"""Gracefully handle shutdown signal (e.g., CTRL+C)."""
print("Graceful shutdown initiated...")
sys.exit(0)
def run_program():
"""Run the main program logic, handling continuous execution if required."""
if not ALLOW_DUPLICATES:
cached_codes_init()
driver = getDriver()
for type in RETAILERS:
print(f'{"-" * 70}\nRetrieving {type.upper()} Promo Code Offers...\n')
driver.get(f'https://www.wired.com/coupons/{type}')
coupons = driver.find_element(By.CLASS_NAME, 'coupons-list').find_elements(By.TAG_NAME, 'a')
ids = [coupon.get_attribute('href') for coupon in coupons]
print(f'{len(coupons)} {type.upper()} Promo Codes/Coupons were found!')
for id in ids:
try:
driver.get(id)
driver.refresh()
time.sleep(2)
print('-' * 50)
title = driver.find_element(By.XPATH, '//*[@id="my-modal"]/div/div/div/h3')
code = driver.find_element(By.XPATH, '//*[@id="my-modal"]/div/div/div/div[2]/span')
try:
link = driver.find_element(By.CLASS_NAME, 'modal-clickout__link').get_attribute("href")
except:
link = "(Error retrieving link)"
print(traceback.format_exc())
data = f'{title.text}:\n\t{code.text} \n{link}\n'
print(data)
if APPRISE_ALERTS:
alerts.notify(title=f'{type.upper()} Coupon', body=f'{title.text}\n\n{code.text}\n{link}')
data = data.replace("\n", " - ")
if not ALLOW_DUPLICATES:
if check_cached_codes(data):
print(f"Code {code.text} for {title.text} already exists in cache, skipping")
continue
else:
append_cached_code(data)
except WebDriverException as e:
print(traceback.format_exc())
if "no such window: target window already closed" in str(e):
continue
except Exception as e:
print(f"Error processing coupon: {e}")
finally:
print('-' * 50)
print(f'{"-" * 70}\nFinished retrieving promotions.\n{"-" * 70}')
print("\nWant coupons on the go? Become a Gold Sponsor for exclusive Discord Bot support!")
print("Gain instant access to coupon notifications and more. Visit our [Gold Sponsor page](https://github.com/sponsors/Prem-ium) for more details.")
def main():
"""Main entry point for running the program."""
run_program()
if __name__ == '__main__':
signal.signal(signal.SIGINT, handle_shutdown_signal)
if not KEEP_ALIVE:
main()
else:
while True:
try:
main()
time.sleep(3600)
except:
print(traceback.format_exc())
time.sleep(60)