Added scheduling to data collection script

api/main.py

@@ -1,6 +1,14 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from config.settings import API_HOST, API_PORT
+
+# Scheduler imports
+from contextlib import asynccontextmanager
+from scheduler.scheduler import scheduler
+from scheduler.task_registry import task_registry
+from scheduler.tasks.cijene_task import CijeneDataFetchTask
+from scheduler.tasks.cleanup_task import DatabaseCleanupTask
+
+# Routes imports
 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
@@ -8,8 +16,52 @@ from api.cijene_routes import router as cijene_router
 from api.similarity_routes import router as similarity_router
 from api.geocoding_routes import router as geocoding_router
 
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Register task classes
+    task_registry.register_task_class("cijene_fetch", CijeneDataFetchTask)
+    task_registry.register_task_class("database_cleanup", DatabaseCleanupTask)
+    
+    # Create task instances
+    fetch_task = task_registry.create_task(
+        "cijene_fetch", 
+        "daily_cijene_fetch",
+        name="Daily Cijene Data Fetch"
+    )
+    
+    cleanup_task = task_registry.create_task(
+        "database_cleanup",
+        "daily_cleanup", 
+        name="Daily Database Cleanup",
+        dependencies=["daily_cijene_fetch"]  # Run after fetch
+    )
+    
+    # Schedule the job chain
+    scheduler.add_daily_job(
+        "daily_data_pipeline",
+        ["daily_cijene_fetch", "daily_cleanup"],
+        hour=16, minute=58
+    )
+    
+    # Add event handlers
+    async def job_completed_handler(job_id: str, results: dict):
+        print(f"✅ Job {job_id} completed successfully!")
+    
+    async def job_failed_handler(job_id: str, results: dict):
+        print(f"❌ Job {job_id} failed!")
+    
+    scheduler.register_event_handler('job_completed', job_completed_handler)
+    scheduler.register_event_handler('job_failed', job_failed_handler)
+    
+    # Start scheduler
+    scheduler.start()
+    yield
+    
+    # Shutdown
+    scheduler.stop()
+
 # Initialize FastAPI
