Add charts for Overview and Data Quality Panel

Signed-off-by: Sampurna Pyne <sampurnapyne1710@gmail.com>

Author: Samk1710

insights/charts/data_quality_panel.py

@@ -0,0 +1,238 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from collections import defaultdict
+from datetime import timedelta
+from itertools import accumulate
+from typing import Any
+from typing import Dict
+
+from django.db.models import Count
+from django.db.models.functions import TruncMonth
+from django.utils import timezone
+
+from insights.models import DataQualityIssueByDatasourceInsight
+from insights.models import DataQualityToDosResolutionInsight
+from insights.utils import format_issue_type_label
+from vulnerabilities.models import AdvisoryToDoV2
+
+# Ignore Phantom Importers that don't collect Affected Packages
+IGNORED_IMPORTERS = {
+    "epss_importer_v2",
+    "epss",
+    "vulnrichment_importer_v2",
+    "vulnrichment",
+    "suse_importer_v2",
+    "suse_score",
+}
+
+
+# Open Issues by Type and Datasource Contribution per Open Issue
+def open_issues_to_datasource_queryset():
+    """Return a query set of open issue counts by type and datasource."""
+    return (
+        AdvisoryToDoV2.objects.filter(is_resolved=False, advisories__datasource_id__isnull=False)
+        .exclude(advisories__datasource_id__in=IGNORED_IMPORTERS)
+        .values("issue_type", "advisories__datasource_id")
+        .annotate(count=Count("todo_id", distinct=True))
+        .iterator()
+    )
+
+
+def iter_open_issues_to_datasource_insights():
+    """Yield DataQualityIssueByDatasourceInsight objects."""
+    for row in open_issues_to_datasource_queryset():
+        yield DataQualityIssueByDatasourceInsight(
+            issue_type=row["issue_type"],
+            datasource_id=row["advisories__datasource_id"],
+            count=row["count"],
+        )
+
+
+def collect_open_issues_to_datasource(pipeline: Any) -> None:
+    """Collect open issue counts by type and datasource."""
+    pipeline.data_quality_issue_types = list(iter_open_issues_to_datasource_insights())
+
+
+def build_issue_type_columns(issue_counts: dict) -> Dict[str, Any]:
+    """Helper to build column data for the issue type bar chart as expected by Billboard"""
+    issue_types = list(issue_counts.keys())
+    open_issue_counts = list(issue_counts.values())
+
+    return {
+        "columns": [
+            ["x"] + issue_types,
+            ["To-Dos"] + open_issue_counts,
+        ],
+        "x_label": "Type of Issue",
+        "y_label": "To-Dos Open Issue Count",
+        "color": "var(--bulma-danger)",
+    }
+
+
+def format_issue_type_bar(snapshot: Any) -> Dict[str, Any]:
+    """Format open issues by type per datasource for the colored bar chart."""
+    open_issue_counts_by_datasource = defaultdict(dict)
+    total_issues_by_type = defaultdict(int)
+
+    for insight in snapshot.data_quality_issue_types.all():
+        label = format_issue_type_label(insight.issue_type)
+        open_issue_counts_by_datasource[insight.datasource_id][label] = insight.count
+        total_issues_by_type[label] += insight.count
+
+    data = {}
+    for datasource_id, counts in open_issue_counts_by_datasource.items():
+        data[datasource_id] = build_issue_type_columns(counts)
+
+    if total_issues_by_type:
+        data["global"] = build_issue_type_columns(total_issues_by_type)
+
+    return data
+
+
+def format_issue_contribution_donut(snapshot: Any) -> Dict[str, Any]:
+    """Format datasource contribution per open issue for the donut chart as expected by Billboard"""
+    datasource_counts_by_issue = defaultdict(dict)
+    total_issues_by_datasource = defaultdict(int)
+
+    for insight in snapshot.data_quality_issue_types.all():
+        label = format_issue_type_label(insight.issue_type)
+        datasource_counts_by_issue[label][insight.datasource_id] = insight.count
+        total_issues_by_datasource[insight.datasource_id] += insight.count
+
+    data = {}
+    for issue_type, counts in datasource_counts_by_issue.items():
+        columns = [[datasource_id, count] for datasource_id, count in counts.items()]
+        data[issue_type] = {"columns": columns}
+
+    if total_issues_by_datasource:
+        global_columns = [
+            [datasource_id, count] for datasource_id, count in total_issues_by_datasource.items()
+        ]
+        data["global"] = {"columns": global_columns}
+
+    return data
+
+
+# To-Dos Issue Resolution Timeline
+def data_quality_todos_resolutions_queryset():
+    """Return resolution rates by month for open and resolved to-dos."""
+    start_date = timezone.now() - timedelta(days=365)  # Collect last 12 months only
+
+    open_todos = (
+        AdvisoryToDoV2.objects.filter(
+            created_at__isnull=False,
+            created_at__gte=start_date,
+            advisories__datasource_id__isnull=False,
+        )
+        .exclude(advisories__datasource_id__in=IGNORED_IMPORTERS)
+        .annotate(month=TruncMonth("created_at"))
+        .values("month", "advisories__datasource_id")
+        .annotate(count=Count("todo_id", distinct=True))
+        .iterator()
+    )
+
+    resolved_todos = (
+        AdvisoryToDoV2.objects.filter(
+            is_resolved=True,
+            resolved_at__isnull=False,
+            resolved_at__gte=start_date,
+            advisories__datasource_id__isnull=False,
+        )
+        .exclude(advisories__datasource_id__in=IGNORED_IMPORTERS)
+        .annotate(month=TruncMonth("resolved_at"))
+        .values("month", "advisories__datasource_id")
+        .annotate(count=Count("todo_id", distinct=True))
+        .iterator()
+    )
+    return open_todos, resolved_todos
+
+
+def iter_data_quality_todos_resolutions_insights():
+    """Yield DataQualityToDosResolutionInsight objects."""
+    open_todos, resolved_todos = data_quality_todos_resolutions_queryset()
+
+    open_counts = defaultdict(int)
+    for open_record in open_todos:
+        datasource = open_record["advisories__datasource_id"]
+        month = open_record["month"].date()
+        open_counts[datasource, month] += open_record["count"]
+
+    resolved_counts = defaultdict(int)
+    for resolved_record in resolved_todos:
+        datasource = resolved_record["advisories__datasource_id"]
+        month = resolved_record["month"].date()
+        resolved_counts[datasource, month] += resolved_record["count"]
+
+    all_keys = set(open_counts.keys()) | set(
+        resolved_counts.keys()
+    )  # All unique (datasource, month) pairs across opened and resolved to-dos
+
+    for datasource_id, month_date in sorted(all_keys):
+        yield DataQualityToDosResolutionInsight(
+            datasource_id=datasource_id,
+            month=month_date,
+            open_count=open_counts[(datasource_id, month_date)],
+            resolved_count=resolved_counts[(datasource_id, month_date)],
+        )
+
+
+def collect_data_quality_todos_resolutions(pipeline: Any) -> None:
+    """Collect historical resolution rates by month."""
+    pipeline.data_quality_todos_resolutions = list(iter_data_quality_todos_resolutions_insights())
+
+
+def build_todos_resolution_columns(month_counts: Dict[str, Dict[str, int]]) -> Dict[str, Any]:
+    """Helper to build issue resolution line chart as expected by Billboard"""
+    sorted_months = sorted(month_counts.keys())
+    new_open_counts = [month_counts[month]["open"] for month in sorted_months]
+    new_resolved_counts = [month_counts[month]["resolved"] for month in sorted_months]
+
+    return {
+        "columns": [
+            ["x"] + sorted_months,
+            ["Open"] + list(accumulate(new_open_counts)),
+            ["Resolved"] + list(accumulate(new_resolved_counts)),
+        ],
+        "new_open_counts": new_open_counts,
+        "new_resolved_counts": new_resolved_counts,
+        "y_label": "Cumulative Count",
+    }
+
+
+def format_todos_resolution_timeline(snapshot: Any) -> Dict[str, Any]:
+    """Format cumulative to-dos resolution line chart"""
+    resolution_insights = snapshot.data_quality_todos_resolutions.all().order_by("month")
+
+    datasource_month_counts = defaultdict(dict)
+    global_month_counts = {}
+
+    for insight in resolution_insights:
+        formatted_month = insight.month.strftime("%Y-%m-%d")
+        datasource_id = insight.datasource_id
+
+        # Initialize global month count if not exists
+        if formatted_month not in global_month_counts:
+            global_month_counts[formatted_month] = {"open": 0, "resolved": 0}
+
+        global_month_counts[formatted_month]["open"] += insight.open_count
+        global_month_counts[formatted_month]["resolved"] += insight.resolved_count
+
+        datasource_month_counts[datasource_id][formatted_month] = {
+            "open": insight.open_count,
+            "resolved": insight.resolved_count,
+        }
+
+    data = {}
+    if global_month_counts:
+        data["global"] = build_todos_resolution_columns(global_month_counts)
+        for datasource_id, month_counts in datasource_month_counts.items():
+            data[datasource_id] = build_todos_resolution_columns(month_counts)
+
+    return data