busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

js - ts + fix

Browse Source
This commit is contained in:
busya
2026-06-01 14:40:17 +03:00
parent 1a7f368324
commit b4b0deb856
19 changed files with 1337 additions and 473 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
backend/src/plugins/llm_analysis/_topology.py Normal file
View File

@@ -0,0 +1,290 @@
# #region DashboardTopology [C:3] [TYPE Module] [SEMANTICS superset, topology, dashboard, layout, parsing]
# @BRIEF Dashboard topology extraction — parses Superset position_json into human-readable
# text descriptions of Tab → Row → Chart hierarchy for LLM prompt context.
# @LAYER Plugin
# @RELATION DEPENDS_ON -> [json]
# @INVARIANT All functions are pure (no DB, no I/O beyond json parsing).
# @DATA_CONTRACT Input: Dashboard position_json + Charts → Output: Topology text.
import json
from typing import Any
# #region _chart_name [C:1] [TYPE Function]
# @BRIEF Return the display name for a chart.
def _chart_name(chart: dict) -> str:
return chart.get("slice_name", chart.get("title", f"Chart {chart.get('id')}"))
# #endregion _chart_name
# #region _chart_params [C:1] [TYPE Function]
# @BRIEF Return parsed chart params dict.
def _chart_params(chart: dict) -> dict:
params = chart.get("params", {})
if isinstance(params, str):
try:
return json.loads(params)
except (json.JSONDecodeError, TypeError):
return {}
return params
# #endregion _chart_params
# #region _describe_chart_dataset [C:2] [TYPE Function]
# @BRIEF Return dataset description line if available.
# @PRE chart and params are chart dicts.
# @POST Returns string with dataset line or empty string.
def _describe_chart_dataset(chart: dict, params: dict) -> str:
ds_id = chart.get("datasource_id") or params.get("datasource_id")
if ds_id:
return f"\n Dataset: {ds_id}"
return ""
# #endregion _describe_chart_dataset
# #region _describe_chart_metrics [C:2] [TYPE Function]
# @BRIEF Return metrics description line if available.
# @PRE params is a parsed chart params dict.
# @POST Returns string with metrics line or empty string.
def _describe_chart_metrics(params: dict) -> str:
metrics = params.get("metrics", [])
if not metrics:
return ""
metric_names: list[str] = []
for m in metrics:
if isinstance(m, dict):
metric_names.append(m.get("label", m.get("expression", str(m))))
else:
metric_names.append(str(m))
if metric_names:
return f"\n Metrics: {', '.join(metric_names[:3])}"
return ""
# #endregion _describe_chart_metrics
# #region _describe_chart_groupby [C:2] [TYPE Function]
# @BRIEF Return group-by description line if available.
# @PRE params is a parsed chart params dict.
# @POST Returns string with group-by line or empty string.
def _describe_chart_groupby(params: dict) -> str:
groupby = params.get("groupby", [])
if groupby:
return f"\n Group by: {', '.join(groupby[:3])}"
return ""
# #endregion _describe_chart_groupby
# #region _describe_chart_filters [C:2] [TYPE Function]
# @BRIEF Return adhoc filters description line if available.
# @PRE params is a parsed chart params dict.
# @POST Returns string with filters line or empty string.
def _describe_chart_filters(params: dict) -> str:
adhoc_filters = params.get("adhoc_filters", [])
if not adhoc_filters:
return ""
filter_parts: list[str] = []
for filt in adhoc_filters[:3]:
if isinstance(filt, dict):
subject = filt.get("subject", "")
operator = filt.get("operator", "")
comparator = filt.get("comparator", "")
if subject and comparator:
filter_parts.append(f"{subject} {operator} {comparator}")
else:
filter_parts.append(filt.get("clause", "where"))
if filter_parts:
return f"\n Filters: {', '.join(filter_parts)}"
return ""
# #endregion _describe_chart_filters
# #region _describe_chart [C:2] [TYPE Function]
# @BRIEF Return a multi-line description of a single chart.
# @PRE chart is a chart detail dict with slice_name, viz_type, params.
# @POST Returns formatted chart description string.
def _describe_chart(chart: dict) -> str:
name = _chart_name(chart)
viz_type = chart.get("viz_type", "unknown")
params = _chart_params(chart)
desc = f' Chart "{name}" ({viz_type})'
desc += _describe_chart_dataset(chart, params)
desc += _describe_chart_metrics(params)
desc += _describe_chart_groupby(params)
desc += _describe_chart_filters(params)
return desc
# #endregion _describe_chart
# #region _sort_chart_key [C:1] [TYPE Function]
# @BRIEF Sort key for chart lookup items: numeric IDs first, then string keys.
def _sort_chart_key(item):
k, _ = item
try:
return (0, int(k))
except (ValueError, TypeError):
return (1, str(k))
# #endregion _sort_chart_key
# #region _resolve_child [C:2] [TYPE Function]
# @BRIEF Resolve a child reference which may be a string key or inline dict.
# @PRE ref is a child reference from position_json, pos_json is the full layout.
# @POST Returns the resolved child dict or None.
def _resolve_child(ref, pos_json: dict) -> dict | None:
if isinstance(ref, dict):
return ref
if isinstance(ref, str) and ref in pos_json:
return pos_json[ref]
return None
# #endregion _resolve_child
# #region _process_tab [C:2] [TYPE Function]
# @BRIEF Process a TAB type position_json node.
# @POST Returns indented lines for the tab and its children.
def _process_tab(
child: dict, depth: int, pos_json: dict, chart_lookup: dict[str, dict]
) -> list[str]:
meta = child.get("meta", {}) or {}
tab_name = meta.get("text", meta.get("label", "Tab"))
sub_lines = _process_children(child, depth + 1, pos_json, chart_lookup)
chart_count = sum(1 for line in sub_lines if line.strip().startswith("Chart"))
indent = " " * depth
return [f'{indent}─── Tab: "{tab_name}" ({chart_count} charts) ───', *sub_lines]
# #endregion _process_tab
# #region _process_chart [C:2] [TYPE Function]
# @BRIEF Process a CHART type position_json node.
# @POST Returns indented line for the chart description or empty list.
def _process_chart(
child: dict, depth: int, chart_lookup: dict[str, dict]
) -> list[str]:
meta = child.get("meta", {}) or {}
chart_id = meta.get("chartId") or meta.get("slice_id")
if chart_id is not None and str(chart_id) in chart_lookup:
indent = " " * depth
return [f"{indent}{_describe_chart(chart_lookup[str(chart_id)])}"]
return []
# #endregion _process_chart
# #region _process_children [C:2] [TYPE Function]
# @BRIEF Recursively process position_json children building indented lines.
# @PRE parent is a position_json node dict, pos_json is the full layout, chart_lookup maps IDs.
# @POST Returns list of formatted topology lines.
def _process_children(
parent: dict, depth: int, pos_json: dict, chart_lookup: dict[str, dict]
) -> list[str]:
result: list[str] = []
children = parent.get("children", []) if isinstance(parent, dict) else []
for child_ref in children:
child = _resolve_child(child_ref, pos_json)
if not child or not isinstance(child, dict):
continue
child_type = child.get("type", "")
if child_type == "TAB":
result.extend(_process_tab(child, depth, pos_json, chart_lookup))
elif child_type == "ROW":
result.extend(_process_children(child, depth, pos_json, chart_lookup))
elif child_type == "CHART":
result.extend(_process_chart(child, depth, chart_lookup))
return result
# #endregion _process_children
# #region _extract_topology_children [C:2] [TYPE Function]
# @BRIEF Extract formatted topology lines from position_json.
# @PRE position_json is a parsed layout dict, chart_lookup maps chart IDs to chart dicts.
# @POST Returns list of formatted topology description lines.
def _extract_topology_children(
position_json: dict, chart_lookup: dict[str, dict]
) -> list[str]:
lines: list[str] = []
grid_nodes = [
v
for v in position_json.values()
if isinstance(v, dict) and v.get("type") in ("GRID", "ROOT", "CONTAINER")
]
if grid_nodes:
for grid in grid_nodes:
lines.extend(_process_children(grid, depth=1, pos_json=position_json, chart_lookup=chart_lookup))
else:
for key, value in position_json.items():
if (
isinstance(value, dict)
and value.get("type") in ("TAB", "ROW", "CHART")
and key != "DASHBOARD_VERSION_KEY"
):
lines.extend(_process_children({"children": [key]}, depth=0, pos_json=position_json, chart_lookup=chart_lookup))
return lines
# #endregion _extract_topology_children
# #region build_dashboard_topology [C:2] [TYPE Function]
# @BRIEF Build a structured text description of dashboard layout from position_json.
# @PARAM dashboard (dict) - Dashboard metadata dict with position_json, dashboard_title.
# @PARAM charts (list[dict]) - List of fetched chart detail dicts for name/viz lookup.
# @POST Returns a human-readable text description of the dashboard topology.
# @RELATION DEPENDS_ON -> [_extract_topology_children]
# @RELATION DEPENDS_ON -> [_sort_chart_key]
def build_dashboard_topology(dashboard: dict, charts: list[dict]) -> str:
"""
Build a structured text description of dashboard layout from position_json.
Returns a human-readable text like:
Dashboard: "Sales Overview"
--- Tab: "Revenue" (3 charts) ---
Chart "Total Revenue YTD" (big_number_total)
Dataset: 5
Metrics: SUM(revenue)
...
--- Tab: "Pipeline" (2 charts) ---
...
"""
lines: list[str] = []
title = dashboard.get(
"dashboard_title",
dashboard.get("title", f"Dashboard {dashboard.get('id', '')}"),
)
lines.append(f'Dashboard: "{title}"')
position_json = dashboard.get("position_json", {})
if isinstance(position_json, str):
try:
position_json = json.loads(position_json)
except (json.JSONDecodeError, TypeError):
position_json = {}
if not isinstance(position_json, dict):
position_json = {}
chart_lookup: dict[str, dict] = {}
for c in charts:
cid = c.get("id") or c.get("slice_id")
if cid is not None:
chart_lookup[str(cid)] = c
try:
children = _extract_topology_children(position_json, chart_lookup)
lines.extend(children)
except Exception as e:
lines.append(f"\n[Topology parse warning: {e}]")
if chart_lookup:
lines.append(f"\nCharts: {len(chart_lookup)} total")
for chart_id, c in sorted(chart_lookup.items(), key=_sort_chart_key):
lines.append(
f" - {c.get('slice_name', f'chart_{chart_id}')} "
f"({c.get('viz_type', 'unknown')})"
)
return "\n".join(lines)
# #endregion build_dashboard_topology
# #endregion DashboardTopology

