ai-agent-admin/backend-fastapi/online_dev/report_manager/engine/chart_data.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""报表图表数据聚合（对齐 JNPF ChartUtil）"""
from __future__ import annotations

from collections import defaultdict
from decimal import Decimal
from typing import Any, Dict, List, Optional, Set, Tuple


def _parse_field(field: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
    """alias.field -> (alias, field_name)"""
    if not field or not isinstance(field, str):
        return None, None
    parts = field.split(".", 1)
    if len(parts) == 2:
        return parts[0], parts[1]
    return None, field


def _collect_rows(datasets: Dict[str, List[Dict[str, Any]]], dataset_names: Set[str]) -> List[Dict[str, Any]]:
    rows: List[Dict[str, Any]] = []
    for name in dataset_names:
        rows.extend(datasets.get(name) or [])
    return rows


def _aggregate(values: List[Any], summary_type: str) -> str:
    if not values:
        return ""
    st = (summary_type or "none").lower()
    nums: List[Decimal] = []
    for v in values:
        try:
            nums.append(Decimal(str(v)))
        except Exception:
            pass
    if st == "sum" and nums:
        return str(sum(nums))
    if st == "avg" and nums:
        return str(sum(nums) / len(nums))
    if st == "max" and nums:
        return str(max(nums))
    if st == "min" and nums:
        return str(min(nums))
    if st == "count":
        return str(len(values))
    return str(values[-1]) if values else ""


def _build_field(
    data_list: List[Dict[str, Any]],
    classify_key: Optional[str],
    series_name_key: Optional[str],
    series_data_key: Optional[str],
    max_key: Optional[str],
    summary_type: str,
) -> Dict[str, Any]:
    chart_map: Dict[Any, Dict[Any, List[Any]]] = defaultdict(lambda: defaultdict(list))
    max_map: Dict[Any, List[Any]] = defaultdict(list)

    for row in data_list:
        if not classify_key:
            continue
        classify = row.get(classify_key)
        if classify is None:
            continue
        value = row.get(series_data_key) if series_data_key else None
        if value is None:
            continue
        series = row.get(series_name_key) if series_name_key else ""
        chart_map[series][classify].append(value)
        if max_key:
            mx = row.get(max_key)
            if mx is not None:
                max_map[classify].append(mx)

    series_name_list: List[str] = []
    classify_map: Dict[Any, List[List[str]]] = defaultdict(list)
    max_counts = [0]

    for series, classify_name_map in chart_map.items():
        series_name_list.append(str(series))
        for classify, value_list in classify_name_map.items():
            agg = _aggregate(value_list, summary_type)
            classify_map[classify].append([agg])
            max_counts.append(len(classify_map[classify]))

    classify_name_list = sorted(str(k) for k in classify_map.keys())
    max_field_list: List[str] = []
    for classify in classify_name_list:
        objects = max_map.get(classify) or [0]
        max_field_list.append(_aggregate(objects, "max"))

    max_len = max(max_counts) if max_counts else 0
    series_data_list: List[List[str]] = []
    for i in range(max_len):
        row_data: List[str] = []
        for category in classify_name_list:
            category_list = classify_map.get(category) or []
            category_data = category_list[i] if i < len(category_list) else []
            row_data.append(category_data[0] if category_data else "")
        series_data_list.append(row_data)

    result: Dict[str, Any] = {
        "classifyNameField": classify_name_list,
        "seriesDataField": series_data_list,
    }
    if series_name_key:
        result["seriesNameField"] = series_name_list
    if max_key:
        result["maxField"] = max_field_list
    return result


def _echart_configs_from_cells(cells: Dict[str, Any]) -> List[Dict[str, Any]]:
    configs: List[Dict[str, Any]] = []
    if not cells:
        return configs
    for key, store in (
        ("floatEcharts", cells.get("floatEcharts")),
        ("cellEcharts", cells.get("cellEcharts")),
    ):
        if not isinstance(store, dict):
            continue
        for drawing_id, item in store.items():
            if not isinstance(item, dict):
                continue
            option = item.get("option") or {}
            configs.append(
                {
                    "drawingId": item.get("drawingId") or drawing_id,
                    "option": option,
                    "source": key,
                }
            )
    return configs


def build_chart_data(
    cells: Dict[str, Any],
    datasets: Dict[str, List[Dict[str, Any]]],
) -> List[Dict[str, Any]]:
    """
    生成预览用 chartData 列表。
    每项: { drawingId, field: { classifyNameField, seriesNameField, seriesDataField, maxField? } }
    """
    result: List[Dict[str, Any]] = []
    for cfg in _echart_configs_from_cells(cells):
        drawing_id = cfg.get("drawingId")
        option = cfg.get("option") or {}
        dataset_names: Set[str] = set()
        classify_alias, classify_field = _parse_field(option.get("classifyNameField"))
        series_alias, series_name_field = _parse_field(option.get("seriesNameField"))
        data_alias, series_data_field = _parse_field(option.get("seriesDataField"))
        max_alias, max_field = _parse_field(option.get("maxField"))
        for alias in (classify_alias, series_alias, data_alias, max_alias):
            if alias:
                dataset_names.add(alias)
        if not dataset_names:
            continue
        rows = _collect_rows(datasets, dataset_names)
        field = _build_field(
            rows,
            classify_field,
            series_name_field,
            series_data_field,
            max_field,
            option.get("summaryType") or "none",
        )
        result.append({"drawingId": drawing_id, "field": field})
    return result