"""
SonarQube trend report generator for analyzing metric trends over time.
"""
import logging
import requests
import json
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
class TrendReportGenerator:
"""Generates HTML trend reports from SonarQube historical metrics."""
# Core metrics to track
CORE_METRICS = [
"bugs",
"vulnerabilities",
"code_smells",
"coverage",
"duplicated_lines_density",
"ncloc"
]
# Metrics where lower is better (improving means decreasing)
LOWER_IS_BETTER = {"bugs", "vulnerabilities", "code_smells", "duplicated_lines_density"}
def __init__(self, sonar_url: str, sonar_token: str, reports_dir: str = "./reports/trends"):
"""
Initialize trend report generator.
Args:
sonar_url: SonarQube server URL
sonar_token: SonarQube authentication token
reports_dir: Directory where trend reports will be saved
"""
self.sonar_url = sonar_url.rstrip("/")
self.sonar_token = sonar_token
self.reports_dir = Path(reports_dir)
self.reports_dir.mkdir(parents=True, exist_ok=True)
self.headers = {"Authorization": f"Bearer {self.sonar_token}"}
def fetch_historical_metrics(
self,
project_key: str,
metric_keys: List[str],
days: int = 30
) -> Dict[str, List[Dict]]:
"""
Fetch historical metric data from SonarQube API.
Args:
project_key: SonarQube project key
metric_keys: List of metric keys to fetch
days: Number of days of history to fetch
Returns:
Dictionary mapping metric keys to lists of historical data points
"""
historical_data = {}
# Calculate date range
to_date = datetime.now()
from_date = to_date - timedelta(days=days)
logger.info(f"Fetching historical metrics for {project_key} from {from_date.date()} to {to_date.date()}")
for metric_key in metric_keys:
try:
url = f"{self.sonar_url}/api/measures/search_history"
# SonarQube API expects dates in YYYY-MM-DD format
params = {
"component": project_key,
"metrics": metric_key,
"from": from_date.strftime("%Y-%m-%d"),
"to": to_date.strftime("%Y-%m-%d")
}
response = requests.get(url, params=params, headers=self.headers, timeout=30)
if response.status_code == 200:
data = response.json()
# Extract history data
measures = data.get("measures", [])
if measures:
metric_data = measures[0]
history = metric_data.get("history", [])
if history:
historical_data[metric_key] = history
logger.debug(f"Fetched {len(history)} data points for {metric_key}")
else:
logger.warning(f"No history array found for metric {metric_key}")
historical_data[metric_key] = []
else:
logger.warning(f"No measures found for metric {metric_key}")
historical_data[metric_key] = []
elif response.status_code == 404:
logger.warning(f"Project or metric not found: {metric_key}")
historical_data[metric_key] = []
else:
logger.warning(f"Failed to fetch {metric_key}: HTTP {response.status_code} - {response.text[:200]}")
historical_data[metric_key] = []
except Exception as e:
logger.error(f"Error fetching historical data for {metric_key}: {e}")
historical_data[metric_key] = []
return historical_data
def calculate_trends(self, historical_data: Dict[str, List[Dict]]) -> Dict[str, Dict]:
"""
Calculate upward/downward trends for each metric.
Args:
historical_data: Dictionary mapping metric keys to historical data points
Returns:
Dictionary with trend information for each metric
"""
trends = {}
for metric_key, history in historical_data.items():
if not history or len(history) < 2:
# Not enough data for trend calculation
trends[metric_key] = {
"current": None,
"previous": None,
"change": None,
"change_percent": None,
"trend": "insufficient_data",
"is_improving": None
}
continue
# Sort by date (most recent first)
sorted_history = sorted(history, key=lambda x: x.get("date", ""), reverse=True)
# Get current and previous values
current_point = sorted_history[0]
previous_point = sorted_history[-1] if len(sorted_history) > 1 else sorted_history[0]
current_value = self._parse_metric_value(current_point.get("value"))
previous_value = self._parse_metric_value(previous_point.get("value"))
if current_value is None or previous_value is None:
trends[metric_key] = {
"current": current_value,
"previous": previous_value,
"change": None,
"change_percent": None,
"trend": "no_data",
"is_improving": None
}
continue
# Calculate change
change = current_value - previous_value
# Calculate percentage change
if previous_value != 0:
change_percent = (change / previous_value) * 100
else:
change_percent = 100.0 if change > 0 else -100.0 if change < 0 else 0.0
# Determine if trend is improving
# For "lower is better" metrics, decreasing is improving
# For "higher is better" metrics (like coverage), increasing is improving
is_lower_better = metric_key in self.LOWER_IS_BETTER
if is_lower_better:
is_improving = change < 0 # Decreasing is good
else:
is_improving = change > 0 # Increasing is good
# Determine trend direction
if abs(change_percent) < 0.01: # Essentially no change
trend = "stable"
elif is_improving:
trend = "improving"
else:
trend = "degrading"
trends[metric_key] = {
"current": current_value,
"previous": previous_value,
"change": change,
"change_percent": change_percent,
"trend": trend,
"is_improving": is_improving,
"history": sorted_history # Include full history for charts
}
return trends
def _parse_metric_value(self, value: Optional[str]) -> Optional[float]:
"""
Parse metric value from string to float.
Args:
value: Metric value as string
Returns:
Float value or None if invalid
"""
if value is None:
return None
try:
# Handle percentage values (e.g., "45.5%")
if isinstance(value, str) and value.endswith("%"):
return float(value.rstrip("%"))
return float(value)
except (ValueError, TypeError):
return None
def _format_metric_name(self, metric_key: str) -> str:
"""Format metric key for display."""
return metric_key.replace("_", " ").title()
def _get_trend_indicator(self, trend_data: Dict) -> Tuple[str, str]:
"""
Get trend indicator symbol and color.
Returns:
Tuple of (symbol, color_class)
"""
trend = trend_data.get("trend", "unknown")
is_improving = trend_data.get("is_improving")
if trend == "improving":
return ("↑", "text-success")
elif trend == "degrading":
return ("↓", "text-danger")
elif trend == "stable":
return ("→", "text-secondary")
else:
return ("—", "text-muted")
def generate_html_report(
self,
project_key: str,
project_name: Optional[str] = None,
trends_data: Dict[str, Dict] = None,
days: int = 30
) -> Path:
"""
Generate HTML trend report.
Args:
project_key: SonarQube project key
project_name: Project display name (optional)
trends_data: Calculated trends data
days: Number of days analyzed
Returns:
Path to generated HTML file
"""
if project_name is None:
project_name = project_key
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{project_key}_trends_{timestamp}.html"
filepath = self.reports_dir / filename
logger.info(f"Generating HTML trend report: {filepath}")
# Generate HTML content
html_content = self._build_html_content(project_key, project_name, trends_data, days)
# Write to file
with open(filepath, "w", encoding="utf-8") as f:
f.write(html_content)
logger.info(f"Trend report generated successfully: {filepath}")
return filepath
def _build_html_content(
self,
project_key: str,
project_name: str,
trends_data: Dict[str, Dict],
days: int
) -> str:
"""Build complete HTML content."""
# Build summary table rows
summary_rows = ""
chart_configs = []
for metric_key in self.CORE_METRICS:
trend_info = trends_data.get(metric_key, {})
symbol, color_class = self._get_trend_indicator(trend_info)
current = trend_info.get("current")
previous = trend_info.get("previous")
change = trend_info.get("change")
change_percent = trend_info.get("change_percent")
# Format values
current_str = self._format_value(current, metric_key)
previous_str = self._format_value(previous, metric_key)
if change is not None and change_percent is not None:
change_str = f"{change:+.2f}" if isinstance(change, float) else str(change)
change_percent_str = f"{change_percent:+.2f}%"
else:
change_str = "N/A"
change_percent_str = "N/A"
summary_rows += f"""
{self._format_metric_name(metric_key)}
{current_str}
{previous_str}
{change_str}
{change_percent_str}
{symbol}
"""
# Prepare chart data
history = trend_info.get("history", [])
if history:
chart_configs.append(self._build_chart_config(metric_key, history))
# Build chart HTML
charts_html = ""
if chart_configs:
for i, chart_config in enumerate(chart_configs):
chart_id = f"chart_{i}"
charts_html += f"""
'
# Build chart JavaScript
chart_js = ""
if chart_configs:
for i, chart_config in enumerate(chart_configs):
chart_id = f"chart_{i}"
# The config is already a JSON string, which is valid JavaScript
chart_js += f"""
var ctx_{i} = document.getElementById('{chart_id}').getContext('2d');
var config_{i} = {chart_config['config']};
new Chart(ctx_{i}, config_{i});
"""
html = f"""
SonarQube Trend Report - {project_name}
"""
return html
def _format_value(self, value: Optional[float], metric_key: str) -> str:
"""Format metric value for display."""
if value is None:
return "N/A"
# Format based on metric type
if metric_key == "coverage" or metric_key == "duplicated_lines_density":
return f"{value:.2f}%"
elif metric_key == "ncloc":
return f"{int(value):,}" if isinstance(value, (int, float)) else str(value)
else:
return f"{int(value):,}" if isinstance(value, (int, float)) and value == int(value) else f"{value:.2f}"
def _build_chart_config(self, metric_key: str, history: List[Dict]) -> Dict:
"""Build Chart.js configuration for a metric."""
# Sort history by date
sorted_history = sorted(history, key=lambda x: x.get("date", ""))
# Extract labels and data
labels = []
data = []
for point in sorted_history:
date_str = point.get("date", "")
value = self._parse_metric_value(point.get("value"))
if date_str and value is not None:
# Format date for display
try:
# Handle various date formats from SonarQube API
if "T" in date_str:
# ISO format with time
date_obj = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
else:
# Simple date format YYYY-MM-DD
date_obj = datetime.strptime(date_str[:10], "%Y-%m-%d")
labels.append(date_obj.strftime("%Y-%m-%d"))
except Exception as e:
logger.debug(f"Date parsing error for {date_str}: {e}")
# Use first 10 chars as fallback
labels.append(date_str[:10] if len(date_str) >= 10 else date_str)
data.append(value)
# Determine if lower is better for color
is_lower_better = metric_key in self.LOWER_IS_BETTER
border_color = "rgb(220, 53, 69)" if is_lower_better else "rgb(40, 167, 69)"
background_color = "rgba(220, 53, 69, 0.1)" if is_lower_better else "rgba(40, 167, 69, 0.1)"
config = {
"type": "line",
"data": {
"labels": labels,
"datasets": [{
"label": self._format_metric_name(metric_key),
"data": data,
"borderColor": border_color,
"backgroundColor": background_color,
"borderWidth": 2,
"fill": True,
"tension": 0.4
}]
},
"options": {
"responsive": True,
"maintainAspectRatio": True,
"scales": {
"y": {
"beginAtZero": False
}
},
"plugins": {
"legend": {
"display": True
}
}
}
}
return {
"metric": metric_key,
"config": json.dumps(config) # Convert to JSON string
}
def generate_report(self, project_key: str, days: int = 30) -> Optional[Path]:
"""
Main entry point to generate trend report.
Args:
project_key: SonarQube project key
days: Number of days of history to analyze
Returns:
Path to generated HTML file or None if failed
"""
try:
# Fetch project name
project_name = self._get_project_name(project_key)
# Fetch historical metrics
logger.info(f"Fetching historical metrics for project: {project_key}")
historical_data = self.fetch_historical_metrics(
project_key,
self.CORE_METRICS,
days
)
# Calculate trends
logger.info("Calculating trends...")
trends_data = self.calculate_trends(historical_data)
# Generate HTML report
logger.info("Generating HTML report...")
report_path = self.generate_html_report(
project_key,
project_name,
trends_data,
days
)
return report_path
except Exception as e:
logger.error(f"Failed to generate trend report: {e}")
return None
def _get_project_name(self, project_key: str) -> Optional[str]:
"""Get project name from SonarQube API."""
try:
url = f"{self.sonar_url}/api/projects/search"
params = {"projects": project_key}
response = requests.get(url, params=params, headers=self.headers, timeout=30)
if response.status_code == 200:
data = response.json()
components = data.get("components", [])
for component in components:
if component.get("key") == project_key:
return component.get("name", project_key)
return project_key
except Exception as e:
logger.warning(f"Failed to fetch project name: {e}")
return project_key