From 2528e7a631538e66ac0015fc415466e0ca7bae9e Mon Sep 17 00:00:00 2001 From: Lexi-Energy <258596309+Lexi-Energy@users.noreply.github.com> Date: Wed, 1 Apr 2026 18:16:06 +0000 Subject: [PATCH 1/5] feat: LLM summaries via litellm, --summarize flag, graceful degradation, 32 new tests --- CHANGELOG.md | 7 + tests/test_cli.py | 14 + tests/test_server.py | 52 +++ tests/test_summarizer.py | 325 +++++++++++++++++++ visualpy/cli.py | 37 +++ visualpy/models.py | 1 + visualpy/summarizer/__init__.py | 4 + visualpy/summarizer/llm.py | 151 +++++++++ visualpy/templates/overview.html | 3 + visualpy/templates/partials/script_card.html | 3 + visualpy/templates/script.html | 3 + 11 files changed, 600 insertions(+) create mode 100644 tests/test_summarizer.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f90f84e..73553f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this ### Added +- LLM-generated plain-English summaries via litellm (BYOK — bring your own key) +- `--summarize` CLI flag for `analyze` and `serve` commands +- Per-script summaries from structured AST data (steps, services, triggers) +- Per-project executive summary from script summaries and connections +- Model override via `VISUALPY_MODEL` environment variable (default: `gemini/gemini-2.5-flash`) +- Summary rendering in web UI: overview header, script cards, script headers +- Graceful degradation when litellm not installed or API key missing - README with badges, personas, quick start, roadmap, and acknowledgments - CONTRIBUTING.md with non-dev-friendly contribution guide - CODE_OF_CONDUCT.md (Contributor Covenant v2.1) diff --git a/tests/test_cli.py b/tests/test_cli.py index ff7eb33..aea3f86 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -59,6 +59,20 @@ def test_analyze_nonexistent_path(): assert "Error" in result.stderr +def test_analyze_summarize_flag(hello_script): + """--summarize flag is accepted and summary fields are present in output.""" + result = subprocess.run( + [sys.executable, "-m", "visualpy", "analyze", str(hello_script), "--summarize"], + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0 + data = json.loads(result.stdout) + assert "summary" in data + assert "summary" in data["scripts"][0] + + def test_version(): result = subprocess.run( [sys.executable, "-m", "visualpy", "--version"], diff --git a/tests/test_server.py b/tests/test_server.py index e7151fc..d7a9c48 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -306,6 +306,58 @@ async def test_mermaid_rerender_function_present(client): assert "reRenderMermaid" in resp.text +# --- LLM summary rendering --- + + +@pytest.mark.anyio +async def test_overview_shows_project_summary(): + project = AnalyzedProject( + path="/tmp/test", + scripts=[AnalyzedScript(path="a.py")], + summary="This project automates lead generation.", + ) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/") + assert "This project automates lead generation." in resp.text + + +@pytest.mark.anyio +async def test_overview_no_summary_no_crash(): + project = AnalyzedProject(path="/tmp/test", scripts=[AnalyzedScript(path="a.py")]) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/") + assert resp.status_code == 200 + assert "lead generation" not in resp.text + + +@pytest.mark.anyio +async def test_script_view_shows_summary(): + script = AnalyzedScript( + path="example.py", + summary="Fetches data from an API and saves it locally.", + ) + project = AnalyzedProject(path="/tmp/test", scripts=[script]) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/script/example.py") + assert "Fetches data from an API and saves it locally." in resp.text + + +@pytest.mark.anyio +async def test_script_card_shows_summary(): + script = AnalyzedScript( + path="example.py", + summary="Automates data fetching.", + ) + project = AnalyzedProject(path="/tmp/test", scripts=[script]) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/") + assert "Automates data fetching." in resp.text + + # --- Integration with real fixtures --- @pytest.mark.anyio diff --git a/tests/test_summarizer.py b/tests/test_summarizer.py new file mode 100644 index 0000000..60ecd9d --- /dev/null +++ b/tests/test_summarizer.py @@ -0,0 +1,325 @@ +"""Tests for LLM summarizer.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from visualpy.models import ( + AnalyzedProject, + AnalyzedScript, + ScriptConnection, + Service, + Step, + Trigger, +) +from visualpy.summarizer.llm import ( + _build_project_prompt, + _build_script_prompt, + summarize_project, + summarize_script, +) + + +# --- Fixtures --- + + +@pytest.fixture +def sample_script(): + return AnalyzedScript( + path="fetch_data.py", + is_entry_point=True, + steps=[ + Step( + line_number=10, + type="api_call", + description="requests.get(url)", + function_name="fetch", + ), + Step( + line_number=15, + type="file_io", + description="json.dump(data, f)", + function_name="save", + ), + Step( + line_number=20, + type="decision", + description="if verbose", + function_name="main", + ), + Step( + line_number=22, + type="output", + description="print(result)", + function_name="main", + ), + ], + services=[Service(name="HTTP Client", library="requests")], + secrets=["API_KEY"], + triggers=[Trigger(type="cli", detail="__main__ guard")], + ) + + +@pytest.fixture +def sample_project(sample_script): + sample_script.summary = "Fetches data from an API and saves it locally." + return AnalyzedProject( + path="/tmp/test", + scripts=[sample_script], + connections=[ + ScriptConnection( + source="fetch_data.py", + target="upload.py", + type="file_io", + detail="fetch_data.py writes data.json -> upload.py reads data.json", + ) + ], + entry_points=["fetch_data.py"], + services=[Service(name="HTTP Client", library="requests")], + secrets=["API_KEY"], + ) + + +# --- Prompt building (deterministic, no LLM) --- + + +class TestBuildScriptPrompt: + def test_returns_two_messages(self, sample_script): + messages = _build_script_prompt(sample_script) + assert isinstance(messages, list) + assert len(messages) == 2 + assert messages[0]["role"] == "system" + assert messages[1]["role"] == "user" + + def test_includes_script_path(self, sample_script): + messages = _build_script_prompt(sample_script) + assert "fetch_data.py" in messages[1]["content"] + + def test_includes_services(self, sample_script): + messages = _build_script_prompt(sample_script) + assert "HTTP Client" in messages[1]["content"] + + def test_includes_steps(self, sample_script): + messages = _build_script_prompt(sample_script) + content = messages[1]["content"] + assert "requests.get(url)" in content + assert "json.dump(data, f)" in content + + def test_includes_secrets(self, sample_script): + messages = _build_script_prompt(sample_script) + assert "API_KEY" in messages[1]["content"] + + def test_includes_triggers(self, sample_script): + messages = _build_script_prompt(sample_script) + assert "cli" in messages[1]["content"] + + def test_groups_steps_by_function(self, sample_script): + messages = _build_script_prompt(sample_script) + content = messages[1]["content"] + assert "In fetch:" in content + assert "In save:" in content + assert "In main:" in content + + def test_empty_script(self): + script = AnalyzedScript(path="empty.py") + messages = _build_script_prompt(script) + assert len(messages) == 2 + assert "empty.py" in messages[1]["content"] + assert "No steps detected" in messages[1]["content"] + + def test_system_prompt_targets_non_technical(self, sample_script): + messages = _build_script_prompt(sample_script) + system = messages[0]["content"].lower() + assert "non-technical" in system or "business" in system + + +class TestBuildProjectPrompt: + def test_returns_two_messages(self, sample_project): + messages = _build_project_prompt(sample_project) + assert isinstance(messages, list) + assert len(messages) == 2 + assert messages[0]["role"] == "system" + assert messages[1]["role"] == "user" + + def test_includes_script_summaries(self, sample_project): + messages = _build_project_prompt(sample_project) + assert "Fetches data from an API" in messages[1]["content"] + + def test_includes_project_path(self, sample_project): + messages = _build_project_prompt(sample_project) + assert "/tmp/test" in messages[1]["content"] + + def test_includes_connections(self, sample_project): + messages = _build_project_prompt(sample_project) + assert "data.json" in messages[1]["content"] + + def test_includes_services(self, sample_project): + messages = _build_project_prompt(sample_project) + assert "HTTP Client" in messages[1]["content"] + + def test_script_without_summary(self): + project = AnalyzedProject( + path="/tmp/bare", + scripts=[AnalyzedScript(path="bare.py")], + ) + messages = _build_project_prompt(project) + assert "No summary available" in messages[1]["content"] + + +# --- Graceful degradation --- + + +class TestGracefulDegradation: + @patch("visualpy.summarizer.llm._call_llm", return_value=None) + def test_script_returns_none_on_failure(self, mock_llm, sample_script): + result = summarize_script(sample_script) + assert result is None + + @patch("visualpy.summarizer.llm._call_llm", return_value=None) + def test_project_returns_none_on_failure(self, mock_llm, sample_project): + result = summarize_project(sample_project) + assert result is None + + @patch("visualpy.summarizer.llm._call_llm", return_value="A helpful summary.") + def test_script_returns_string_on_success(self, mock_llm, sample_script): + result = summarize_script(sample_script) + assert result == "A helpful summary." + mock_llm.assert_called_once() + + @patch("visualpy.summarizer.llm._call_llm", return_value="A helpful summary.") + def test_project_returns_string_on_success(self, mock_llm, sample_project): + result = summarize_project(sample_project) + assert result == "A helpful summary." + mock_llm.assert_called_once() + + @patch("visualpy.summarizer.llm._call_llm") + def test_model_env_var_override(self, mock_llm, sample_script): + mock_llm.return_value = "summary" + with patch.dict("os.environ", {"VISUALPY_MODEL": "openai/gpt-4o-mini"}): + summarize_script(sample_script) + assert mock_llm.call_args[0][1] == "openai/gpt-4o-mini" + + @patch("visualpy.summarizer.llm._call_llm") + def test_explicit_model_parameter(self, mock_llm, sample_script): + mock_llm.return_value = "summary" + summarize_script(sample_script, model="openai/gpt-4o-mini") + assert mock_llm.call_args[0][1] == "openai/gpt-4o-mini" + + @patch("visualpy.summarizer.llm._call_llm") + def test_default_model(self, mock_llm, sample_script): + mock_llm.return_value = "summary" + with patch.dict("os.environ", {}, clear=True): + summarize_script(sample_script) + assert "gemini" in mock_llm.call_args[0][1] + + +class TestCallLlm: + @patch.dict("sys.modules", {"litellm": None}) + def test_missing_litellm_returns_none(self): + # Force ImportError by removing litellm from modules + from visualpy.summarizer.llm import _call_llm + + with patch("builtins.__import__", side_effect=ImportError("no litellm")): + result = _call_llm([{"role": "user", "content": "test"}], "test/model") + assert result is None + + def test_llm_exception_returns_none(self): + from visualpy.summarizer.llm import _call_llm + + mock_litellm = MagicMock() + mock_litellm.completion.side_effect = RuntimeError("API error") + + with patch.dict("sys.modules", {"litellm": mock_litellm}): + result = _call_llm( + [{"role": "user", "content": "test"}], "test/model" + ) + assert result is None + + def test_none_content_returns_none(self): + from visualpy.summarizer.llm import _call_llm + + mock_response = MagicMock() + mock_response.choices[0].message.content = None + mock_litellm = MagicMock() + mock_litellm.completion.return_value = mock_response + + with patch.dict("sys.modules", {"litellm": mock_litellm}): + result = _call_llm( + [{"role": "user", "content": "test"}], "test/model" + ) + assert result is None + + def test_empty_content_returns_none(self): + from visualpy.summarizer.llm import _call_llm + + mock_response = MagicMock() + mock_response.choices[0].message.content = " " + mock_litellm = MagicMock() + mock_litellm.completion.return_value = mock_response + + with patch.dict("sys.modules", {"litellm": mock_litellm}): + result = _call_llm( + [{"role": "user", "content": "test"}], "test/model" + ) + assert result is None + + def test_successful_call(self): + from visualpy.summarizer.llm import _call_llm + + mock_response = MagicMock() + mock_response.choices[0].message.content = " A summary. " + + mock_litellm = MagicMock() + mock_litellm.completion.return_value = mock_response + + with patch.dict("sys.modules", {"litellm": mock_litellm}): + result = _call_llm( + [{"role": "user", "content": "test"}], "test/model" + ) + assert result == "A summary." + mock_litellm.completion.assert_called_once_with( + model="test/model", + messages=[{"role": "user", "content": "test"}], + max_tokens=200, + ) + + +# --- Integration with real LLM (marked slow) --- + + +@pytest.mark.slow +class TestRealLLM: + def test_summarize_script(self, hello_script, fixtures_dir): + from visualpy.analyzer.ast_parser import analyze_file + + script = analyze_file(hello_script, fixtures_dir) + result = summarize_script(script) + assert result is not None + assert len(result) > 10 + assert len(result) < 500 + + def test_summarize_project(self, fixtures_dir): + from visualpy.analyzer.cross_file import resolve_connections + from visualpy.analyzer.scanner import scan_project + + agentic = fixtures_dir / "agentic_workflows" + scripts = scan_project(agentic) + connections = resolve_connections(scripts, agentic) + + for script in scripts: + script.summary = summarize_script(script) + + project = AnalyzedProject( + path=str(agentic), + scripts=scripts, + connections=connections, + entry_points=[s.path for s in scripts if s.is_entry_point], + services=[svc for s in scripts for svc in s.services], + secrets=sorted({sec for s in scripts for sec in s.secrets}), + ) + + result = summarize_project(project) + assert result is not None + assert len(result) > 20 diff --git a/visualpy/cli.py b/visualpy/cli.py index 5ae311a..53e356f 100644 --- a/visualpy/cli.py +++ b/visualpy/cli.py @@ -29,6 +29,11 @@ def app(): analyze_parser.add_argument( "--output", "-o", help="Output JSON file (default: stdout)" ) + analyze_parser.add_argument( + "--summarize", + action="store_true", + help="Generate LLM summaries (requires API key, install with pip install visualpy[llm])", + ) # serve command serve_parser = subparsers.add_parser( @@ -37,6 +42,11 @@ def app(): serve_parser.add_argument("path", help="Path to folder to analyze") serve_parser.add_argument("--port", type=int, default=8000, help="Port (default: 8000)") serve_parser.add_argument("--host", default="127.0.0.1", help="Host (default: 127.0.0.1)") + serve_parser.add_argument( + "--summarize", + action="store_true", + help="Generate LLM summaries (requires API key, install with pip install visualpy[llm])", + ) args = parser.parse_args() @@ -83,6 +93,25 @@ def _build_project(target: Path) -> AnalyzedProject: ) +def _summarize_project(project: AnalyzedProject) -> None: + """Populate LLM summaries on the project (mutates in place).""" + from visualpy.summarizer import summarize_project, summarize_script + + print("[visualpy] Generating summaries...", file=sys.stderr) + + for i, script in enumerate(project.scripts, 1): + print( + f"[visualpy] Script {i}/{len(project.scripts)}: {script.path}", + file=sys.stderr, + ) + script.summary = summarize_script(script) + + project.summary = summarize_project(project) + + count = sum(1 for s in project.scripts if s.summary) + (1 if project.summary else 0) + print(f"[visualpy] Generated {count} summaries", file=sys.stderr) + + def _run_analyze(args: argparse.Namespace) -> None: """Run the analysis pipeline and output JSON.""" target = Path(args.path).resolve() @@ -92,6 +121,10 @@ def _run_analyze(args: argparse.Namespace) -> None: sys.exit(1) project = _build_project(target) + + if args.summarize: + _summarize_project(project) + output = json.dumps(dataclasses.asdict(project), indent=2) if args.output: @@ -119,6 +152,10 @@ def _run_serve(args: argparse.Namespace) -> None: print(f"[visualpy] Analyzing {target}...", file=sys.stderr) project = _build_project(target) + + if args.summarize: + _summarize_project(project) + print( f"[visualpy] Found {len(project.scripts)} scripts, " f"{len(project.connections)} connections", diff --git a/visualpy/models.py b/visualpy/models.py index 35a77ac..6a5ca9b 100644 --- a/visualpy/models.py +++ b/visualpy/models.py @@ -71,3 +71,4 @@ class AnalyzedProject: services: list[Service] = field(default_factory=list) secrets: list[str] = field(default_factory=list) entry_points: list[str] = field(default_factory=list) + summary: str | None = None # LLM-generated executive summary diff --git a/visualpy/summarizer/__init__.py b/visualpy/summarizer/__init__.py index 73a0917..06ca7d8 100644 --- a/visualpy/summarizer/__init__.py +++ b/visualpy/summarizer/__init__.py @@ -1 +1,5 @@ """LLM summarizer — optional plain-English descriptions via BYOK LLM.""" + +from visualpy.summarizer.llm import summarize_project, summarize_script + +__all__ = ["summarize_script", "summarize_project"] diff --git a/visualpy/summarizer/llm.py b/visualpy/summarizer/llm.py index 747f73a..a9e044b 100644 --- a/visualpy/summarizer/llm.py +++ b/visualpy/summarizer/llm.py @@ -1 +1,152 @@ """LLM-based summarization — feed AST structure, get non-technical descriptions.""" + +from __future__ import annotations + +import os +import sys +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from visualpy.models import AnalyzedProject, AnalyzedScript + +DEFAULT_MODEL = "gemini/gemini-2.5-flash" + +_SYSTEM_PROMPT = ( + "You are a technical writer who explains software to non-technical business " + "stakeholders. You write clear, jargon-free summaries. You never mention " + "programming concepts like functions, variables, loops, or imports. Instead, " + "describe what the script accomplishes in business terms." +) + + +def summarize_script( + script: AnalyzedScript, model: str | None = None +) -> str | None: + """Generate a plain-English summary for a single script.""" + model = model or os.environ.get("VISUALPY_MODEL", DEFAULT_MODEL) + messages = _build_script_prompt(script) + return _call_llm(messages, model) + + +def summarize_project( + project: AnalyzedProject, model: str | None = None +) -> str | None: + """Generate an executive summary for the entire project.""" + model = model or os.environ.get("VISUALPY_MODEL", DEFAULT_MODEL) + messages = _build_project_prompt(project) + return _call_llm(messages, model) + + +def _build_script_prompt(script: AnalyzedScript) -> list[dict]: + """Format a script's structured data into LLM messages.""" + triggers = ", ".join(f"{t.type}: {t.detail}" for t in script.triggers) or "None" + services = ", ".join(s.name for s in script.services) or "None" + secrets = ", ".join(script.secrets) or "None" + + # Group steps by function + steps_by_func: dict[str, list[str]] = {} + for step in script.steps: + key = step.function_name or "(top-level)" + steps_by_func.setdefault(key, []).append( + f"- [{step.type}] {step.description}" + ) + + steps_text = "" + for func, lines in steps_by_func.items(): + steps_text += f"\nIn {func}:\n" + "\n".join(lines) + "\n" + + if not steps_text.strip(): + steps_text = "No steps detected." + + user_msg = ( + f"Objective: Write a 1-2 sentence plain-English summary of what this " + f"automation script does.\n\n" + f"Script: {script.path}\n" + f"Entry point: {'Yes' if script.is_entry_point else 'No'}\n" + f"Triggers: {triggers}\n" + f"Services used: {services}\n" + f"Secrets/API keys needed: {secrets}\n\n" + f"Steps (in execution order):\n{steps_text}\n" + f"Instructions:\n" + f"- Describe what this script accomplishes, not how it works\n" + f"- Mention the key services and data sources by name\n" + f"- Keep it under 2 sentences\n" + f"- Write for someone who has never seen code before\n\n" + f"Expected Output: A 1-2 sentence plain-English description." + ) + + return [ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": user_msg}, + ] + + +def _build_project_prompt(project: AnalyzedProject) -> list[dict]: + """Format a project's structured data into LLM messages.""" + services = ", ".join(s.name for s in project.services) or "None" + entry_points = ", ".join(project.entry_points) or "None" + + script_lines = [] + for s in project.scripts: + desc = s.summary or "No summary available" + script_lines.append(f"- {s.path}: {desc}") + scripts_text = "\n".join(script_lines) or "No scripts." + + conn_lines = [] + for c in project.connections: + conn_lines.append(f"- {c.source} -> {c.target}: {c.detail}") + conns_text = "\n".join(conn_lines) or "No connections." + + user_msg = ( + f"Objective: Write a 2-3 sentence executive summary of this automation " + f"project.\n\n" + f"Project: {project.path}\n" + f"Scripts: {len(project.scripts)}\n" + f"Services: {services}\n" + f"Entry points: {entry_points}\n\n" + f"Script summaries:\n{scripts_text}\n\n" + f"Connections between scripts:\n{conns_text}\n\n" + f"Instructions:\n" + f"- Describe the overall purpose of this project as a system\n" + f"- Mention how the scripts work together\n" + f"- Highlight the main services and data flows\n" + f"- Keep it under 3 sentences\n" + f"- Write for a business stakeholder, not a developer\n\n" + f"Expected Output: A 2-3 sentence executive summary." + ) + + return [ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": user_msg}, + ] + + +def _call_llm(messages: list[dict], model: str) -> str | None: + """Call the LLM via litellm. Returns None on any failure.""" + try: + import litellm + except ImportError: + print( + "[visualpy] Warning: litellm not installed. " + "Install with: pip install visualpy[llm]", + file=sys.stderr, + ) + return None + + litellm.suppress_debug_info = True + + try: + response = litellm.completion( + model=model, + messages=messages, + max_tokens=200, + # Omit temperature — lets litellm use each provider's default. + # Avoids known issues with models that misbehave at low temps. + ) + content = response.choices[0].message.content + if not content or not content.strip(): + return None + return content.strip() + except Exception as exc: + print(f"[visualpy] Warning: LLM call failed: {exc}", file=sys.stderr) + return None diff --git a/visualpy/templates/overview.html b/visualpy/templates/overview.html index 4478db7..df4c89f 100644 --- a/visualpy/templates/overview.html +++ b/visualpy/templates/overview.html @@ -24,6 +24,9 @@

