justin/Python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added support for webhooks.

Browse Source
This commit is contained in:
JustinOros
2025-07-24 11:00:56 -07:00
parent b91c86e01f
commit d1927036b9
1 changed files with 53 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
diff-web.py
+53 -46
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
# Description: Monitor a website for changes.
# Usage: python3 diff-web.py https://example.com --time 60 --email user@example.com
# Usage: python3 diff-web.py https://example.com --time 300 --email user@example.com --hook https://example.com/webhook
# Author: Justin Oros
# Source: https://github.com/JustinOros
@@ -8,6 +8,7 @@ import requests
import hashlib
import os
import time
import sys
import argparse
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
from datetime import datetime
@@ -21,12 +22,30 @@ SMTP_USERNAME = os.getenv("EMAIL_USER")
SMTP_PASSWORD = os.getenv("EMAIL_PASS")
EMAIL_FROM = SMTP_USERNAME
sys.argv = [arg if not arg.startswith('/') else '-' + arg[1:] for arg in sys.argv]
parser = argparse.ArgumentParser(description="Monitor a website for changes.", add_help=False)
parser.add_argument("--help", action="help", help="Show this help message and exit")
parser.add_argument("url", nargs='?', help="The website URL or domain to monitor.")
parser.add_argument("-d", "--domain", help="Domain to monitor.")
parser.add_argument("-t", "--time", type=int, default=60,
help="Time interval between checks in seconds (default: 60).")
parser.add_argument("-l", "--log", type=str,
help="Optional log file name (default: domain.tld.log)")
parser.add_argument("-q", "--quiet", action="store_true",
help="Suppress console output.")
parser.add_argument("-e", "--email", nargs='+',
help="Email address(es) to notify on changes.")
parser.add_argument("-h", "--hook", type=str,
help="Webhook URL to POST to on changes.")
args = parser.parse_args()
def normalize_url(url):
parsed = urlparse(url)
if not parsed.scheme:
url = 'https://' + url
parsed = urlparse(url)
query = parse_qs(parsed.query)
stripped_query = {k: v for k, v in query.items() if not k.startswith(('utm_', 'fbclid', 'gclid'))}
new_query = urlencode(stripped_query, doseq=True)
@@ -82,21 +101,25 @@ def send_email(subject, body, recipients):
except Exception as e:
print(f"Failed to send email: {e}")
def monitor_website(url, interval, log_file, quiet=False, recipients=None):
def send_webhook(url, payload):
try:
response = requests.post(url, json=payload, timeout=10)
response.raise_for_status()
except Exception as e:
print(f"Failed to send webhook to {url}: {e}")
def monitor_website(url, interval, log_file, quiet=False, recipients=None, webhook_url=None):
url = normalize_url(url)
domain = urlparse(url).netloc
cache_file = get_cache_filename(url)
if not os.path.exists(log_file):
with open(log_file, 'w') as f:
pass
initial_message = f"[{timestamp()}] Monitoring {domain}"
if not quiet:
print(initial_message)
with open(log_file, 'a') as f:
f.write(initial_message + '\n')
while True:
try:
current_content = fetch_content(url)
@@ -105,7 +128,6 @@ def monitor_website(url, interval, log_file, quiet=False, recipients=None):
log_message(f"[{timestamp()}] [ERROR] Failed to fetch {url}: {e}", log_file, quiet)
time.sleep(interval)
continue
is_changed = True
if os.path.exists(cache_file):
with open(cache_file, 'r') as f:
@@ -117,49 +139,34 @@ def monitor_website(url, interval, log_file, quiet=False, recipients=None):
subject = f"Website Change Detected: {domain}"
body = f"A change was detected on {url} at {timestamp()}."
send_email(subject, body, recipients)
if webhook_url:
payload = {
"url": url,
"domain": domain,
"timestamp": timestamp(),
"message": "Change detected on monitored website."
}
send_webhook(webhook_url, payload)
else:
log_message(f"[{timestamp()}] First-time check for {url}: storing baseline.", log_file, quiet)
with open(cache_file, 'w') as f:
f.write(current_hash)
time.sleep(interval)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Monitor a website for changes.")
parser.add_argument("url", nargs='?', help="The website URL or domain to monitor.")
parser.add_argument("-d", "--domain", help="Domain to monitor.")
parser.add_argument("-t", "--time", type=int, default=60,
help="Time interval between checks in seconds (default: 60).")
parser.add_argument("-l", "--log", type=str,
help="Optional log file name (default: domain.tld.log)")
parser.add_argument("-q", "--quiet", action="store_true",
help="Suppress console output.")
parser.add_argument("-e", "--email", nargs='+',
help="Email address(es) to notify on changes.")
args = parser.parse_args()
target = args.domain if args.domain else args.url
if not target:
parser.error("You must specify a URL/domain to monitor either as positional argument or with -d/--domain.")
# Normalize target for filename extraction
if not urlparse(target).scheme:
target = "https://" + target
log_file = get_log_filename(target, args.log)
if not os.path.exists(log_file):
print(f"Log file created at {os.path.abspath(log_file)}")
try:
monitor_website(target, args.time, log_file, args.quiet, args.email)
except KeyboardInterrupt:
message = f"[{timestamp()}] Monitoring halted by user (^C)."
if not args.quiet:
print(message)
with open(log_file, 'a') as f:
f.write(message + '\n')
target = args.domain if args.domain else args.url
if not target:
parser.error("You must specify a URL/domain to monitor either as positional argument or with -d/--domain.")
if not urlparse(target).scheme:
target = "https://" + target
log_file = get_log_filename(target, args.log)
if not os.path.exists(log_file):
print(f"Log file created at {os.path.abspath(log_file)}")
try:
monitor_website(target, args.time, log_file, args.quiet, args.email, args.hook)
except KeyboardInterrupt:
message = f"[{timestamp()}] Monitoring halted by user (^C)."
if not args.quiet:
print(message)
with open(log_file, 'a') as f:
f.write(message + '\n')