sql_query_tool.py

# sql_query_tool.py
from crewai.tools import tool
import sqlite3
import json, os
from typing import List, Dict, Any, Optional
# Get the project root directory
project_root = os.path.dirname(os.path.abspath(__file__))

# Construct the relative path to tickets.db
db_path = os.path.join(project_root, "tickets.db")

# Use db_path wherever the database path is required
print(f"Database Path: {db_path}")

@tool("ticket_query_tool")
def ticket_query_tool(ticket_number: str):
    """
    Tool to query the ticket database by ticket number.
    Args:
        ticket_number: str - The ticket number to search for (e.g., TKT-1024).
    Returns:
        list: A list of ticket details matching the ticket number.
    """
    # db_path = "C:/Users/anike/PycharmProjects/AI_Agents_crew/tickets.db"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # Query for specific ticket number
        query = """
        SELECT *
        FROM tickets 
        WHERE ticket_number = ? OR ticket_number LIKE ?
        """

        cursor.execute(query, (ticket_number, f"%{ticket_number}%"))
        results = cursor.fetchall()

        if not results:
            return [{"message": f"No ticket found with number: {ticket_number}"}]

        # Format results
        tickets = []
        for row in results:
            ticket = {
                "ticket_number": row[0],
                "opened_at": row[1],
                "closed_at": row[2],
                "description": row[3],
                "resolution_comment": row[4],
                "closed_by": row[5],
                "type": row[6],
                "status": row[7],
                "severity": row[8],
                "reported_by": row[9],
                "assigned_to": row[10],
                "last_updated": row[11],
                "environment": row[12],

            }
            tickets.append(ticket)

        conn.close()
        return tickets

    except Exception as e:
        return [{"error": f"Database error: {str(e)}"}]