Project Overview

{{ project.entry_points|length }} entry point{{ "s" if project.entry_points|length != 1 }} + {% if project.summary %} +

{{ project.summary }}

+ {% endif %} diff --git a/visualpy/templates/partials/script_card.html b/visualpy/templates/partials/script_card.html index 7df7f1a..620dd1b 100644 --- a/visualpy/templates/partials/script_card.html +++ b/visualpy/templates/partials/script_card.html @@ -18,6 +18,9 @@ {% if '/' in script.path %}

{{ script.path }}

{% endif %} + {% if script.summary %} +

{{ script.summary }}

+ {% endif %}
{{ script.steps|length }} steps {% if script.steps %} diff --git a/visualpy/templates/script.html b/visualpy/templates/script.html index 10cfe9e..3a7c630 100644 --- a/visualpy/templates/script.html +++ b/visualpy/templates/script.html @@ -42,6 +42,9 @@

{{ script.path }}

main({% for name, info in script.signature.items() %}{{ name }}{% if not loop.last %}, {% endif %}{% endfor %})

{% endif %} + {% if script.summary %} +

{{ script.summary }}

+ {% endif %}
From 59d9c34d5037aa6bfe7dfe696507964be07ef579 Mon Sep 17 00:00:00 2001 From: Lexi-Energy <258596309+Lexi-Energy@users.noreply.github.com> Date: Wed, 1 Apr 2026 20:34:38 +0000 Subject: [PATCH 2/5] fix: max_tokens for thinking models, onclick quoting, sticky sidebar --- tests/test_summarizer.py | 2 +- visualpy/summarizer/llm.py | 5 ++++- visualpy/templates/script.html | 5 +++-- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/test_summarizer.py b/tests/test_summarizer.py index 60ecd9d..ccf7354 100644 --- a/tests/test_summarizer.py +++ b/tests/test_summarizer.py @@ -282,7 +282,7 @@ def test_successful_call(self): mock_litellm.completion.assert_called_once_with( model="test/model", messages=[{"role": "user", "content": "test"}], - max_tokens=200, + max_tokens=2048, ) diff --git a/visualpy/summarizer/llm.py b/visualpy/summarizer/llm.py index a9e044b..4940037 100644 --- a/visualpy/summarizer/llm.py +++ b/visualpy/summarizer/llm.py @@ -139,7 +139,10 @@ def _call_llm(messages: list[dict], model: str) -> str | None: response = litellm.completion( model=model, messages=messages, - max_tokens=200, + max_tokens=2048, + # Higher than needed for text output (~50 tokens) because some + # models (Gemini 2.5) use thinking tokens that count against this + # limit. 2048 accommodates thinking overhead comfortably. # Omit temperature — lets litellm use each provider's default. # Avoids known issues with models that misbehave at low temps. ) diff --git a/visualpy/templates/script.html b/visualpy/templates/script.html index 3a7c630..9158d0e 100644 --- a/visualpy/templates/script.html +++ b/visualpy/templates/script.html @@ -79,7 +79,7 @@

Flow

All Steps

{% for step in script.steps %} -
-
+

Step Detail

@@ -173,6 +173,7 @@

Imports< return; } htmx.ajax('GET', '/partials/step/' + encodeURIComponent(scriptPath) + '/' + line, {target: content, swap: 'innerHTML'}); + document.getElementById('step-detail').scrollIntoView({behavior: 'smooth', block: 'nearest'}); } {% endblock %} From 6eb7ad2a57e2eabe72cdb1a118484e137b0e2892 Mon Sep 17 00:00:00 2001 From: Lexi-Energy <258596309+Lexi-Energy@users.noreply.github.com> Date: Wed, 1 Apr 2026 22:17:53 +0000 Subject: [PATCH 3/5] feat: compact mode, importance scoring, flow toggle, 20 new tests --- CHANGELOG.md | 4 + tests/test_mermaid.py | 187 +++++++++++++++++++ tests/test_server.py | 87 +++++++++ visualpy/mermaid.py | 68 ++++++- visualpy/server.py | 30 ++- visualpy/templates/overview.html | 2 +- visualpy/templates/partials/script_card.html | 11 +- visualpy/templates/script.html | 46 ++++- 8 files changed, 422 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73553f5..7b49d2f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this ### Added +- Compact mode for script flow diagrams — functions with >8 steps collapse to summary nodes +- Compact/Detailed toggle button on script pages (auto-shown for scripts with >30 steps) +- Importance scoring for scripts — overview sorted by connectivity, entry points, services +- "Key" badge on top scripts in overview - LLM-generated plain-English summaries via litellm (BYOK — bring your own key) - `--summarize` CLI flag for `analyze` and `serve` commands - Per-script summaries from structured AST data (steps, services, triggers) diff --git a/tests/test_mermaid.py b/tests/test_mermaid.py index 887131b..491e13c 100644 --- a/tests/test_mermaid.py +++ b/tests/test_mermaid.py @@ -4,9 +4,11 @@ from visualpy.analyzer.cross_file import resolve_connections from visualpy.analyzer.scanner import scan_project from visualpy.mermaid import ( + _compact_function_node, _escape_label, _sanitize_id, _step_node, + importance_score, project_graph, script_flow, ) @@ -320,6 +322,191 @@ def test_script_flow_all_step_types(): assert '[["Transform: f"]]:::transform' in result +# --- _compact_function_node --- + + +def test_compact_function_node_label(): + steps = [ + Step(line_number=i, type="api_call", description=f"call{i}") + for i in range(10) + ] + result = _compact_function_node("fetch_data", steps, "test") + assert "fetch_data()" in result + assert "10 steps" in result + assert "10 API" in result + + +def test_compact_function_node_shape(): + steps = [Step(line_number=1, type="transform", description="x")] + result = _compact_function_node("func", steps, "test") + assert '[[' in result + assert ']]' in result + assert ":::compact" in result + + +def test_compact_function_node_multiple_types(): + steps = [ + Step(line_number=1, type="api_call", description="a"), + Step(line_number=2, type="api_call", description="b"), + Step(line_number=3, type="file_io", description="c"), + Step(line_number=4, type="decision", description="d"), + ] + result = _compact_function_node("func", steps, "test") + assert "4 steps" in result + assert "2 API" in result + assert "1 File I/O" in result + assert "1 Decision" in result + + +# --- script_flow compact mode --- + + +def test_compact_collapses_large_functions(): + """Functions with more steps than threshold are collapsed.""" + steps = [ + Step(line_number=i, type="api_call", description=f"call{i}", function_name="big_func") + for i in range(12) + ] + script = AnalyzedScript(path="test.py", steps=steps) + result = script_flow(script, compact=True, compact_threshold=8) + assert "big_func()" in result + assert "12 steps" in result + assert ":::compact" in result + # Should NOT have individual step nodes + assert "n_test_0" not in result + # Should NOT have a subgraph + assert "subgraph" not in result + + +def test_compact_keeps_small_functions(): + """Functions with steps at or below threshold stay expanded.""" + steps = [ + Step(line_number=i, type="api_call", description=f"call{i}", function_name="small_func") + for i in range(5) + ] + script = AnalyzedScript(path="test.py", steps=steps) + result = script_flow(script, compact=True, compact_threshold=8) + # Should have individual step nodes (not collapsed) + assert "n_test_0" in result + assert "subgraph" in result + assert ":::compact" not in result + + +def test_compact_module_level_never_collapsed(): + """Module-level steps (_module_) are never collapsed even in compact mode.""" + steps = [ + Step(line_number=i, type="output", description=f"print{i}") + for i in range(20) + ] + script = AnalyzedScript(path="test.py", steps=steps) + result = script_flow(script, compact=True, compact_threshold=8) + # Module level steps should all be present individually + assert "n_test_0" in result + assert "n_test_19" in result + assert ":::compact" not in result + + +def test_compact_threshold_boundary(): + """Function with exactly threshold steps stays expanded; threshold+1 collapses.""" + steps_at = [ + Step(line_number=i, type="api_call", description=f"c{i}", function_name="func") + for i in range(8) + ] + script_at = AnalyzedScript(path="test.py", steps=steps_at) + result_at = script_flow(script_at, compact=True, compact_threshold=8) + assert ":::compact" not in result_at # stays expanded + + steps_over = steps_at + [ + Step(line_number=8, type="api_call", description="c8", function_name="func") + ] + script_over = AnalyzedScript(path="test.py", steps=steps_over) + result_over = script_flow(script_over, compact=True, compact_threshold=8) + assert ":::compact" in result_over # collapsed + + +def test_compact_no_internal_edges(): + """Collapsed functions should have no --> edges.""" + steps = [ + Step(line_number=i, type="api_call", description=f"c{i}", function_name="big") + for i in range(12) + ] + script = AnalyzedScript(path="test.py", steps=steps) + result = script_flow(script, compact=True, compact_threshold=8) + assert "-->" not in result + + +def test_compact_classdefs(): + steps = [ + Step(line_number=i, type="api_call", description=f"c{i}", function_name="big") + for i in range(12) + ] + script = AnalyzedScript(path="test.py", steps=steps) + result = script_flow(script, compact=True) + assert "classDef compact" in result + + +def test_compact_default_is_false(): + """script_flow() without compact arg produces same output as compact=False.""" + steps = [ + Step(line_number=i, type="api_call", description=f"c{i}", function_name="big") + for i in range(12) + ] + script = AnalyzedScript(path="test.py", steps=steps) + result_default = script_flow(script) + result_explicit = script_flow(script, compact=False) + assert result_default == result_explicit + # Should have individual steps, not compact + assert ":::compact" not in result_default + assert "n_test_0" in result_default + + +# --- importance_score --- + + +def test_importance_entry_point_bonus(): + script = AnalyzedScript(path="main.py", is_entry_point=True) + project = AnalyzedProject(path="/tmp", scripts=[script], entry_points=["main.py"]) + score = importance_score(script, project) + assert score >= 2 # entry point bonus + + +def test_importance_connections(): + scripts = [AnalyzedScript(path="a.py"), AnalyzedScript(path="b.py")] + connections = [ + ScriptConnection(source="a.py", target="b.py", type="import", detail=""), + ScriptConnection(source="b.py", target="a.py", type="file_io", detail=""), + ] + project = AnalyzedProject(path="/tmp", scripts=scripts, connections=connections) + score_a = importance_score(scripts[0], project) + assert score_a >= 2 # 1 out + 1 in + + +def test_importance_services(): + script = AnalyzedScript( + path="api.py", + services=[Service(name="S1", library="l1"), Service(name="S2", library="l2")], + ) + project = AnalyzedProject(path="/tmp", scripts=[script]) + score = importance_score(script, project) + assert score >= 2 # 2 services + + +def test_importance_step_bonus_capped(): + steps = [Step(line_number=i, type="output", description="x") for i in range(200)] + script = AnalyzedScript(path="big.py", steps=steps) + project = AnalyzedProject(path="/tmp", scripts=[script]) + score = importance_score(script, project) + # 200 steps // 20 = 10, but capped at 5 + assert score == 5 + + +def test_importance_isolated_script(): + script = AnalyzedScript(path="lonely.py", steps=[Step(line_number=1, type="output", description="x")]) + project = AnalyzedProject(path="/tmp", scripts=[script]) + score = importance_score(script, project) + assert score == 0 # 1 step // 20 = 0, no connections, no entry, no services + + # --- Integration with real fixtures --- diff --git a/tests/test_server.py b/tests/test_server.py index d7a9c48..8eb5b98 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -358,6 +358,93 @@ async def test_script_card_shows_summary(): assert "Automates data fetching." in resp.text +# --- Compact toggle + importance --- + + +@pytest.mark.anyio +async def test_script_view_has_toggle_for_large_script(): + """Script with >30 steps should have a compact/detailed toggle button.""" + steps = [ + Step(line_number=i, type="api_call", description=f"call{i}", function_name="big") + for i in range(35) + ] + script = AnalyzedScript(path="big.py", steps=steps) + project = AnalyzedProject(path="/tmp", scripts=[script]) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/script/big.py") + assert "flow-toggle" in resp.text + assert "switchFlow" in resp.text + + +@pytest.mark.anyio +async def test_script_view_no_toggle_for_small_script(client): + """Script with <=30 steps should NOT have a toggle button element.""" + async with AsyncClient(transport=ASGITransport(app=client), base_url="http://test") as ac: + resp = await ac.get("/script/example.py") + # The button element should not be rendered (JS function still exists but that's fine) + assert 'id="flow-toggle"' not in resp.text + + +@pytest.mark.anyio +async def test_script_view_passes_both_flows(): + """Response should contain both compact and detailed flow data.""" + steps = [ + Step(line_number=i, type="api_call", description=f"call{i}", function_name="big") + for i in range(35) + ] + script = AnalyzedScript(path="big.py", steps=steps) + project = AnalyzedProject(path="/tmp", scripts=[script]) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/script/big.py") + assert "flow-detailed" in resp.text + assert "flow-compact" in resp.text + + +@pytest.mark.anyio +async def test_overview_scripts_sorted_by_importance(): + """Scripts should be sorted by importance (entry points first in card grid).""" + scripts = [ + AnalyzedScript(path="helper.py"), + AnalyzedScript(path="main.py", is_entry_point=True), + ] + project = AnalyzedProject( + path="/tmp", scripts=scripts, entry_points=["main.py"] + ) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/") + # Look at the script cards section (after "Scripts" heading) + cards_section = resp.text[resp.text.index("Scripts