-app = FastAPI(title="SupaKuna API")
+app = FastAPI(title="SupaKuna API", lifespan=lifespan)
 
 # CORS configuration
 app.add_middleware(

api/scheduler_routes.py

@@ -0,0 +1,28 @@
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import JSONResponse
+from scheduler.scheduler import scheduler
+from scheduler.task_registry import task_registry
+from datetime import datetime, timedelta
+
+router = APIRouter(prefix="/scheduler", tags=["Scheduler Management"])
+
+@router.get("/status")
+async def get_scheduler_status():
+    """Get current scheduler status"""
+    return scheduler.get_job_status()
+
+@router.post("/run-job/{job_id}")
+async def run_job_now(job_id: str):
+    """Manually trigger a job"""
+    job_config = scheduler.job_configs.get(job_id)
+    if not job_config:
+        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
+    
+    # Run job immediately
+    await scheduler._execute_job(
+        job_id, 
+        job_config['task_ids'], 
+        job_config['context']
+    )
+    
+    return {"message": f"Job {job_id} executed successfully"}

app.py

@@ -1,4 +1,3 @@
-from api.main import app
 import uvicorn
 from config.settings import API_HOST, API_PORT
 

requirements.txt

@@ -13,6 +13,7 @@ supabase
 rembg
 httpx
 unidecode
+apscheduler
 # Similarity Engine Dependencies
 requests
 python-dateutil

scheduler/scheduler.py

@@ -0,0 +1,149 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.date import DateTrigger
+from apscheduler.triggers.interval import IntervalTrigger
+from typing import Dict, Any, List, Callable, Optional
+from datetime import datetime
+import logging
+from scheduler.task_registry import TaskRegistry, task_registry
+from scheduler.task_executor import TaskExecutor
+from scheduler.tasks.base_task import TaskStatus
+
+logger = logging.getLogger(__name__)
+
+class ModularScheduler:
+    def __init__(self, registry: TaskRegistry = None):
+        self.scheduler = AsyncIOScheduler()
+        self.registry = registry or task_registry
+        self.executor = TaskExecutor(self.registry)
+        self.job_configs: Dict[str, Dict] = {}
+        self.event_handlers: Dict[str, List[Callable]] = {}
+    
+    def add_daily_job(self, job_id: str, task_ids: List[str], 
+                     hour: int = 2, minute: int = 0, 
+                     context: Dict[str, Any] = None):
+        """Add a daily scheduled job"""
+        self.scheduler.add_job(
+            self._execute_job,
+            CronTrigger(hour=hour, minute=minute),
+            args=[job_id, task_ids, context],
+            id=job_id,
+            replace_existing=True
+        )
+        
+        self.job_configs[job_id] = {
+            'task_ids': task_ids,
+            'schedule': f"Daily at {hour:02d}:{minute:02d}",
+            'context': context or {}
+        }
+        
+        logger.info(f"Added daily job '{job_id}' scheduled for {hour:02d}:{minute:02d}")
+    
+    def add_interval_job(self, job_id: str, task_ids: List[str], 
+                        minutes: int = None, hours: int = None,
+                        context: Dict[str, Any] = None):
+        """Add an interval-based job"""
+        self.scheduler.add_job(
+            self._execute_job,
+            IntervalTrigger(minutes=minutes, hours=hours),
+            args=[job_id, task_ids, context],
+            id=job_id,
+            replace_existing=True
+        )
+        
+        self.job_configs[job_id] = {
+            'task_ids': task_ids,
+            'schedule': f"Every {hours or 0}h {minutes or 0}m",
+            'context': context or {}
+        }
+        
+        logger.info(f"Added interval job '{job_id}' every {hours or 0}h {minutes or 0}m")
+    
+    def add_one_time_job(self, job_id: str, task_ids: List[str], 
+                        run_date: datetime, context: Dict[str, Any] = None):
+        """Add a one-time job"""
+        self.scheduler.add_job(
+            self._execute_job,
+            DateTrigger(run_date=run_date),
+            args=[job_id, task_ids, context],
+            id=job_id,
+            replace_existing=True
+        )
+        
+        self.job_configs[job_id] = {
+            'task_ids': task_ids,
+            'schedule': f"One-time at {run_date}",
+            'context': context or {}
+        }
+        
+        logger.info(f"Added one-time job '{job_id}' for {run_date}")
+    
+    def register_event_handler(self, event: str, handler: Callable):
+        """Register event handler for job completion/failure"""
+        if event not in self.event_handlers:
+            self.event_handlers[event] = []
+        self.event_handlers[event].append(handler)
+    
+    async def _execute_job(self, job_id: str, task_ids: List[str], 
+                          context: Dict[str, Any] = None):
+        """Execute job with task chain"""
+        try:
+            logger.info(f"Starting scheduled job: {job_id}")
+            start_time = datetime.now()
+            
+            results = await self.executor.execute_task_chain(task_ids, context)
+            
+            end_time = datetime.now()
+            duration = (end_time - start_time).total_seconds()
+            
+            # Determine overall job status
+            failed_tasks = [task_id for task_id, result in results.items() 
+                          if result.status.value in ['failed']]
+            
+            if failed_tasks:
+                logger.error(f"Job {job_id} completed with failures in {duration:.2f}s: {failed_tasks}")
+                await self._trigger_event('job_failed', job_id, results)
+            else:
+                logger.info(f"Job {job_id} completed successfully in {duration:.2f}s")
+                await self._trigger_event('job_completed', job_id, results)
+            
+            # Clean up completed tasks
+            self.registry.clear_completed_tasks()
+            
+        except Exception as e:
+            logger.error(f"Critical error in job {job_id}: {str(e)}")
+            await self._trigger_event('job_error', job_id, {'error': str(e)})
+    
+    async def _trigger_event(self, event: str, job_id: str, data: Any):
+        """Trigger registered event handlers"""
+        if event in self.event_handlers:
+            for handler in self.event_handlers[event]:
+                try:
+                    if asyncio.iscoroutinefunction(handler):
+                        await handler(job_id, data)
+                    else:
+                        handler(job_id, data)
+                except Exception as e:
+                    logger.error(f"Error in event handler for {event}: {str(e)}")
+    
+    def start(self):
+        """Start the scheduler"""
+        self.scheduler.start()
+        logger.info("Modular scheduler started")
+        logger.info(f"Active jobs: {list(self.job_configs.keys())}")
+    
+    def stop(self):
+        """Stop the scheduler"""
+        self.scheduler.shutdown()
+        logger.info("Scheduler stopped")
+    
+    def get_job_status(self) -> Dict[str, Any]:
+        """Get status of all jobs"""
+        return {
+            'running': self.scheduler.running,
+            'jobs': self.job_configs,
+            'active_tasks': len(self.registry.get_all_tasks())
+        }
+
+# Global scheduler instance
+scheduler = ModularScheduler()

scheduler/task_executor.py

@@ -0,0 +1,86 @@
+from typing import List, Dict, Any, Optional
+from scheduler.tasks.base_task import BaseTask, TaskStatus, TaskResult
+from scheduler.task_registry import TaskRegistry
+import asyncio
+import logging
+
+logger = logging.getLogger(__name__)
+
+class TaskExecutor:
+    def __init__(self, registry: TaskRegistry):
+        self.registry = registry
+        self.execution_context: Dict[str, Any] = {}
+    
+    async def execute_task_chain(self, task_ids: List[str], 
+                                context: Dict[str, Any] = None) -> Dict[str, TaskResult]:
+        """Execute tasks in dependency order"""
+        if context:
+            self.execution_context.update(context)
+        
+        tasks = [self.registry.get_task(task_id) for task_id in task_ids]
+        tasks = [task for task in tasks if task is not None]
+        
+        if not tasks:
+            logger.warning("No valid tasks found to execute")
+            return {}
+        
+        # Sort tasks by dependencies (topological sort)
+        sorted_tasks = self._topological_sort(tasks)
+        completed_task_ids = []
+        results = {}
+        
+        logger.info(f"Starting execution chain with {len(sorted_tasks)} tasks")
+        
+        for task in sorted_tasks:
+            if task.can_execute(completed_task_ids):
+                # Add results from previous tasks to context
+                task_context = self.execution_context.copy()
+                task_context['previous_results'] = results
+                
+                result = await task.run(task_context)
+                results[task.task_id] = result
+                
+                if result.status == TaskStatus.COMPLETED:
+                    completed_task_ids.append(task.task_id)
+                    # Add task result data to global context for next tasks
+                    if result.data:
+                        self.execution_context.update(result.data)
+                elif result.status == TaskStatus.FAILED:
+                    logger.error(f"Task chain stopped due to failure in task: {task.task_id}")
+                    break
+            else:
+                logger.warning(f"Skipping task {task.task_id} - dependencies not met")
+                results[task.task_id] = TaskResult(TaskStatus.SKIPPED, 
+                                                 error="Dependencies not satisfied")
+        
+        logger.info(f"Task chain execution completed. {len(completed_task_ids)} tasks succeeded")
+        return results
+    
+    def _topological_sort(self, tasks: List[BaseTask]) -> List[BaseTask]:
+        """Sort tasks based on dependencies using topological sort"""
+        task_dict = {task.task_id: task for task in tasks}
+        visited = set()
+        temp_visited = set()
+        result = []
+        
+        def visit(task_id: str):
+            if task_id in temp_visited:
+                raise ValueError(f"Circular dependency detected involving task: {task_id}")
+            if task_id in visited:
+                return
+            
+            temp_visited.add(task_id)
+            task = task_dict.get(task_id)
+            if task:
+                for dep_id in task.dependencies:
+                    if dep_id in task_dict:
+                        visit(dep_id)
+                visited.add(task_id)
+                result.append(task)
+            temp_visited.remove(task_id)
+        
+        for task in tasks:
+            if task.task_id not in visited:
+                visit(task.task_id)
+        
+        return result

scheduler/task_registry.py

@@ -0,0 +1,48 @@
+from typing import Dict, Type, List
+from scheduler.tasks.base_task import BaseTask
+import logging
+
+logger = logging.getLogger(__name__)
+
+class TaskRegistry:
+    def __init__(self):
+        self._task_classes: Dict[str, Type[BaseTask]] = {}
+        self._task_instances: Dict[str, BaseTask] = {}
+    
+    def register_task_class(self, task_type: str, task_class: Type[BaseTask]):
+        """Register a task class by type"""
+        self._task_classes[task_type] = task_class
+        logger.debug(f"Registered task class: {task_type}")
+    
+    def create_task(self, task_type: str, task_id: str, **kwargs) -> BaseTask:
+        """Create a task instance from registered class"""
+        if task_type not in self._task_classes:
+            raise ValueError(f"Task type '{task_type}' not registered")
+        
+        task_class = self._task_classes[task_type]
+        task_instance = task_class(task_id=task_id, **kwargs)
+        self._task_instances[task_id] = task_instance
+        
+        logger.debug(f"Created task instance: {task_id} of type {task_type}")
+        return task_instance
+    
+    def get_task(self, task_id: str) -> BaseTask:
+        """Get task instance by ID"""
+        return self._task_instances.get(task_id)
+    
+    def get_all_tasks(self) -> List[BaseTask]:
+        """Get all task instances"""
+        return list(self._task_instances.values())
+    
+    def clear_completed_tasks(self):
+        """Remove completed tasks from registry"""
+        completed_tasks = [
+            task_id for task_id, task in self._task_instances.items()
+            if task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED]
+        ]
+        for task_id in completed_tasks:
+            del self._task_instances[task_id]
+        logger.info(f"Cleared {len(completed_tasks)} completed tasks")
+
+# Global registry instance
+task_registry = TaskRegistry()

