E-commerce Price Monitoring System

import requests
import sqlite3
import smtplib
import time
import json
from datetime import datetime
from email.mime.text import MimeText
from typing import List, Dict

class EcommerceMonitor:
    def __init__(self, VENYM_SEARCH_api_key: str, db_path: str = "ecommerce.db"):
        self.api_key = VENYM_SEARCH_api_key
        self.db_path = db_path
        self.base_url = "https://www.search.venym.io/api/v1"
        self.init_database()

    def init_database(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute('''
            CREATE TABLE IF NOT EXISTS products (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                search_query TEXT NOT NULL,
                target_price REAL,
                current_price REAL,
                url TEXT,
                last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                price_history TEXT
            )
        ''')

        cursor.execute('''
            CREATE TABLE IF NOT EXISTS alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                product_id INTEGER,
                alert_type TEXT,
                message TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (product_id) REFERENCES products (id)
            )
        ''')

        conn.commit()
        conn.close()

    def add_product(self, name: str, search_query: str, target_price: float = None):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute('''
            INSERT INTO products (name, search_query, target_price, price_history)
            VALUES (?, ?, ?, ?)
        ''', (name, search_query, target_price, '[]'))

        product_id = cursor.lastrowid
        conn.commit()
        conn.close()

        return product_id

    def search_product_prices(self, search_query: str) -> List[Dict]:
        response = requests.post(
            f"{self.base_url}/search",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "query": f"{search_query} price buy",
                "max_results": 20,
                "country": "us"
            }
        )

        if response.status_code != 200:
            return []

        results = response.json()["search_results"]

        ecommerce_domains = [
            'amazon.com', 'ebay.com', 'walmart.com', 'target.com',
            'bestbuy.com', 'homedepot.com', 'lowes.com', 'costco.com',
            'newegg.com', 'etsy.com', 'shopify.com'
        ]

        filtered_results = []
        for result in results:
            domain = result['link'].split('/')[2].replace('www.', '')
            if any(ecom in domain for ecom in ecommerce_domains):
                filtered_results.append(result)

        return filtered_results

    def extract_price_from_page(self, url: str) -> Dict:
        response = requests.post(
            f"{self.base_url}/scrape",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "url": url,
                "extract_options": ["title", "text", "metadata"],
                "wait_for_selector": "body",
                "remove_selectors": [".ads", ".popup", ".modal"]
            }
        )

        if response.status_code != 200:
            return {}

        data = response.json()
        content = data.get("primary_content", {})

        import re
        text = content.get("text", "")

        price_patterns = [
            r'\$([0-9,]+\.?[0-9]*)',
            r'USD\s*([0-9,]+\.?[0-9]*)',
            r'Price:\s*\$([0-9,]+\.?[0-9]*)',
        ]

        prices = []
        for pattern in price_patterns:
            matches = re.findall(pattern, text)
            for match in matches:
                try:
                    price = float(match.replace(',', ''))
                    if 1 < price < 10000:
                        prices.append(price)
                except ValueError:
                    continue

        return {
            "url": url,
            "title": content.get("title", ""),
            "prices": list(set(prices)),
            "lowest_price": min(prices) if prices else None,
            "extracted_at": datetime.now().isoformat()
        }

def main():
    monitor = EcommerceMonitor("your-venym-search-api-key")
    monitor.add_product("iPhone 15 Pro", "iPhone 15 Pro 256GB", target_price=900.00)
    monitor.add_product("Sony WH-1000XM5", "Sony WH-1000XM5 headphones", target_price=300.00)

if __name__ == "__main__":
    main()

import streamlit as st
import sqlite3
import pandas as pd
import plotly.express as px
import json

st.set_page_config(page_title="E-commerce Price Monitor", layout="wide")
st.title("E-commerce Price Monitor Dashboard")

@st.cache_resource
def get_connection():
    return sqlite3.connect("ecommerce.db", check_same_thread=False)

conn = get_connection()
cursor = conn.cursor()

cursor.execute("SELECT COUNT(*) FROM products")
total_products = cursor.fetchone()[0]

cursor.execute("SELECT COUNT(*) FROM alerts WHERE created_at > datetime('now', '-24 hours')")
alerts_24h = cursor.fetchone()[0]

col1, col2 = st.columns(2)
col1.metric("Products Monitored", total_products)
col2.metric("Alerts (24h)", alerts_24h)

cursor.execute('SELECT id, name, current_price, target_price FROM products')
products_data = cursor.fetchall()
if products_data:
    df = pd.DataFrame(products_data, columns=['ID', 'Product', 'Current Price', 'Target Price'])
    st.dataframe(df, use_container_width=True)