Task: workarena.servicenow.single-chart-value-retrieval
Goal: In the Computer count by Class report, what is the value of "UNIX Server" in the chart (in percent)?
Task configuration: url: /now/nav/ui/classic/params/target/sys_report_template.do%3Fjvar_report_id=00d5dd60931f7a106162fb67dd03d696
question: value; percent; UNIX Server

Validation code:
import json
import math
import os
import re
from typing import Any, Dict, List, Optional
from urllib.parse import parse_qsl, unquote, urlparse

import requests

def _multi_unquote(s: str, rounds: int = 3) -> str:
    for _ in range(rounds):
        new = unquote(s)
        if new == s:
            break
        s = new
    return s


def _api_get(endpoint: str, **kwargs) -> requests.Response:
    return requests.get(
        os.getenv("SERVICENOW_INSTANCE_URL", "").rstrip("/") + endpoint,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            **json.loads(os.getenv("SERVICENOW_EXTRA_HTTP_HEADERS", "{}")),
        },
        **kwargs,
    )


def extract_dashboard_sys_id(url: str) -> str | None:
    raw = _multi_unquote(url)
    parsed = urlparse(raw)
    if parsed.query:
        val = dict(parse_qsl(parsed.query)).get("sysparm_dashboard")
        if val:
            return val


def get_widget(dashboard_sys_id: str, title: str) -> Optional[Dict[str, Any]]:
    # 1) dashboard -> tabs (m2m)
    m2m = _api_get(
        "/api/now/table/pa_m2m_dashboard_tabs",
        params={
            "sysparm_query": f"dashboard={dashboard_sys_id}",
            "sysparm_fields": "tab",
            "sysparm_limit": "1000",
        },
    ).json()["result"]
    tab_ids = [
        row["tab"]["value"] if isinstance(row["tab"], dict) else row["tab"]
        for row in m2m
    ]
    for tab_id in tab_ids:
        # 2) tab -> canvas_page
        tab = _api_get(
            f"/api/now/table/pa_tabs/{tab_id}",
            params={"sysparm_fields": "name,canvas_page"},
        ).json()["result"]
        canvas_id = (
            tab["canvas_page"]["value"]
            if isinstance(tab.get("canvas_page"), dict)
            else tab.get("canvas_page")
        )

        # 3) canvas_page -> panes -> portal widgets
        panes = _api_get(
            "/api/now/table/sys_grid_canvas_pane",
            params={
                "sysparm_query": f"grid_canvas={canvas_id}",
                "sysparm_fields": "portal_widget",
                "sysparm_limit": "1000",
                "sysparm_display_value": "all",
            },
        ).json()["result"]
        portal_widget_ids = [
            (
                p["portal_widget"]["value"]
                if isinstance(p["portal_widget"], dict)
                else p["portal_widget"]
            )
            for p in panes
            if p.get("portal_widget")
        ]
        if not portal_widget_ids:
            continue
        all_pw_id = ",".join(portal_widget_ids)

        prefs = _api_get(
            "/api/now/table/sys_portal_preferences",
            params={"sysparm_query": f"portal_sectionIN{all_pw_id}^name=sys_id"},
        ).json()["result"]
        prefs_sys_ids = [p["value"] for p in prefs]
        if not prefs_sys_ids:
            continue
        all_prefs_sys_id = ",".join(prefs_sys_ids)

        # Try PA widgets first
        pa = _api_get(
            "/api/now/table/pa_widgets",
            params={
                "sysparm_query": f"sys_idIN{all_prefs_sys_id}",
                "sysparm_limit": "1000",
                "sysparm_fields": "sys_id,name,indicator,visualization",
            },
        ).json()["result"]
        filtered_pa = [p for p in pa if p.get("name") == title]
        if filtered_pa:
            w = filtered_pa[0]
            return {
                "type": "pa_widget",
                "sys_id": w["sys_id"],
                "name": w.get("name"),
                "indicator": (
                    w["indicator"]["value"]
                    if isinstance(w.get("indicator"), dict)
                    else w.get("indicator")
                ),
                "visualization": w.get("visualization"),
            }

        # Try Reports
        rpt = _api_get(
            "/api/now/table/sys_report",
            params={
                "sysparm_query": f"sys_idIN{all_prefs_sys_id}",
                "sysparm_limit": "1000",
                "sysparm_fields": "sys_id,title,type,table,field,trend_field,filter,active",
                "sysparm_display_value": "true",
            },
        ).json()["result"]
        filtered_rpt = [r for r in rpt if r.get("title") == title]
        if filtered_rpt:
            r = filtered_rpt[0]
            return {
                "type": "report",
                "sys_id": r["sys_id"],
                "title": r.get("title"),
                "report_type": r.get("type"),
                "table": r.get("table"),
                "field_name": r.get("field"),
                "trend_field": r.get("trend_field"),
                "filter": r.get("filter"),
                "active": r.get("active"),
            }

    return None