"):] + main_pos = cards_section.index("/script/main.py") + helper_pos = cards_section.index("/script/helper.py") + assert main_pos < helper_pos + + +@pytest.mark.anyio +async def test_overview_key_badge(): + """Top scripts should have a 'key' badge.""" + scripts = [ + AnalyzedScript( + path="hub.py", + is_entry_point=True, + services=[], + steps=[Step(line_number=i, type="api_call", description="x") for i in range(50)], + ), + AnalyzedScript(path="leaf1.py"), + AnalyzedScript(path="leaf2.py"), + ] + project = AnalyzedProject( + path="/tmp", scripts=scripts, entry_points=["hub.py"] + ) + app = create_app(project) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + resp = await ac.get("/") + # hub.py should have the key badge (top 1/3 = at least 1) + assert ">key<" in resp.text + + # --- Integration with real fixtures --- @pytest.mark.anyio diff --git a/visualpy/mermaid.py b/visualpy/mermaid.py index fd6fa62..1ef5959 100644 --- a/visualpy/mermaid.py +++ b/visualpy/mermaid.py @@ -4,10 +4,21 @@ import re import sys +from collections import Counter from pathlib import PurePosixPath from visualpy.models import AnalyzedProject, AnalyzedScript, ScriptConnection, Step +# Short labels for step types (used in compact nodes and potentially templates). +_TYPE_LABELS: dict[str, str] = { + "api_call": "API", + "file_io": "File I/O", + "db_op": "DB", + "decision": "Decision", + "output": "Output", + "transform": "Transform", +} + # Step type → Mermaid node shape template. # Placeholders: {id} = sanitized node ID, {label} = escaped description. _STEP_SHAPES: dict[str, str] = { @@ -37,7 +48,8 @@ classDef decision fill:#ffedd5,stroke:#ea580c,color:#7c2d12 classDef output fill:#f3f4f6,stroke:#6b7280,color:#1f2937 classDef transform fill:#ccfbf1,stroke:#0d9488,color:#134e4a -classDef entry fill:#dcfce7,stroke:#16a34a,stroke-width:3px,color:#14532d""" +classDef entry fill:#dcfce7,stroke:#16a34a,stroke-width:3px,color:#14532d +classDef compact fill:#f0f4ff,stroke:#6366f1,color:#312e81,stroke-width:2px""" _SPECIAL_CHARS = re.compile(r"[^a-zA-Z0-9_]") _WARNED_TYPES: set[str] = set() @@ -100,6 +112,42 @@ def _step_node(step: Step, script_stem: str) -> str: return node_def +def _compact_function_node(func_name: str, steps: list[Step], stem: str) -> str: + """Generate a single Mermaid node summarising a collapsed function.""" + node_id = f"n_{_SPECIAL_CHARS.sub('_', stem)}_{_SPECIAL_CHARS.sub('_', func_name)}_compact" + + counts = Counter(s.type for s in steps) + # Top 4 types by count, descending. + top = counts.most_common(4) + parts = [f"{n} {_TYPE_LABELS.get(t, t)}" for t, n in top] + if len(counts) > 4: + parts.append(f"+{len(counts) - 4} more") + type_summary = ", ".join(parts) + total = len(steps) + + raw_label = f"{func_name}() — {total} steps ({type_summary})" + label = _escape_label(raw_label, max_len=100) + return f'{node_id}[["{label}"]]:::compact' + + +def importance_score(script: AnalyzedScript, project: AnalyzedProject) -> int: + """Heuristic importance score for a script within a project. + + Higher = more important. Used to sort scripts in the overview. + """ + score = 0 + for conn in project.connections: + if conn.source == script.path: + score += 1 + if conn.target == script.path: + score += 1 + if script.path in project.entry_points: + score += 2 + score += len(script.services) + score += min(len(script.steps) // 20, 5) + return score + + def project_graph(project: AnalyzedProject) -> str: """Generate a Mermaid graph showing scripts as nodes and connections as edges.""" lines: list[str] = ["graph LR"] @@ -146,8 +194,17 @@ def project_graph(project: AnalyzedProject) -> str: return "\n".join(lines) -def script_flow(script: AnalyzedScript) -> str: - """Generate a Mermaid flowchart for a single script's steps.""" +def script_flow( + script: AnalyzedScript, + *, + compact: bool = False, + compact_threshold: int = 8, +) -> str: + """Generate a Mermaid flowchart for a single script's steps. + + When *compact* is True, functions with more than *compact_threshold* + steps are collapsed into a single summary node. + """ lines: list[str] = ["graph TB"] if not script.steps: @@ -167,6 +224,11 @@ def script_flow(script: AnalyzedScript) -> str: # Emit subgraphs per function. click_lines: list[str] = [] for func_name, steps in func_steps.items(): + # Compact mode: collapse large functions into a single node. + if compact and func_name != "_module_" and len(steps) > compact_threshold: + lines.append(f" {_compact_function_node(func_name, steps, stem)}") + continue + if func_name != "_module_": sg_id = _sanitize_id(f"{stem}_{func_name}") label = _escape_label(func_name + "()") diff --git a/visualpy/server.py b/visualpy/server.py index 1a9ecc0..5516a9d 100644 --- a/visualpy/server.py +++ b/visualpy/server.py @@ -10,7 +10,7 @@ from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates -from visualpy.mermaid import project_graph, script_flow +from visualpy.mermaid import importance_score, project_graph, script_flow from visualpy.models import AnalyzedProject _PACKAGE_DIR = Path(__file__).parent @@ -54,12 +54,22 @@ async def unhandled_error(request: Request, exc: Exception): @app.get("/", response_class=HTMLResponse) async def overview(request: Request): + scored = sorted( + project.scripts, + key=lambda s: importance_score(s, project), + reverse=True, + ) + # Top ~30% of scripts are "key" scripts (at least 1). + key_count = max(1, len(scored) // 3) + key_paths = {s.path for s in scored[:key_count]} return templates.TemplateResponse( request, "overview.html", context={ "project": project, "graph": app.state.project_graph, + "sorted_scripts": scored, + "key_paths": key_paths, }, ) @@ -78,17 +88,27 @@ async def script_view(request: Request, path: str): status_code=404, ) try: - flow = script_flow(script) + flow_detailed = script_flow(script) + except Exception as exc: + print(f"[visualpy] Warning: failed to build detailed flow for {path}: {exc}", file=sys.stderr) + flow_detailed = 'graph TB\n error["Flow generation failed for this script"]' + try: + flow_compact = script_flow(script, compact=True) except Exception as exc: - print(f"[visualpy] Warning: failed to build flow for {path}: {exc}", file=sys.stderr) - flow = 'graph TB\n error["Flow generation failed for this script"]' + print(f"[visualpy] Warning: failed to build compact flow for {path}: {exc}", file=sys.stderr) + flow_compact = 'graph TB\n error["Compact flow generation failed"]' + total_steps = len(script.steps) return templates.TemplateResponse( request, "script.html", context={ "project": project, "script": script, - "flow": flow, + "flow": flow_compact if total_steps > 30 else flow_detailed, + "flow_detailed": flow_detailed, + "flow_compact": flow_compact, + "total_steps": total_steps, + "default_compact": total_steps > 30, }, ) diff --git a/visualpy/templates/overview.html b/visualpy/templates/overview.html index df4c89f..2aa2382 100644 --- a/visualpy/templates/overview.html +++ b/visualpy/templates/overview.html @@ -62,7 +62,7 @@

