generated from Templates/Docker_Image
227 lines
8.4 KiB
Python
227 lines
8.4 KiB
Python
"""Agent orchestrator for software generation."""
|
|
|
|
import asyncio
|
|
from typing import Optional
|
|
from ai_software_factory.agents.git_manager import GitManager
|
|
from ai_software_factory.agents.ui_manager import UIManager
|
|
from ai_software_factory.agents.gitea import GiteaAPI
|
|
from ai_software_factory.agents.database_manager import DatabaseManager
|
|
from ai_software_factory.config import settings
|
|
from datetime import datetime
|
|
import os
|
|
|
|
|
|
class AgentOrchestrator:
|
|
"""Orchestrates the software generation process with full audit trail."""
|
|
|
|
def __init__(
|
|
self,
|
|
project_id: str,
|
|
project_name: str,
|
|
description: str,
|
|
features: list,
|
|
tech_stack: list,
|
|
db = None
|
|
):
|
|
"""Initialize orchestrator."""
|
|
self.project_id = project_id
|
|
self.project_name = project_name
|
|
self.description = description
|
|
self.features = features
|
|
self.tech_stack = tech_stack
|
|
self.status = "initialized"
|
|
self.progress = 0
|
|
self.current_step = None
|
|
self.message = ""
|
|
self.logs = []
|
|
self.ui_data = {}
|
|
self.db = db
|
|
|
|
# Initialize agents
|
|
self.git_manager = GitManager(project_id)
|
|
self.ui_manager = UIManager(project_id)
|
|
self.gitea_api = GiteaAPI(
|
|
token=settings.GITEA_TOKEN,
|
|
base_url=settings.GITEA_URL
|
|
)
|
|
|
|
# Initialize database manager if db session provided
|
|
self.db_manager = None
|
|
self.history = None
|
|
if db:
|
|
self.db_manager = DatabaseManager(db)
|
|
# Log project start to database
|
|
self.history = self.db_manager.log_project_start(
|
|
project_id=project_id,
|
|
project_name=project_name,
|
|
description=description
|
|
)
|
|
# Re-fetch with new history_id
|
|
self.db_manager = DatabaseManager(db)
|
|
|
|
async def run(self) -> dict:
|
|
"""Run the software generation process with full audit logging."""
|
|
try:
|
|
# Step 1: Initialize project
|
|
self.progress = 5
|
|
self.current_step = "Initializing project"
|
|
self.message = "Setting up project structure..."
|
|
self.logs.append(f"[{datetime.utcnow().isoformat()}] Initializing project.")
|
|
|
|
# Step 2: Create project structure (skip git operations)
|
|
self.progress = 15
|
|
self.current_step = "Creating project structure"
|
|
self.message = "Creating project files..."
|
|
await self._create_project_structure()
|
|
|
|
# Step 3: Generate initial code
|
|
self.progress = 25
|
|
self.current_step = "Generating initial code"
|
|
self.message = "Generating initial code with Ollama..."
|
|
await self._generate_code()
|
|
|
|
# Step 4: Test the code
|
|
self.progress = 50
|
|
self.current_step = "Testing code"
|
|
self.message = "Running tests..."
|
|
await self._run_tests()
|
|
|
|
# Step 5: Commit to git (skip in test env)
|
|
self.progress = 75
|
|
self.current_step = "Committing to git"
|
|
self.message = "Skipping git operations in test environment..."
|
|
|
|
# Step 6: Create PR (skip in test env)
|
|
self.progress = 90
|
|
self.current_step = "Creating PR"
|
|
self.message = "Skipping PR creation in test environment..."
|
|
|
|
# Step 7: Complete
|
|
self.progress = 100
|
|
self.current_step = "Completed"
|
|
self.message = "Software generation complete!"
|
|
self.logs.append(f"[{datetime.utcnow().isoformat()}] Software generation complete!")
|
|
|
|
# Log completion to database if available
|
|
if self.db_manager and self.history:
|
|
self.db_manager.log_project_complete(
|
|
history_id=self.history.id,
|
|
message="Software generation complete!"
|
|
)
|
|
|
|
return {
|
|
"status": "completed",
|
|
"progress": self.progress,
|
|
"message": self.message,
|
|
"current_step": self.current_step,
|
|
"logs": self.logs,
|
|
"ui_data": self.ui_manager.ui_data,
|
|
"history_id": self.history.id if self.history else None
|
|
}
|
|
|
|
except Exception as e:
|
|
self.status = "error"
|
|
self.message = f"Error: {str(e)}"
|
|
self.logs.append(f"[{datetime.utcnow().isoformat()}] Error: {str(e)}")
|
|
|
|
# Log error to database if available
|
|
if self.db_manager and self.history:
|
|
self.db_manager.log_error(
|
|
history_id=self.history.id,
|
|
error=str(e)
|
|
)
|
|
|
|
return {
|
|
"status": "error",
|
|
"progress": self.progress,
|
|
"message": self.message,
|
|
"current_step": self.current_step,
|
|
"logs": self.logs,
|
|
"error": str(e),
|
|
"ui_data": self.ui_manager.ui_data,
|
|
"history_id": self.history.id if self.history else None
|
|
}
|
|
|
|
async def _create_project_structure(self) -> None:
|
|
"""Create initial project structure."""
|
|
project_dir = self.project_id
|
|
|
|
# Create .gitignore
|
|
gitignore_path = f"{project_dir}/.gitignore"
|
|
try:
|
|
os.makedirs(project_dir, exist_ok=True)
|
|
with open(gitignore_path, "w") as f:
|
|
f.write("# Python\n__pycache__/\n*.pyc\n*.pyo\n*.pyd\n.Python\n*.env\n.venv/\nnode_modules/\n.env\nbuild/\ndist/\n.pytest_cache/\n.mypy_cache/\n.coverage\nhtmlcov/\n.idea/\n.vscode/\n*.swp\n*.swo\n*~\n.DS_Store\n.git\n")
|
|
except Exception as e:
|
|
self.logs.append(f"[{datetime.utcnow().isoformat()}] Failed to create .gitignore: {str(e)}")
|
|
|
|
# Create README.md
|
|
readme_path = f"{project_dir}/README.md"
|
|
try:
|
|
with open(readme_path, "w") as f:
|
|
f.write(f"# {self.project_name}\n\n{self.description}\n\n## Features\n")
|
|
for feature in self.features:
|
|
f.write(f"- {feature}\n")
|
|
f.write(f"\n## Tech Stack\n")
|
|
for tech in self.tech_stack:
|
|
f.write(f"- {tech}\n")
|
|
except Exception as e:
|
|
self.logs.append(f"[{datetime.utcnow().isoformat()}] Failed to create README.md: {str(e)}")
|
|
|
|
async def _generate_code(self) -> None:
|
|
"""Generate code using Ollama."""
|
|
# This would call Ollama API to generate code
|
|
# For now, create a placeholder file
|
|
try:
|
|
main_py_path = f"{self.project_id}/main.py"
|
|
os.makedirs(self.project_id, exist_ok=True)
|
|
with open(main_py_path, "w") as f:
|
|
f.write("# Generated by AI Software Factory\n")
|
|
f.write("print('Hello, World!')\n")
|
|
except Exception as e:
|
|
self.logs.append(f"[{datetime.utcnow().isoformat()}] Failed to create main.py: {str(e)}")
|
|
|
|
# Log code change to audit trail
|
|
if self.db_manager and self.history:
|
|
self.db_manager.log_code_change(
|
|
project_id=self.project_id,
|
|
change_type="CREATE",
|
|
file_path="main.py",
|
|
actor="agent",
|
|
actor_type="agent",
|
|
details="Generated main.py file"
|
|
)
|
|
|
|
async def _run_tests(self) -> None:
|
|
"""Run tests for the generated code."""
|
|
# This would run pytest or other test framework
|
|
# For now, simulate test success
|
|
pass
|
|
|
|
async def _commit_to_git(self) -> None:
|
|
"""Commit changes to git."""
|
|
pass # Skip git operations in test environment
|
|
|
|
async def _create_pr(self) -> None:
|
|
"""Create pull request."""
|
|
pass # Skip PR creation in test environment
|
|
|
|
def update_status(self, status: str, progress: int, message: str) -> None:
|
|
"""Update status and progress."""
|
|
self.status = status
|
|
self.progress = progress
|
|
self.message = message
|
|
|
|
def get_ui_data(self) -> dict:
|
|
"""Get UI data."""
|
|
return self.ui_manager.ui_data
|
|
|
|
def render_dashboard(self) -> str:
|
|
"""Render dashboard HTML."""
|
|
return self.ui_manager.render_dashboard()
|
|
|
|
def get_history(self) -> Optional[dict]:
|
|
"""Get project history from database."""
|
|
if self.db_manager and self.history:
|
|
return self.db_manager.get_project_audit_data(self.history.project_id)
|
|
return None |