src/utils/visualization.py - master - yakobsonsa/bi_agent

src/utils/visualization.py
834 строки35 KB
Yakobsonsa
Use grouped bars for monthly plan/fact charts
11 фев 2026, 00:56
11 фев 2026, 00:56da9583a
"""Data visualization utilities."""

import logging
import io
import textwrap
from typing import List, Dict, Any, Optional, Tuple
from pathlib import Path
import json
from src.settings import get_settings

logger = logging.getLogger(__name__)

# Lazy imports for visualization libraries
_matplotlib_imported = False
_bokeh_imported = False

def _import_matplotlib():
    """Lazy import matplotlib."""
    global _matplotlib_imported
    if not _matplotlib_imported:
        import matplotlib
        matplotlib.use('Agg')
        _matplotlib_imported = True

def _import_bokeh():
    """Lazy import bokeh."""
    global _bokeh_imported
    if not _bokeh_imported:
        try:
            import bokeh
            _bokeh_imported = True
            return True
        except ImportError:
            logger.warning("Bokeh not installed, falling back to matplotlib")
            return False
    return True


class Visualizer:
    """Create visualizations from query results."""

    def __init__(self, chart_type: str = 'matplotlib'):
        """Initialize visualizer."""
        settings = get_settings()
        self.chart_type = chart_type
        self.max_points = settings.visualizer.max_points_on_chart
        self.figure_width = settings.visualizer.figure_width
        self.figure_height = settings.visualizer.figure_height
        self.label_fontsize = settings.visualizer.label_fontsize
        self.title_fontsize = settings.visualizer.title_fontsize
        self.dpi = settings.visualizer.dpi
        self.table_max_rows = settings.visualizer.table_max_rows
        self.pdf_a4_width_mm = settings.visualizer.pdf_a4_width_mm
        self.pdf_a4_height_mm = settings.visualizer.pdf_a4_height_mm
        
        # Enable Bokeh support
        self.bokeh_enabled = _import_bokeh()
        
        logger.info(f"Visualizer initialized with {chart_type} (Bokeh: {self.bokeh_enabled})")

    def create_chart(
        self,
        data: List[Dict[str, Any]],
        chart_name: str = 'chart',
        title: str = 'Data Visualization',
        x_column: Optional[str] = None,
        y_column: Optional[str] = None,
        chart_type_override: Optional[str] = None
    ) -> Optional[str]:
        """Create chart from data and return file path.

        chart_type_override examples:
        - "matplotlib" (auto)
        - "matplotlib:line", "matplotlib:bar", "matplotlib:barh",
          "matplotlib:stacked", "matplotlib:box", "matplotlib:heatmap"
        - "bokeh", "bokeh:bar", "bokeh:line", "bokeh:grouped_bar"
        - "plotly" (bar by default)
        - "line", "bar", "barh", "stacked", "box", "heatmap" (shortcut for matplotlib)
        """
        if not data:
            logger.warning("No data provided for chart")
            return None

        chart_type_raw = (chart_type_override or self.chart_type or '').lower()
        base_type, variant = self._parse_chart_type(chart_type_raw)

        try:
            if base_type == 'plotly':
                return self._create_plotly_chart(
                    data, chart_name, title, x_column, y_column
                )
            elif base_type == 'bokeh' and self.bokeh_enabled:
                bokeh_path = self._create_bokeh_chart(
                    data, chart_name, title, x_column, y_column, variant
                )
                if bokeh_path:
                    return bokeh_path
                logger.warning("Bokeh render failed, falling back to matplotlib")
                return self._create_matplotlib_chart(
                    data, chart_name, title, x_column, y_column, variant
                )
            else:
                return self._create_matplotlib_chart(
                    data, chart_name, title, x_column, y_column, variant
                )
        except Exception as e:
            logger.error(f"Error creating chart: {e}")
            return None

    def _parse_chart_type(self, chart_type_raw: str) -> Tuple[str, Optional[str]]:
        """Split base type and variant."""
        if not chart_type_raw:
            return 'matplotlib', None
        if chart_type_raw.startswith('plotly'):
            return 'plotly', None
        if chart_type_raw.startswith('bokeh'):
            parts = chart_type_raw.split(':', 1)
            variant = parts[1] if len(parts) > 1 else None
            return 'bokeh', variant
        if chart_type_raw.startswith('matplotlib'):
            parts = chart_type_raw.split(':', 1)
            variant = parts[1] if len(parts) > 1 else None
            return 'matplotlib', variant
        # Shortcut variants map to matplotlib
        return 'matplotlib', chart_type_raw
    
    def _detect_data_type(self, x_data: List[str], y_data: List[float]) -> str:
        """Detect if data is time series, monthly, or categorical."""
        if not x_data:
            return 'categorical'
        
        # Check if looks like monthly data
        months_ru = ['янв', 'фев', 'мар', 'апр', 'май', 'июн', 'июл', 'авг', 'сен', 'окт', 'ноя', 'дек']
        if any(month in str(x_data[0]).lower() for month in months_ru):
            return 'monthly'
        
        # Check if starts with date pattern
        if any(c.isdigit() for c in str(x_data[0])):
            return 'timeseries'
        
        return 'categorical'
    
    def _calculate_trend(self, y_data: List[float]) -> Tuple[List[float], float]:
        """Calculate polynomial trend for data."""
        try:
            import numpy as np
            if len(y_data) < 2:
                return y_data, 0.0
            
            x = np.arange(len(y_data))
            # Fit polynomial (degree 2 for smooth trend)
            z = np.polyfit(x, y_data, min(2, len(y_data) - 1))
            p = np.poly1d(z)
            trend = p(x).tolist()
            
            # Calculate slope for trend direction
            slope = (y_data[-1] - y_data[0]) / len(y_data) if len(y_data) > 1 else 0.0
            return trend, slope
        except Exception as e:
            logger.debug(f"Could not calculate trend: {e}")
            return y_data, 0.0
    
    def _find_extremes(self, y_data: List[float]) -> Tuple[int, int]:
        """Find indices of max and min values."""
        if not y_data:
            return -1, -1
        max_idx = y_data.index(max(y_data))
        min_idx = y_data.index(min(y_data))
        return max_idx, min_idx
    
    @staticmethod
    def _format_number_short(value, pos=None) -> str:
        """Format numbers for chart axes (billions → млрд, millions → млн, thousands → тыс)."""
        try:
            num = float(value)
            if abs(num) >= 1_000_000_000:
                return f'{num / 1_000_000_000:.1f} млрд'
            elif abs(num) >= 1_000_000:
                return f'{num / 1_000_000:.1f} млн'
            elif abs(num) >= 1_000:
                return f'{num / 1_000:.0f} тыс'
            else:
                return f'{num:.0f}'
        except (ValueError, TypeError):
            return str(value)

    def _create_matplotlib_chart(
        self,
        data: List[Dict[str, Any]],
        chart_name: str,
        title: str,
        x_column: Optional[str],
        y_column: Optional[str],
        variant: Optional[str]
    ) -> str:
        """Create matplotlib chart with variants."""
        try:
            import matplotlib.pyplot as plt
            import matplotlib
            import numpy as np
            matplotlib.use('Agg')  # Non-interactive backend
        except ImportError:
            logger.error("matplotlib not installed")
            return None

        # Auto-detect columns - prefer numeric for Y axis
        group_column = None  # For grouped bar charts
        force_grouped_bar = False
        numeric_cols = []
        text_cols = []
        month_cols = []
        
        if not x_column or not y_column:
            keys = list(data[0].keys())
            
            # Find numeric columns
            for col in keys:
                col_lower = col.lower()
                # Check if it's a month/time column
                if col_lower in ['month', 'месяц', 'date', 'дата', 'период', 'period']:
                    month_cols.append(col)
                    text_cols.append(col)  # Also add to text_cols
                elif any(word in col_lower for word in ['month', 'месяц', 'date', 'дата']):
                    month_cols.append(col)
                    text_cols.append(col)
                else:
                    try:
                        float(data[0].get(col, 0))
                        numeric_cols.append(col)
                    except (ValueError, TypeError):
                        text_cols.append(col)
            
            logger.debug(f"Column auto-detect - numeric: {numeric_cols}, text: {text_cols}, month: {month_cols}")
            
            # 🔥 PRIORITY 1: If month column exists → TIME-SERIES chart (ignore grouped logic)
            if month_cols:
                x_column = month_cols[0]  # Month/time on X-axis
                y_column = numeric_cols[0] if numeric_cols else keys[-1]
                # If there's another text column (not month), use it for grouping (multiple lines)
                non_month_text = [col for col in text_cols if col not in month_cols]
                if len(non_month_text) >= 1:
                    group_column = non_month_text[0]  # e.g., product_name for multiple lines
                logger.info(f"Detected TIME-SERIES data: X={x_column} (time), Group={group_column}, Y={y_column}")
            
            # PRIORITY 2: If we have 3+ columns and 2+ text columns (no month), use grouped chart
            elif len(keys) >= 3 and len(text_cols) >= 2:
                x_column = text_cols[0]  # First text = X axis (e.g., mrf)
                group_column = text_cols[1]  # Second text = Groups (e.g., product_name)
                y_column = numeric_cols[0] if numeric_cols else keys[-1]
                logger.info(f"Detected GROUPED BAR data: X={x_column}, Group={group_column}, Y={y_column}")
            else:
                if not x_column:
                    # Use first text column for X, or first column
                    x_column = text_cols[0] if text_cols else (keys[0] if keys else None)
                
                if not y_column:
                    # Use first numeric column for Y
                    y_column = numeric_cols[0] if numeric_cols else (keys[1] if len(keys) > 1 else keys[0])
            
            if not x_column or not y_column:
                logger.error("Could not auto-detect X and Y columns")
                return None

        # Special case: monthly data with multiple numeric metrics (e.g., plan vs fact)
        if month_cols and len(numeric_cols) >= 2:
            x_column = month_cols[0]
            group_column = "__metric__"
            y_column = "__value__"
            force_grouped_bar = True
            expanded_data = []
            for row in data:
                for metric in numeric_cols:
                    expanded_data.append(
                        {
                            x_column: row.get(x_column, ""),
                            group_column: metric,
                            y_column: row.get(metric, 0),
                        }
                    )
            data = expanded_data

        # Prepare data - handle grouped vs simple charts
        if group_column:
            # Grouped bar chart data (e.g., products by regions)
            # Data structure: [{mrf: 'X', product_name: 'A', revenue: 100}, ...]
            # Need to transform to: X-categories, grouped by product
            
            # Extract unique values
            from collections import defaultdict
            x_values = []
            x_seen = set()
            for row in data:
                x_val = str(row.get(x_column, ''))
                if x_val not in x_seen:
                    x_values.append(x_val)
                    x_seen.add(x_val)
            
            # Group data by group_column (SUM values for same x+group combination)
            grouped_data = defaultdict(lambda: defaultdict(float))
            for row in data:
                x_val = str(row.get(x_column, ''))
                group_val = str(row.get(group_column, ''))
                y_val = float(row.get(y_column, 0))
                grouped_data[group_val][x_val] += y_val  # ← SUM instead of overwrite!
            
            groups = list(grouped_data.keys())
            logger.info(f"Creating grouped bar chart: {len(x_values)} X-categories × {len(groups)} groups")
        else:
            # Simple chart data
            x_data = [str(row.get(x_column, '')) for row in data]
            y_data_raw = [row.get(y_column, 0) for row in data]
            
            logger.debug(f"Chart data - X column: {x_column}, Y column: {y_column}")
            logger.debug(f"Y data raw (first 3): {y_data_raw[:3]}")

            try:
                y_data = [float(v) if v is not None else 0.0 for v in y_data_raw]
            except (ValueError, TypeError) as e:
                logger.error(f"Y data is not numeric: {e}, raw sample: {y_data_raw[:3]}")
                return None

            # Downsample long series
            if len(x_data) > self.max_points:
                step = max(1, len(x_data) // self.max_points)
                x_data = x_data[::step]
                y_data = y_data[::step]
                logger.info(f"Downsampled series to {len(x_data)} points (step={step})")

        fig, ax = plt.subplots(figsize=(self.figure_width, self.figure_height))
        variant = (variant or 'auto').lower()
        
        # Apply number formatting to Y-axis
        from matplotlib.ticker import FuncFormatter
        ax.yaxis.set_major_formatter(FuncFormatter(self._format_number_short))

        # GROUPED BAR CHART (when we have group column)
        if group_column:
            import numpy as np
            
            # Extended color palette for more groups (up to 20 colors)
            colors = [
                '#3498db', '#e74c3c', '#2ecc71', '#f39c12', '#9b59b6', 
                '#1abc9c', '#e67e22', '#34495e', '#16a085', '#c0392b',
                '#8e44ad', '#27ae60', '#d35400', '#2980b9', '#c0ca33',
                '#00897b', '#6a1b9a', '#f57c00', '#5e35b1', '#00acc1'
            ]
            
            # Calculate total for each group
            group_totals = {}
            for group, group_vals in grouped_data.items():
                group_totals[group] = sum(group_vals.values())
            
            # Sort groups by total (descending)
            sorted_groups = sorted(group_totals.items(), key=lambda x: x[1], reverse=True)
            
            # 🔥 Check if X-axis is TIME-SERIES → use LINE CHART instead of BAR CHART
            x_col_lower = x_column.lower()
            is_time_series = any(word in x_col_lower for word in ['month', 'месяц', 'date', 'дата', 'период', 'period'])
            if force_grouped_bar:
                is_time_series = False
            
            # Smart limit: time-series can show more groups (lines don't overlap much)
            if is_time_series:
                MAX_LEGEND_ITEMS = 15  # Show more for time-series (lines are readable)
            else:
                MAX_LEGEND_ITEMS = 7   # Fewer for bar charts (too many bars = clutter)
            
            top_groups = [g[0] for g in sorted_groups[:MAX_LEGEND_ITEMS]]
            groups_to_show = top_groups if len(sorted_groups) > MAX_LEGEND_ITEMS else [g[0] for g in sorted_groups]
            
            if is_time_series:
                # TIME-SERIES: Line chart with multiple lines (one per group)
                for idx, group in enumerate(groups_to_show):
                    group_data = grouped_data[group]
                    y_values = [group_data.get(x, 0) for x in x_values]
                    
                    ax.plot(x_values, y_values, marker='o', linestyle='-', linewidth=2.5,
                           label=group, color=colors[idx % len(colors)], markersize=6, alpha=0.85)
                
                ax.set_xlabel(x_column, fontsize=self.label_fontsize)
                ax.set_ylabel(y_column, fontsize=self.label_fontsize)
                ax.legend(title=group_column, fontsize=10, loc='best')
                ax.grid(True, alpha=0.3)
                plt.xticks(rotation=45, ha='right')
                
                if len(groups) > MAX_LEGEND_ITEMS:
                    logger.info(f"Created TIME-SERIES line chart with top {len(groups_to_show)}/{len(groups)} groups × {len(x_values)} time points")
                else:
                    logger.info(f"Created TIME-SERIES line chart with {len(groups_to_show)} groups × {len(x_values)} time points")
            else:
                # CATEGORICAL: Grouped bar chart
                x_positions = np.arange(len(x_values))
                bar_width = 0.8 / len(groups_to_show)
                
                for idx, group in enumerate(groups_to_show):
                    group_data = grouped_data[group]
                    y_values = [group_data.get(x, 0) for x in x_values]
                    offset = (idx - len(groups_to_show)/2 + 0.5) * bar_width
                    
                    ax.bar(x_positions + offset, y_values, bar_width, 
                          label=group, color=colors[idx % len(colors)], 
                          alpha=0.85, edgecolor='black', linewidth=0.5)
                
                ax.set_xticks(x_positions)
                ax.set_xticklabels(x_values, rotation=45, ha='right')
                ax.set_xlabel(x_column, fontsize=self.label_fontsize)
                ax.set_ylabel(y_column, fontsize=self.label_fontsize)
                ax.legend(title=group_column, fontsize=10, loc='best')
                ax.grid(True, alpha=0.3, axis='y')
                
                if len(groups) > MAX_LEGEND_ITEMS:
                    logger.info(f"Created grouped bar chart with top {len(groups_to_show)}/{len(groups)} groups × {len(x_values)} categories")
                else:
                    logger.info(f"Created grouped bar chart with {len(groups_to_show)} groups × {len(x_values)} categories")
        
        # SIMPLE CHARTS (no grouping)
        else:
            # Auto-detect best chart type if not overridden
            if not variant or variant == 'auto':
                data_type = self._detect_data_type(x_data, y_data)
                # Use line+trend for time/monthly series, bars for categorical
                if data_type in ('monthly', 'timeseries') and len(x_data) >= 5:
                    variant = 'combo'  # Line with trend
                else:
                    variant = 'bar'
            
            # Calculate trend for combo/line charts
            trend_line, slope = self._calculate_trend(y_data)
            
            if variant in ('bar',) or (variant == 'auto' and len(x_data) <= 20):
                # Colored bars based on trend
                colors = ['#2ecc71' if (i == 0 or y_data[i] >= y_data[i-1]) else '#e74c3c' 
                         for i in range(len(y_data))]
                ax.bar(x_data, y_data, color=colors, alpha=0.8, edgecolor='black', linewidth=0.5)
                ax.set_ylabel(y_column, fontsize=self.label_fontsize)
                logger.info(f"Created bar chart with {len(x_data)} bars")
                
            elif variant in ('combo', 'line+trend'):
                # Line chart with trend overlay
                ax.plot(x_data, y_data, marker='o', linestyle='-', linewidth=2.5, 
                       label='Фактические данные', color='#3498db', markersize=6)
                
                # Add trend line
                ax.plot(x_data, trend_line, linestyle='--', linewidth=2, 
                       label='Тренд', color='#e74c3c', alpha=0.7)
                
                # Fill area between
                ax.fill_between(range(len(y_data)), y_data, trend_line, 
                               alpha=0.2, color='#95a5a6')
                
                # Annotate extremes
                try:
                    max_idx, min_idx = self._find_extremes(y_data)
                    if max_idx >= 0:
                        formatted_max = self._format_number_short(y_data[max_idx])
                        ax.annotate(f'Макс: {formatted_max}', 
                                  xy=(max_idx, y_data[max_idx]),
                                  xytext=(0, 10), textcoords='offset points',
                                  ha='center', fontsize=9, color='#27ae60',
                                  bbox=dict(boxstyle='round,pad=0.3', facecolor='#d5f4e6', alpha=0.7))
                    if min_idx >= 0 and min_idx != max_idx:
                        formatted_min = self._format_number_short(y_data[min_idx])
                        ax.annotate(f'Мин: {formatted_min}',
                                  xy=(min_idx, y_data[min_idx]),
                                  xytext=(0, -15), textcoords='offset points',
                                  ha='center', fontsize=9, color='#c0392b',
                                  bbox=dict(boxstyle='round,pad=0.3', facecolor='#fadbd8', alpha=0.7))
                except Exception as e:
                    logger.debug(f"Could not annotate extremes: {e}")
                
                ax.legend(loc='upper left', fontsize=10)
                ax.set_ylabel(y_column, fontsize=self.label_fontsize)
                logger.info(f"Created combo line+trend chart with {len(x_data)} points")
                
            elif variant in ('line',):
                ax.plot(x_data, y_data, marker='o', linestyle='-', linewidth=2)
                ax.set_ylabel(y_column, fontsize=self.label_fontsize)
            elif variant in ('barh', 'horizontal'):
                ax.barh(x_data, y_data)
            elif variant in ('stacked', 'area'):
                ax.fill_between(range(len(y_data)), y_data, step='pre', alpha=0.4)
                ax.plot(range(len(y_data)), y_data, linewidth=2)
                ax.set_xticks(range(len(x_data)))
                ax.set_xticklabels(x_data, rotation=45, ha='right')
            elif variant in ('box', 'boxplot'):
                numeric_cols = self._numeric_columns(data)
                if not numeric_cols:
                    logger.error("No numeric columns for boxplot")
                    return None
                values = [[row.get(col) for row in data if isinstance(row.get(col), (int, float))] for col in numeric_cols]
                ax.boxplot(values, labels=numeric_cols, vert=True)
            elif variant in ('heatmap', 'correlation', 'corr'):
                numeric_cols = self._numeric_columns(data)
                if len(numeric_cols) < 2:
                    logger.error("Need at least 2 numeric columns for heatmap")
                    return None
                matrix = np.array([[row.get(col, 0) if isinstance(row.get(col), (int, float)) else 0 for col in numeric_cols] for row in data])
                if matrix.size == 0:
                    logger.error("No numeric data for heatmap")
                    return None
                corr = np.corrcoef(matrix, rowvar=False)
                cax = ax.imshow(corr, cmap='coolwarm', vmin=-1, vmax=1)
                ax.set_xticks(range(len(numeric_cols)))
                ax.set_yticks(range(len(numeric_cols)))
                ax.set_xticklabels(numeric_cols, rotation=45, ha='right')
                ax.set_yticklabels(numeric_cols)
                fig.colorbar(cax, ax=ax, fraction=0.046, pad=0.04)
            else:
                # default bar
                ax.bar(x_data, y_data)

        # Set common chart properties (if not already set by grouped chart)
        if not group_column:
            ax.set_xlabel(x_column, fontsize=self.label_fontsize)
        
        ax.set_title(title, fontsize=self.title_fontsize, fontweight='bold')
        
        if not group_column:
            ax.grid(True, alpha=0.3)
            if len(x_data) > 10 and variant not in ('heatmap', 'box'):
                plt.xticks(rotation=45, ha='right')

        plt.tight_layout()

        output_dir = Path('./output')
        output_dir.mkdir(exist_ok=True)
        file_path = output_dir / f'{chart_name}.png'

        plt.savefig(file_path, dpi=self.dpi, bbox_inches='tight')
        plt.close(fig)

        logger.info(f"Chart saved to {file_path}")
        return str(file_path)

    def _create_plotly_chart(
        self,
        data: List[Dict[str, Any]],
        chart_name: str,
        title: str,
        x_column: Optional[str],
        y_column: Optional[str]
    ) -> str:
        """Create plotly chart (interactive)."""
        try:
            import plotly.graph_objects as go
        except ImportError:
            logger.error("plotly not installed")
            return None
        
        # Auto-detect columns
        if not x_column or not y_column:
            keys = list(data[0].keys())
            if len(keys) >= 2:
                # Smart type detection
                numeric_cols = []
                text_cols = []
                for col in keys:
                    try:
                        float(data[0].get(col, 0))
                        numeric_cols.append(col)
                    except (ValueError, TypeError):
                        text_cols.append(col)
                
                x_column = x_column or (text_cols[0] if text_cols else keys[0])
                y_column = y_column or (numeric_cols[0] if numeric_cols else keys[1])
                logger.info(f"Auto-detected columns - X: {x_column} (text), Y: {y_column} (numeric)")
            else:
                logger.error("Not enough columns for chart")
                return None
        
        # Extract data
        try:
            x_data = [str(row.get(x_column, '')) for row in data]
            y_data = [float(row.get(y_column, 0)) for row in data]
        except (ValueError, TypeError) as e:
            logger.error(f"Data type error in plotly chart: {e}")
            return None
        
        fig = go.Figure(data=[go.Bar(x=x_data, y=y_data)])
        fig.update_layout(
            title=title,
            xaxis_title=x_column,
            yaxis_title=y_column,
            height=600,
            showlegend=True
        )
        
        # Save to file
        output_dir = Path('./output')
        output_dir.mkdir(exist_ok=True)
        file_path = output_dir / f'{chart_name}.html'
        
        fig.write_html(file_path)
        logger.info(f"Interactive chart saved to {file_path}")
        return str(file_path)

    def export_pdf(
        self,
        text_sections: List[Tuple[str, str]],
        chart_paths: Optional[List[str]] = None,
        output_name: str = 'report.pdf'
    ) -> Optional[str]:
        """Export analysis and charts to a PDF using matplotlib PdfPages."""
        try:
            import matplotlib.pyplot as plt
            import matplotlib
            from matplotlib.backends.backend_pdf import PdfPages
            import matplotlib.image as mpimg
            matplotlib.use('Agg')
        except ImportError:
            logger.error("matplotlib not installed for PDF export")
            return None

        output_dir = Path('./output')
        output_dir.mkdir(exist_ok=True)
        pdf_path = output_dir / output_name

        chart_paths = chart_paths or []

        # Convert mm to inches (A4: 210mm x 297mm = 8.27" x 11.69")
        a4_portrait_width = self.pdf_a4_width_mm / 25.4
        a4_portrait_height = self.pdf_a4_height_mm / 25.4
        a4_landscape_width = self.pdf_a4_height_mm / 25.4
        a4_landscape_height = self.pdf_a4_width_mm / 25.4

        with PdfPages(pdf_path) as pdf:
            # Text page
            fig, ax = plt.subplots(figsize=(a4_portrait_width, a4_portrait_height))
            ax.axis('off')
            ax.set_xlim(0, 1)
            ax.set_ylim(0, 1)
            
            y = 0.95
            line_height = 0.025  # Height per line of text (increased)
            section_spacing = 0.08  # Space between sections (increased)
            
            for title, body in text_sections:
                # Title
                ax.text(0.05, y, title, fontsize=11, fontweight='bold', 
                       va='top', wrap=True, family='sans-serif')
                y -= 0.05  # Space after title
                
                # Body text with proper wrapping - smaller width for better spacing
                wrapped = textwrap.fill(body or "", width=75)
                num_lines = len(wrapped.split('\n'))
                
                ax.text(0.08, y, wrapped, fontsize=8, va='top', wrap=True, 
                       family='sans-serif', linespacing=2.2)
                
                # Move down by number of lines + spacing
                y -= (num_lines * line_height + section_spacing)
            
            pdf.savefig(fig, bbox_inches='tight')
            plt.close(fig)

            # Charts pages
            for chart in chart_paths:
                try:
                    if chart.endswith('.html'):
                        logger.warning("Skipping HTML chart in PDF export: %s", chart)
                        continue
                    img = mpimg.imread(chart)
                    fig, ax = plt.subplots(figsize=(a4_landscape_width, a4_landscape_height))
                    ax.axis('off')
                    ax.imshow(img)
                    pdf.savefig(fig, bbox_inches='tight')
                    plt.close(fig)
                except Exception as e:
                    logger.error(f"Failed to embed chart {chart} into PDF: {e}")

        logger.info(f"PDF report saved to {pdf_path}")
        return str(pdf_path)

    def format_table(self, data: List[Dict[str, Any]]) -> str:
        """Format data as text table."""
        if not data:
            return "No data"
        
        # Limit rows using settings
        displayed_data = data[:self.table_max_rows]
        
        # Get columns
        columns = list(displayed_data[0].keys())
        
        # Create header
        header = " | ".join(str(c)[:15] for c in columns)
        separator = "-" * len(header)
        
        # Create rows
        rows = []
        for row in displayed_data:
            row_str = " | ".join(str(row.get(c, ''))[:15] for c in columns)
            rows.append(row_str)
        
        result = f"{header}\n{separator}\n" + "\n".join(rows)
        
        if len(data) > self.table_max_rows:
            result += f"\n... and {len(data) - self.table_max_rows} more rows"
        
        return result

    def _create_bokeh_chart(
        self,
        data: List[Dict[str, Any]],
        chart_name: str,
        title: str,
        x_column: Optional[str],
        y_column: Optional[str],
        variant: Optional[str]
    ) -> Optional[str]:
        """Create Bokeh chart and export to PNG for Telegram."""
        from bokeh.plotting import figure
        from bokeh.models import ColumnDataSource, HoverTool
        from bokeh.transform import dodge
        from bokeh.palettes import Category20_20
        from bokeh.io import export_png
        import pandas as pd
        import os
        
        try:
            # Configure webdriver for export_png
            from selenium import webdriver
            from selenium.webdriver.chrome.options import Options
            from selenium.webdriver.chrome.service import Service
            
            chrome_options = Options()
            chrome_options.add_argument('--headless')
            chrome_options.add_argument('--no-sandbox')
            chrome_options.add_argument('--disable-gpu')
            chrome_options.add_argument('--disable-dev-shm-usage')
            
            chrome_bin = os.environ.get('CHROME_BIN', '/usr/bin/chromium')
            chromedriver_path = os.environ.get('BOKEH_CHROMEDRIVER', '/usr/bin/chromedriver')
            
            chrome_options.binary_location = chrome_bin
            service = Service(executable_path=chromedriver_path)
            driver = webdriver.Chrome(service=service, options=chrome_options)
            df = pd.DataFrame(data)
            
            # Auto-detect columns
            if not x_column or not y_column:
                # Separate numeric and text columns
                numeric_cols = df.select_dtypes(include=['number']).columns.tolist()
                text_cols = df.select_dtypes(include=['object', 'string']).columns.tolist()
                
                if not x_column:
                    x_column = text_cols[0] if len(text_cols) > 0 else df.columns[0]
                if not y_column:
                    y_column = numeric_cols[0] if len(numeric_cols) > 0 else df.columns[1]
                
                logger.info(f"Auto-detected columns in bokeh - X: {x_column}, Y: {y_column}")
            
            output_dir = Path('output')
            output_dir.mkdir(exist_ok=True)
            output_path = output_dir / f'{chart_name}.png'
            
            # Check for grouped data (multiple dimensions)
            is_grouped = len(df.columns) > 2
            
            if variant == 'grouped_bar' or (is_grouped and not variant):
                # Grouped bar chart for multi-dimensional data
                p = figure(
                    x_range=list(df[x_column].unique()),
                    title=title,
                    width=self.figure_width * 80,
                    height=self.figure_height * 80,
                    toolbar_location=None
                )
                
                # Get grouping column (third column)
                group_col = df.columns[2] if len(df.columns) > 2 else None
                if group_col:
                    groups = df[group_col].unique()[:10]  # Limit to 10 groups
                    colors = Category20_20[:len(groups)]
                    
                    for idx, (group, color) in enumerate(zip(groups, colors)):
                        group_data = df[df[group_col] == group]
                        p.vbar(
                            x=dodge(group_col, -0.25 + idx * 0.5 / len(groups), range=p.x_range),
                            top=y_column,
                            width=0.4 / len(groups),
                            source=ColumnDataSource(group_data),
                            color=color,
                            legend_label=str(group)
                        )
                
                p.legend.location = 'top_right'
                p.legend.click_policy = 'hide'
                
            else:
                # Simple bar chart
                p = figure(
                    x_range=list(df[x_column].astype(str)[:self.max_points]),
                    title=title,
                    width=self.figure_width * 80,
                    height=self.figure_height * 80,
                    toolbar_location=None
                )
                
                p.vbar(
                    x=x_column,
                    top=y_column,
                    width=0.8,
                    source=ColumnDataSource(df[:self.max_points]),
                    color='skyblue'
                )
            
            # Add hover tool
            hover = HoverTool(tooltips=[
                (x_column, f'@{x_column}'),
                (y_column, f'@{y_column}')
            ])
            p.add_tools(hover)
            
            p.xaxis.major_label_orientation = 0.785  # 45 degrees
            p.xaxis.axis_label = x_column
            p.yaxis.axis_label = y_column
            
            export_png(p, filename=str(output_path), webdriver=driver)
            driver.quit()
            logger.info(f"Bokeh chart saved to {output_path}")
            return str(output_path)
            
        except Exception as e:
            logger.error(f"Error creating Bokeh chart: {e}")
            return None

    @staticmethod
    def _numeric_columns(data: List[Dict[str, Any]]) -> List[str]:
        """Detect numeric columns in dataset."""
        if not data:
            return []
        sample = data[0]
        return [k for k, v in sample.items() if isinstance(v, (int, float))]