Dependency Graph

Scripts

{% if project.scripts %}
- {% for script in project.scripts %} + {% for script in sorted_scripts|default(project.scripts) %} {% include "partials/script_card.html" %} {% endfor %}
diff --git a/visualpy/templates/partials/script_card.html b/visualpy/templates/partials/script_card.html index 620dd1b..891a30f 100644 --- a/visualpy/templates/partials/script_card.html +++ b/visualpy/templates/partials/script_card.html @@ -11,9 +11,14 @@ class="block bg-white dark:bg-gray-800 rounded-lg shadow p-4 hover:shadow-md hover:ring-1 hover:ring-blue-300 dark:hover:ring-blue-600 transition">
{{ script.path.split('/')[-1] }} - {% if script.is_entry_point %} - entry - {% endif %} + + {% if key_paths is defined and script.path in key_paths %} + key + {% endif %} + {% if script.is_entry_point %} + entry + {% endif %} +
{% if '/' in script.path %}

{{ script.path }}

diff --git a/visualpy/templates/script.html b/visualpy/templates/script.html index 9158d0e..0631d7d 100644 --- a/visualpy/templates/script.html +++ b/visualpy/templates/script.html @@ -52,9 +52,20 @@

{{ script.path }}

-

Flow

+
+

Flow

+ {% if total_steps is defined and total_steps > 30 %} + + {% endif %} +

