feat: add story image generation modules and tests

Author: anyulled

scripts/config.py

@@ -0,0 +1,61 @@
+"""
+Configuration settings for story image generation
+"""
+import os
+from dataclasses import dataclass, field
+from typing import Dict, Tuple
+
+
+@dataclass
+class Config:
+    # Paths
+    LOGO_PATH: str = "../public/images/logo.png"
+    FONT_PRIMARY: str = "../src/fonts/DejaVu Sans Bold.ttf"
+    FONT_SECONDARY: str = "../src/fonts/Square 721 Regular.otf"
+    OUTPUT_DIR: str = "output"
+
+    # Image settings
+    STORY_SIZE: Tuple[int, int] = (1080, 1920)
+    SAFE_MARGIN: int = 100
+    BLUR_PERCENTAGE: float = 0.3
+
+    # Colors
+    BG_COLOR: str = "#002454"
+    BG_GRADIENT_BOTTOM: str = "#00152E"
+    COLOR_WHITE: str = "#FFFFFF"
+    COLOR_YELLOW: str = "#FFCC00"
+    COLOR_GREEN: str = "#00FFAA"
+
+    # Font sizes
+    FONT_SIZE_NAME: int = 60
+    FONT_SIZE_ROLE: int = 40
+    FONT_SIZE_SESSION: int = 60
+    FONT_SIZE_EVENT: int = 30
+    ROLE_OFFSET_Y: int = 60
+
+    # API endpoints
+    SPEAKERS_API: str = "https://sessionize.com/api/v2/xhudniix/view/Speakers"
+    SESSIONS_API: str = "https://sessionize.com/api/v2/xhudniix/view/Sessions"
+
+    # Event details
+    EVENT_DETAILS: str = "July 8–10, 2025 — La Farga, Barcelona"
+
+    # Track gradients
+    TRACK_GRADIENTS: Dict[str, Tuple[str, str]] = field(
+        default_factory=lambda: {
+            "Java | JVM | Cloud": ("#003366", "#001122"),
+            "DevOps, VMs, Kubernetes": ("#004422", "#001911"),
+            "Frontend, JS, WASM": ("#223366", "#111122"),
+            "Leadership, Agile, Diversity": ("#662244", "#221122"),
+            "Big Data, ML, AI, Python": ("#222244", "#111122"),
+            "No Track": ("#002454", "#00152E"),
+        })
+
+    def get_track_gradient(self, track_name: str) -> Tuple[str, str]:
+        """Get gradient colors for a specific track"""
+        return self.TRACK_GRADIENTS.get(track_name, (self.BG_COLOR,
+                                                     self.BG_GRADIENT_BOTTOM))
+
+    def ensure_output_dir(self) -> None:
+        """Create output directory if it doesn't exist"""
+        os.makedirs(self.OUTPUT_DIR, exist_ok=True)

scripts/data_fetcher.py

@@ -0,0 +1,111 @@
+"""
+Module for fetching data from APIs and local resources
+"""
+import logging
+import requests
+from io import BytesIO
+from typing import Dict, List, Any, Optional
+from PIL import Image
+
+logger = logging.getLogger(__name__)
+
+
+class DataFetcher:
+    """Class for fetching data from various sources"""
+
+    def __init__(self, request_timeout: int = 10):
+        self.request_timeout = request_timeout
+
+    def fetch_json(self, url: str) -> Any:
+        """Fetch JSON data from a URL
+        
+        Args:
+            url: The URL to fetch data from
+            
+        Returns:
+            Parsed JSON data
+            
+        Raises:
+            requests.RequestException: If the request fails
+        """
+        try:
+            response = requests.get(url, timeout=self.request_timeout)
+            response.raise_for_status()
+            return response.json()
+        except requests.RequestException as e:
+            logger.error(f"Failed to fetch JSON from {url}: {e}")
+            raise
+
+    def fetch_image(self, url: str) -> Image.Image:
+        """Fetch an image from a URL
+        
+        Args:
+            url: The URL to fetch the image from
+            
+        Returns:
+            PIL Image object
+            
+        Raises:
+            requests.RequestException: If the request fails
+            IOError: If the image cannot be processed
+        """
+        try:
+            response = requests.get(url, timeout=self.request_timeout)
+            response.raise_for_status()
+            return Image.open(BytesIO(response.content)).convert("RGBA")
+        except requests.RequestException as e:
+            logger.error(f"Failed to fetch image from {url}: {e}")
+            raise
+        except IOError as e:
+            logger.error(f"Failed to process image from {url}: {e}")
+            raise
+
+    def load_local_image(self, path: str) -> Image.Image:
+        """Load an image from a local file
+        
+        Args:
+            path: Path to the image file
+            
+        Returns:
+            PIL Image object
+            
+        Raises:
+            FileNotFoundError: If the file does not exist
+            IOError: If the image cannot be processed
+        """
+        try:
+            return Image.open(path).convert("RGBA")
+        except (FileNotFoundError, IOError) as e:
+            logger.error(f"Failed to load image from {path}: {e}")
+            raise
+
+    def fetch_speakers(self, api_url: str) -> Dict[str, Dict]:
+        """Fetch speaker data and index by ID
+        
+        Args:
+            api_url: URL for the speakers API
+            
+        Returns:
+            Dictionary of speakers indexed by ID
+        """
+        speakers_data = self.fetch_json(api_url)
+        return {s['id']: s for s in speakers_data}
+
+    def fetch_sessions_by_track(self, api_url: str) -> Dict[str, List[Dict]]:
+        """Fetch session data grouped by track
+        
+        Args:
+            api_url: URL for the sessions API
+            
+        Returns:
+            Dictionary of sessions grouped by track name
+        """
+        sessions_grouped = self.fetch_json(api_url)
+        tracks = {}
+        
+        for group in sessions_grouped:
+            track_name = group.get("title", "No Track")
+            for session in group.get("sessions", []):
+                tracks.setdefault(track_name, []).append(session)
+                
+        return tracks