def extract_report_sys_id(url: str) -> str | None:
    raw = _multi_unquote(url)
    parsed = urlparse(raw)
    if parsed.query:
        val = dict(parse_qsl(parsed.query)).get("jvar_report_id")
        if val:
            return val


def get_report(report_sys_id: str) -> Dict[str, Any]:
    r = _api_get(
        f"/api/now/table/sys_report/{report_sys_id}",
        params={
            "sysparm_fields": "sys_id,title,type,table,field,trend_field,filter,active",
            "sysparm_display_value": "true",
        },
    ).json()["result"]
    return {
        "type": "report",
        "sys_id": r["sys_id"],
        "title": r.get("title"),
        "report_type": r.get("type"),
        "table": r.get("table"),
        "field_name": r.get("field"),
        "trend_field": r.get("trend_field"),
        "filter": r.get("filter"),
        "active": r.get("active"),
    }


def get_chart_values(chart: Dict[str, Any]) -> List[Dict[str, Any]]:
    query_filter = chart.get("filter", "")
    results = _api_get(
        f"/api/now/stats/{chart['table']}",
        params={
            "sysparm_query": query_filter,
            "sysparm_group_by": chart["field_name"],
            "sysparm_count": "true",
            "sysparm_display_value": "true",
        },
    ).json()["result"]
    return results


def get_trend_values(chart: Dict[str, Any], interval: str = "month") -> List[Dict[str, Any]]:
    """Query table records and bucket by trend_field into time periods."""
    from collections import Counter
    from datetime import datetime, timedelta

    query_filter = chart.get("filter", "")
    trend_field = chart["field_name"]  # already swapped to trend_field by caller

    # Fetch all matching records, only the trend field
    offset = 0
    limit = 1000
    all_timestamps = []
    while True:
        records = _api_get(
            f"/api/now/table/{chart['table']}",
            params={
                "sysparm_query": query_filter,
                "sysparm_fields": trend_field,
                "sysparm_limit": str(limit),
                "sysparm_offset": str(offset),
            },
        ).json()["result"]
        for rec in records:
            ts = rec.get(trend_field, "")
            if ts:
                all_timestamps.append(ts)
        if len(records) < limit:
            break
        offset += limit

    # Bucket timestamps by interval
    buckets: Counter = Counter()
    for ts in all_timestamps:
        try:
            dt = datetime.strptime(ts[:19], "%Y-%m-%d %H:%M:%S")
        except ValueError:
            continue

        if interval == "month":
            label = dt.strftime("%b/%Y")
        elif interval == "week":
            monday = dt - timedelta(days=dt.weekday())
            label = monday.strftime("%Y-%m-%d")
        elif interval == "year":
            label = dt.strftime("%Y")
        elif interval == "day":
            label = dt.strftime("%Y-%m-%d")
        else:
            label = dt.strftime("%b/%Y")

        buckets[label] += 1

    total = sum(buckets.values())
    points = []
    for label, count in buckets.items():
        percent = round(count / total * 100, 4) if total > 0 else 0
        points.append({"label": label, "count": count, "percent": percent})
    return points