408
backend/src/plugins/llm_analysis/plugin.py
View File

@@ -14,6 +14,7 @@
from datetime import UTC, datetime, timedelta
import json
import os
import re
from typing import Any
from ...core.async_superset_client import AsyncSupersetClient # noqa: F401 — available for async client wiring
@@ -22,7 +23,7 @@ from ...core.logger import belief_scope, logger
from ...core.plugin_base import PluginBase
from ...core.superset_client import SupersetClient
from ...core.task_manager.context import TaskContext
from ...models.llm import ValidationPolicy, ValidationRecord
from ...models.llm import ValidationPolicy, ValidationRecord, ValidationRun
from ...services.llm_prompt_templates import (
DEFAULT_LLM_PROMPTS,
normalize_llm_settings,
@@ -31,6 +32,7 @@ from ...services.llm_prompt_templates import (
from ...services.llm_provider import LLMProviderService
from ...services.notifications.service import NotificationService
from .models import DetectedIssue, LLMProviderType, ValidationResult, ValidationStatus
from ._topology import build_dashboard_topology
from .service import DatasetHealthChecker, LLMClient, ScreenshotService
@@ -82,7 +84,8 @@ JSON_FORMAT_INSTRUCTION = (
' {\n'
' "severity": "WARN" | "FAIL",\n'
' "message": "Description of the issue",\n'
' "location": "Optional location info"\n'
' "location": "Where the issue is (tab name, chart name, or screen region)",\n'
' "tab_index": "0-based index of the tab screenshot where the issue appears (-1 if unknown or not tab-specific)"\n'
' }\n'
' ]\n'
'}'
@@ -103,34 +106,55 @@ def _ensure_json_prompt(prompt: str | None) -> str:
# #endregion _ensure_json_prompt
# #region _update_run_status [TYPE Function]
# #region _update_run_status [C:3] [TYPE Function]
# @BRIEF Aggregate ValidationRecord statuses into the ValidationRun.
# Only marks run as finished when ALL dashboards have a record.
# Marks run as completed when ALL expected dashboards have a record.
# Computes pass/warn/fail/unknown counters from individual record statuses.
# @RATIONALE Before the fix, this function used `from ..models.llm import ValidationRun`
# which resolved to src.plugins.models.llm (doesn't exist) — the ModuleNotFoundError
# was silently swallowed by `except Exception: pass`, causing the run to stay "running"
# indefinitely. Fixed by importing ValidationRun at module level.
# @REJECTED Leaving the run in "running" status rejected — blocks future runs for the same
# policy and shows an incomplete state in the UI. Setting run.status = worst record status
# rejected — the run lifecycle status must be "completed"/"partial"/"failed", not "FAIL".
# @PRE db session is active, run_id may be None.
# @POST If run_id is set and all expected dashboards are recorded,
# ValidationRun.status is updated to the worst record status.
# ValidationRun.status is set to 'completed', counters are computed.
# duration_ms becomes available via finished_at.
def _update_run_status(db, run_id: str | None) -> None:
if not run_id:
return
from ..models.llm import ValidationRecord as VR, ValidationRun as VRun
try:
records = db.query(VR).filter(VR.run_id == run_id).all()
run = db.query(VRun).filter(VRun.id == run_id).first()
if not records or not run:
return
# Only mark complete when ALL expected dashboards have a record
total_expected = run.dashboard_count or 0
if total_expected > 0 and len(records) < total_expected:
return
# Determine worst status: FAIL > WARN > UNKNOWN > PASS
priority = {"FAIL": 0, "WARN": 1, "UNKNOWN": 2, "PASS": 3}
worst = min(records, key=lambda r: priority.get(r.status, 99))
run.status = worst.status
with belief_scope("_update_run_status", f"run_id={run_id}"):
from datetime import datetime, timezone
run.finished_at = datetime.now(timezone.utc)
db.commit()
except Exception:
pass
try:
records = db.query(ValidationRecord).filter(ValidationRecord.run_id == run_id).all()
run = db.query(ValidationRun).filter(ValidationRun.id == run_id).first()
if not records or not run:
logger.warning(f"[_update_run_status] Run {run_id} not found or has no records")
return
total_expected = run.dashboard_count or 0
if total_expected > 0 and len(records) < total_expected:
logger.info(
f"[_update_run_status] Run {run_id}: {len(records)}/{total_expected} records, not yet complete"
)
return
pass_count = sum(1 for r in records if r.status == 'PASS')
warn_count = sum(1 for r in records if r.status == 'WARN')
fail_count = sum(1 for r in records if r.status == 'FAIL')
unknown_count = sum(1 for r in records if r.status == 'UNKNOWN')
run.status = 'completed'
run.finished_at = datetime.now(timezone.utc)
run.pass_count = pass_count
run.warn_count = warn_count
run.fail_count = fail_count
run.unknown_count = unknown_count
db.commit()
logger.info(
f"[_update_run_status] Run {run_id} completed: "
f"PASS={pass_count} WARN={warn_count} FAIL={fail_count} UNKNOWN={unknown_count}"
)
except Exception as e:
logger.error(f"[_update_run_status] Failed to finalize run {run_id}: {e}")
# #endregion _update_run_status
@@ -186,9 +210,14 @@ class DashboardValidationPlugin(PluginBase):
# @PARAM context (Optional[TaskContext]) - Task context for logging with source attribution.
# @PRE params contains dashboard_id, environment_id, and provider_id.
# @POST Returns a dictionary with validation results and persists them to the database.
# Guarantees _update_run_status is called even if individual dashboard processing fails.
# @SIDE_EFFECT When screenshot_enabled=True: captures a screenshot, calls multimodal LLM API.
# When screenshot_enabled=False: fetches chart/dataset metadata, calls text-only LLM API.
# Both paths write to the database and dispatch notifications.
# @REJECTED Early exit from the dashboard loop on exception rejected — previously an error
# in one dashboard caused _update_run_status to never be called, leaving the run stuck
# in "running" status forever. Now each dashboard is wrapped in try/except and the
# loop always proceeds to _update_run_status.
async def execute(self, params: dict[str, Any], context: TaskContext | None = None):
with belief_scope("execute", f"plugin_id={self.id}"):
log = context.logger if context else logger
@@ -259,6 +288,7 @@ class DashboardValidationPlugin(PluginBase):
# Process each dashboard sequentially, collecting results
results: list[dict[str, Any]] = []
last_error: BaseException | None = None
for idx, dashboard_id in enumerate(all_dashboard_ids):
log.info(f"Processing dashboard {idx + 1}/{len(all_dashboard_ids)}: {dashboard_id}")
params["dashboard_id"] = dashboard_id
@@ -292,18 +322,26 @@ class DashboardValidationPlugin(PluginBase):
params["source_invalid"] = True
params["source_error"] = str(e)
if screenshot_enabled:
result = await self._execute_path_a(
params, context, db, db_provider, api_key, env, config_mgr, log
)
else:
result = await self._execute_path_b(
params, context, db, db_provider, api_key, env, config_mgr, log
)
results.append(result)
try:
if screenshot_enabled:
result = await self._execute_path_a(
params, context, db, db_provider, api_key, env, config_mgr, log
)
else:
result = await self._execute_path_b(
params, context, db, db_provider, api_key, env, config_mgr, log
)
results.append(result)
except Exception as e:
log.error(f"Dashboard {dashboard_id} processing failed: {e}")
last_error = e
# Continue to next dashboard — records for already-processed ones are saved
# After ALL dashboards processed, update run status to reflect completion
# After ALL dashboards processed (even if some failed), update run status
_update_run_status(db, params.get("run_id"))
if last_error:
raise last_error # Re-raise so TaskManager marks the task as FAILED
return {"dashboards": results, "total": len(results)}
finally:
@@ -384,11 +422,21 @@ class DashboardValidationPlugin(PluginBase):
dashboard_prompt = _ensure_json_prompt(raw_prompt)
llm_call_started_at = datetime.now(UTC)
if jpeg_paths:
# Extract tab labels from JPEG filenames for prompt context
# Filename format: {dashboard_id}_{tab_label}_{timestamp}_d{depth}_llm.jpg
tab_labels = []
for jp in jpeg_paths:
base = os.path.splitext(os.path.basename(jp))[0]
# Remove _llm suffix and parse
clean = base.rsplit("_llm", 1)[0] if base.endswith("_llm") else base
m = re.match(r"\d+_(.+)_\d+_d\d+$", clean)
tab_labels.append(m.group(1).replace("_", " ") if m else f"tab {len(tab_labels)}")
analysis = await llm_client.analyze_dashboard_multimodal(
screenshot_paths=jpeg_paths,
logs=logs,
prompt_template=dashboard_prompt,
max_images=db_provider.max_images,
tab_labels=tab_labels,
)
else:
# Fallback: text-only analysis if no screenshots
@@ -478,7 +526,9 @@ class DashboardValidationPlugin(PluginBase):
summary=validation_result.summary,
issues=[issue.model_dump() for issue in validation_result.issues],
screenshot_path=validation_result.screenshot_path,
screenshot_paths=webp_paths or jpeg_paths,
screenshot_paths=webp_paths or jpeg_paths or [],
logs_sent_to_llm=logs,
execution_path="screenshot",
raw_response=json.dumps(result_payload, ensure_ascii=False),
)
db.add(db_record)
@@ -594,7 +644,7 @@ class DashboardValidationPlugin(PluginBase):
)
# 5. Build topology text
topology = self._build_dashboard_topology(dashboard, chart_data)
topology = build_dashboard_topology(dashboard, chart_data)
# 6. Build text prompt and call LLM (text-only)
llm_client = LLMClient(
@@ -689,6 +739,9 @@ class DashboardValidationPlugin(PluginBase):
summary=validation_result.summary,
issues=[issue.model_dump() for issue in validation_result.issues],
screenshot_path=None,
screenshot_paths=[],
logs_sent_to_llm=logs,
execution_path="text_only",
raw_response=json.dumps(result_payload, ensure_ascii=False),
)
db.add(db_record)
@@ -789,290 +842,7 @@ class DashboardValidationPlugin(PluginBase):
# endregion DashboardValidationPlugin._fetch_dashboard_logs
# region DashboardValidationPlugin._build_dashboard_topology [TYPE Function]
# @PURPOSE: Build a structured text description of dashboard layout from position_json.
# Recursively parses position_json to build Tab -> Row -> Chart hierarchy.
# @PARAM dashboard (dict) - Dashboard metadata dict with position_json, dashboard_title.
# @PARAM charts (list[dict]) - List of fetched chart detail dicts for name/viz lookup.
# @POST Returns a human-readable text description of the dashboard topology.
# @RELATION DEPENDS_ON -> [json]
def _build_dashboard_topology(self, dashboard: dict, charts: list[dict]) -> str:
"""
Build a structured text description of dashboard layout from position_json.
Returns a human-readable text like:
Dashboard: "Sales Overview"
--- Tab: "Revenue" (3 charts) ---
Chart "Total Revenue YTD" (big_number_total)
Dataset: 5
Metrics: SUM(revenue)
...
--- Tab: "Pipeline" (2 charts) ---
...
"""
lines: list[str] = []
title = dashboard.get(
"dashboard_title",
dashboard.get("title", f"Dashboard {dashboard.get('id', '')}"),
)
lines.append(f'Dashboard: "{title}"')
position_json = dashboard.get("position_json", {})
if isinstance(position_json, str):
try:
position_json = json.loads(position_json)
except (json.JSONDecodeError, TypeError):
position_json = {}
if not isinstance(position_json, dict):
position_json = {}
# Build chart lookup: chart_id -> chart dict
chart_lookup: dict[str, dict] = {}
for c in charts:
cid = c.get("id") or c.get("slice_id")
if cid is not None:
chart_lookup[str(cid)] = c
# Process the position_json layout
try:
children = _extract_topology_children(position_json, chart_lookup)
lines.extend(children)
except Exception as e:
lines.append(f"\n[Topology parse warning: {e}]")
# Add chart summary at the end
if chart_lookup:
lines.append(f"\nCharts: {len(chart_lookup)} total")
for chart_id, c in sorted(chart_lookup.items(), key=_sort_chart_key):
lines.append(
f" - {c.get('slice_name', f'chart_{chart_id}')} "
f"({c.get('viz_type', 'unknown')})"
)
return "\n".join(lines)
# endregion DashboardValidationPlugin._build_dashboard_topology
# #region _chart_name [TYPE Function]
# @BRIEF Return the display name for a chart.
def _chart_name(chart: dict) -> str:
return chart.get("slice_name", chart.get("title", f"Chart {chart.get('id')}"))
# #endregion _chart_name
# #region _chart_params [TYPE Function]
# @BRIEF Return parsed chart params dict.
def _chart_params(chart: dict) -> dict:
params = chart.get("params", {})
if isinstance(params, str):
try:
return json.loads(params)
except (json.JSONDecodeError, TypeError):
return {}
return params
# #endregion _chart_params
# #region _describe_chart_dataset [TYPE Function]
# @BRIEF Return dataset description line if available.
# @PRE chart and params are chart dicts.
# @POST Returns string with dataset line or empty string.
def _describe_chart_dataset(chart: dict, params: dict) -> str:
ds_id = chart.get("datasource_id") or params.get("datasource_id")
if ds_id:
return f"\n Dataset: {ds_id}"
return ""
# #endregion _describe_chart_dataset
# #region _describe_chart_metrics [TYPE Function]
# @BRIEF Return metrics description line if available.
# @PRE params is a parsed chart params dict.
# @POST Returns string with metrics line or empty string.
def _describe_chart_metrics(params: dict) -> str:
metrics = params.get("metrics", [])
if not metrics:
return ""
metric_names: list[str] = []
for m in metrics:
if isinstance(m, dict):
metric_names.append(m.get("label", m.get("expression", str(m))))
else:
metric_names.append(str(m))
if metric_names:
return f"\n Metrics: {', '.join(metric_names[:3])}"
return ""
# #endregion _describe_chart_metrics
# #region _describe_chart_groupby [TYPE Function]
# @BRIEF Return group-by description line if available.
# @PRE params is a parsed chart params dict.
# @POST Returns string with group-by line or empty string.
def _describe_chart_groupby(params: dict) -> str:
groupby = params.get("groupby", [])
if groupby:
return f"\n Group by: {', '.join(groupby[:3])}"
return ""
# #endregion _describe_chart_groupby
# #region _describe_chart_filters [TYPE Function]
# @BRIEF Return adhoc filters description line if available.
# @PRE params is a parsed chart params dict.
# @POST Returns string with filters line or empty string.
def _describe_chart_filters(params: dict) -> str:
adhoc_filters = params.get("adhoc_filters", [])
if not adhoc_filters:
return ""
filter_parts: list[str] = []
for filt in adhoc_filters[:3]:
if isinstance(filt, dict):
subject = filt.get("subject", "")
operator = filt.get("operator", "")
comparator = filt.get("comparator", "")
if subject and comparator:
filter_parts.append(f"{subject} {operator} {comparator}")
else:
filter_parts.append(filt.get("clause", "where"))
if filter_parts:
return f"\n Filters: {', '.join(filter_parts)}"
return ""
# #endregion _describe_chart_filters
# #region _describe_chart [TYPE Function]
# @BRIEF Return a multi-line description of a single chart.
# @PRE chart is a chart detail dict with slice_name, viz_type, params.
# @POST Returns formatted chart description string.
def _describe_chart(chart: dict) -> str:
name = _chart_name(chart)
viz_type = chart.get("viz_type", "unknown")
params = _chart_params(chart)
desc = f' Chart "{name}" ({viz_type})'
desc += _describe_chart_dataset(chart, params)
desc += _describe_chart_metrics(params)
desc += _describe_chart_groupby(params)
desc += _describe_chart_filters(params)
return desc
# #endregion _describe_chart
# #region _sort_chart_key [TYPE Function]
# @BRIEF Sort key for chart lookup items: numeric IDs first, then string keys.
def _sort_chart_key(item):
k, _ = item
try:
return (0, int(k))
except (ValueError, TypeError):
return (1, str(k))
# #endregion _sort_chart_key
# #region _resolve_child [TYPE Function]
# @BRIEF Resolve a child reference which may be a string key or inline dict.
# @PRE ref is a child reference from position_json, pos_json is the full layout.
# @POST Returns the resolved child dict or None.
def _resolve_child(ref, pos_json: dict) -> dict | None:
if isinstance(ref, dict):
return ref
if isinstance(ref, str) and ref in pos_json:
return pos_json[ref]
return None
# #endregion _resolve_child
# #region _process_tab [TYPE Function]
# @BRIEF Process a TAB type position_json node.
# @POST Returns indented lines for the tab and its children.
def _process_tab(
child: dict, depth: int, pos_json: dict, chart_lookup: dict[str, dict]
) -> list[str]:
meta = child.get("meta", {}) or {}
tab_name = meta.get("text", meta.get("label", "Tab"))
sub_lines = _process_children(child, depth + 1, pos_json, chart_lookup)
chart_count = sum(1 for line in sub_lines if line.strip().startswith("Chart"))
indent = " " * depth
return [f'{indent}─── Tab: "{tab_name}" ({chart_count} charts) ───', *sub_lines]
# #endregion _process_tab
# #region _process_chart [TYPE Function]
# @BRIEF Process a CHART type position_json node.
# @POST Returns indented line for the chart description or empty list.
def _process_chart(
child: dict, depth: int, chart_lookup: dict[str, dict]
) -> list[str]:
meta = child.get("meta", {}) or {}
chart_id = meta.get("chartId") or meta.get("slice_id")
if chart_id is not None and str(chart_id) in chart_lookup:
indent = " " * depth
return [f"{indent}{_describe_chart(chart_lookup[str(chart_id)])}"]
return []
# #endregion _process_chart
# #region _process_children [TYPE Function]
# @BRIEF Recursively process position_json children building indented lines.
# @PRE parent is a position_json node dict, pos_json is the full layout, chart_lookup maps IDs.
# @POST Returns list of formatted topology lines.
def _process_children(
parent: dict, depth: int, pos_json: dict, chart_lookup: dict[str, dict]
) -> list[str]:
result: list[str] = []
children = parent.get("children", []) if isinstance(parent, dict) else []
for child_ref in children:
child = _resolve_child(child_ref, pos_json)
if not child or not isinstance(child, dict):
continue
child_type = child.get("type", "")
if child_type == "TAB":
result.extend(_process_tab(child, depth, pos_json, chart_lookup))
elif child_type == "ROW":
result.extend(_process_children(child, depth, pos_json, chart_lookup))
elif child_type == "CHART":
result.extend(_process_chart(child, depth, chart_lookup))
return result
# #endregion _process_children
# #region _extract_topology_children [TYPE Function]
# @BRIEF Extract formatted topology lines from position_json.
# @PRE position_json is a parsed layout dict, chart_lookup maps chart IDs to chart dicts.
# @POST Returns list of formatted topology description lines.
def _extract_topology_children(
position_json: dict, chart_lookup: dict[str, dict]
) -> list[str]:
lines: list[str] = []
# Find grid/root containers that hold the layout
grid_nodes = [
v
for v in position_json.values()
if isinstance(v, dict) and v.get("type") in ("GRID", "ROOT", "CONTAINER")
]
if grid_nodes:
for grid in grid_nodes:
lines.extend(_process_children(grid, depth=1, pos_json=position_json, chart_lookup=chart_lookup))
else:
# No explicit grid — try processing all top-level layout components
for key, value in position_json.items():
if (
isinstance(value, dict)
and value.get("type") in ("TAB", "ROW", "CHART")
and key != "DASHBOARD_VERSION_KEY"
):
lines.extend(_process_children({"children": [key]}, depth=0, pos_json=position_json, chart_lookup=chart_lookup))
return lines
# #endregion _extract_topology_children
# #endregion DashboardValidationPlugin
# endregion DashboardValidationPlugin
# #region DocumentationPlugin [TYPE Class]

13
backend/src/plugins/llm_analysis/service.py
View File

@@ -629,7 +629,7 @@ class ScreenshotService:
if not safe_tab:
safe_tab = f"tab_{depth}_{i}"
tab_filename = f"{dashboard_id}_{safe_tab}_{timestamp}.png"
tab_filename = f"{dashboard_id}_{safe_tab}_{timestamp}_d{depth}.png"
tab_path = os.path.join(output_dir, tab_filename)
try:
@@ -1280,6 +1280,7 @@ class LLMClient:
# region LLMClient.analyze_dashboard_multimodal [TYPE Function] [C:3]
# @PURPOSE Path A: send screenshots + logs to multimodal LLM, with chunking support.
# @PRE screenshot_paths is a non-empty list of paths.
# tab_labels, if provided, must have the same length as screenshot_paths.
# @POST Returns dict {status, summary, issues} with optional chunk_count.
# @SIDE_EFFECT Compresses images, calls external LLM API (possibly multiple times for chunks).
# @RATIONALE Screenshots are split into chunks of max_images to respect provider image limits.
@@ -1293,6 +1294,7 @@ class LLMClient:
max_width: int = 1024,
image_quality: int = 60,
max_images: int | None = None,
tab_labels: list[str] | None = None,
) -> dict[str, Any]:
with belief_scope("analyze_dashboard_multimodal"):
if not screenshot_paths:
@@ -1302,7 +1304,14 @@ class LLMClient:
encoded_images = self._optimize_images(screenshot_paths, max_width, image_quality)
log_text = "\n".join(logs)
prompt = render_prompt(prompt_template, {"logs": log_text})
tab_list_text = "\n".join(
f" Screenshot {i}: {label}" for i, label in enumerate(tab_labels or [])
) or "Screenshots are in order."
prompt = render_prompt(prompt_template, {
"logs": log_text,
"tab_list": tab_list_text,
"total_chunks": str(len(encoded_images)),
})
# 2. Determine chunking
# Default to 8 images per chunk as a safe fallback when max_images is 0 or None