bruno/dss
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(analysis): Implement project analysis engine and CI/CD workflow

Browse Source 
This commit introduces a new project analysis engine to the DSS.

Key features include:
- A new analysis module in `dss-mvp1/dss/analyze` that can parse React projects and generate a dependency graph.
- A command-line interface (`dss-mvp1/dss-cli.py`) to run the analysis, designed for use in CI/CD pipelines.
- A new `dss_project_export_context` tool in the Claude MCP server to allow AI agents to access the analysis results.
- A `.gitlab-ci.yml` file to automate the analysis on every push, ensuring the project context is always up-to-date.
- Tests for the new analysis functionality.

This new architecture enables DSS to have a deep, version-controlled understanding of a project's structure, which can be used to power more intelligent agents and provide better developer guidance. The analysis is no longer automatically triggered on `init`, but is designed to be run manually or by a CI/CD pipeline.
This commit is contained in:
Digital Production Factory
2025-12-10 11:05:27 -03:00
parent 842cce133c
commit d53b61008c
15 changed files with 952 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
tools/analysis/project_analyzer.py Normal file
View File

@@ -0,0 +1,85 @@
import os
import json
import networkx as nx
from pyast_ts import parse
import cssutils
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
# Configure cssutils to ignore noisy error messages
cssutils.log.setLevel(logging.CRITICAL)
def analyze_react_project(project_path: str) -> dict:
"""
Analyzes a React project, building a graph of its components and styles.
Args:
project_path: The root path of the React project.
Returns:
A dictionary containing the component graph and analysis report.
"""
log.info(f"Starting analysis of project at: {project_path}")
graph = nx.DiGraph()
# Supported extensions for react/js/ts files
supported_exts = ('.js', '.jsx', '.ts', '.tsx')
for root, _, files in os.walk(project_path):
for file in files:
file_path = os.path.join(root, file)
relative_path = os.path.relpath(file_path, project_path)
if file.endswith(supported_exts):
graph.add_node(relative_path, type='file', language='typescript')
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# Placeholder for AST parsing and analysis
# ast = parse(content)
# For now, we'll just add the node
except Exception as e:
log.error(f"Could not process file {file_path}: {e}")
elif file.endswith('.css'):
graph.add_node(relative_path, type='file', language='css')
try:
# Placeholder for CSS parsing
# sheet = cssutils.parseFile(file_path)
pass
except Exception as e:
log.error(f"Could not parse css file {file_path}: {e}")
log.info(f"Analysis complete. Found {graph.number_of_nodes()} files.")
# Convert graph to a serializable format
serializable_graph = nx.node_link_data(graph)
return serializable_graph
def save_analysis(project_path: str, analysis_data: dict):
"""
Saves the analysis data to a file in the project's .dss directory.
"""
dss_dir = os.path.join(project_path, '.dss')
os.makedirs(dss_dir, exist_ok=True)
output_path = os.path.join(dss_dir, 'analysis_graph.json')
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(analysis_data, f, indent=2)
log.info(f"Analysis data saved to {output_path}")
if __name__ == '__main__':
# Example usage:
# Replace '.' with the actual path to a React project for testing.
# In a real scenario, this would be called by the MCP.
target_project_path = '.'
analysis_result = analyze_react_project(target_project_path)
save_analysis(target_project_path, analysis_result)

25
tools/dss_mcp/handler.py
View File

