392bcfa2ef
Major enhancements to Clover CLI: ✨ New Features: - AI agent with multi-turn conversation capabilities - Tool calling system with 11+ tools for file operations, Git, linting, etc. - Step-by-step AI assistance with play-by-play commentary - Enhanced interactive mode with better UX 🔧 Core Components Added: - ai_agent.py: Main AI agent with conversation management - models/: API client and model management system - Comprehensive tool system for development tasks 🛠️ Tools Available: - File operations (create, read, update, delete) - Command execution with safety checks - Git operations (status, diff, commit, push) - Code linting and formatting - Project structure analysis - Security scanning and dependency management 💡 User Experience: - Real-time tool execution summaries - File creation with full path visibility - Error handling and retry mechanisms - Clean conversation flow until task completion 🧹 Repository Cleanup: - Added comprehensive .gitignore - Removed __pycache__ directories and build artifacts - Organized project structure The AI can now actually create files, run commands, and work through complex development tasks step-by-step with full transparency.
619 lines
19 KiB
Python
619 lines
19 KiB
Python
"""
|
|
Project operation tools for Clover - A terminal assistant for AI-powered project management
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
# Add the current directory to Python path for imports
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from config.settings import load_config
|
|
from models.api_client import APIClient
|
|
from tools.file_tools import create_file, list_files, read_file
|
|
|
|
|
|
class ProjectSummarizer:
|
|
"""Handle project summarization operations with LLM integration"""
|
|
|
|
def __init__(self):
|
|
self.config = load_config()
|
|
self.api_client = APIClient()
|
|
self.max_threads = self.config.get("threads", 5)
|
|
self.timeout = self.config.get("timeout", 300)
|
|
|
|
def _call_llm_for_summary(self, content: str, filepath: str) -> str:
|
|
"""
|
|
Call LLM to generate a summary of file content
|
|
|
|
Args:
|
|
content (str): File content to summarize
|
|
filepath (str): Path of the file being summarized
|
|
|
|
Returns:
|
|
str: Generated summary
|
|
"""
|
|
try:
|
|
# Prepare prompt for file summarization
|
|
prompt = f"""
|
|
Please provide a concise summary of the following file ({filepath}):
|
|
|
|
```
|
|
{content}
|
|
```
|
|
|
|
Focus on:
|
|
- Main purpose and functionality
|
|
- Key components, classes, or functions
|
|
- Important dependencies or imports
|
|
- Overall role in the project
|
|
|
|
Keep the summary under 200 words and make it useful for understanding the project structure.
|
|
"""
|
|
|
|
# Call the LLM API
|
|
response = self.api_client.generate_text(
|
|
prompt=prompt, model=self.config.get("model", "qwen2.5-coder:7b")
|
|
)
|
|
|
|
if "error" in response:
|
|
return f"Error generating summary for {filepath}: {response['error']}"
|
|
|
|
# Extract the response content
|
|
if "choices" in response and len(response["choices"]) > 0:
|
|
return response["choices"][0]["message"]["content"].strip()
|
|
else:
|
|
return (
|
|
f"Generated summary for {filepath}: Basic file analysis completed."
|
|
)
|
|
|
|
except Exception as e:
|
|
return f"Error summarizing {filepath}: {str(e)}"
|
|
|
|
def _generate_project_structure_with_llm(self, file_list: List[str]) -> str:
|
|
"""
|
|
Generate project structure using LLM analysis
|
|
|
|
Args:
|
|
file_list (List[str]): List of files in the project
|
|
|
|
Returns:
|
|
str: Generated project structure description
|
|
"""
|
|
try:
|
|
# Create a formatted file list
|
|
file_tree = "\n".join([f"- {f}" for f in sorted(file_list)])
|
|
|
|
prompt = f"""
|
|
Analyze the following project file structure and create a comprehensive project structure document:
|
|
|
|
Files in the project:
|
|
{file_tree}
|
|
|
|
Please provide:
|
|
1. A brief description of what this project appears to be
|
|
2. Key directories and their purposes
|
|
3. Main entry points or important files
|
|
4. Technology stack based on file extensions
|
|
5. Project organization patterns
|
|
|
|
Format as a structured markdown document.
|
|
"""
|
|
|
|
response = self.api_client.generate_text(
|
|
prompt=prompt, model=self.config.get("model", "qwen2.5-coder:7b")
|
|
)
|
|
|
|
if "error" in response:
|
|
return f"# Project Structure\n\nError generating structure: {response['error']}"
|
|
|
|
if "choices" in response and len(response["choices"]) > 0:
|
|
return response["choices"][0]["message"]["content"].strip()
|
|
else:
|
|
return "# Project Structure\n\nBasic project analysis completed."
|
|
|
|
except Exception as e:
|
|
return f"# Project Structure\n\nError analyzing project: {str(e)}"
|
|
|
|
|
|
def summarize_file(filepath: str) -> Dict[str, Any]:
|
|
"""
|
|
Use LLM to generate a summary of a specific file
|
|
|
|
Args:
|
|
filepath (str): Path to the file to summarize
|
|
|
|
Returns:
|
|
Dict containing summary and metadata
|
|
"""
|
|
try:
|
|
# Check if file exists
|
|
if not os.path.exists(filepath):
|
|
return {
|
|
"error": f"File {filepath} does not exist",
|
|
"filepath": filepath,
|
|
"summary": None,
|
|
}
|
|
|
|
# Read file content
|
|
try:
|
|
content = read_file(filepath)
|
|
except Exception as e:
|
|
return {
|
|
"error": f"Could not read file {filepath}: {str(e)}",
|
|
"filepath": filepath,
|
|
"summary": None,
|
|
}
|
|
|
|
# Check file size - avoid very large files
|
|
if len(content) > 50000: # 50KB limit
|
|
content = content[:50000] + "\n... [File truncated for analysis]"
|
|
|
|
# Initialize summarizer and generate summary
|
|
summarizer = ProjectSummarizer()
|
|
summary = summarizer._call_llm_for_summary(content, filepath)
|
|
|
|
return {
|
|
"filepath": filepath,
|
|
"summary": summary,
|
|
"file_size": len(content),
|
|
"lines": len(content.splitlines()) if content else 0,
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"error": f"Error summarizing file {filepath}: {str(e)}",
|
|
"filepath": filepath,
|
|
"summary": None,
|
|
}
|
|
|
|
|
|
def get_project_structure(project_path: str = ".") -> Dict[str, Any]:
|
|
"""
|
|
Look for structure.md file or generate it using LLM
|
|
|
|
Args:
|
|
project_path (str): Path to the project directory
|
|
|
|
Returns:
|
|
Dict containing project structure information
|
|
"""
|
|
try:
|
|
structure_file = os.path.join(project_path, "structure.md")
|
|
|
|
# Check if structure.md already exists
|
|
if os.path.exists(structure_file):
|
|
try:
|
|
existing_content = read_file(structure_file)
|
|
return {
|
|
"structure": existing_content,
|
|
"source": "existing_file",
|
|
"file_path": structure_file,
|
|
}
|
|
except Exception as e:
|
|
print(f"Warning: Could not read existing structure.md: {e}")
|
|
|
|
# Generate new structure using LLM
|
|
print("Generating project structure using AI...")
|
|
|
|
# Get list of all files in project
|
|
all_files = []
|
|
ignore_dirs = {
|
|
".git",
|
|
"__pycache__",
|
|
"node_modules",
|
|
".pytest_cache",
|
|
"venv",
|
|
"env",
|
|
"clover_env",
|
|
}
|
|
ignore_files = {".DS_Store", ".gitignore", ".pyc"}
|
|
|
|
for root, dirs, files in os.walk(project_path):
|
|
# Filter out ignored directories
|
|
dirs[:] = [d for d in dirs if d not in ignore_dirs]
|
|
|
|
for file in files:
|
|
if not any(file.endswith(ext) for ext in ignore_files):
|
|
rel_path = os.path.relpath(os.path.join(root, file), project_path)
|
|
all_files.append(rel_path)
|
|
|
|
# Generate structure using LLM
|
|
summarizer = ProjectSummarizer()
|
|
structure_content = summarizer._generate_project_structure_with_llm(all_files)
|
|
|
|
# Save the generated structure
|
|
try:
|
|
create_file(structure_file, structure_content)
|
|
print(f"Created structure.md with AI-generated project analysis")
|
|
except Exception as e:
|
|
print(f"Warning: Could not save structure.md: {e}")
|
|
|
|
return {
|
|
"structure": structure_content,
|
|
"source": "generated",
|
|
"file_path": structure_file,
|
|
"files_analyzed": len(all_files),
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"error": f"Error getting project structure: {str(e)}",
|
|
"structure": None,
|
|
"source": "error",
|
|
}
|
|
|
|
|
|
def aggregate_summaries(summaries: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""
|
|
Collect summaries from all files and create a combined project summary
|
|
|
|
Args:
|
|
summaries (List[Dict]): List of file summaries
|
|
|
|
Returns:
|
|
Dict containing aggregated project summary
|
|
"""
|
|
try:
|
|
if not summaries:
|
|
return {"error": "No summaries provided", "aggregated_summary": None}
|
|
|
|
# Filter out summaries with errors
|
|
valid_summaries = [
|
|
s for s in summaries if "error" not in s and s.get("summary")
|
|
]
|
|
|
|
if not valid_summaries:
|
|
return {
|
|
"error": "No valid summaries to aggregate",
|
|
"aggregated_summary": None,
|
|
}
|
|
|
|
# Prepare content for LLM aggregation
|
|
summary_text = ""
|
|
for i, summary_data in enumerate(valid_summaries, 1):
|
|
filepath = summary_data.get("filepath", "unknown")
|
|
summary = summary_data.get("summary", "No summary available")
|
|
summary_text += f"\n{i}. File: {filepath}\n Summary: {summary}\n"
|
|
|
|
# Use LLM to create aggregated summary
|
|
summarizer = ProjectSummarizer()
|
|
prompt = f"""
|
|
Based on the following individual file summaries, create a comprehensive project overview:
|
|
|
|
{summary_text}
|
|
|
|
Please provide:
|
|
1. Overall project purpose and functionality
|
|
2. Main components and architecture
|
|
3. Key technologies and dependencies
|
|
4. Project organization and structure
|
|
5. Notable features or capabilities
|
|
|
|
Keep it concise but comprehensive (under 500 words).
|
|
"""
|
|
|
|
response = summarizer.api_client.generate_text(
|
|
prompt=prompt, model=summarizer.config.get("model", "qwen2.5-coder:7b")
|
|
)
|
|
|
|
if "error" in response:
|
|
return {
|
|
"error": f"Error generating aggregated summary: {response['error']}",
|
|
"aggregated_summary": None,
|
|
"files_processed": len(valid_summaries),
|
|
}
|
|
|
|
if "choices" in response and len(response["choices"]) > 0:
|
|
aggregated_content = response["choices"][0]["message"]["content"].strip()
|
|
else:
|
|
aggregated_content = "Project summary aggregation completed."
|
|
|
|
return {
|
|
"aggregated_summary": aggregated_content,
|
|
"files_processed": len(valid_summaries),
|
|
"total_files": len(summaries),
|
|
"failed_files": len(summaries) - len(valid_summaries),
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"error": f"Error aggregating summaries: {str(e)}",
|
|
"aggregated_summary": None,
|
|
}
|
|
|
|
|
|
def incremental_summarization(
|
|
changed_files: List[str], project_path: str = "."
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Re-summarize only changed files to save tokens and time
|
|
|
|
Args:
|
|
changed_files (List[str]): List of changed files to summarize
|
|
project_path (str): Path to the project directory
|
|
|
|
Returns:
|
|
Dict containing incremental summary results
|
|
"""
|
|
try:
|
|
if not changed_files:
|
|
return {
|
|
"message": "No changed files to process",
|
|
"summaries": [],
|
|
"files_processed": 0,
|
|
}
|
|
|
|
results = []
|
|
successful = 0
|
|
failed = 0
|
|
|
|
# Process each changed file
|
|
for filepath in changed_files:
|
|
full_path = (
|
|
os.path.join(project_path, filepath)
|
|
if not os.path.isabs(filepath)
|
|
else filepath
|
|
)
|
|
|
|
print(f"Summarizing changed file: {filepath}")
|
|
summary_result = summarize_file(full_path)
|
|
|
|
if "error" not in summary_result:
|
|
successful += 1
|
|
else:
|
|
failed += 1
|
|
|
|
results.append(summary_result)
|
|
|
|
# Create summary of changes
|
|
change_summary = f"Processed {len(changed_files)} changed files. {successful} successful, {failed} failed."
|
|
|
|
return {
|
|
"summary": change_summary,
|
|
"changed_files": changed_files,
|
|
"summaries": results,
|
|
"files_processed": successful,
|
|
"files_failed": failed,
|
|
"total_files": len(changed_files),
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"error": f"Error in incremental summarization: {str(e)}",
|
|
"changed_files": changed_files,
|
|
"summaries": [],
|
|
}
|
|
|
|
|
|
def summarize_entire_project(
|
|
project_path: str = ".", max_workers: int = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Summarize all files in a project using concurrent processing
|
|
|
|
Args:
|
|
project_path (str): Path to the project directory
|
|
max_workers (int): Maximum number of concurrent threads
|
|
|
|
Returns:
|
|
Dict containing complete project summary
|
|
"""
|
|
try:
|
|
config = load_config()
|
|
if max_workers is None:
|
|
max_workers = config.get("threads", 5)
|
|
|
|
# Get all relevant files in the project
|
|
all_files = []
|
|
ignore_dirs = {
|
|
".git",
|
|
"__pycache__",
|
|
"node_modules",
|
|
".pytest_cache",
|
|
"venv",
|
|
"env",
|
|
"clover_env",
|
|
}
|
|
ignore_extensions = {".pyc", ".pyo", ".pyd", ".so", ".dll", ".exe"}
|
|
text_extensions = {
|
|
".py",
|
|
".js",
|
|
".ts",
|
|
".html",
|
|
".css",
|
|
".md",
|
|
".txt",
|
|
".json",
|
|
".yml",
|
|
".yaml",
|
|
".xml",
|
|
".sql",
|
|
}
|
|
|
|
for root, dirs, files in os.walk(project_path):
|
|
# Filter out ignored directories
|
|
dirs[:] = [d for d in dirs if d not in ignore_dirs]
|
|
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
|
|
# Skip ignored file types
|
|
if any(file.endswith(ext) for ext in ignore_extensions):
|
|
continue
|
|
|
|
# Only process text files or known code files
|
|
if (
|
|
any(file.endswith(ext) for ext in text_extensions)
|
|
or "." not in file
|
|
):
|
|
all_files.append(file_path)
|
|
|
|
if not all_files:
|
|
return {
|
|
"error": "No suitable files found to summarize",
|
|
"project_summary": None,
|
|
}
|
|
|
|
print(
|
|
f"Found {len(all_files)} files to summarize using {max_workers} threads..."
|
|
)
|
|
|
|
# Process files concurrently
|
|
summaries = []
|
|
successful = 0
|
|
failed = 0
|
|
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
# Submit all summarization tasks
|
|
future_to_file = {
|
|
executor.submit(summarize_file, filepath): filepath
|
|
for filepath in all_files
|
|
}
|
|
|
|
# Collect results as they complete
|
|
for future in as_completed(future_to_file):
|
|
filepath = future_to_file[future]
|
|
try:
|
|
result = future.result()
|
|
summaries.append(result)
|
|
|
|
if "error" not in result:
|
|
successful += 1
|
|
print(f"✓ Summarized: {filepath}")
|
|
else:
|
|
failed += 1
|
|
print(
|
|
f"✗ Failed: {filepath} - {result.get('error', 'Unknown error')}"
|
|
)
|
|
|
|
except Exception as e:
|
|
failed += 1
|
|
print(f"✗ Exception processing {filepath}: {str(e)}")
|
|
summaries.append(
|
|
{"filepath": filepath, "error": str(e), "summary": None}
|
|
)
|
|
|
|
print(f"Completed file summarization: {successful} successful, {failed} failed")
|
|
|
|
# Aggregate all summaries
|
|
print("Creating aggregated project summary...")
|
|
aggregation_result = aggregate_summaries(summaries)
|
|
|
|
# Get or generate project structure
|
|
structure_result = get_project_structure(project_path)
|
|
|
|
# Compile final project summary
|
|
final_summary = {
|
|
"project_path": project_path,
|
|
"files_analyzed": len(all_files),
|
|
"files_successful": successful,
|
|
"files_failed": failed,
|
|
"individual_summaries": summaries,
|
|
"aggregated_summary": aggregation_result,
|
|
"project_structure": structure_result,
|
|
"timestamp": str(Path().absolute()),
|
|
}
|
|
|
|
return final_summary
|
|
|
|
except Exception as e:
|
|
return {
|
|
"error": f"Error summarizing entire project: {str(e)}",
|
|
"project_summary": None,
|
|
}
|
|
|
|
|
|
def create_project_summary_file(
|
|
project_path: str = ".", output_file: str = "clover.md"
|
|
) -> bool:
|
|
"""
|
|
Create a comprehensive project summary file
|
|
|
|
Args:
|
|
project_path (str): Path to the project directory
|
|
output_file (str): Name of the output summary file
|
|
|
|
Returns:
|
|
bool: True if successful, False otherwise
|
|
"""
|
|
try:
|
|
print("Generating comprehensive project summary...")
|
|
|
|
# Generate complete project summary
|
|
summary_data = summarize_entire_project(project_path)
|
|
|
|
if "error" in summary_data:
|
|
print(f"Error generating project summary: {summary_data['error']}")
|
|
return False
|
|
|
|
# Format the summary as markdown
|
|
content = f"""# Project Summary - {Path(project_path).absolute().name}
|
|
|
|
## Overview
|
|
{summary_data.get("aggregated_summary", {}).get("aggregated_summary", "No summary available")}
|
|
|
|
## Project Statistics
|
|
- **Files Analyzed**: {summary_data.get("files_analyzed", 0)}
|
|
- **Successfully Processed**: {summary_data.get("files_successful", 0)}
|
|
- **Failed to Process**: {summary_data.get("files_failed", 0)}
|
|
|
|
## Project Structure
|
|
{summary_data.get("project_structure", {}).get("structure", "No structure information available")}
|
|
|
|
## Individual File Summaries
|
|
|
|
"""
|
|
|
|
# Add individual file summaries
|
|
individual_summaries = summary_data.get("individual_summaries", [])
|
|
for summary in individual_summaries:
|
|
if "error" not in summary and summary.get("summary"):
|
|
filepath = summary.get("filepath", "Unknown")
|
|
file_summary = summary.get("summary", "No summary")
|
|
content += f"### {filepath}\n{file_summary}\n\n"
|
|
|
|
content += f"""
|
|
---
|
|
*Generated by Clover CLI on {summary_data.get("timestamp", "unknown time")}*
|
|
"""
|
|
|
|
# Write to output file
|
|
output_path = os.path.join(project_path, output_file)
|
|
create_file(output_path, content)
|
|
|
|
print(f"Project summary saved to: {output_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error creating project summary file: {str(e)}")
|
|
return False
|
|
|
|
|
|
# Convenience function for backward compatibility
|
|
def summarize_project(project_path: str = ".") -> str:
|
|
"""
|
|
Simple project summarization function
|
|
|
|
Args:
|
|
project_path (str): Path to the project directory
|
|
|
|
Returns:
|
|
str: Project summary text
|
|
"""
|
|
try:
|
|
result = summarize_entire_project(project_path)
|
|
if "error" in result:
|
|
return f"Error: {result['error']}"
|
|
|
|
aggregated = result.get("aggregated_summary", {})
|
|
return aggregated.get("aggregated_summary", "Project analysis completed.")
|
|
|
|
except Exception as e:
|
|
return f"Error summarizing project: {str(e)}"
|