OverviewFilesCommitsIssuesDocsReleasesPeers

ollama_semantic_function_calling / ollamatools-new / ollamatools / plugins / standard / issues.py

190 lines · python ref:
"""
Issue tracking functionality for the ModularOllamaAgent.

This module provides functions for managing issues, including
loading, saving, adding, listing, and updating issues.
"""

import os
import json
import time
from typing import Dict, Any, List, Optional

from .utils import print_info, print_error, print_success

def load_issues(agent) -> None:
    """
    Load saved issues from file.
    
    Args:
        agent: The agent instance
    """
    if os.path.exists(agent.issues_file):
        try:
            with open(agent.issues_file, 'r') as f:
                data = json.load(f)
                agent.issues = data.get('issues', [])
                agent.issue_id_counter = data.get('next_id', 1)
                agent.print_info(f"Loaded {len(agent.issues)} issues from {agent.issues_file}")
        except Exception as e:
            agent.print_error(f"Failed to load issues: {str(e)}")
            agent.issues = []
            agent.issue_id_counter = 1

def save_issues(agent) -> None:
    """
    Save issues to file.
    
    Args:
        agent: The agent instance
    """
    try:
        with open(agent.issues_file, 'w') as f:
            json.dump({
                'issues': agent.issues,
                'next_id': agent.issue_id_counter
            }, f, indent=2)
        agent.print_info(f"Saved {len(agent.issues)} issues to {agent.issues_file}")
    except Exception as e:
        agent.print_error(f"Failed to save issues: {str(e)}")

def add_issue(agent, title: str, description: str, severity: str = 'medium') -> int:
    """
    Add a new issue to the tracker.
    
    Args:
        agent: The agent instance
        title: Issue title
        description: Issue description
        severity: Issue severity (low, medium, high)
        
    Returns:
        int: Issue ID
    """
    # Validate severity
    if severity not in ['low', 'medium', 'high']:
        severity = 'medium'
    
    # Create issue
    issue = {
        'id': agent.issue_id_counter,
        'title': title,
        'description': description,
        'severity': severity,
        'status': 'open',
        'created_at': time.strftime('%Y-%m-%d %H:%M:%S'),
        'updated_at': time.strftime('%Y-%m-%d %H:%M:%S')
    }
    
    # Add to list and increment counter
    agent.issues.append(issue)
    agent.issue_id_counter += 1
    
    # Save issues
    save_issues(agent)
    
    agent.print_success(f"Issue #{issue['id']} added: {title}")
    return issue['id']

def list_issues(agent, status: Optional[str] = None) -> List[Dict[str, Any]]:
    """
    List all issues, optionally filtered by status.
    
    Args:
        agent: The agent instance
        status: Optional status filter (open, resolved)
        
    Returns:
        List[Dict[str, Any]]: List of issues
    """
    # Filter issues if status provided
    filtered_issues = agent.issues
    if status:
        filtered_issues = [i for i in agent.issues if i['status'] == status]
    
    # Print issues
    if not filtered_issues:
        agent.print_info("No issues found")
        return []
    
    agent.print_info(f"Issues ({len(filtered_issues)}):")
    print("\nID  | Severity | Status   | Title")
    print("----|----------|----------|-------------------")
    
    for issue in filtered_issues:
        print(f"{issue['id']:<4}| {issue['severity']:<9}| {issue['status']:<9}| {issue['title']}")
    
    return filtered_issues

def get_issue(agent, issue_id: int) -> Optional[Dict[str, Any]]:
    """
    Get details of a specific issue.
    
    Args:
        agent: The agent instance
        issue_id: Issue ID
        
    Returns:
        Optional[Dict[str, Any]]: Issue details or None if not found
    """
    # Find issue
    for issue in agent.issues:
        if issue['id'] == issue_id:
            # Print issue details
            agent.print_info(f"Issue #{issue['id']}: {issue['title']}")
            print(f"Status: {issue['status']}")
            print(f"Severity: {issue['severity']}")
            print(f"Created: {issue['created_at']}")
            print(f"Updated: {issue['updated_at']}")
            print("\nDescription:")
            print(issue['description'])
            return issue
    
    agent.print_error(f"Issue #{issue_id} not found")
    return None

def update_issue(
    agent, 
    issue_id: int, 
    status: Optional[str] = None, 
    severity: Optional[str] = None, 
    description: Optional[str] = None
) -> bool:
    """
    Update an existing issue.
    
    Args:
        agent: The agent instance
        issue_id: Issue ID
        status: New status (open, resolved)
        severity: New severity (low, medium, high)
        description: New description
        
    Returns:
        bool: True if updated successfully
    """
    # Find issue
    for i, issue in enumerate(agent.issues):
        if issue['id'] == issue_id:
            # Update fields
            if status and status in ['open', 'resolved']:
                agent.issues[i]['status'] = status
            
            if severity and severity in ['low', 'medium', 'high']:
                agent.issues[i]['severity'] = severity
            
            if description:
                agent.issues[i]['description'] = description
            
            # Update timestamp
            agent.issues[i]['updated_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
            
            # Save issues
            save_issues(agent)
            
            agent.print_success(f"Issue #{issue_id} updated")
            return True
    
    agent.print_error(f"Issue #{issue_id} not found")
    return False