| |
| |
| """ |
| Coverage Trend Export Utility |
| |
| Purpose: Export trending data for external analysis (Excel, BI tools, custom scripts). |
| Supports multiple formats (CSV, JSON, Excel) with date range filtering. |
| |
| Usage: |
| python coverage_trend_export.py [options] |
| |
| Options: |
| --trending-file PATH Path to cross_platform_trend.json (default: relative path) |
| --output PATH Path to output file (default: coverage_export.csv) |
| --format FORMAT Export format: csv|json|excel (default: csv) |
| --days INT Number of days to export (default: 30, 0 for all history) |
| |
| Example: |
| python coverage_trend_export.py --format csv --days 30 |
| python coverage_trend_export.py --format excel --days 90 --output quarterly_report.xlsx |
| python coverage_trend_export.py --format json --days 0 --output full_history.json |
| """ |
|
|
| import argparse |
| import csv |
| import json |
| import logging |
| import sys |
| from datetime import datetime, timedelta, timezone |
| from pathlib import Path |
| from typing import Dict, List, Optional |
|
|
| |
| try: |
| import openpyxl |
| from openpyxl import Workbook |
| from openpyxl.styles import Font, Alignment, PatternFill |
| EXCEL_SUPPORT = True |
| except ImportError: |
| EXCEL_SUPPORT = False |
|
|
| |
| logging.basicConfig( |
| level=logging.INFO, |
| format='%(levelname)s: %(message)s' |
| ) |
| logger = logging.getLogger(__name__) |
|
|
| |
| TREND_FILE = Path("tests/coverage_reports/metrics/cross_platform_trend.json") |
|
|
|
|
| def load_trending_data(trend_file: Path) -> dict: |
| """ |
| Load trending data from cross_platform_trend.json. |
| |
| Returns empty dict with history=[] if file doesn't exist. |
| """ |
| try: |
| with open(trend_file, 'r') as f: |
| data = json.load(f) |
| logger.info(f"Loaded trending data from {trend_file}") |
| return data |
| except FileNotFoundError: |
| logger.warning(f"Trending file not found: {trend_file}") |
| return {"history": [], "latest": None} |
| except json.JSONDecodeError as e: |
| logger.error(f"Invalid JSON in trending file: {e}") |
| return {"history": [], "latest": None} |
|
|
|
|
| def filter_by_date(history: list, days: int) -> list: |
| """ |
| Filter history entries to last N days. |
| |
| Args: |
| history: List of trend entries |
| days: Number of days to filter (0 for all history) |
| |
| Returns: |
| Filtered list of entries |
| """ |
| if days == 0: |
| return history |
|
|
| cutoff_date = datetime.now(timezone.utc) - timedelta(days=days) |
| cutoff_iso = cutoff_date.isoformat().replace('+00:00', 'Z') |
|
|
| filtered = [entry for entry in history if entry.get('timestamp', '') >= cutoff_iso] |
|
|
| logger.info(f"Filtered {len(history)} entries to {len(filtered)} entries (last {days} days)") |
| return filtered |
|
|
|
|
| def calculate_summary_stats(trending_data: dict) -> dict: |
| """ |
| Calculate summary statistics for trending data. |
| |
| Returns dict with per-platform min/max/avg/current stats. |
| """ |
| history = trending_data.get('history', []) |
|
|
| if not history: |
| return {} |
|
|
| platforms = ['backend', 'frontend', 'mobile', 'desktop'] |
| stats = {} |
|
|
| for platform in platforms: |
| coverages = [ |
| entry.get('platforms', {}).get(platform, 0.0) |
| for entry in history |
| if platform in entry.get('platforms', {}) |
| ] |
|
|
| if coverages: |
| stats[platform] = { |
| 'min': round(min(coverages), 2), |
| 'max': round(max(coverages), 2), |
| 'avg': round(sum(coverages) / len(coverages), 2), |
| 'current': round(coverages[-1], 2) if coverages else 0.0, |
| 'first': round(coverages[0], 2) if coverages else 0.0, |
| 'last': round(coverages[-1], 2) if coverages else 0.0, |
| 'count': len(coverages) |
| } |
|
|
| |
| stats['overall'] = { |
| 'entry_count': len(history), |
| 'date_range': { |
| 'start': history[0].get('timestamp') if history else None, |
| 'end': history[-1].get('timestamp') if history else None |
| } |
| } |
|
|
| return stats |
|
|
|
|
| def export_to_csv(trending_data: dict, output_file: Path, days: int = 30) -> None: |
| """ |
| Export trending data to CSV format. |
| |
| CSV columns: |
| - timestamp, overall_coverage, backend, frontend, mobile, desktop, commit_sha, branch |
| """ |
| history = trending_data.get('history', []) |
| filtered = filter_by_date(history, days) |
|
|
| if not filtered: |
| logger.warning("No data to export") |
| return |
|
|
| try: |
| with open(output_file, 'w', newline='') as f: |
| writer = csv.writer(f) |
|
|
| |
| writer.writerow([ |
| 'timestamp', |
| 'overall_coverage', |
| 'backend', |
| 'frontend', |
| 'mobile', |
| 'desktop', |
| 'commit_sha', |
| 'branch' |
| ]) |
|
|
| |
| for entry in filtered: |
| platforms = entry.get('platforms', {}) |
| writer.writerow([ |
| entry.get('timestamp', ''), |
| entry.get('overall_coverage', 0.0), |
| platforms.get('backend', 0.0), |
| platforms.get('frontend', 0.0), |
| platforms.get('mobile', 0.0), |
| platforms.get('desktop', 0.0), |
| entry.get('commit_sha', ''), |
| entry.get('branch', '') |
| ]) |
|
|
| logger.info(f"Exported {len(filtered)} rows to {output_file}") |
|
|
| |
| if filtered: |
| start_date = filtered[0].get('timestamp', 'Unknown') |
| end_date = filtered[-1].get('timestamp', 'Unknown') |
| logger.info(f"Date range: {start_date} to {end_date}") |
|
|
| except IOError as e: |
| logger.error(f"Failed to write CSV: {e}") |
| sys.exit(1) |
|
|
|
|
| def export_to_json(trending_data: dict, output_file: Path, days: int = 30) -> None: |
| """ |
| Export trending data to JSON format with metadata. |
| |
| JSON structure: |
| { |
| "export_time": "2026-03-07T19:30:00Z", |
| "total_entries": 100, |
| "filtered_entries": 30, |
| "date_range": {...}, |
| "summary_stats": {...}, |
| "history": [...] |
| } |
| """ |
| history = trending_data.get('history', []) |
| filtered = filter_by_date(history, days) |
|
|
| if not filtered: |
| logger.warning("No data to export") |
| return |
|
|
| |
| summary_stats = calculate_summary_stats({'history': filtered}) |
|
|
| |
| export_data = { |
| 'export_time': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'), |
| 'total_entries': len(history), |
| 'filtered_entries': len(filtered), |
| 'date_range': { |
| 'start': filtered[0].get('timestamp') if filtered else None, |
| 'end': filtered[-1].get('timestamp') if filtered else None |
| }, |
| 'summary_stats': summary_stats, |
| 'history': filtered |
| } |
|
|
| try: |
| with open(output_file, 'w') as f: |
| json.dump(export_data, f, indent=2) |
|
|
| logger.info(f"Exported {len(filtered)} entries to {output_file}") |
|
|
| |
| if filtered: |
| start_date = filtered[0].get('timestamp', 'Unknown') |
| end_date = filtered[-1].get('timestamp', 'Unknown') |
| logger.info(f"Date range: {start_date} to {end_date}") |
|
|
| except IOError as e: |
| logger.error(f"Failed to write JSON: {e}") |
| sys.exit(1) |
|
|
|
|
| def export_to_excel(trending_data: dict, output_file: Path, days: int = 30) -> None: |
| """ |
| Export trending data to Excel format with multiple sheets. |
| |
| Sheets: |
| 1. Summary - Overall stats, platform breakdown |
| 2. History - Time series data with all columns |
| """ |
| if not EXCEL_SUPPORT: |
| logger.error("Excel export requires openpyxl. Install with: pip install openpyxl") |
| sys.exit(1) |
|
|
| history = trending_data.get('history', []) |
| filtered = filter_by_date(history, days) |
|
|
| if not filtered: |
| logger.warning("No data to export") |
| return |
|
|
| try: |
| wb = Workbook() |
|
|
| |
| wb.remove(wb.active) |
|
|
| |
| ws_summary = wb.create_sheet('Summary') |
|
|
| |
| ws_summary['A1'] = 'Coverage Trend Summary' |
| ws_summary['A1'].font = Font(bold=True, size=14) |
| ws_summary.merge_cells('A1:B1') |
|
|
| ws_summary['A3'] = 'Export Date:' |
| ws_summary['B3'] = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC') |
|
|
| ws_summary['A5'] = 'Total Entries:' |
| ws_summary['B5'] = len(history) |
|
|
| ws_summary['A6'] = 'Filtered Entries:' |
| ws_summary['B6'] = len(filtered) |
|
|
| ws_summary['A7'] = 'Date Range:' |
| ws_summary['B7'] = f"{filtered[0].get('timestamp', 'Unknown')} to {filtered[-1].get('timestamp', 'Unknown')}" |
|
|
| |
| ws_summary['A9'] = 'Platform' |
| ws_summary['B9'] = 'Current' |
| ws_summary['C9'] = 'Min' |
| ws_summary['D9'] = 'Max' |
| ws_summary['E9'] = 'Average' |
|
|
| |
| for col in ['A', 'B', 'C', 'D', 'E']: |
| cell = ws_summary[f'{col}9'] |
| cell.font = Font(bold=True) |
| cell.fill = PatternFill(start_color='CCCCCC', end_color='CCCCCC', fill_type='solid') |
|
|
| |
| platforms = ['backend', 'frontend', 'mobile', 'desktop'] |
| stats = calculate_summary_stats({'history': filtered}) |
|
|
| row = 10 |
| for platform in platforms: |
| if platform in stats: |
| ws_summary[f'A{row}'] = platform.capitalize() |
| ws_summary[f'B{row}'] = stats[platform]['current'] |
| ws_summary[f'C{row}'] = stats[platform]['min'] |
| ws_summary[f'D{row}'] = stats[platform]['max'] |
| ws_summary[f'E{row}'] = stats[platform]['avg'] |
| row += 1 |
|
|
| |
| ws_summary.column_dimensions['A'].width = 15 |
| ws_summary.column_dimensions['B'].width = 12 |
| ws_summary.column_dimensions['C'].width = 12 |
| ws_summary.column_dimensions['D'].width = 12 |
| ws_summary.column_dimensions['E'].width = 12 |
|
|
| |
| ws_history = wb.create_sheet('History') |
|
|
| |
| headers = ['Timestamp', 'Overall', 'Backend', 'Frontend', 'Mobile', 'Desktop', 'Commit SHA', 'Branch'] |
| for col, header in enumerate(headers, start=1): |
| cell = ws_history.cell(row=1, column=col) |
| cell.value = header |
| cell.font = Font(bold=True) |
| cell.fill = PatternFill(start_color='CCCCCC', end_color='CCCCCC', fill_type='solid') |
|
|
| |
| for row_idx, entry in enumerate(filtered, start=2): |
| platforms = entry.get('platforms', {}) |
| ws_history.cell(row=row_idx, column=1).value = entry.get('timestamp', '') |
| ws_history.cell(row=row_idx, column=2).value = entry.get('overall_coverage', 0.0) |
| ws_history.cell(row=row_idx, column=3).value = platforms.get('backend', 0.0) |
| ws_history.cell(row=row_idx, column=4).value = platforms.get('frontend', 0.0) |
| ws_history.cell(row=row_idx, column=5).value = platforms.get('mobile', 0.0) |
| ws_history.cell(row=row_idx, column=6).value = platforms.get('desktop', 0.0) |
| ws_history.cell(row=row_idx, column=7).value = entry.get('commit_sha', '') |
| ws_history.cell(row=row_idx, column=8).value = entry.get('branch', '') |
|
|
| |
| ws_history.column_dimensions['A'].width = 25 |
| for col in ['B', 'C', 'D', 'E', 'F']: |
| ws_history.column_dimensions[col].width = 12 |
| ws_history.column_dimensions['G'].width = 20 |
| ws_history.column_dimensions['H'].width = 15 |
|
|
| |
| for row in range(2, len(filtered) + 2): |
| for col in [2, 3, 4, 5, 6]: |
| ws_history.cell(row=row, column=col).number_format = '0.00' |
|
|
| |
| wb.save(output_file) |
|
|
| logger.info(f"Exported {len(filtered)} entries to {output_file}") |
|
|
| |
| if filtered: |
| start_date = filtered[0].get('timestamp', 'Unknown') |
| end_date = filtered[-1].get('timestamp', 'Unknown') |
| logger.info(f"Date range: {start_date} to {end_date}") |
|
|
| except Exception as e: |
| logger.error(f"Failed to write Excel: {e}") |
| sys.exit(1) |
|
|
|
|
| def main(): |
| """Main entry point for CLI.""" |
| parser = argparse.ArgumentParser( |
| description='Export coverage trending data for external analysis', |
| formatter_class=argparse.RawDescriptionHelpFormatter, |
| epilog=""" |
| Examples: |
| # Export last 30 days to CSV |
| python coverage_trend_export.py --format csv --days 30 |
| |
| # Export all history to JSON |
| python coverage_trend_export.py --format json --days 0 |
| |
| # Export last 90 days to Excel |
| python coverage_trend_export.py --format excel --days 90 --output quarterly_report.xlsx |
| """ |
| ) |
|
|
| parser.add_argument( |
| '--trending-file', |
| type=Path, |
| default=TREND_FILE, |
| help='Path to cross_platform_trend.json' |
| ) |
|
|
| parser.add_argument( |
| '--output', |
| type=Path, |
| default=Path('coverage_export.csv'), |
| help='Path to output file' |
| ) |
|
|
| parser.add_argument( |
| '--format', |
| choices=['csv', 'json', 'excel'], |
| default='csv', |
| help='Export format (default: csv)' |
| ) |
|
|
| parser.add_argument( |
| '--days', |
| type=int, |
| default=30, |
| help='Number of days to export (default: 30, 0 for all history)' |
| ) |
|
|
| args = parser.parse_args() |
|
|
| |
| trending_data = load_trending_data(args.trending_file) |
|
|
| if not trending_data.get('history'): |
| logger.error("No trending data found") |
| sys.exit(1) |
|
|
| |
| if args.format == 'csv': |
| export_to_csv(trending_data, args.output, args.days) |
| elif args.format == 'json': |
| export_to_json(trending_data, args.output, args.days) |
| elif args.format == 'excel': |
| export_to_excel(trending_data, args.output, args.days) |
|
|
| logger.info("Export complete") |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|