Working script that fetches products by brand

api/cijene_routes.py

@@ -0,0 +1,426 @@
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import JSONResponse
+import httpx
+import asyncio
+import logging
+import sys
+from typing import Dict, List
+from datetime import datetime
+from db.cijene_repository import CijeneRepository
+
+# Configure logger with proper UTF-8 encoding
+class UTFFormatter(logging.Formatter):
+    def format(self, record):
+        # Ensure the message is properly encoded
+        if hasattr(record, 'msg') and isinstance(record.msg, str):
+            try:
+                record.msg = record.msg.encode('utf-8', errors='replace').decode('utf-8')
+            except:
+                record.msg = record.msg.encode('ascii', errors='replace').decode('ascii')
+        return super().format(record)
+
+# Configure logging with UTF-8 support
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.StreamHandler(sys.stdout),  # Use stdout instead of stderr
+        logging.FileHandler('cijene_processing.log', encoding='utf-8')  # Specify UTF-8 encoding
+    ]
+)
+
+# Apply custom formatter to handle Croatian characters
+for handler in logging.getLogger().handlers:
+    handler.setFormatter(UTFFormatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/cijene", tags=["Cijene API"])
+
+@router.post("/populate-database")
+async def populate_database_from_cijene():
+    """
+    Populate database with products and prices from Cijene API
+    Goes through all brands and fetches their products
+    """
+    logger.info("=== Starting database population from Cijene API ===")
+    repository = CijeneRepository()
+    
+    try:
+        # Get all brands and store chains
+        logger.info("Initializing data fetching...")
+        brands = repository.get_all_brands()
+        store_chains = repository.get_all_store_chains()
+        
+        if not brands:
+            logger.error("No brands found in database")
+            raise HTTPException(status_code=404, detail="No brands found in database")
+        
+        logger.info(f"Found {len(brands)} brands and {len(store_chains)} store chains")
+        
+        results = {
+            "total_brands_processed": 0,
+            "total_products_found": 0,
+            "total_products_inserted": 0,
+            "total_products_updated": 0,
+            "total_store_products_created": 0,
+            "total_price_records_inserted": 0,
+            "errors": []
+        }
+        
+        # API configuration
+        base_url = "https://api.cijene.dev/v1/products/"
+        headers = {
+            "Authorization": "Bearer aibeeboh2cuataChi2doMa2Aacah0eli",
+            "Content-Type": "application/json"
+        }
+        
+        logger.info("Starting brand processing...")
+        
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            for index, brand in enumerate(brands, 1):
+                brand_name = brand['brand_name']
+                brand_id = brand['brand_id']
+                
+                safe_brand_name = repository.safe_log_string(brand_name)
+                logger.info(f"Processing brand {index}/{len(brands)}: '{safe_brand_name}' (ID: {brand_id})")
+                
+                try:
+                    # Make API request for this brand
+                    url = f"{base_url}?q={brand_name}"
+                    logger.info(f"Making API request to: {url}")
+                    
+                    response = await client.get(url, headers=headers)
+                    
+                    if response.status_code != 200:
+                        error_msg = f"API error for brand {safe_brand_name}: HTTP {response.status_code}"
+                        logger.error(error_msg)
+                        results["errors"].append(error_msg)
+                        continue
+                    
+                    logger.info(f"API request successful for brand '{safe_brand_name}'")
+                    data = response.json()
+                    products = data.get("products", [])
+                    
+                    results["total_brands_processed"] += 1
+                    results["total_products_found"] += len(products)
+                    
+                    logger.info(f"Found {len(products)} products for brand '{safe_brand_name}'")
+                    
+                    # Process each product
+                    for product_index, product in enumerate(products, 1):
+                        try:
+                            safe_product_name = repository.safe_log_string(product.get('name', 'unknown'))
+                            logger.debug(f"Processing product {product_index}/{len(products)} for brand '{safe_brand_name}': {safe_product_name}")
+                            await process_product(product, brand_id, store_chains, repository, results)
+                        except Exception as e:
+                            error_msg = f"Error processing product {product.get('ean', 'unknown')} for brand '{safe_brand_name}': {str(e)}"
+                            logger.error(error_msg)
+                            results["errors"].append(error_msg)
+                            continue
+                    
+                    logger.info(f"Completed processing brand '{safe_brand_name}' - Products: {len(products)}")
+                    
+                    # Add small delay to avoid rate limiting
+                    await asyncio.sleep(0.1)
+                    
+                    # Log progress every 10 brands
+                    if index % 10 == 0:
+                        logger.info(f"Progress update: {index}/{len(brands)} brands processed")
+                        logger.info(f"Current stats - Products inserted: {results['total_products_inserted']}, Updated: {results['total_products_updated']}, Store products: {results['total_store_products_created']}, Price records: {results['total_price_records_inserted']}")
+                    
+                except Exception as e:
+                    error_msg = f"Error processing brand {safe_brand_name}: {str(e)}"
+                    logger.error(error_msg)
+                    results["errors"].append(error_msg)
+                    continue
+        
+        logger.info("=== Database population completed ===")
+        logger.info(f"Final results: {results}")
+        
+        return JSONResponse(content={
+            "message": "Database population completed",
+            "results": results
+        })
+        
+    except Exception as e:
+        logger.error(f"Critical error during database population: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to populate database: {str(e)}")
+
+async def process_product(product: Dict, brand_id: str, store_chains: List[Dict], 
+                         repository: CijeneRepository, results: Dict):
+    """Process a single product and its price data"""
+    
+    ean = product.get("ean", "")
+    name = product.get("name", "")
+    quantity = product.get("quantity", "")
+    unit = product.get("unit", "")
+    chains = product.get("chains", [])
+    
+    if not ean:
+        safe_name = repository.safe_log_string(name)
+        logger.warning(f"Skipping product without EAN: {safe_name}")
+        return
+    
+    safe_name = repository.safe_log_string(name)
+    logger.debug(f"Processing product: {safe_name} (EAN: {ean})")
+    
+    # Check if product already exists (by EAN or product_name+brand_id)
+    existing_product_id = repository.check_product_exists(ean, name, brand_id)
+    
+    if existing_product_id:
+        product_id = existing_product_id
+        results["total_products_updated"] += 1
+        logger.debug(f"Product exists, using existing ID: {product_id}")
+    else:
+        # Create new product
+        weight = repository.combine_quantity_unit(quantity, unit)
+        
+        product_data = {
+            "product_ean": ean,
+            "product_name": name,
+            "product_weight": weight,
+            "product_quantity": 1,  # Add default quantity
+            "brand_id": brand_id
+        }
+        
+        product_id = repository.insert_product(product_data)
+        results["total_products_inserted"] += 1
+        logger.debug(f"Created new product with ID: {product_id}")
+    
+    # Process price data for each chain
+    today = datetime.now().date().isoformat()
+    logger.debug(f"Processing {len(chains)} price entries for product {safe_name}")
+    
+    price_records_for_product = 0
+    
+    for chain_data in chains:
+        chain_name = chain_data.get("chain", "")
+        min_price = chain_data.get("min_price")
+        max_price = chain_data.get("max_price")
+        avg_price = chain_data.get("avg_price")
+        
+        if not chain_name:
+            logger.debug("Skipping chain entry without name")
+            continue
+        
+        # Find matching store chain
+        store_chain_id = repository.find_matching_store_chain(chain_name, store_chains)
+        
+        if not store_chain_id:
+            safe_chain_name = repository.safe_log_string(chain_name)
+            logger.debug(f"No matching store chain found for '{safe_chain_name}', skipping price entry")
+            continue
+        
+        # Get or create store_product mapping
+        try:
+            store_product_id = repository.get_or_create_store_product(product_id, store_chain_id)
+            results["total_store_products_created"] += 1
+        except Exception as e:
+            logger.error(f"Failed to get/create store_product mapping: {str(e)}")
+            continue
+        
+        # Validate price data
+        try:
+            min_price_float = float(min_price) if min_price else None
+            max_price_float = float(max_price) if max_price else None
+            avg_price_float = float(avg_price) if avg_price else None
+            
+            if not any([min_price_float, max_price_float, avg_price_float]):
+                safe_chain_name = repository.safe_log_string(chain_name)
+                logger.debug(f"No valid price data for chain '{safe_chain_name}', skipping")
+                continue
+                
+        except (ValueError, TypeError) as e:
+            safe_chain_name = repository.safe_log_string(chain_name)
+            logger.warning(f"Invalid price data for chain '{safe_chain_name}': {e}")
+            continue
+        
+        # Insert price history - FIXED: use store_product_id and correct column name
+        price_data = {
+            "store_product_id": store_product_id,  # CHANGED from product_id
+            "price_date": today,
+            "min_price": min_price_float,
+            "max_price": max_price_float,
+            "avg_price": avg_price_float
+        }
+        
+        repository.insert_price_history(price_data)
+        results["total_price_records_inserted"] += 1
+        price_records_for_product += 1
+    
+    logger.debug(f"Completed processing product {safe_name}: {price_records_for_product} price records inserted")
+
+@router.post("/test-brand/{brand_name}")
+async def test_single_brand(brand_name: str):
+    """
+    Test endpoint that processes a single brand the same way as populate-database
+    Fetches products and inserts them into the database
+    """
+    safe_brand_name = CijeneRepository().safe_log_string(brand_name)
+    logger.info(f"=== Starting single brand processing for: '{safe_brand_name}' ===")
+    repository = CijeneRepository()
+    
+    try:
+        # Get all store chains
+        logger.info("Fetching store chains for processing...")
+        store_chains = repository.get_all_store_chains()
+        
+        # Find the specific brand in database
+        logger.info(f"Looking up brand '{safe_brand_name}' in database...")
+        brands = repository.get_all_brands()
+        target_brand = None
+        
+        for brand in brands:
+            if brand['brand_name'].lower() == brand_name.lower():
+                target_brand = brand
+                break
+        
+        if not target_brand:
+            logger.error(f"Brand '{safe_brand_name}' not found in database")
+            raise HTTPException(status_code=404, detail=f"Brand '{brand_name}' not found in database")
+        
+        brand_id = target_brand['brand_id']
+        logger.info(f"Found brand '{safe_brand_name}' with ID: {brand_id}")
+        
+        results = {
+            "brand_name": brand_name,
+            "brand_id": brand_id,
+            "total_products_found": 0,
+            "total_products_inserted": 0,
+            "total_products_updated": 0,
+            "total_store_products_created": 0,
+            "total_price_records_inserted": 0,
+            "store_chains_matched": [],
+            "store_chains_not_matched": [],
+            "errors": []
+        }
+        
+        # API configuration
+        base_url = "https://api.cijene.dev/v1/products/"
+        headers = {
+            "Authorization": "Bearer aibeeboh2cuataChi2doMa2Aacah0eli",
+            "Content-Type": "application/json"
+        }
+        
+        # Make API request for this brand
+        url = f"{base_url}?q={brand_name}"
+        logger.info(f"Making API request to: {url}")
+        
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(url, headers=headers)
+            
+            if response.status_code != 200:
+                error_msg = f"API error for brand {safe_brand_name}: HTTP {response.status_code}"
+                logger.error(error_msg)
+                raise HTTPException(status_code=response.status_code, detail=error_msg)
+            
+            logger.info(f"API request successful for brand '{safe_brand_name}'")
+            data = response.json()
+            products = data.get("products", [])
+            
+            results["total_products_found"] = len(products)
+            logger.info(f"Found {len(products)} products for brand '{safe_brand_name}'")
+            
+            # Process each product
+            for product_index, product in enumerate(products, 1):
+                try:
+                    safe_product_name = repository.safe_log_string(product.get('name', 'unknown'))
+                    logger.info(f"Processing product {product_index}/{len(products)}: {safe_product_name}")
+                    
+                    # Process product using the same function as populate-database
+                    await process_product(product, brand_id, store_chains, repository, results)
+                    
+                except Exception as e:
+                    error_msg = f"Error processing product {product.get('ean', 'unknown')}: {str(e)}"
+                    logger.error(error_msg)
+                    results["errors"].append(error_msg)
+                    continue
+            
+            # Determine which store chains were matched vs not matched
+            api_chain_names = set()
+            for product in products:
+                for chain_data in product.get("chains", []):
+                    if chain_data.get("chain"):
+                        api_chain_names.add(chain_data.get("chain"))
+            
+            db_chain_names = {repository.normalize_string(sc['store_chain_name']): sc['store_chain_name'] 
+                             for sc in store_chains}
+            
+            for api_chain in api_chain_names:
+                normalized_api_chain = repository.normalize_string(api_chain)
+                if normalized_api_chain in db_chain_names:
+                    if api_chain not in [item['api_name'] for item in results["store_chains_matched"]]:
+                        results["store_chains_matched"].append({
+                            "api_name": api_chain,
+                            "db_name": db_chain_names[normalized_api_chain]
+                        })
+                else:
+                    if api_chain not in results["store_chains_not_matched"]:
+                        results["store_chains_not_matched"].append(api_chain)
+        
+        logger.info(f"=== Completed processing brand '{safe_brand_name}' ===")
+        # Use safe logging for results
+        safe_results = {
+            **results,
+            "brand_name": repository.safe_log_string(results["brand_name"])
+        }
+        logger.info(f"Results: {safe_results}")
+        
+        return JSONResponse(content={
+            "message": f"Successfully processed brand '{brand_name}'",
+            "results": results
+        })
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Critical error processing brand '{safe_brand_name}': {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to process brand: {str(e)}")
+
+# Keep the original test function as a separate endpoint for just fetching data
+@router.get("/fetch-brand-data/{brand_name}")
+async def fetch_brand_data_only(brand_name: str):
+    """Fetch raw API data for a single brand without processing"""
+    safe_brand_name = CijeneRepository().safe_log_string(brand_name)
+    logger.info(f"Fetching raw data for brand: {safe_brand_name}")
+    
+    try:
+        url = f"https://api.cijene.dev/v1/products/?q={brand_name}"
+        headers = {
+            "Authorization": "Bearer aibeeboh2cuataChi2doMa2Aacah0eli",
+            "Content-Type": "application/json"
+        }
+        
+        logger.info(f"Making API request to: {url}")
+        
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(url, headers=headers)
+            
+            if response.status_code != 200:
+                logger.error(f"API request failed with status: {response.status_code}")
+                raise HTTPException(status_code=response.status_code, detail="API request failed")
+            
+            data = response.json()
+            products_count = len(data.get("products", []))
+            logger.info(f"Successfully fetched {products_count} products for brand '{safe_brand_name}'")
+            
+            return data
+            
+    except Exception as e:
+        logger.error(f"Error fetching data for brand '{safe_brand_name}': {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Error fetching data: {str(e)}")
+
+@router.get("/processing-logs")
+async def get_processing_logs():
+    """Get recent processing logs"""
+    try:
+        # Read last 100 lines from log file
+        with open('cijene_processing.log', 'r', encoding='utf-8') as f:
+            lines = f.readlines()
+            return {"logs": lines[-100:]}  # Return last 100 lines
+    except FileNotFoundError:
+        return {"logs": ["No log file found"]}
+    except Exception as e:
+        logger.error(f"Error reading logs: {str(e)}")
+        raise HTTPException(status_code=500, detail="Error reading logs")

api/main.py

@@ -4,6 +4,7 @@ from config.settings import API_HOST, API_PORT
 from api.product_routes import router as product_router
 from api.receipt_routes import router as receipt_router
 from api.scrape_routes import router as scrape_router
+from api.cijene_routes import router as cijene_router
 
 # Initialize FastAPI
 app = FastAPI(title="SupaKuna API")
@@ -22,6 +23,7 @@ app.add_middleware(
 app.include_router(product_router)
 app.include_router(receipt_router)
 app.include_router(scrape_router)
+app.include_router(cijene_router)
 
 @app.get("/", tags=["Health"])
 def health_check():

db/cijene_repository.py

@@ -0,0 +1,169 @@
+from datetime import datetime
+from typing import List, Dict, Optional
+import re
+import unidecode
+import logging
+from db.supabase_client import SupabaseClient
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+class CijeneRepository:
+    def __init__(self):
+        self.supabase = SupabaseClient().get_client()
+    
+    def get_all_brands(self) -> List[Dict]:
+        """Get all brands from the database"""
+        try:
+            logger.info("Fetching all brands from database...")
+            response = self.supabase.table("brands").select("brand_id, brand_name").execute()
+            logger.info(f"Successfully fetched {len(response.data)} brands")
+            return response.data
+        except Exception as e:
+            logger.error(f"Error fetching brands: {str(e)}")
+            raise Exception(f"Error fetching brands: {str(e)}")
+    
+    def get_all_store_chains(self) -> List[Dict]:
+        """Get all store chains from the database"""
+        try:
+            logger.info("Fetching all store chains from database...")
+            response = self.supabase.table("store_chains").select("store_chain_id, store_chain_name").execute()
+            logger.info(f"Successfully fetched {len(response.data)} store chains")
+            return response.data
+        except Exception as e:
+            logger.error(f"Error fetching store chains: {str(e)}")
+            raise Exception(f"Error fetching store chains: {str(e)}")
+    
+    def normalize_string(self, text: str) -> str:
+        """Normalize string for comparison by removing special characters and converting to lowercase"""
+        if not text:
+            return ""
+        # Remove special characters and convert to lowercase
+        normalized = unidecode.unidecode(text)
+        normalized = re.sub(r'[^a-zA-Z0-9\s]', '', normalized)
+        normalized = re.sub(r'\s+', ' ', normalized).strip().lower()
+        return normalized
+    
+    def safe_log_string(self, text: str) -> str:
+        """Safely encode string for logging to avoid Unicode errors"""
+        if not text:
+            return ""
+        try:
+            # Try to encode/decode to handle special characters
+            return text.encode('utf-8', errors='replace').decode('utf-8')
+        except:
+            # Fallback to ASCII if UTF-8 fails
+            return unidecode.unidecode(text)
+    
+    def find_matching_store_chain(self, chain_name: str, store_chains: List[Dict]) -> Optional[str]:
+        """Find matching store chain ID by comparing normalized names"""
+        normalized_chain = self.normalize_string(chain_name)
+        safe_chain_name = self.safe_log_string(chain_name)
+        logger.debug(f"Looking for store chain match for: '{safe_chain_name}' (normalized: '{normalized_chain}')")
+        
+        for store_chain in store_chains:
+            normalized_store_chain = self.normalize_string(store_chain['store_chain_name'])
+            if normalized_chain == normalized_store_chain:
+                safe_store_name = self.safe_log_string(store_chain['store_chain_name'])
+                logger.debug(f"Found match: '{safe_chain_name}' -> '{safe_store_name}' (ID: {store_chain['store_chain_id']})")
+                return store_chain['store_chain_id']
+        
+        logger.warning(f"No matching store chain found for: '{safe_chain_name}'")
+        return None
+    
+    def check_product_exists(self, ean: str, product_name: str, brand_id: str) -> Optional[str]:
+        """Check if product already exists by EAN or (product_name, brand_id) and return product_id if found"""
+        try:
+            # First check by EAN
+            logger.debug(f"Checking if product exists with EAN: {ean}")
+            response = self.supabase.table("products").select("product_id").eq("product_ean", ean).execute()
+            if response.data:
+                logger.debug(f"Product exists with EAN {ean}, ID: {response.data[0]['product_id']}")
+                return response.data[0]['product_id']
+            
+            # Then check by product_name and brand_id
+            safe_name = self.safe_log_string(product_name)
+            logger.debug(f"Checking if product exists with name '{safe_name}' and brand_id: {brand_id}")
+            response = self.supabase.table("products").select("product_id").eq("product_name", product_name).eq("brand_id", brand_id).execute()
+            if response.data:
+                logger.debug(f"Product exists with name and brand, ID: {response.data[0]['product_id']}")
+                return response.data[0]['product_id']
+            
+            logger.debug(f"Product with EAN {ean} or name '{safe_name}' does not exist")
+            return None
+        except Exception as e:
+            logger.error(f"Error checking product existence for EAN {ean}: {str(e)}")
+            raise Exception(f"Error checking product existence: {str(e)}")
+    
+    def insert_product(self, product_data: Dict) -> str:
+        """Insert a new product and return the product_id"""
+        try:
+            safe_name = self.safe_log_string(product_data['product_name'])
+            logger.info(f"Inserting new product: {safe_name} (EAN: {product_data['product_ean']})")
+            response = self.supabase.table("products").insert(product_data).execute()
+            product_id = response.data[0]['product_id']
+            logger.info(f"Successfully inserted product with ID: {product_id}")
+            return product_id
+        except Exception as e:
+            safe_name = self.safe_log_string(product_data.get('product_name', 'unknown'))
+            logger.error(f"Error inserting product {safe_name}: {str(e)}")
+            raise Exception(f"Error inserting product: {str(e)}")
+    
+    def get_or_create_store_product(self, product_id: str, store_chain_id: str) -> str:
+        """Get existing store_product or create new one, return store_product_id"""
+        try:
+            # Check if store_product mapping already exists
+            logger.debug(f"Checking if store_product exists for product {product_id} and store_chain {store_chain_id}")
+            response = self.supabase.table("store_products").select("store_product_id").eq("product_id", product_id).eq("store_chain_id", store_chain_id).execute()
+            
+            if response.data:
+                store_product_id = response.data[0]['store_product_id']
+                logger.debug(f"Store_product exists with ID: {store_product_id}")
+                return store_product_id
+            
+            # Create new store_product mapping
+            logger.debug(f"Creating new store_product mapping for product {product_id} and store_chain {store_chain_id}")
+            store_product_data = {
+                "product_id": product_id,
+                "store_chain_id": store_chain_id
+            }
+            response = self.supabase.table("store_products").insert(store_product_data).execute()
+            store_product_id = response.data[0]['store_product_id']
+            logger.debug(f"Successfully created store_product with ID: {store_product_id}")
+            return store_product_id
+            
+        except Exception as e:
+            logger.error(f"Error getting/creating store_product for product {product_id} and store_chain {store_chain_id}: {str(e)}")
+            raise Exception(f"Error getting/creating store_product: {str(e)}")
+    
+    def insert_price_history(self, price_data: Dict) -> None:
+        """Insert price history record"""
+        try:
+            logger.debug(f"Inserting price history for store_product {price_data['store_product_id']}")
+            self.supabase.table("product_price_history").insert(price_data).execute()
+            logger.debug("Price history inserted successfully")
+        except Exception as e:
+            logger.error(f"Error inserting price history for store_product {price_data.get('store_product_id', 'unknown')}: {str(e)}")
+            raise Exception(f"Error inserting price history: {str(e)}")
+    
+    def combine_quantity_unit(self, quantity: str, unit: str) -> str:
+        """Combine quantity and unit into a single weight string without spaces"""
+        if not quantity and not unit:
+            return ""
+        
+        # Clean quantity (remove commas, extra spaces)
+        clean_quantity = quantity.replace(',', '.').strip() if quantity else ""
+        clean_unit = unit.lower().strip() if unit else ""
+        
+        # Combine without space
+        if clean_quantity and clean_unit:
+            result = f"{clean_quantity}{clean_unit}"
+        elif clean_quantity:
+            result = clean_quantity
+        elif clean_unit:
+            result = clean_unit
+        else:
+            result = ""
+        
+        logger.debug(f"Combined quantity '{quantity}' and unit '{unit}' -> '{result}'")
+        return result

requirements.txt

@@ -10,4 +10,6 @@ python-multipart
 google-cloud-vision
 python-dotenv
 supabase
-rembg
+rembg
+httpx
+unidecode