scheduler/tasks/base_task.py

@@ -0,0 +1,65 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any, List, Optional
+from enum import Enum
+import logging
+from datetime import datetime
+import asyncio
+
+logger = logging.getLogger(__name__)
+
+class TaskStatus(Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+class TaskResult:
+    def __init__(self, status: TaskStatus, data: Dict[Any, Any] = None, error: str = None):
+        self.status = status
+        self.data = data or {}
+        self.error = error
+        self.timestamp = datetime.now()
+
+class BaseTask(ABC):
+    def __init__(self, task_id: str, name: str, dependencies: List[str] = None):
+        self.task_id = task_id
+        self.name = name
+        self.dependencies = dependencies or []
+        self.status = TaskStatus.PENDING
+        self.result: Optional[TaskResult] = None
+        self.created_at = datetime.now()
+        self.started_at: Optional[datetime] = None
+        self.completed_at: Optional[datetime] = None
+    
+    @abstractmethod
+    async def execute(self, context: Dict[str, Any] = None) -> TaskResult:
+        """Execute the task and return result"""
+        pass
+    
+    def can_execute(self, completed_tasks: List[str]) -> bool:
+        """Check if all dependencies are satisfied"""
+        return all(dep in completed_tasks for dep in self.dependencies)
+    
+    async def run(self, context: Dict[str, Any] = None) -> TaskResult:
+        """Wrapper method that handles status tracking"""
+        try:
+            self.status = TaskStatus.RUNNING
+            self.started_at = datetime.now()
+            logger.info(f"Starting task: {self.name} ({self.task_id})")
+            
+            self.result = await self.execute(context or {})
+            self.status = self.result.status
+            self.completed_at = datetime.now()
+            
+            duration = (self.completed_at - self.started_at).total_seconds()
+            logger.info(f"Task {self.name} completed with status {self.status.value} in {duration:.2f}s")
+            
+            return self.result
+            
+        except Exception as e:
+            self.status = TaskStatus.FAILED
+            self.completed_at = datetime.now()
+            self.result = TaskResult(TaskStatus.FAILED, error=str(e))
+            logger.error(f"Task {self.name} failed: {str(e)}")
+            return self.result

scheduler/tasks/cijene_task.py

@@ -0,0 +1,226 @@
+from scheduler.tasks.base_task import BaseTask, TaskResult, TaskStatus
+from typing import Dict, Any, List
+import httpx
+import asyncio
+from datetime import datetime
+from db.cijene_repository import CijeneRepository
+import logging
+
+logger = logging.getLogger(__name__)
+
+class CijeneDataFetchTask(BaseTask):
+    def __init__(self, task_id: str, name: str = "Cijene Data Fetch", 
+                 dependencies: list = None, brands_limit: int = None):
+        super().__init__(task_id, name, dependencies)
+        self.repository = CijeneRepository()
+        self.brands_limit = brands_limit
+
+    async def execute(self, context: Dict[str, Any] = None) -> TaskResult:
+        """Execute the Cijene data fetching process using existing logic"""
+        try:
+            logger.info("Starting Cijene data fetch process")
+            
+            # Get all brands and store chains
+            brands = self.repository.get_all_brands()
+            store_chains = self.repository.get_all_store_chains()
+            
+            # Apply limit for testing
+            if self.brands_limit:
+                brands = brands[:self.brands_limit]
+                logger.info(f"Limited to {self.brands_limit} brands for testing")
+            
+            if not brands:
+                return TaskResult(TaskStatus.FAILED, error="No brands found in database")
+            
+            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,
+                "processing_duration_seconds": 0,
+                "errors": []
+            }
+            
+            # API configuration
+            base_url = "https://api.cijene.dev/v1/products/"
+            headers = {
+                "Authorization": "Bearer aibeeboh2cuataChi2doMa2Aacah0eli",
+                "Content-Type": "application/json"
+            }
+            
+            start_time = asyncio.get_event_loop().time()
+            
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                for index, brand in enumerate(brands, 1):
+                    try:
+                        brand_name = brand['brand_name']
+                        brand_id = brand['brand_id']
+                        
+                        safe_brand_name = self.repository.safe_log_string(brand_name)
+                        logger.info(f"Processing brand {index}/{len(brands)}: {safe_brand_name}")
+                        
+                        url = f"{base_url}?q={brand_name}"
+                        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
+                        
+                        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 using the existing logic
+                        for product_index, product in enumerate(products, 1):
+                            try:
+                                safe_product_name = self.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}")
+                                
+                                # Use the existing process_product logic
+                                await self._process_product(product, brand_id, store_chains, 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)}")
+                        
+                        # 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']}")
+                        
+                    except Exception as e:
+                        error_msg = f"Error processing brand {brand.get('brand_name', 'unknown')}: {str(e)}"
+                        logger.error(error_msg)
+                        results["errors"].append(error_msg)
+                        continue
+            
+            end_time = asyncio.get_event_loop().time()
+            results["processing_duration_seconds"] = round(end_time - start_time, 2)
+            
+            logger.info(f"Cijene fetch completed: {results}")
+            return TaskResult(TaskStatus.COMPLETED, data=results)
+            
+        except Exception as e:
+            logger.error(f"Critical error in Cijene fetch: {str(e)}")
+            return TaskResult(TaskStatus.FAILED, error=str(e))
+
+    async def _process_product(self, product: Dict, brand_id: str, 
+                              store_chains: List[Dict], results: Dict):
+        """
+        Process a single product and its price data
+        This is the EXACT same logic from your working cijene_routes.py
+        """
+        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 = self.repository.safe_log_string(name)
+            logger.warning(f"Skipping product without EAN: {safe_name}")
+            return
+        
+        safe_name = self.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 = self.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 = self.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 = self.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 = self.repository.find_matching_store_chain(chain_name, store_chains)
+            
+            if not store_chain_id:
+                safe_chain_name = self.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 = self.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 = self.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 = self.repository.safe_log_string(chain_name)
+                logger.warning(f"Invalid price data for chain '{safe_chain_name}': {e}")
+                continue
+            
+            # Insert price history
+            price_data = {
+                "store_product_id": store_product_id,
+                "price_date": today,
+                "min_price": min_price_float,
+                "max_price": max_price_float,
+                "avg_price": avg_price_float
+            }
+            
+            self.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")

