diff --git a/README.md b/README.md index f01b51c..a2ddafa 100644 --- a/README.md +++ b/README.md @@ -6,12 +6,14 @@ A comprehensive web scraper for tracking product prices across multiple e-commer - **Multi-site Price Tracking**: Monitor prices across JJ Food Service, A to Z Catering, and Amazon UK - **Beautiful Web UI**: Clean, responsive interface for managing products and viewing price history +- **Daily Shopping Lists**: Automatically generate shopping lists with the best prices for each store - **Price Alerts**: Get notified when products reach your target price - **Historical Data**: View price trends with interactive charts - **Automated Scraping**: Schedule regular price checks -- **Multiple Notifications**: Email and webhook notifications +- **Multiple Notifications**: Email and webhook notifications for price alerts and shopping lists - **Robust Scraping**: Built-in retry logic, rotating user agents, and rate limiting - **Special Pricing Detection**: Automatically detects and prioritizes delivery prices and special offers +- **Smart Shopping**: Compare prices across stores and get recommendations for the best deals ## Quick Start 🚀 @@ -37,6 +39,7 @@ A comprehensive web scraper for tracking product prices across multiple e-commer The web interface provides: - **Dashboard**: Overview of all tracked products with current prices +- **Shopping Lists**: Daily shopping lists showing the best deals for each store - **Add Products**: Easy form to add new products with URLs from multiple sites - **Product Details**: Detailed view with price history charts and statistics - **Settings**: Configuration management and system health checks @@ -256,3 +259,46 @@ This project is for educational purposes. Please review the terms of service of --- **Happy price tracking! 🛍️** + +## Shopping Lists 🛍️ + +The price tracker automatically generates daily shopping lists showing the best deals for each store: + +### Features +- **Best Price Selection**: Automatically finds the lowest current price for each product +- **Store-Specific Lists**: Separate shopping lists for each tracked store +- **Savings Calculation**: Shows how much you save compared to target prices +- **Customizable Preferences**: Set minimum savings thresholds, maximum items, and delivery schedules +- **Multiple Formats**: View online, print, export to CSV, or email daily lists + +### Daily Automation +Set up automatic daily shopping lists: + +```bash +# Generate and send shopping lists manually +python shopping_list_scheduler.py --force + +# Test without sending (dry run) +python shopping_list_scheduler.py --dry-run + +# Set up daily cron job (9 AM daily) +echo "0 9 * * * cd /path/to/price-tracker && python shopping_list_scheduler.py" | crontab - +``` + +### Shopping List Preferences + +For each store, you can configure: +- **Enable/Disable**: Turn shopping lists on/off per store +- **Minimum Savings**: Only include items with savings above threshold +- **Maximum Items**: Limit list size to avoid overwhelming +- **Out of Stock**: Include/exclude unavailable items +- **Delivery Time**: Schedule when to send daily lists +- **Notifications**: Email and/or webhook delivery + +### Web Interface + +Access shopping lists via the web interface: +- **View All Lists**: See current shopping lists for all stores +- **Store Details**: Detailed view with full product information +- **Send Immediately**: Manually trigger email/webhook delivery +- **Manage Preferences**: Configure settings per store diff --git a/main.py b/main.py index b17e48e..882d25b 100644 --- a/main.py +++ b/main.py @@ -83,6 +83,47 @@ async def run_scraper(): raise +def run_shopping_lists(): + """Generate and optionally send daily shopping lists.""" + from src.config import Config + from src.database import DatabaseManager + from src.notification import NotificationManager + from src.shopping_list import AutoShoppingListGenerator + + config = Config() + db_manager = DatabaseManager(config.database_path) + notification_manager = NotificationManager(config) + shopping_generator = AutoShoppingListGenerator(db_manager, notification_manager) + + print("🛒 Generating automated shopping lists...") + + # Generate shopping lists + shopping_lists = shopping_generator.generate_shopping_lists() + summary = shopping_generator.get_summary_stats() + + # Display results + print(f"\n📊 Summary:") + print(f" • {summary['total_products']} products tracked") + print(f" • £{summary['total_cost']:.2f} total cost") + print(f" • £{summary['total_savings']:.2f} total savings") + print(f" • {summary['store_count']} stores involved") + + for store_list in shopping_lists: + print(f"\n🏪 {store_list.store_display_name}: {store_list.item_count} items (£{store_list.total_cost:.2f})") + + # Send daily list if email is configured + email_config = config.notification_config.get('email', {}) + if email_config.get('enabled') and email_config.get('recipient_email'): + print(f"\n📧 Sending daily shopping list...") + success = shopping_generator.send_daily_shopping_list() + if success: + print(f"✅ Shopping list sent successfully!") + else: + print(f"❌ Failed to send shopping list") + + return shopping_lists + + def run_web_ui(): """Run the web UI for managing products and viewing price history.""" import os @@ -99,14 +140,16 @@ def run_web_ui(): def main(): parser = argparse.ArgumentParser(description='Price Tracker') - parser.add_argument('--mode', choices=['scrape', 'web'], default='web', - help='Run mode: scrape prices or start web UI') + parser.add_argument('--mode', choices=['scrape', 'web', 'shopping'], default='web', + help='Run mode: scrape prices, start web UI, or generate shopping lists') parser.add_argument('--config', help='Path to config file') args = parser.parse_args() if args.mode == 'scrape': asyncio.run(run_scraper()) + elif args.mode == 'shopping': + run_shopping_lists() else: run_web_ui() diff --git a/shopping_list_scheduler.py b/shopping_list_scheduler.py new file mode 100755 index 0000000..ca7c6f9 --- /dev/null +++ b/shopping_list_scheduler.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +""" +Daily shopping list scheduler for price tracker +Run this script daily (e.g., via cron) to automatically generate and send shopping lists. +""" + +import sys +import os +import asyncio +import logging +from datetime import datetime, time + +# Add the src directory to Python path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) + +from src.config import Config +from src.database import DatabaseManager +from src.notification import NotificationManager +from src.shopping_list import ShoppingListManager + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler('shopping_list_scheduler.log'), + logging.StreamHandler() + ] +) + +logger = logging.getLogger(__name__) + + +async def main(): + """Main scheduler function.""" + logger.info("Starting shopping list scheduler") + + try: + # Initialize components + config = Config() + + if config.has_config_error(): + logger.error(f"Configuration error: {config.get_config_error()}") + return + + db_manager = DatabaseManager(config.database_path) + notification_manager = NotificationManager(config) + shopping_list_manager = ShoppingListManager(db_manager, notification_manager) + + # Generate shopping lists for all enabled stores + logger.info("Generating shopping lists for all enabled stores") + shopping_lists = shopping_list_manager.generate_all_shopping_lists() + + if not shopping_lists: + logger.warning("No shopping lists generated - no enabled stores or no products") + return + + logger.info(f"Generated {len(shopping_lists)} shopping lists") + + # Check which stores should send at this time + current_time = datetime.now().time() + lists_to_send = {} + + for store_name, shopping_list in shopping_lists.items(): + preferences = shopping_list_manager.get_store_preferences(store_name) + + # Parse send time + send_time_str = preferences.get('send_time', '09:00') + try: + send_time = time.fromisoformat(send_time_str) + # Allow a 30-minute window around the scheduled time + time_diff = abs((current_time.hour * 60 + current_time.minute) - + (send_time.hour * 60 + send_time.minute)) + + if time_diff <= 30: # Within 30 minutes + lists_to_send[store_name] = shopping_list + logger.info(f"Scheduling {store_name} for sending (scheduled: {send_time_str}, current: {current_time.strftime('%H:%M')})") + else: + logger.info(f"Skipping {store_name} - not scheduled time (scheduled: {send_time_str}, current: {current_time.strftime('%H:%M')})") + + except ValueError: + logger.warning(f"Invalid send time format for {store_name}: {send_time_str}") + # Default to sending if time format is invalid + lists_to_send[store_name] = shopping_list + + if not lists_to_send: + logger.info("No shopping lists scheduled for this time") + return + + # Send the scheduled shopping lists + logger.info(f"Sending {len(lists_to_send)} shopping lists") + results = await shopping_list_manager.send_shopping_lists(lists_to_send) + + # Log results + successful = 0 + for store_name, success in results.items(): + if success: + successful += 1 + logger.info(f"Successfully sent shopping list for {store_name}") + else: + logger.error(f"Failed to send shopping list for {store_name}") + + logger.info(f"Shopping list scheduler completed: {successful}/{len(results)} sent successfully") + + except Exception as e: + logger.error(f"Shopping list scheduler failed: {str(e)}", exc_info=True) + raise + + +def force_send_all(): + """Force send all shopping lists regardless of schedule.""" + logger.info("Force sending all shopping lists") + + async def force_send(): + config = Config() + + if config.has_config_error(): + logger.error(f"Configuration error: {config.get_config_error()}") + return + + db_manager = DatabaseManager(config.database_path) + notification_manager = NotificationManager(config) + shopping_list_manager = ShoppingListManager(db_manager, notification_manager) + + shopping_lists = shopping_list_manager.generate_all_shopping_lists() + + if shopping_lists: + results = await shopping_list_manager.send_shopping_lists(shopping_lists) + successful = sum(1 for success in results.values() if success) + logger.info(f"Force send completed: {successful}/{len(results)} sent successfully") + else: + logger.warning("No shopping lists to send") + + asyncio.run(force_send()) + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Daily shopping list scheduler") + parser.add_argument('--force', action='store_true', + help='Force send all shopping lists regardless of schedule') + parser.add_argument('--dry-run', action='store_true', + help='Generate lists but do not send them') + + args = parser.parse_args() + + if args.force: + force_send_all() + elif args.dry_run: + logger.info("Dry run mode - generating lists without sending") + + config = Config() + if config.has_config_error(): + logger.error(f"Configuration error: {config.get_config_error()}") + sys.exit(1) + + db_manager = DatabaseManager(config.database_path) + notification_manager = NotificationManager(config) + shopping_list_manager = ShoppingListManager(db_manager, notification_manager) + + shopping_lists = shopping_list_manager.generate_all_shopping_lists() + + for store_name, shopping_list in shopping_lists.items(): + logger.info(f"{store_name}: {shopping_list.item_count} items, " + f"£{shopping_list.total_cost:.2f} total, " + f"£{shopping_list.total_savings:.2f} savings") + else: + asyncio.run(main()) diff --git a/src/database.py b/src/database.py index 6a9cd0c..0fde2da 100644 --- a/src/database.py +++ b/src/database.py @@ -235,3 +235,7 @@ class DatabaseManager: } return stats + + def get_connection(self): + """Get a database connection.""" + return sqlite3.connect(self.db_path) diff --git a/src/shopping_list.py b/src/shopping_list.py new file mode 100644 index 0000000..191c0e3 --- /dev/null +++ b/src/shopping_list.py @@ -0,0 +1,830 @@ +""" +Automated shopping list generator for price tracker +Analyzes scraped prices to determine cheapest store for each product +""" + +import json +import logging +from datetime import datetime, timedelta +from typing import Dict, List, Tuple, Optional, Any +from dataclasses import dataclass + +from .database import DatabaseManager +from .notification import NotificationManager + +logger = logging.getLogger(__name__) + + +@dataclass +class ShoppingItem: + """Represents an item in a shopping list.""" + product_id: int + product_name: str + current_price: float + store_name: str + store_url: str + last_updated: datetime + savings_vs_most_expensive: float = 0.0 + + +@dataclass +class StoreShoppingList: + """Represents a complete shopping list for one store.""" + store_name: str + store_display_name: str + base_url: str + items: List[ShoppingItem] + total_cost: float + total_savings: float + item_count: int + + +class AutoShoppingListGenerator: + """Generates automated shopping lists based on current best prices.""" + + def __init__(self, db_manager: DatabaseManager, notification_manager: NotificationManager = None): + self.db_manager = db_manager + self.notification_manager = notification_manager + + # Store display names and URLs + self.store_info = { + 'jjfoodservice': { + 'display_name': 'JJ Food Service', + 'base_url': 'https://www.jjfoodservice.com' + }, + 'atoz_catering': { + 'display_name': 'A to Z Catering', + 'base_url': 'https://www.atoz-catering.co.uk' + }, + 'amazon_uk': { + 'display_name': 'Amazon UK', + 'base_url': 'https://www.amazon.co.uk' + } + } + + def get_current_best_prices(self) -> Dict[int, Dict]: + """Get the current cheapest price for each product across all stores.""" + conn = self.db_manager.get_connection() + cursor = conn.cursor() + + # Get latest price for each product from each store + query = """ + WITH latest_prices AS ( + SELECT + p.id as product_id, + p.name as product_name, + ph.site_name, + ph.price, + ph.timestamp, + p.urls, + ROW_NUMBER() OVER (PARTITION BY p.id, ph.site_name ORDER BY ph.timestamp DESC) as rn + FROM products p + LEFT JOIN price_history ph ON p.id = ph.product_id + WHERE ph.price IS NOT NULL AND ph.price > 0 AND p.active = 1 + ), + current_prices AS ( + SELECT * FROM latest_prices WHERE rn = 1 + ), + cheapest_per_product AS ( + SELECT + product_id, + product_name, + MIN(price) as min_price, + MAX(price) as max_price + FROM current_prices + GROUP BY product_id, product_name + ) + SELECT + cp.product_id, + cp.product_name, + cp.site_name, + cp.price, + cp.timestamp, + cp.urls, + cpp.min_price, + cpp.max_price + FROM current_prices cp + JOIN cheapest_per_product cpp ON cp.product_id = cpp.product_id + WHERE cp.price = cpp.min_price + ORDER BY cp.product_name, cp.site_name + """ + + cursor.execute(query) + results = cursor.fetchall() + conn.close() + + # Group by product (handle ties where multiple stores have same lowest price) + best_prices = {} + for row in results: + product_id = row[0] + if product_id not in best_prices: + best_prices[product_id] = [] + + # Parse URLs from JSON + urls = json.loads(row[5]) if row[5] else {} + site_name = row[2] + store_url = urls.get(site_name, "") + + best_prices[product_id].append({ + 'product_name': row[1], + 'store_name': site_name, + 'price': row[3], + 'scraped_at': datetime.fromisoformat(row[4]), + 'store_url': store_url, + 'min_price': row[6], + 'max_price': row[7] + }) + + return best_prices + + def generate_shopping_lists(self) -> List[StoreShoppingList]: + """Generate automated shopping lists for each store.""" + best_prices = self.get_current_best_prices() + + # Group items by store + store_lists = {} + + for product_id, price_options in best_prices.items(): + # If multiple stores have the same lowest price, prefer in order: JJ Food Service, A to Z, Amazon + store_priority = {'jjfoodservice': 1, 'atoz_catering': 2, 'amazon_uk': 3} + best_option = min(price_options, key=lambda x: (x['price'], store_priority.get(x['store_name'], 999))) + + store_name = best_option['store_name'] + if store_name not in store_lists: + store_lists[store_name] = [] + + # Calculate savings vs most expensive option + savings = best_option['max_price'] - best_option['price'] + + shopping_item = ShoppingItem( + product_id=product_id, + product_name=best_option['product_name'], + current_price=best_option['price'], + store_name=store_name, + store_url=best_option['store_url'], + last_updated=best_option['scraped_at'], + savings_vs_most_expensive=savings + ) + + store_lists[store_name].append(shopping_item) + + # Convert to StoreShoppingList objects + shopping_lists = [] + for store_name, items in store_lists.items(): + store_info = self.store_info.get(store_name, { + 'display_name': store_name.title(), + 'base_url': '' + }) + + total_cost = sum(item.current_price for item in items) + total_savings = sum(item.savings_vs_most_expensive for item in items) + + shopping_list = StoreShoppingList( + store_name=store_name, + store_display_name=store_info['display_name'], + base_url=store_info['base_url'], + items=sorted(items, key=lambda x: x.product_name.lower()), + total_cost=total_cost, + total_savings=total_savings, + item_count=len(items) + ) + + shopping_lists.append(shopping_list) + + # Sort by total cost (cheapest store first) + return sorted(shopping_lists, key=lambda x: x.total_cost) + + def get_summary_stats(self) -> Dict: + """Get summary statistics about the current shopping recommendations.""" + shopping_lists = self.generate_shopping_lists() + + total_products = sum(sl.item_count for sl in shopping_lists) + total_cost = sum(sl.total_cost for sl in shopping_lists) + total_savings = sum(sl.total_savings for sl in shopping_lists) + + # Find the most recommended store + best_store = max(shopping_lists, key=lambda x: x.item_count) if shopping_lists else None + + return { + 'total_products': total_products, + 'total_cost': total_cost, + 'total_savings': total_savings, + 'store_count': len(shopping_lists), + 'most_items_store': best_store.store_display_name if best_store else None, + 'most_items_count': best_store.item_count if best_store else 0, + 'generated_at': datetime.now() + } + + def send_daily_shopping_list(self) -> bool: + """Send daily shopping list via email/webhook.""" + if not self.notification_manager: + return False + + try: + shopping_lists = self.generate_shopping_lists() + summary = self.get_summary_stats() + + if not shopping_lists: + return False + + # Generate email content + subject = f"Daily Shopping List - £{summary['total_cost']:.2f} across {summary['store_count']} stores" + + html_content = self._generate_email_html(shopping_lists, summary) + text_content = self._generate_email_text(shopping_lists, summary) + + # Send email notification + success = self.notification_manager.send_email( + subject=subject, + message=text_content, + html_message=html_content + ) + + # Send webhook if configured + if self.notification_manager.config.get('webhook', {}).get('enabled'): + webhook_data = { + 'type': 'daily_shopping_list', + 'summary': summary, + 'store_lists': [ + { + 'store': sl.store_display_name, + 'items': len(sl.items), + 'total': sl.total_cost, + 'savings': sl.total_savings + } + for sl in shopping_lists + ] + } + self.notification_manager.send_webhook(webhook_data) + + return success + + except Exception as e: + logger.error(f"Error sending daily shopping list: {str(e)}") + return False + + def _generate_email_html(self, shopping_lists: List[StoreShoppingList], summary: Dict) -> str: + """Generate HTML email content for shopping lists.""" + html = f""" + +
+ + + +Best prices found for {summary['generated_at'].strftime('%B %d, %Y')}
+{store_list.item_count} items - £{store_list.total_cost:.2f} total (Save £{store_list.total_savings:.2f})
+Generated: {shopping_list.generated_at.strftime('%Y-%m-%d %H:%M')}
+Items: {shopping_list.item_count}
+Total Cost: £{shopping_list.total_cost:.2f}
+Total Savings: £{shopping_list.total_savings:.2f}
+{item.description}
' + + html += f'Target Price: £{item.target_price:.2f}
' + + if item.savings: + html += f'Save £{item.savings:.2f}!
' + + if not item.availability: + html += 'OUT OF STOCK
' + + if item.url: + html += f'' + + html += 'Total Shopping Cost: £{shopping_list.total_cost:.2f}
+Total Savings: £{shopping_list.total_savings:.2f}
+Happy shopping!
+ + +""" + return html + + def _generate_webhook_data(self, shopping_list: StoreShoppingList) -> Dict[str, Any]: + """Generate webhook data for shopping list.""" + return { + "type": "daily_shopping_list", + "store_name": shopping_list.store_name, + "generated_at": shopping_list.generated_at.isoformat(), + "summary": { + "item_count": shopping_list.item_count, + "total_cost": shopping_list.total_cost, + "total_savings": shopping_list.total_savings + }, + "items": [ + { + "product_id": item.product_id, + "name": item.product_name, + "price": item.best_price, + "savings": item.savings, + "availability": item.availability, + "url": item.url + } + for item in shopping_list.items + ] + } diff --git a/src/web_ui.py b/src/web_ui.py index 70f7a5b..76df37e 100644 --- a/src/web_ui.py +++ b/src/web_ui.py @@ -18,6 +18,7 @@ from .database import DatabaseManager from .config import Config from .scraper_manager import ScraperManager from .notification import NotificationManager +from .shopping_list import AutoShoppingListGenerator from .utils import format_price, group_results_by_status @@ -68,6 +69,7 @@ def create_app(): db_manager = DatabaseManager(config.database_path) scraper_manager = ScraperManager(config) notification_manager = NotificationManager(config) + shopping_list_generator = AutoShoppingListGenerator(db_manager, notification_manager) class ProductForm(FlaskForm): name = StringField('Product Name', validators=[DataRequired()]) @@ -366,5 +368,110 @@ def create_app(): flash(f'Error deleting product: {str(e)}', 'error') return redirect(url_for('index')) + + @app.route('/shopping-lists') + def shopping_lists(): + """Display automated shopping lists based on best prices.""" + try: + shopping_lists = shopping_list_generator.generate_shopping_lists() + summary = shopping_list_generator.get_summary_stats() + + return render_template('shopping_lists.html', + shopping_lists=shopping_lists, + summary=summary) + except Exception as e: + flash(f'Error generating shopping lists: {str(e)}', 'danger') + return redirect(url_for('index')) + + @app.route('/shopping-list/Items to Buy
+Total Cost
+Total Savings
+Savings Rate
+| # | +Product | +Price | +Savings | +Last Updated | +Link | +
|---|---|---|---|---|---|
| + {{ loop.index }} + | +
+ {{ item.product_name }}
+ {% if item.savings_vs_most_expensive > 0 %}
+ + Best price! + + {% endif %} + |
+ + + £{{ "%.2f"|format(item.current_price) }} + + | ++ {% if item.savings_vs_most_expensive > 0 %} + + £{{ "%.2f"|format(item.savings_vs_most_expensive) }} + + {% else %} + - + {% endif %} + | ++ + {{ item.last_updated.strftime('%Y-%m-%d %H:%M') if item.last_updated else 'Unknown' }} + + | ++ {% if item.store_url %} + + + + {% else %} + - + {% endif %} + | +
| Totals: | ++ £{{ "%.2f"|format(shopping_list.total_cost) }} + | ++ £{{ "%.2f"|format(shopping_list.total_savings) }} + | ++ | ||
Total Products
+Total Cost
+Total Savings
+Stores to Visit
+No price data found to generate shopping lists. Make sure you have:
+The system automatically analyzes all your tracked products and finds the store with the lowest current price for each item.
+ +Products are grouped by store, so you know exactly what to buy from each location for maximum savings.
+See how much you save compared to buying each item at the most expensive store.
+ +Set up daily email notifications to get your optimized shopping list delivered automatically.
+