"""Task template management and parametrization.""" import csv import hashlib import random from datetime import datetime from pathlib import Path from typing import Any import yaml from jinja2 import Template from shared.config import settings from shared.logger import setup_logger from shared.utils import encode_data_uri, generate_task_id logger = setup_logger(__name__) class TaskTemplate: """Single task template with parametrization.""" def __init__(self, data: dict[str, Any]) -> None: """Initialize task template. Args: data: Template data from YAML """ self.id = data["id"] self.brief = data["brief"] self.attachments = data.get("attachments", []) self.checks = data.get("checks", []) self.round2 = data.get("round2", []) def parametrize(self, seed: str) -> dict[str, Any]: """Parametrize template with seed. Args: seed: Seed string for randomization Returns: Parametrized task data """ # Set random seed for reproducibility random.seed(seed) # Generate data based on template context = self._generate_context(seed) # Parametrize brief brief = Template(self.brief).render(**context) # Parametrize attachments attachments = [] for att_template in self.attachments: att_name = Template(att_template["name"]).render(**context) att_url_template = att_template["url"] # Generate attachment content if "data:text/csv" in att_url_template: content = self._generate_csv_data(seed) att_url = encode_data_uri("text/csv", content.encode()) elif "data:text/markdown" in att_url_template: content = self._generate_markdown_data(seed) att_url = encode_data_uri("text/markdown", content.encode()) elif "data:application/json" in att_url_template: content = self._generate_json_data(seed) att_url = encode_data_uri("application/json", content.encode()) else: att_url = Template(att_url_template).render(**context) attachments.append({"name": att_name, "url": att_url}) # Parametrize checks checks = [Template(check).render(**context) for check in self.checks] return { "brief": brief, "attachments": attachments, "checks": checks, "context": context, } def parametrize_round2(self, seed: str, round2_index: int = 0) -> dict[str, Any]: """Parametrize round 2 task. Args: seed: Seed string round2_index: Index of round2 option to use Returns: Parametrized round 2 task data """ if not self.round2 or round2_index >= len(self.round2): raise ValueError(f"No round2 option at index {round2_index}") round2_task = self.round2[round2_index] random.seed(seed + f"-round2-{round2_index}") context = self._generate_context(seed) brief = Template(round2_task["brief"]).render(**context) checks = [Template(check).render(**context) for check in round2_task.get("checks", [])] attachments = [] for att_template in round2_task.get("attachments", []): att_name = Template(att_template["name"]).render(**context) att_url = Template(att_template["url"]).render(**context) attachments.append({"name": att_name, "url": att_url}) return { "brief": brief, "attachments": attachments, "checks": checks, "context": context, } def _generate_context(self, seed: str) -> dict[str, Any]: """Generate context variables for template. Args: seed: Seed string Returns: Context dictionary """ # Generate deterministic values based on seed hash_val = int(hashlib.md5(seed.encode()).hexdigest(), 16) return { "seed": seed, "hash": hash_val % 10000, "result": (hash_val % 9000) + 1000, # Result between 1000-9999 } def _generate_csv_data(self, seed: str) -> str: """Generate sample CSV data. Args: seed: Seed for randomization Returns: CSV string """ random.seed(seed) products = ["Widget", "Gadget", "Doohickey", "Thingamajig", "Gizmo"] regions = ["North", "South", "East", "West", "Central"] lines = ["product,region,sales"] for _ in range(random.randint(10, 20)): product = random.choice(products) region = random.choice(regions) sales = round(random.uniform(100, 1000), 2) lines.append(f"{product},{region},{sales}") return "\n".join(lines) def _generate_markdown_data(self, seed: str) -> str: """Generate sample Markdown data. Args: seed: Seed for randomization Returns: Markdown string """ random.seed(seed) content = f"""# Sample Document {seed[:8]} ## Introduction This is a sample markdown document generated for testing. ## Code Example ```python def hello(): print("Hello, world!") ``` ## List - Item 1 - Item 2 - Item 3 ## Conclusion This document contains {random.randint(50, 150)} words. """ return content def _generate_json_data(self, seed: str) -> str: """Generate sample JSON data. Args: seed: Seed for randomization Returns: JSON string """ import json random.seed(seed) data = { "USD": 1.0, "EUR": round(random.uniform(0.8, 0.95), 2), "GBP": round(random.uniform(0.7, 0.85), 2), "JPY": round(random.uniform(110, 140), 2), } return json.dumps(data, indent=2) class TaskTemplateManager: """Manage task templates.""" def __init__(self, templates_dir: Path | None = None) -> None: """Initialize template manager. Args: templates_dir: Directory containing template YAML files """ self.templates_dir = templates_dir or settings.task_templates_dir self.templates: dict[str, TaskTemplate] = {} self.load_templates() def load_templates(self) -> None: """Load all templates from directory.""" if not self.templates_dir.exists(): logger.warning(f"Templates directory not found: {self.templates_dir}") return for yaml_file in self.templates_dir.glob("*.yaml"): try: with open(yaml_file) as f: data = yaml.safe_load(f) template = TaskTemplate(data) self.templates[template.id] = template logger.info(f"Loaded template: {template.id}") except Exception as e: logger.error(f"Failed to load template {yaml_file}: {e}") logger.info(f"Loaded {len(self.templates)} templates") def get_template(self, template_id: str) -> TaskTemplate: """Get template by ID. Args: template_id: Template identifier Returns: Task template Raises: KeyError: If template not found """ if template_id not in self.templates: raise KeyError(f"Template not found: {template_id}") return self.templates[template_id] def get_random_template(self) -> TaskTemplate: """Get random template. Returns: Random task template """ if not self.templates: raise ValueError("No templates available") return random.choice(list(self.templates.values())) def generate_task( self, email: str, template_id: str | None = None, round_num: int = 1 ) -> dict[str, Any]: """Generate a complete task from template. Args: email: Student email template_id: Template ID (random if None) round_num: Round number Returns: Complete task data """ # Get template if template_id: template = self.get_template(template_id) else: template = self.get_random_template() # Generate seed now = datetime.utcnow() seed = f"{email}-{now.strftime('%Y-%m-%d-%H')}" # Parametrize if round_num == 1: task_data = template.parametrize(seed) else: # For round 2, pick random round2 option round2_index = random.randint(0, len(template.round2) - 1) if template.round2 else 0 task_data = template.parametrize_round2(seed, round2_index) # Generate task ID task_id = generate_task_id(template.id, task_data["brief"], task_data["attachments"]) return { "template_id": template.id, "task_id": task_id, **task_data, }