@tool("status_query_tool")
def status_query_tool(status: str):
    """
    Tool to query tickets by status.
    Args:
        status: str - The status to search for (e.g., 'open', 'in progress', 'closed', 'resolved').
    Returns:
        list: A list of tickets with the specified status.
    """
    # db_path = "C:/Users/anike/PycharmProjects/AI_Agents_crew/tickets.db"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        query = """
        SELECT ticket_number, opened_at, closed_at, description, resolution_comment, 
               closed_by, type, status, severity, reported_by, assigned_to, 
               last_updated, environment, tags
        FROM tickets 
        WHERE LOWER(status) = LOWER(?) OR LOWER(status) LIKE LOWER(?)
        ORDER BY last_updated DESC
        LIMIT 20
        """

        cursor.execute(query, (status, f"%{status}%"))
        results = cursor.fetchall()

        if not results:
            return [{"message": f"No tickets found with status: {status}"}]

        tickets = []
        for row in results:
            ticket = {
                "ticket_number": row[0],
                "status": row[7],
                "severity": row[8],
                "description": row[3][:100] + "..." if len(row[3]) > 100 else row[3],
                "assigned_to": row[10],
                "last_updated": row[11],

            }
            tickets.append(ticket)

        conn.close()
        return tickets

    except Exception as e:
        return [{"error": f"Database error: {str(e)}"}]


@tool("severity_query_tool")
def severity_query_tool(severity: str):
    """
    Tool to query tickets by severity level.
    Args:
        severity: str - The severity level to search for (e.g., 'low', 'medium', 'high', 'critical').
    Returns:
        list: A list of tickets with the specified severity.
    """
    # db_path = "C:/Users/anike/PycharmProjects/AI_Agents_crew/tickets.db"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        query = """
        SELECT ticket_number, opened_at, closed_at, description, resolution_comment, 
               closed_by, type, status, severity, reported_by, assigned_to, 
               last_updated
        FROM tickets 
        WHERE LOWER(severity) = LOWER(?)
        ORDER BY 
            CASE 
                WHEN LOWER(severity) = 'critical' THEN 1
                WHEN LOWER(severity) = 'high' THEN 2
                WHEN LOWER(severity) = 'medium' THEN 3
                WHEN LOWER(severity) = 'low' THEN 4
                ELSE 5
            END,
            last_updated DESC
        LIMIT 20
        """

        cursor.execute(query, (severity,))
        results = cursor.fetchall()
        print(results)
        if not results:
            return [{"message": f"No tickets found with severity: {severity}"}]

        tickets = []
        for row in results:
            ticket = {
                "ticket_number": row[0],
                "status": row[7],
                "severity": row[8],
                "description": row[3][:100] + "..." if len(row[3]) > 100 else row[3],
                "assigned_to": row[10],
                "opened_at": row[1],
                "last_updated": row[11],

            }
            tickets.append(ticket)

        conn.close()
        return tickets

    except Exception as e:
        return [{"error": f"Database error: {str(e)}"}]


@tool("tag_query_tool")
def tag_query_tool(tag: str):
    """
    Tool to query tickets by tags.
    Args:
        tag: str - The tag to search for (e.g., 'backend', 'API', 'frontend', 'database').
    Returns:
        list: A list of tickets containing the specified tag.
    """
    # db_path = "C:/Users/anike/PycharmProjects/AI_Agents_crew/tickets.db"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # Search for tickets where the tag appears in the tags JSON array
        query = """
        SELECT ticket_number, opened_at, closed_at, description, resolution_comment, 
               closed_by, type, status, severity, reported_by, assigned_to, 
               last_updated, environment, tags
        FROM tickets 
        WHERE LOWER(tags) LIKE LOWER(?)
        ORDER BY last_updated DESC
        LIMIT 20
        """

        cursor.execute(query, (f'%"{tag}"%',))
        results = cursor.fetchall()
        print(results)

        if not results:
            return [{"message": f"No tickets found with tag: {tag}"}]
        print(results)
        tickets = []
        for row in results:
            tags_list = json.loads(row[13]) if row[13] else []
            # Double-check that the tag actually matches (case-insensitive)
            if any(tag.lower() in t.lower() for t in tags_list):
                ticket = {
                    "ticket_number": row[0],
                    "status": row[7],
                    "severity": row[8],
                    "description": row[3][:100] + "..." if len(row[3]) > 100 else row[3],
                    "assigned_to": row[10],
                    "last_updated": row[11],

                }
                tickets.append(ticket)

        conn.close()
        return tickets

    except Exception as e:
        return [{"error": f"Database error: {str(e)}"}]


# tag_query_tool("backend")

@tool("keyword_search_tool")
def keyword_search_tool(keywords: str):
    """
    Tool to search tickets by keywords in description or resolution comments.
    Args:
        keywords: str - Keywords to search for in ticket descriptions and resolution comments.
    Returns:
        list: A list of tickets matching the keywords.
    """
    # db_path = "C:/Users/anike/PycharmProjects/AI_Agents_crew/tickets.db"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # Split keywords and search in description and resolution_comment
        keyword_list = [k.strip() for k in keywords.split() if k.strip()]

        # Build dynamic query with multiple keywords
        conditions = []
        params = []

        for keyword in keyword_list:
            conditions.append("(LOWER(description) LIKE LOWER(?) OR LOWER(resolution_comment) LIKE LOWER(?))")
            params.extend([f"%{keyword}%", f"%{keyword}%"])

        if not conditions:
            return [{"message": "No keywords provided for search"}]

        query = f"""
        SELECT ticket_number, opened_at, closed_at, description, resolution_comment, 
               closed_by, type, status, severity, reported_by, assigned_to, 
               last_updated, environment, tags
        FROM tickets 
        WHERE {' OR '.join(conditions)}
        ORDER BY last_updated DESC
        LIMIT 20
        """

        cursor.execute(query, params)
        results = cursor.fetchall()

        if not results:
            return [{"message": f"No tickets found matching keywords: {keywords}"}]

        tickets = []
        for row in results:
            ticket = {
                "ticket_number": row[0],
                "status": row[7],
                "severity": row[8],
                "description": row[3][:150] + "..." if len(row[3]) > 150 else row[3],
                "resolution_comment": row[4][:100] + "..." if row[4] and len(row[4]) > 100 else row[4],
                "assigned_to": row[10],
                "last_updated": row[11],

            }
            tickets.append(ticket)

        conn.close()
        return tickets

    except Exception as e:
        return [{"error": f"Database error: {str(e)}"}]


@tool("advanced_ticket_search")
def advanced_ticket_search(
        ticket_number: Optional[str] = None,
        status: Optional[str] = None,
        severity: Optional[str] = None,
        tag: Optional[str] = None,
        assigned_to: Optional[str] = None,
        keywords: Optional[str] = None
):
    """
    Advanced tool to search tickets with multiple criteria.
    Args:
        ticket_number: Optional ticket number to search for
        status: Optional status filter
        severity: Optional severity filter  
        tag: Optional tag filter
        assigned_to: Optional assignee filter
        keywords: Optional keywords to search in description
    Returns:
        list: Tickets matching the specified criteria
    """
    # db_path = "C:/Users/anike/PycharmProjects/AI_Agents_crew/tickets.db"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # Build dynamic query
        conditions = []
        params = []

        base_query = """
        SELECT ticket_number, opened_at, closed_at, description, resolution_comment, 
               closed_by, type, status, severity, reported_by, assigned_to, 
               last_updated, environment, tags
        FROM tickets WHERE 1=1
        """

        if ticket_number:
            conditions.append("AND (ticket_number = ? OR ticket_number LIKE ?)")
            params.extend([ticket_number, f"%{ticket_number}%"])

        if status:
            conditions.append("AND LOWER(status) LIKE LOWER(?)")
            params.append(f"%{status}%")

        if severity:
            conditions.append("AND LOWER(severity) = LOWER(?)")
            params.append(severity)

        if tag:
            conditions.append("AND LOWER(tags) LIKE LOWER(?)")
            params.append(f'%"{tag}"%')

        if assigned_to:
            conditions.append("AND LOWER(assigned_to) LIKE LOWER(?)")
            params.append(f"%{assigned_to}%")

        if keywords:
            keyword_list = [k.strip() for k in keywords.split() if k.strip()]
            keyword_conditions = []
            for keyword in keyword_list:
                keyword_conditions.append(
                    "(LOWER(description) LIKE LOWER(?) OR LOWER(resolution_comment) LIKE LOWER(?))")
                params.extend([f"%{keyword}%", f"%{keyword}%"])
            if keyword_conditions:
                conditions.append(f"AND ({' OR '.join(keyword_conditions)})")

        query = base_query + " " + " ".join(conditions) + " ORDER BY last_updated DESC LIMIT 20"

        cursor.execute(query, params)
        results = cursor.fetchall()

        if not results:
            return [{"message": "No tickets found matching the specified criteria"}]
        print(results)
        tickets = []
        for row in results:
            ticket = {
                "ticket_number": row[0],
                "opened_at": row[1],
                "closed_at": row[2],
                "description": row[3],
                "resolution_comment": row[4],
                "closed_by": row[5],
                "type": row[6],
                "status": row[7],
                "severity": row[8],
                "reported_by": row[9],
                "assigned_to": row[10],
                "last_updated": row[11],
                "environment": row[12],

            }
            tickets.append(ticket)

        conn.close()
        return tickets

    except Exception as e:
        return [{"error": f"Database error: {str(e)}"}]