Python Product Price Monitor

Python Product Price Monitor

1. Features

With the help of ChatGPT, I wrote a simple price monitoring script in Python.

It supports the following features:

  1. Get the current price
  2. Send an alert when the price drops below a target
  3. Send email notifications
  4. Send a daily email to confirm that the program is still running

2. Code

import requests
from bs4 import BeautifulSoup
import smtplib
from email.mime.text import MIMEText
from email.header import Header
import time
from datetime import datetime, timedelta
import pytz

# Email account information
my_email = "yours@mutou.men"
my_password = "123456"
recipient_email = "abc@outlook.com"

# Product links and target price list. Fill in your desired price after target_price.
products = [
    {"url": "https://www.usa.canon.com/shop/p/refurbished-rf50mm-f1-8-stm", "target_price": 100, "reached_target": False, "last_price": None},
    {"url": "https://www.usa.canon.com/shop/p/refurbished-eos-r8?color=Black&type=Refurbished", "target_price": 600, "reached_target": False, "last_price": None}
]

def get_price(url):
    """Get the product price from the specified URL."""
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        soup = BeautifulSoup(response.content, "html.parser")
        
        if "canon.com" in url:
            price_tag = soup.find("span", {"class": "price"})
            if price_tag:
                price_str = price_tag.text.strip().replace("$", "").replace(",", "")
                return float(price_str)
            else:
                raise ValueError("Price tag not found")
        else:
            raise ValueError("Unsupported URL format")
        
    except requests.RequestException as e:
        print(f"Error getting price for {url}: network issue - {e}")
    except Exception as e:
        print(f"Error getting price for {url}: {e}")
    return None

def send_email(subject, message):
    """Send an email notification with the specified subject and message."""
    try:
        msg = MIMEText(message, "plain", "utf-8")
        msg["Subject"] = Header(subject, "utf-8")
        msg["From"] = my_email
        msg["To"] = recipient_email
        
        server = smtplib.SMTP_SSL("smtp.exmail.qq.com", 465)
        server.login(my_email, my_password)
        server.sendmail(my_email, recipient_email, msg.as_string())
        server.quit()
        
        print("Email sent successfully.")
    except Exception as e:
        print(f"Error sending email: {e}")

def main():
    """Main function for monitoring product prices and sending notifications."""
    # Set timezone to Beijing time
    tz = pytz.timezone('Asia/Shanghai')
    
    while True:
        now = datetime.now(tz)
        current_time = now.strftime("%H:%M:%S")
        
        # Send a confirmation email every morning at 8:00
        if current_time == "08:00:00":
            subject = "Program Running Confirmation"
            message = "Your monitoring program is still running."
            send_email(subject, message)
        
        print("Starting monitoring. Current time:", now)
        
        for p in products:
            current_price = get_price(p["url"])
            if current_price is not None:
                print(f"Product {p['url']} current price is ${current_price}")
                if current_price <= p["target_price"]:
                    if not p["reached_target"] or (p["last_price"] is not None and current_price > p["last_price"]):
                        subject = "Product price reached target!"
                        message = f"Product {p['url']} current price is ${current_price}, which has reached your target price of ${p['target_price']}."
                        send_email(subject, message)
                        print(f"Product {p['url']} current price is ${current_price}")
                        p["reached_target"] = True
                else:
                    p["reached_target"] = False  # Resume monitoring after the price rises
                p["last_price"] = current_price
        
        # Print next check time and wait for 1 hour
        next_check_time = now + timedelta(hours=1)
        print(f"This monitoring cycle has ended. Next check time: {next_check_time} (1-hour interval)")
        time.sleep(3600)

if __name__ == "__main__":
    main()

3. Run the Script in the Background on a VPS

nohup python3 your_script.py > script.log 2>&1 &

4. Check Background Processes

ps aux | grep your_script.py