def _stats_to_chart_data(stats_results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform ServiceNow stats API results to chart_data format."""
    total = sum(int(r["stats"]["count"]) for r in stats_results)
    points = []
    for r in stats_results:
        count = int(r["stats"]["count"])
        gf = r.get("groupby_fields", [])
        label = gf[0].get("value", "") if gf else ""
        percent = round(count / total * 100, 4) if total > 0 else 0
        points.append({"label": label, "count": count, "percent": percent})
    return points


def _parse_trend_interval(filter_str: str) -> str:
    """Extract the trend interval from a TRENDBY directive in the filter string."""
    match = re.search(r"\^TRENDBY\w+,(\w+)", filter_str)
    if match:
        return match.group(1)
    return "month"



def analyze_chart(
    chart_data: List[Dict[str, Any]],
    question: str,
    chart_series: str,
    response: str,
) -> int:
    """Analyze chart data against the assistant's numeric answer.

    chart_data can be:
      - flat list of points: [{"label": str, "count": num, "percent": num}, ...]
      - multi-series: [{"name": str, "data": [point, ...]}, ...]
    """
    if not chart_data or not isinstance(chart_data, list):
        return 0

    if not response:
        return 0

    # Resolve multi-series to flat list of points
    if isinstance(chart_data[0], dict) and "data" in chart_data[0]:
        if len(chart_data) == 1:
            points = chart_data[0]["data"]
        else:
            if not chart_series:
                return 0
            matches = [s for s in chart_data if s.get("name") == chart_series]
            if not matches:
                return 0
            points = matches[0].get("data", [])
    else:
        points = chart_data

    if not points:
        return 0

    # Remove labels to avoid counting their numbers, then extract floats
    labels = {p.get("label", "") for p in points}
    scrubbed = str(response)
    for label in labels:
        if label:
            scrubbed = scrubbed.replace(label, "")
    # extract numbers like 1,234 or 12.5 or 12,5 (we normalize commas away)
    nums = re.findall(r"[\d]+(?:[.,]\d+)?", scrubbed.replace(",", ""))
    try:
        response_vals = sorted({float(x) for x in nums})
    except ValueError:
        response_vals = []
    if not response_vals:
        return 0

    def get_count_list():
        return [float(p.get("count", 0.0)) for p in points]

    def get_expected_value_for(label: str, fmt: str) -> float | None:
        for p in points:
            if p.get("label") == label:
                if fmt == "count":
                    return float(p.get("count", 0.0))
                return float(p.get("percent", 0.0))
        return None

    def close(a: float, b: float) -> bool:
        return math.isclose(a, b, rel_tol=1e-3, abs_tol=1e-6)

    q = question.strip().lower()

    # "value; format; label"
    if q.startswith("value"):
        parts = [s.strip() for s in question.split(";")]
        if len(parts) < 3:
            return 0
        fmt, label = parts[1], parts[2]
        expected = get_expected_value_for(label, fmt)
        if expected is None:
            return 0
        return 1 if any(close(expected, v) for v in response_vals) else 0

    # max/min
    if "max" in q or "min" in q:
        want_max = "max" == q
        counts = get_count_list()
        if not counts:
            return 0
        target = (max if want_max else min)(counts)
        resp_lower = response.lower()
        for p in points:
            if float(p.get("count", 0.0)) == target:
                # Normalize label: replace special chars with space, check each token
                label = re.sub(r"[^a-zA-Z0-9]", " ", p.get("label", "")).lower()
                tokens = [t for t in label.split() if t]
                label_match = tokens and all(t in resp_lower for t in tokens)
                if label_match and any(close(target, v) for v in response_vals):
                    return 1
        return 0

    # mean/median/mode
    if any(k in q for k in ("mean", "median", "mode")):
        counts = get_count_list()
        if not counts:
            return 0

        if "mean" in q:
            target = sum(counts) / len(counts)
        elif "median" in q:
            s = sorted(counts)
            mid = len(s) // 2
            target = s[mid] if len(s) % 2 else (s[mid - 1] + s[mid]) / 2.0
        else:  # "mode"
            freq: Dict[float, int] = {}
            for v in counts:
                freq[v] = freq.get(v, 0) + 1
            max_freq = max(freq.values())
            modes = sorted([v for v, c in freq.items() if c == max_freq])
            target = modes[0]

        if len(response_vals) != 1:
            return 0
        return 1 if close(target, response_vals[0]) else 0

    return 0


def validate(messages: List[Dict[str, str]], config: Dict[str, Any]) -> int:
    """
    Validate an assistant's numeric answer against chart data fetched from
    a ServiceNow dashboard or report.

    Expected config keys:
      - url: str           # dashboard or report URL
      - chart_title: str   # title of the chart on the dashboard
      - chart_series: str  # series name (empty string if single-series)
      - question: str      # e.g., "value; count; Label", "max", "min", "mean", "median", "mode"
    """
    if (
        "url" not in config
        or "chart_title" not in config
        or "chart_series" not in config
        or "question" not in config
    ):
        raise ValueError(
            "Target must contain 'url', 'chart_title', 'chart_series' and 'question'"
        )

    if not messages:
        raise ValueError("Messages must not be empty")

    # Extract last assistant message
    response = None
    for msg in reversed(messages):
        if msg.get("role") == "assistant":
            response = msg.get("content") or ""
            break
    if not response:
        return 0

    # Fetch chart data from API
    url = config["url"]
    chart_title = config["chart_title"]

    if "pa_dashboard.do" in url:
        dashboard_sys_id = extract_dashboard_sys_id(url)
        if not dashboard_sys_id:
            raise ValueError(f"Could not extract dashboard sys_id from URL: {url}")
        widget = get_widget(dashboard_sys_id, title=chart_title)
        if widget is None:
            raise ValueError(
                f"Chart '{chart_title}' not found on dashboard {dashboard_sys_id}"
            )
        chart_series = config["chart_series"]
        if chart_series and widget.get("trend_field"):
            widget_for_query = dict(widget)
            series_filter = f"{widget['field_name']}={chart_series}"
            existing_filter = widget.get("filter", "")
            if existing_filter and not existing_filter.startswith("^"):
                series_filter += "^"
            widget_for_query["filter"] = series_filter + existing_filter
            widget_for_query["field_name"] = widget["trend_field"]
            interval = _parse_trend_interval(widget.get("filter", ""))
            points = get_trend_values(widget_for_query, interval)
        elif not chart_series and widget.get("trend_field"):
            widget_for_query = dict(widget)
            widget_for_query["field_name"] = widget["trend_field"]
            interval = _parse_trend_interval(widget.get("filter", ""))
            points = get_trend_values(widget_for_query, interval)
        else:
            chart_values = get_chart_values(widget)
            points = _stats_to_chart_data(chart_values)
    elif "sys_report_template.do" in url:
        report_sys_id = extract_report_sys_id(url)
        if not report_sys_id:
            raise ValueError(f"Could not extract report sys_id from URL: {url}")
        report = get_report(report_sys_id)
        chart_series = config["chart_series"]
        if chart_series and report.get("trend_field"):
            # Trend chart with a specific series: filter by series, group by time
            report_for_query = dict(report)
            series_filter = f"{report['field_name']}={chart_series}"
            existing_filter = report.get("filter", "")
            if existing_filter and not existing_filter.startswith("^"):
                series_filter += "^"
            report_for_query["filter"] = series_filter + existing_filter
            report_for_query["field_name"] = report["trend_field"]
            interval = _parse_trend_interval(report.get("filter", ""))
            points = get_trend_values(report_for_query, interval)
        elif not chart_series and report.get("trend_field"):
            report_for_query = dict(report)
            report_for_query["field_name"] = report["trend_field"]
            interval = _parse_trend_interval(report.get("filter", ""))
            points = get_trend_values(report_for_query, interval)
        else:
            chart_values = get_chart_values(report)
            points = _stats_to_chart_data(chart_values)
    else:
        raise ValueError(f"Unsupported URL format: {url}")

    return analyze_chart(
        chart_data=points,
        question=config["question"],
        chart_series=config["chart_series"],
        response=response,
    )