scheduler/tasks/cleanup_task.py

@@ -0,0 +1,32 @@
+from scheduler.tasks.base_task import BaseTask, TaskResult, TaskStatus
+from typing import Dict, Any
+import logging
+
+logger = logging.getLogger(__name__)
+
+class DatabaseCleanupTask(BaseTask):
+    def __init__(self, task_id: str, name: str = "Database Cleanup", 
+                 dependencies: list = None):
+        super().__init__(task_id, name, dependencies)
+    
+    async def execute(self, context: Dict[str, Any] = None) -> TaskResult:
+        """Placeholder cleanup task"""
+        try:
+            logger.info("Database cleanup completed (placeholder)")
+            return TaskResult(TaskStatus.COMPLETED, data={"cleanup_completed": True})
+        except Exception as e:
+            return TaskResult(TaskStatus.FAILED, error=str(e))
+
+class NotificationTask(BaseTask):
+    def __init__(self, task_id: str, name: str = "Notification", 
+                 dependencies: list = None):
+        super().__init__(task_id, name, dependencies)
+    
+    async def execute(self, context: Dict[str, Any] = None) -> TaskResult:
+        """Placeholder notification task"""
+        try:
+            previous_results = context.get('previous_results', {})
+            logger.info(f"Notification sent (placeholder) - Previous results: {len(previous_results)} tasks")
+            return TaskResult(TaskStatus.COMPLETED, data={"notification_sent": True})
+        except Exception as e:
+            return TaskResult(TaskStatus.FAILED, error=str(e))