mydatascraper/competitor_compare

1import pandas as pd
2import os
3from datetime import datetime
4from sqlalchemy import create_engine, text
5from app.config import settings
6from app.utils.cloud_storage import upload_to_s3, upload_to_gcs
7from typing import Optional, List
8
9
10class BulkExportService:
11
12    def __init__(self):
13        self.engine = create_engine(settings.DATABASE_URL_SYNC)
14        os.makedirs(settings.EXPORT_PATH, exist_ok=True)
15
16    def export_full_dataset(
17        self,
18        retailers: Optional[List[str]] = None,
19        categories: Optional[List[str]] = None,
20        file_format: str = "parquet",
21        destination: str = "local",
22    ) -> dict:
23        """Export the full current price dataset."""
24
25        query = """
26            WITH latest_prices AS (
27                SELECT DISTINCT ON (p.product_id, p.store_id)
28                    p.id as price_id,
29                    pr.upc,
30                    pr.name as product_name,
31                    pr.brand,
32                    pr.category,
33                    pr.subcategory,
34                    pr.size,
35                    pr.unit,
36                    p.retailer,
37                    s.name as store_name,
38                    s.store_id as retailer_store_id,
39                    s.city,
40                    s.state,
41                    s.zip_code,
42                    p.regular_price,
43                    p.sale_price,
44                    p.unit_price,
45                    p.unit_price_unit,
46                    p.in_stock,
47                    p.is_on_sale,
48                    p.promo_description,
49                    p.observed_at,
50                    p.currency
51                FROM prices p
52                JOIN products pr ON p.product_id = pr.id
53                JOIN stores s ON p.store_id = s.id
54                WHERE pr.is_active = true
55                {retailer_filter}
56                {category_filter}
57                ORDER BY p.product_id, p.store_id, p.observed_at DESC
58            )
59            SELECT * FROM latest_prices
60            ORDER BY category, product_name, retailer
61        """
62
63        retailer_filter = ""
64        category_filter = ""
65        if retailers:
66            retailer_list = ", ".join(f"'{r}'" for r in retailers)
67            retailer_filter = f"AND p.retailer IN ({retailer_list})"
68        if categories:
69            category_list = ", ".join(f"'{c}'" for c in categories)
70            category_filter = f"AND pr.category IN ({category_list})"
71
72        query = query.format(
73            retailer_filter=retailer_filter,
74            category_filter=category_filter,
75        )
76
77        df = pd.read_sql(query, self.engine)
78
79        # Generate filename
80        timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
81        filename = f"grocery_prices_{timestamp}"
82
83        if file_format == "parquet":
84            filepath = os.path.join(settings.EXPORT_PATH, f"{filename}.parquet")
85            df.to_parquet(filepath, index=False, engine="pyarrow")
86        elif file_format == "csv":
87            filepath = os.path.join(settings.EXPORT_PATH, f"{filename}.csv")
88            df.to_csv(filepath, index=False)
89        elif file_format == "json":
90            filepath = os.path.join(settings.EXPORT_PATH, f"{filename}.json")
91            df.to_json(filepath, orient="records", lines=True)
92        else:
93            raise ValueError(f"Unsupported format: {file_format}")
94
95        file_size = os.path.getsize(filepath)
96        record_count = len(df)
97
98        # Upload to cloud if requested
99        cloud_path = None
100        if destination == "s3":
101            cloud_path = upload_to_s3(filepath, f"daily/{filename}.{file_format}")
102        elif destination == "gcs":
103            cloud_path = upload_to_gcs(filepath, f"daily/{filename}.{file_format}")
104
105        return {
106            "filepath": filepath,
107            "cloud_path": cloud_path,
108            "record_count": record_count,
109            "file_size_bytes": file_size,
110            "format": file_format,
111            "destination": destination,
112        }
113
114    def export_daily_snapshot(self, destination: str = "local"):
115        """Export both parquet and CSV for daily delivery."""
116        results = []
117        for fmt in ["parquet", "csv"]:
118            result = self.export_full_dataset(
119                file_format=fmt, destination=destination
120            )
121            results.append(result)
122        return results
123
124    def export_category_files(
125        self,
126        destination: str = "local",
127        file_format: str = "parquet",
128    ):
129        """Export separate files per category."""
130        categories = [
131            "produce", "dairy", "meat", "pantry",
132            "household", "personal_care", "baby",
133        ]
134        results = []
135        for cat in categories:
136            result = self.export_full_dataset(
137                categories=[cat],
138                file_format=file_format,
139                destination=destination,
140            )
141            results.append({"category": cat, **result})
142        return results