Smart Batch Task Executor

What It Does

Installation

git clone <repository-url>
   cd smart_batch_executor

pip install -r requirements.txt

Usage

python smart_batch_executor.py --config tasks.json

Source Code

import json
import click
import openai
import logging
from typing import List, Dict

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def load_tasks(config_path: str) -> Dict:
    """Load tasks from a JSON configuration file."""
    try:
        with open(config_path, 'r') as file:
            return json.load(file)
    except FileNotFoundError:
        logging.error("Configuration file not found.")
        raise
    except json.JSONDecodeError:
        logging.error("Invalid JSON format in configuration file.")
        raise

def optimize_task_order(tasks: List[Dict], dependencies: Dict) -> List[Dict]:
    """Use AI to optimize the order of tasks based on dependencies."""
    try:
        prompt = "Optimize the following tasks based on dependencies: " + json.dumps({"tasks": tasks, "dependencies": dependencies})
        response = openai.Completion.create(
            model="text-davinci-003",
            prompt=prompt,
            max_tokens=1000
        )
        optimized_order = json.loads(response.choices[0].text.strip())
        return optimized_order
    except Exception as e:
        logging.error(f"Failed to optimize task order: {e}")
        raise

def execute_task(task: Dict) -> Dict:
    """Execute a single task and return its result."""
    try:
        logging.info(f"Executing task: {task['name']}")
        # Simulate task execution
        return {"task": task["name"], "status": "success", "output": f"Output of {task['name']}"}
    except Exception as e:
        logging.error(f"Error executing task {task['name']}: {e}")
        return {"task": task["name"], "status": "failed", "error": str(e)}

def execute_tasks(tasks: List[Dict], dependencies: Dict) -> List[Dict]:
    """Execute a list of tasks with error handling and retries."""
    results = []
    for task in tasks:
        try:
            result = execute_task(task)
            results.append(result)
        except Exception as e:
            logging.warning(f"Retrying task {task['name']} due to error: {e}")
            try:
                result = execute_task(task)
                results.append(result)
            except Exception as e:
                logging.error(f"Task {task['name']} failed after retry: {e}")
                results.append({"task": task["name"], "status": "failed", "error": str(e)})
    return results

@click.command()
@click.option('--config', type=click.Path(exists=True), required=True, help='Path to the JSON configuration file.')
def main(config):
    """Smart Batch Task Executor"""
    try:
        config_data = load_tasks(config)
        tasks = config_data.get("tasks", [])
        dependencies = config_data.get("dependencies", {})

        if not tasks:
            logging.error("No tasks found in the configuration file.")
            return

        optimized_tasks = optimize_task_order(tasks, dependencies)
        results = execute_tasks(optimized_tasks, dependencies)

        with open('execution_logs.json', 'w') as log_file:
            json.dump(results, log_file, indent=4)

        logging.info("Task execution completed. Logs saved to execution_logs.json.")
    except Exception as e:
        logging.error(f"An error occurred: {e}")

if __name__ == "__main__":
    main()