Spaces:
Sleeping
Sleeping
| """ | |
| Support ticket module for creating Jira tickets. | |
| Handles integration with Jira API for user support requests. | |
| """ | |
| from typing import Dict, Any, Optional | |
| from jira import JIRA | |
| from jira.exceptions import JIRAError | |
| from utils import setup_logger, format_timestamp | |
| from config import JIRA_SERVER, JIRA_EMAIL, JIRA_API_TOKEN, JIRA_PROJECT_KEY | |
| logger = setup_logger(__name__) | |
| class SupportTicketManager: | |
| """ | |
| Manages creation of support tickets in Jira. | |
| """ | |
| def __init__( | |
| self, | |
| server: str = JIRA_SERVER, | |
| email: str = JIRA_EMAIL, | |
| api_token: str = JIRA_API_TOKEN, | |
| project_key: str = JIRA_PROJECT_KEY | |
| ): | |
| """ | |
| Initializes Jira client with credentials. | |
| Inputs: server, email, api_token, project_key (all strings) | |
| Outputs: None | |
| """ | |
| self.server = server | |
| self.email = email | |
| self.api_token = api_token | |
| self.project_key = project_key | |
| self.jira_client: Optional[JIRA] = None | |
| logger.info("SupportTicketManager initialized") | |
| def _connect(self) -> bool: | |
| """ | |
| Establishes connection to Jira. | |
| Inputs: None | |
| Outputs: boolean (success status) | |
| """ | |
| if not all([self.server, self.email, self.api_token]): | |
| logger.error("Jira credentials not configured") | |
| return False | |
| try: | |
| self.jira_client = JIRA( | |
| server=self.server, | |
| basic_auth=(self.email, self.api_token) | |
| ) | |
| logger.info(f"Successfully connected to Jira: {self.server}") | |
| return True | |
| except JIRAError as e: | |
| logger.error(f"Failed to connect to Jira: {str(e)}") | |
| return False | |
| except Exception as e: | |
| logger.error(f"Unexpected error connecting to Jira: {str(e)}") | |
| return False | |
| def create_ticket( | |
| self, | |
| summary: str, | |
| description: str, | |
| issue_type: str = "Task", | |
| priority: str = "Medium" | |
| ) -> Dict[str, Any]: | |
| """ | |
| Creates a support ticket in Jira. | |
| Inputs: summary (string), description (string), issue_type (string), priority (string) | |
| Outputs: dictionary with ticket information | |
| """ | |
| logger.info(f"Creating support ticket: {summary}") | |
| # Connect to Jira if not already connected | |
| if not self.jira_client: | |
| if not self._connect(): | |
| return { | |
| "success": False, | |
| "error": "Failed to connect to Jira. Please check configuration." | |
| } | |
| try: | |
| # Create issue dictionary | |
| issue_dict = { | |
| "project": {"key": self.project_key}, | |
| "summary": summary, | |
| "description": description, | |
| "issuetype": {"name": issue_type}, | |
| } | |
| # Add priority if supported | |
| try: | |
| issue_dict["priority"] = {"name": priority} | |
| except: | |
| pass # Priority field might not be available in all Jira configurations | |
| # Create the issue | |
| new_issue = self.jira_client.create_issue(fields=issue_dict) | |
| ticket_key = new_issue.key | |
| ticket_url = f"{self.server}/browse/{ticket_key}" | |
| logger.info(f"Successfully created ticket: {ticket_key}") | |
| return { | |
| "success": True, | |
| "ticket_key": ticket_key, | |
| "ticket_url": ticket_url, | |
| "message": f"Support ticket created successfully: {ticket_key}" | |
| } | |
| except JIRAError as e: | |
| error_msg = f"Jira error: {str(e)}" | |
| logger.error(error_msg) | |
| return { | |
| "success": False, | |
| "error": error_msg | |
| } | |
| except Exception as e: | |
| error_msg = f"Failed to create ticket: {str(e)}" | |
| logger.error(error_msg) | |
| return { | |
| "success": False, | |
| "error": error_msg | |
| } | |
| def create_ticket_from_conversation( | |
| self, | |
| user_query: str, | |
| conversation_context: str, | |
| user_email: Optional[str] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Creates a support ticket with conversation context. | |
| Inputs: user_query (string), conversation_context (string), user_email (optional string) | |
| Outputs: dictionary with ticket information | |
| """ | |
| timestamp = format_timestamp() | |
| # Build ticket summary | |
| summary = f"Support Request: {user_query[:80]}" | |
| # Build detailed description | |
| description_parts = [ | |
| f"*Support Request - {timestamp}*", | |
| "", | |
| "h3. User Query:", | |
| user_query, | |
| "", | |
| "h3. Conversation Context:", | |
| "{noformat}", | |
| conversation_context, | |
| "{noformat}", | |
| "" | |
| ] | |
| if user_email: | |
| description_parts.extend([ | |
| "h3. Contact:", | |
| f"Email: {user_email}", | |
| "" | |
| ]) | |
| description_parts.append("---") | |
| description_parts.append("_This ticket was automatically created by the Data Insights App._") | |
| description = "\n".join(description_parts) | |
| return self.create_ticket( | |
| summary=summary, | |
| description=description, | |
| issue_type="Task", | |
| priority="Medium" | |
| ) | |
| def test_connection(self) -> Dict[str, Any]: | |
| """ | |
| Tests connection to Jira. | |
| Inputs: None | |
| Outputs: dictionary with connection status | |
| """ | |
| logger.info("Testing Jira connection...") | |
| if self._connect(): | |
| try: | |
| # Try to get server info | |
| server_info = self.jira_client.server_info() | |
| return { | |
| "success": True, | |
| "message": f"Connected to Jira successfully", | |
| "server_version": server_info.get("version", "Unknown") | |
| } | |
| except Exception as e: | |
| return { | |
| "success": False, | |
| "error": f"Connected but failed to get server info: {str(e)}" | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": "Failed to connect to Jira" | |
| } | |