Click any step to see details

+ +
 {{ flow }}
@@ -175,5 +186,38 @@ 

Imports< htmx.ajax('GET', '/partials/step/' + encodeURIComponent(scriptPath) + '/' + line, {target: content, swap: 'innerHTML'}); document.getElementById('step-detail').scrollIntoView({behavior: 'smooth', block: 'nearest'}); } + +var _flowIsCompact = {{ 'true' if default_compact|default(false) else 'false' }}; + +async function switchFlow() { + if (!window.mermaidModule) { + console.warn('[visualpy] Mermaid not loaded yet'); + return; + } + + _flowIsCompact = !_flowIsCompact; + var btn = document.getElementById('flow-toggle'); + if (btn) btn.textContent = _flowIsCompact ? 'Show Detailed' : 'Show Compact'; + + var src = _flowIsCompact + ? document.getElementById('flow-compact') + : document.getElementById('flow-detailed'); + if (!src) return; + + var el = document.querySelector('.mermaid'); + if (!el) return; + + var code = src.textContent.trim(); + el.setAttribute('data-mermaid-src', code); + el.removeAttribute('data-processed'); + el.innerHTML = code; + + try { + await window.mermaidModule.run({nodes: [el]}); + } catch (err) { + console.error('[visualpy] Mermaid re-render failed:', err); + el.innerHTML = '

Diagram failed to render. Try refreshing the page.

'; + } +} {% endblock %} From c30e30647dfef4a5d3442636e8ba23ab9737dc65 Mon Sep 17 00:00:00 2001 From: Lexi-Energy <258596309+Lexi-Energy@users.noreply.github.com> Date: Wed, 1 Apr 2026 23:36:27 +0000 Subject: [PATCH 4/5] feat: --from-json flag, Docker demo at visualpy.lexi-energy.com, README update --- .dockerignore | 13 ++++++ CHANGELOG.md | 2 + Dockerfile | 19 ++++++++ README.md | 30 ++++++------ docker-compose.yml | 15 ++++++ tests/test_cli.py | 111 +++++++++++++++++++++++++++++++++++++++++++++ visualpy/cli.py | 96 ++++++++++++++++++++++++++++++++++----- 7 files changed, 260 insertions(+), 26 deletions(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 docker-compose.yml diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..3bf7ce9 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,13 @@ +.venv/ +__pycache__/ +*.pyc +.git/ +.github/ +*.egg-info/ +dist/ +build/ +docs/ +.claude/ +.env +tests/*.py +tests/__pycache__/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b49d2f..abdcda7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this ### Added +- `--from-json` flag for `serve` command — load pre-computed analysis from JSON file +- Dockerfile + docker-compose for public demo deployment (pre-baked LLM summaries) - Compact mode for script flow diagrams — functions with >8 steps collapse to summary nodes - Compact/Detailed toggle button on script pages (auto-shown for scripts with >30 steps) - Importance scoring for scripts — overview sorted by connectivity, entry points, services diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..f6b210f --- /dev/null +++ b/Dockerfile @@ -0,0 +1,19 @@ +FROM python:3.12-slim +WORKDIR /app + +COPY pyproject.toml README.md ./ +COPY visualpy/ visualpy/ +COPY static/ static/ +RUN pip install --no-cache-dir -e ".[llm]" + +COPY tests/fixtures/agentic_workflows/ /demo_project/ + +ARG GEMINI_API_KEY="" +RUN if [ -n "$GEMINI_API_KEY" ]; then \ + GEMINI_API_KEY=$GEMINI_API_KEY visualpy analyze /demo_project --summarize -o /demo_data.json; \ + else \ + visualpy analyze /demo_project -o /demo_data.json; \ + fi + +EXPOSE 8123 +CMD ["visualpy", "serve", "--from-json", "/demo_data.json", "--host", "0.0.0.0", "--port", "8123"] diff --git a/README.md b/README.md index 93792c3..efe7e43 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,8 @@ Auto-visualise Python automations for non-technical stakeholders. Drop a folder of Python scripts, get a visual breakdown of what they do, how they connect, and what they need. No execution required, no config needed. +**[Live demo](https://visualpy.lexi-energy.com)** — see it in action on a real 8-script lead generation pipeline. + ## Who is this for? - **Operations teams** who inherited a folder of automation scripts and need to understand what each one does before touching anything. @@ -23,6 +25,10 @@ pip install -e . visualpy analyze /path/to/your/scripts # JSON breakdown visualpy serve /path/to/your/scripts # starts a local web UI + +# Optional: add plain-English LLM summaries (needs an API key) +pip install -e ".[llm]" +visualpy serve /path/to/your/scripts --summarize ``` Requires Python 3.12 or later. No config files, no decorators in your code, no setup. Point it at a folder and go. @@ -42,6 +48,9 @@ The result is a structured project map, viewable as JSON or as an interactive we - **Project dependency graph** — see how scripts relate to each other at a glance - **Per-script flow diagrams** — step-by-step visual breakdown of what each file does, grouped by function +- **Compact mode** — large scripts (68+ steps) auto-collapse to readable summaries; toggle between compact and detailed views +- **LLM summaries** — optional plain-English descriptions powered by any LLM provider via litellm (BYOK). `--summarize` flag on both `analyze` and `serve` +- **Importance scoring** — scripts sorted by connectivity; most important scripts highlighted with a "key" badge - **Service and secret detection** — instantly see which external services and API keys are in play - **Entry point detection** — identifies scripts with `if __name__ == "__main__"`, cron triggers, webhooks, and CLI entry points - **Dark mode** — toggle between light and dark themes, persisted across sessions @@ -54,12 +63,15 @@ The result is a structured project map, viewable as JSON or as an interactive we | Sprint | Status | What | |--------|--------|------| | 0: Init | Done | Repo skeleton, models, CLI stubs, test fixtures | -| 1: The Engine | Done | Folder-to-JSON analysis pipeline, 59 tests | +| 1: The Engine | Done | Folder-to-JSON analysis pipeline | | 1.5: Hardening | Done | Transform detection, inputs/outputs enrichment, false positive fixes | -| 2: The Face | Done | Web UI with Mermaid.js graphs, dark mode, HTMX interactivity, 140 tests | +| 2: The Face | Done | Web UI with Mermaid.js graphs, dark mode, HTMX interactivity | | 3: The Community | Done | FOSS prep, docs, issue templates, CI | -| 4: The Voice | Planned | LLM summaries (litellm, BYOK), per-script descriptions, project executive summary | -| 5: The Feedback Loop | Planned | Annotations, human-in-the-loop corrections | +| 4: The Voice | Done | LLM summaries (litellm, BYOK), per-script and project-level descriptions | +| 5: The Scaling Fix | Done | Compact mode, importance scoring, compact/detailed toggle | +| 5.5: The Demo | Done | Docker deployment, pre-baked summaries, [live demo](https://visualpy.lexi-energy.com) | +| 6: The Translation | Next | LLM-powered step descriptions, business language UI | +| 7: The Export | Planned | Static HTML export, summary caching, markdown export | ## Contributing @@ -67,16 +79,6 @@ We'd love your help — whether it's a bug report, a feature idea, or a question See [CONTRIBUTING.md](CONTRIBUTING.md) for how to get started. We've written it specifically for people who might be new to open source. -## Acknowledgments - -visualpy stands on the shoulders of great open-source projects. We studied their patterns and approaches: - -- [pyflowchart](https://github.com/cdfmlr/pyflowchart) (MIT) — function subgraph grouping, AST-to-flowchart patterns -- [code2flow](https://github.com/scottrogowski/code2flow) (MIT) — entry point classification, directory hierarchy, graph organization -- [emerge](https://github.com/glato/emerge) (MIT) — dark mode toggle, data embedding strategy, module separation -- [staticfg](https://github.com/coetaur0/staticfg) (Apache-2.0) — AST visitor pattern for control flow -- [VizTracer](https://github.com/gaogaotiantian/viztracer) (Apache-2.0) — zero-config philosophy (no decorators, no code changes) - ## License [MIT](LICENSE) diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..220fbd7 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,15 @@ +services: + visualpy: + build: + context: . + args: + GEMINI_API_KEY: ${GEMINI_API_KEY} + container_name: visualpy + restart: always + networks: + - caddy_net + +networks: + caddy_net: + external: true + name: caddy_net diff --git a/tests/test_cli.py b/tests/test_cli.py index aea3f86..4c099d7 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,9 +1,22 @@ """Tests for CLI integration.""" +import dataclasses import json import subprocess import sys +import pytest + +from visualpy.cli import _project_from_dict +from visualpy.models import ( + AnalyzedProject, + AnalyzedScript, + ScriptConnection, + Service, + Step, + Trigger, +) + def test_analyze_hello(hello_script): result = subprocess.run( @@ -81,3 +94,101 @@ def test_version(): ) assert result.returncode == 0 assert "0.1.0" in result.stdout + + +# --- --from-json --- + + +def test_project_from_dict_roundtrip(): + """asdict → _project_from_dict should reconstruct equivalent project.""" + project = AnalyzedProject( + path="/tmp/test", + scripts=[ + AnalyzedScript( + path="example.py", + is_entry_point=True, + steps=[ + Step( + line_number=10, + type="api_call", + description="requests.get()", + function_name="fetch", + service=Service(name="HTTP", library="requests"), + inputs=["url"], + outputs=["response"], + ), + ], + services=[Service(name="HTTP", library="requests")], + secrets=["API_KEY"], + triggers=[Trigger(type="cli", detail="__main__")], + signature={"url": "str"}, + summary="Fetches data.", + ), + ], + connections=[ + ScriptConnection(source="a.py", target="b.py", type="import", detail="a→b"), + ], + services=[Service(name="HTTP", library="requests")], + secrets=["API_KEY"], + entry_points=["example.py"], + summary="A test project.", + ) + data = dataclasses.asdict(project) + restored = _project_from_dict(data) + + assert restored.path == project.path + assert restored.summary == project.summary + assert len(restored.scripts) == 1 + assert restored.scripts[0].path == "example.py" + assert restored.scripts[0].is_entry_point is True + assert restored.scripts[0].steps[0].line_number == 10 + assert restored.scripts[0].steps[0].service.name == "HTTP" + assert restored.scripts[0].steps[0].inputs == ["url"] + assert restored.scripts[0].triggers[0].type == "cli" + assert restored.scripts[0].summary == "Fetches data." + assert len(restored.connections) == 1 + assert restored.connections[0].source == "a.py" + assert restored.entry_points == ["example.py"] + + +def test_from_json_roundtrip_via_cli(hello_script, tmp_path): + """analyze → JSON file → serve --from-json should accept the file.""" + json_file = tmp_path / "analysis.json" + # Generate JSON + result = subprocess.run( + [sys.executable, "-m", "visualpy", "analyze", str(hello_script), "-o", str(json_file)], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + assert json_file.exists() + + # Load and verify it reconstructs + data = json.loads(json_file.read_text()) + project = _project_from_dict(data) + assert len(project.scripts) == 1 + assert project.scripts[0].path == "hello.py" + + +def test_from_json_missing_file(): + """--from-json with nonexistent file should fail cleanly.""" + result = subprocess.run( + [sys.executable, "-m", "visualpy", "serve", "--from-json", "/does/not/exist.json"], + capture_output=True, + text=True, + timeout=10, + ) + assert result.returncode != 0 + assert "Error" in result.stderr + + +def test_serve_requires_path_or_from_json(): + """serve with neither path nor --from-json should fail.""" + result = subprocess.run( + [sys.executable, "-m", "visualpy", "serve"], + capture_output=True, + text=True, + timeout=10, + ) + assert result.returncode != 0 + assert "Error" in result.stderr or "required" in result.stderr.lower() diff --git a/visualpy/cli.py b/visualpy/cli.py index 53e356f..fbcd3c9 100644 --- a/visualpy/cli.py +++ b/visualpy/cli.py @@ -9,7 +9,14 @@ from visualpy import __version__ from visualpy.analyzer.cross_file import resolve_connections from visualpy.analyzer.scanner import scan_project -from visualpy.models import AnalyzedProject +from visualpy.models import ( + AnalyzedProject, + AnalyzedScript, + ScriptConnection, + Service, + Step, + Trigger, +) def app(): @@ -39,9 +46,14 @@ def app(): serve_parser = subparsers.add_parser( "serve", help="Start web UI for visual exploration" ) - serve_parser.add_argument("path", help="Path to folder to analyze") + serve_parser.add_argument("path", nargs="?", default=None, help="Path to folder to analyze") serve_parser.add_argument("--port", type=int, default=8000, help="Port (default: 8000)") serve_parser.add_argument("--host", default="127.0.0.1", help="Host (default: 127.0.0.1)") + serve_parser.add_argument( + "--from-json", + dest="from_json", + help="Load pre-computed analysis from JSON file instead of scanning", + ) serve_parser.add_argument( "--summarize", action="store_true", @@ -138,24 +150,84 @@ def _run_analyze(args: argparse.Namespace) -> None: print(output) +def _project_from_dict(data: dict) -> AnalyzedProject: + """Reconstruct an AnalyzedProject from a dict (inverse of dataclasses.asdict).""" + scripts = [] + for s in data.get("scripts", []): + steps = [ + Step( + line_number=st["line_number"], + type=st["type"], + description=st["description"], + function_name=st.get("function_name"), + service=Service(**st["service"]) if st.get("service") else None, + inputs=st.get("inputs", []), + outputs=st.get("outputs", []), + ) + for st in s.get("steps", []) + ] + scripts.append( + AnalyzedScript( + path=s["path"], + is_entry_point=s.get("is_entry_point", False), + steps=steps, + imports_internal=s.get("imports_internal", []), + imports_external=s.get("imports_external", []), + services=[Service(**svc) for svc in s.get("services", [])], + secrets=s.get("secrets", []), + triggers=[Trigger(**t) for t in s.get("triggers", [])], + signature=s.get("signature"), + summary=s.get("summary"), + ) + ) + return AnalyzedProject( + path=data["path"], + scripts=scripts, + connections=[ScriptConnection(**c) for c in data.get("connections", [])], + services=[Service(**svc) for svc in data.get("services", [])], + secrets=data.get("secrets", []), + entry_points=data.get("entry_points", []), + summary=data.get("summary"), + ) + + def _run_serve(args: argparse.Namespace) -> None: """Analyze the target and start the web UI.""" import uvicorn from visualpy.server import create_app - target = Path(args.path).resolve() - - if not target.exists(): - print(f"[visualpy] Error: path does not exist: {args.path}", file=sys.stderr) + if args.from_json: + json_path = Path(args.from_json) + if not json_path.exists(): + print(f"[visualpy] Error: JSON file not found: {args.from_json}", file=sys.stderr) + sys.exit(1) + try: + data = json.loads(json_path.read_text()) + except (json.JSONDecodeError, OSError) as exc: + print(f"[visualpy] Error: could not read JSON file: {exc}", file=sys.stderr) + sys.exit(1) + project = _project_from_dict(data) + print(f"[visualpy] Loaded analysis from {args.from_json}", file=sys.stderr) + if args.summarize: + print( + "[visualpy] Warning: --summarize ignored with --from-json " + "(generate summaries during 'analyze' instead)", + file=sys.stderr, + ) + elif args.path: + target = Path(args.path).resolve() + if not target.exists(): + print(f"[visualpy] Error: path does not exist: {args.path}", file=sys.stderr) + sys.exit(1) + print(f"[visualpy] Analyzing {target}...", file=sys.stderr) + project = _build_project(target) + if args.summarize: + _summarize_project(project) + else: + print("[visualpy] Error: either path or --from-json is required", file=sys.stderr) sys.exit(1) - print(f"[visualpy] Analyzing {target}...", file=sys.stderr) - project = _build_project(target) - - if args.summarize: - _summarize_project(project) - print( f"[visualpy] Found {len(project.scripts)} scripts, " f"{len(project.connections)} connections", From 95b00b57083c4a8b9682a678160a0f94cadfd774 Mon Sep 17 00:00:00 2001 From: Lexi-Energy <258596309+Lexi-Energy@users.noreply.github.com> Date: Wed, 1 Apr 2026 23:37:19 +0000 Subject: [PATCH 5/5] fix: error handling for malformed --from-json, warn on ignored args --- README.md | 2 +- tests/test_cli.py | 2 -- visualpy/cli.py | 15 ++++++++++++++- 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index efe7e43..1d645fc 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,7 @@ The result is a structured project map, viewable as JSON or as an interactive we - **Project dependency graph** — see how scripts relate to each other at a glance - **Per-script flow diagrams** — step-by-step visual breakdown of what each file does, grouped by function -- **Compact mode** — large scripts (68+ steps) auto-collapse to readable summaries; toggle between compact and detailed views +- **Compact mode** — functions with many steps auto-collapse to readable summaries; toggle between compact and detailed views - **LLM summaries** — optional plain-English descriptions powered by any LLM provider via litellm (BYOK). `--summarize` flag on both `analyze` and `serve` - **Importance scoring** — scripts sorted by connectivity; most important scripts highlighted with a "key" badge - **Service and secret detection** — instantly see which external services and API keys are in play diff --git a/tests/test_cli.py b/tests/test_cli.py index 4c099d7..4e384af 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -5,8 +5,6 @@ import subprocess import sys -import pytest - from visualpy.cli import _project_from_dict from visualpy.models import ( AnalyzedProject, diff --git a/visualpy/cli.py b/visualpy/cli.py index fbcd3c9..2e82987 100644 --- a/visualpy/cli.py +++ b/visualpy/cli.py @@ -198,6 +198,11 @@ def _run_serve(args: argparse.Namespace) -> None: from visualpy.server import create_app if args.from_json: + if args.path: + print( + f"[visualpy] Warning: --from-json takes precedence; ignoring path '{args.path}'", + file=sys.stderr, + ) json_path = Path(args.from_json) if not json_path.exists(): print(f"[visualpy] Error: JSON file not found: {args.from_json}", file=sys.stderr) @@ -207,7 +212,15 @@ def _run_serve(args: argparse.Namespace) -> None: except (json.JSONDecodeError, OSError) as exc: print(f"[visualpy] Error: could not read JSON file: {exc}", file=sys.stderr) sys.exit(1) - project = _project_from_dict(data) + try: + project = _project_from_dict(data) + except (KeyError, TypeError, AttributeError) as exc: + print( + f"[visualpy] Error: invalid JSON structure: {exc}\n" + f"[visualpy] Hint: use 'visualpy analyze ... -o file.json' to generate a valid file", + file=sys.stderr, + ) + sys.exit(1) print(f"[visualpy] Loaded analysis from {args.from_json}", file=sys.stderr) if args.summarize: print(