@@ -26,6 +26,7 @@ from storage.json_store import Projects, ActivityLog
from .config import mcp_config, integration_config
from .context.project_context import get_context_manager, ProjectContext
from .tools.project_tools import PROJECT_TOOLS, ProjectTools
from .tools.analysis_tools import ANALYSIS_TOOLS, AnalysisTools
from .integrations.figma import FIGMA_TOOLS, FigmaTools
from .integrations.storybook import STORYBOOK_TOOLS, StorybookTools
from .integrations.jira import JIRA_TOOLS, JiraTools
@@ -86,6 +87,14 @@ class MCPHandler:
"requires_integration": False
}
# Register analysis tools
for tool in ANALYSIS_TOOLS:
self._tool_registry[tool.name] = {
"tool": tool,
"category": "analysis",
"requires_integration": False
}
# Register Figma tools
for tool in FIGMA_TOOLS:
self._tool_registry[tool.name] = {
@@ -212,6 +221,8 @@ class MCPHandler:
# Execute based on category
if category == "project":
result = await self._execute_project_tool(tool_name, arguments, context)
elif category == "analysis":
result = await self._execute_analysis_tool(tool_name, arguments, context)
elif category == "figma":
result = await self._execute_figma_tool(tool_name, arguments, context)
elif category == "storybook":
@@ -293,6 +304,20 @@ class MCPHandler:
project_tools = ProjectTools(context.user_id)
return await project_tools.execute_tool(tool_name, arguments)
async def _execute_analysis_tool(
self,
tool_name: str,
arguments: Dict[str, Any],
context: MCPContext
) -> Dict[str, Any]:
"""Execute an analysis tool"""
# Ensure project_id is set for context if needed, though project_path is explicit
if "project_id" not in arguments:
arguments["project_id"] = context.project_id
analysis_tools = AnalysisTools(context.user_id)
return await analysis_tools.execute_tool(tool_name, arguments)
async def _execute_figma_tool(
self,
tool_name: str,

82
tools/dss_mcp/tools/analysis_tools.py Normal file
View File

@@ -0,0 +1,82 @@
"""
DSS MCP - Code Analysis Tools
"""
import asyncio
from typing import Dict, Any
# Adjust the import path to find the project_analyzer
# This assumes the script is run from the project root.
from tools.analysis.project_analyzer import analyze_react_project, save_analysis
class Tool:
"""Basic tool definition for MCP"""
def __init__(self, name: str, description: str, input_schema: Dict[str, Any]):
self.name = name
self.description = description
self.inputSchema = input_schema
# Define the new tool
analyze_project_tool = Tool(
name="analyze_project",
description="Analyzes a given project's structure, components, and styles. This is a long-running operation.",
input_schema={
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "The absolute path to the project to be analyzed."
}
},
"required": ["project_path"]
}
)
class AnalysisTools:
"""
A wrapper class for analysis-related tools.
"""
def __init__(self, user_id: str = None):
self.user_id = user_id
async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
if tool_name == "analyze_project":
return await self.analyze_project(arguments.get("project_path"))
else:
return {"error": f"Analysis tool '{tool_name}' not found."}
async def analyze_project(self, project_path: str) -> Dict[str, Any]:
"""
Triggers the analysis of a project.
"""
if not project_path:
return {"error": "project_path is a required argument."}
try:
# This is a potentially long-running task.
# In a real scenario, this should be offloaded to a background worker.
# For now, we run it asynchronously.
loop = asyncio.get_event_loop()
# Run the analysis in a separate thread to avoid blocking the event loop
analysis_data = await loop.run_in_executor(
None, analyze_react_project, project_path
)
# Save the analysis data
await loop.run_in_executor(
None, save_analysis, project_path, analysis_data
)
return {
"status": "success",
"message": f"Analysis complete for project at {project_path}.",
"graph_nodes": len(analysis_data.get("nodes", [])),
"graph_edges": len(analysis_data.get("links", []))
}
except Exception as e:
return {"error": f"An error occurred during project analysis: {str(e)}"}
# A list of all tools in this module
ANALYSIS_TOOLS = [
analyze_project_tool
]

63
tools/dss_mcp/tools/project_tools.py
View File

@@ -21,6 +21,7 @@ from ..context.project_context import get_context_manager
from ..security import CredentialVault
from ..audit import AuditLog, AuditEventType
from storage.json_store import Projects, Components, Tokens, ActivityLog # JSON storage
from ..handler import get_mcp_handler, MCPContext
# Tool definitions (metadata for Claude)
@@ -168,7 +169,7 @@ PROJECT_TOOLS = [
},
"root_path": {
"type": "string",
"description": "Root directory path for the project"
"description": "Root directory path for the project. Can be a git URL or a local folder path."
}
},
"required": ["name", "root_path"]
@@ -457,22 +458,28 @@ class ProjectTools:
def __init__(self, user_id: Optional[int] = None):
self.context_manager = get_context_manager()
self.user_id = user_id
self.projects_db = Projects()
async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute a tool by name"""
handlers = {
# Project Management
"dss_create_project": self.create_project,
"dss_list_projects": self.list_projects,
"dss_get_project": self.get_project,
# Read-only tools
"dss_get_project_summary": self.get_project_summary,
"dss_list_components": self.list_components,
"dss_get_component": self.get_component,
"dss_get_design_tokens": self.get_design_tokens,
"dss_get_project_health": self.get_project_health,
"dss_list_styles": self.list_styles,
"dss_get_discovery_data": self.get_discovery_data
"dss_get_discovery_.dat": self.get_discovery_data
}
handler = handlers.get(tool_name)
if not handler:
return {"error": f"Unknown tool: {tool_name}"}
return {"error": f"Unknown or not implemented tool: {tool_name}"}
try:
result = await handler(**arguments)
@@ -480,6 +487,56 @@ class ProjectTools:
except Exception as e:
return {"error": str(e)}
async def create_project(self, name: str, root_path: str, description: str = "") -> Dict[str, Any]:
"""Create a new project and trigger initial analysis."""
project_id = str(uuid.uuid4())
# The `create` method in json_store handles the creation of the manifest
self.projects_db.create(
id=project_id,
name=name,
description=description
)
# We may still want to update the root_path if it's not part of the manifest
self.projects_db.update(project_id, root_path=root_path)
# Trigger the analysis as a background task
# We don't want to block the creation call
mcp_handler = get_mcp_handler()
# Create a context for the tool call
# The user_id might be important for permissions later
mcp_context = MCPContext(project_id=project_id, user_id=self.user_id)
# It's better to run this in the background and not wait for the result here
asyncio.create_task(
mcp_handler.execute_tool(
tool_name="analyze_project",
arguments={"project_path": root_path},
context=mcp_context
)
)
return {
"status": "success",
"message": "Project created successfully. Analysis has been started in the background.",
"project_id": project_id
}
async def list_projects(self, filter_status: Optional[str] = None) -> Dict[str, Any]:
"""List all projects."""
all_projects = self.projects_db.list(status=filter_status)
return {"projects": all_projects}
async def get_project(self, project_id: str) -> Dict[str, Any]:
"""Get a single project by its ID."""
project = self.projects_db.get(project_id)
if not project:
return {"error": f"Project with ID '{project_id}' not found."}
return {"project": project}
async def get_project_summary(
self,
project_id: str,