scripts/file_utils.py

@@ -0,0 +1,53 @@
+"""
+Utility functions for file operations
+"""
+import os
+import logging
+import re
+from typing import Optional
+from PIL import Image
+
+logger = logging.getLogger(__name__)
+
+
+def slugify(value: str) -> str:
+    """
+    Convert a string to a URL-friendly slug
+    
+    Args:
+        value: String to convert
+        
+    Returns:
+        Slugified string
+    """
+    # Remove special characters
+    value = re.sub(r'[^\w\s-]', '', value).strip().lower()
+    # Replace whitespace with hyphens
+    return re.sub(r'[-\s]+', '-', value)
+
+
+def save_image(image: Image.Image, directory: str, filename: str) -> str:
+    """
+    Save an image to a file
+    
+    Args:
+        image: PIL Image to save
+        directory: Directory to save in
+        filename: Base filename (will be slugified)
+        
+    Returns:
+        Path to the saved file
+    """
+    os.makedirs(directory, exist_ok=True)
+    
+    # Ensure the filename is safe
+    safe_filename = f"{slugify(filename)}.png"
+    output_path = os.path.join(directory, safe_filename)
+    
+    try:
+        image.save(output_path)
+        logger.info(f"Saved image to {output_path}")
+        return output_path
+    except Exception as e:
+        logger.error(f"Failed to save image to {output_path}: {e}")
+        raise

scripts/generate-story-images.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""
+Generate conference story images for speakers and sessions.
+
+This script uses the refactored modules to generate story images for the conference,
+following SOLID principles for better maintainability and testability.
+"""
+import argparse
+import logging
+from concurrent.futures import ThreadPoolExecutor
+from typing import Dict, Any, List, Tuple
+
+from config import Config
+from data_fetcher import DataFetcher
+from story_generator import StoryGenerator
+
+# Set up logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+
+def parse_arguments():
+    """Parse command line arguments"""
+    parser = argparse.ArgumentParser(
+        description='Generate story images for conference speakers and sessions')
+    parser.add_argument('--max-workers', type=int, default=10,
+                        help='Maximum number of worker threads')
+    parser.add_argument('--timeout', type=int, default=10,
+                        help='Timeout for HTTP requests in seconds')
+    parser.add_argument('--output-dir', type=str, default='output',
+                        help='Directory to save generated images')
+    return parser.parse_args()
+
+
+def process_session(generator: StoryGenerator, session: Dict[str, Any],
+                    speaker: Dict[str, Any], track_name: str) -> str:
+    """Process a single session and generate its story image
+    
+    Args:
+        generator: StoryGenerator instance
+        session: Session data
+        speaker: Speaker data
+        track_name: Name of the track
+        
+    Returns:
+        Path to the generated image file
+    """
+    try:
+        return generator.generate_story(speaker, session, track_name)
+    except Exception as e:
+        logger.error(
+            f"Failed to process session {session.get('title', 'Unknown')}: {e}")
+        raise
+
+
+def main():
+    """Main entry point"""
+    args = parse_arguments()
+
+    # Initialize configuration
+    config = Config()
+    # Override output directory from command line if provided
+    if args.output_dir:
+        config.OUTPUT_DIR = args.output_dir
+
+    # Ensure output directory exists
+    config.ensure_output_dir()
+
+    # Initialize components
+    data_fetcher = DataFetcher(request_timeout=args.timeout)
+    story_generator = StoryGenerator(config, data_fetcher)
+
+    logger.info("Fetching speakers and sessions...")
+
+    try:
+        # Fetch data
+        speakers_lookup = data_fetcher.fetch_speakers(config.SPEAKERS_API)
+        tracks = data_fetcher.fetch_sessions_by_track(config.SESSIONS_API)
+
+        # Process all sessions
+        with ThreadPoolExecutor(max_workers=args.max_workers) as executor:
+            futures = []
+
+            for track_name, sessions in tracks.items():
+                logger.info(
+                    f"Processing track: {track_name} with {len(sessions)} sessions")
+
+                for session in sessions:
+                    speaker_ids = session.get('speakers', [])
+                    speaker_id = speaker_ids[0]['id'] if speaker_ids else None
+
+                    if speaker_id and speaker_id in speakers_lookup:
+                        speaker = speakers_lookup[speaker_id]
+                        futures.append(
+                            executor.submit(
+                                process_session,
+                                story_generator,
+                                session,
+                                speaker,
+                                track_name
+                            )
+                        )
+
+            # Wait for all tasks to complete
+            completed = 0
+            for future in futures:
+                future.result()
+                completed += 1
+                if completed % 5 == 0:
+                    logger.info(
+                        f"Processed {completed}/{len(futures)} sessions")
+
+        logger.info("✅ All stories generated!")
+
+    except Exception as e:
+        logger.error(f"Error processing stories: {e}")
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    exit(main())