work in progress: FreshDesk CLI

cg/apps/freshdesk/freshdesk_api.py

@@ -0,0 +1,173 @@
+import json
+import logging
+from typing import Dict, List, Optional
+
+import requests
+
+LOG = logging.getLogger(__name__)
+
+
+class Freshdesk:
+    def __init__(self, api_key: str, domain: str):
+        """Initialize the Freshdesk API"""
+        self.api_key = api_key
+        self.domain = domain
+        self.base_url = f"https://{domain}/api/v2/"
+
+    def _make_request(self, method: str, endpoint: str, payload: Optional[Dict] = None) -> Dict:
+        """Make a request to the Freshdesk API"""
+        url = f"{self.base_url}{endpoint}"
+        headers = {"Content-Type": "application/json"}
+        response = requests.request(
+            method,
+            url,
+            auth=(self.api_key, "X"),
+            headers=headers,
+            data=json.dumps(payload) if payload else None,
+        )
+        if response.status_code in {200, 201}:
+            return response.json()
+        else:
+            raise Exception(
+                f"Request failed with status code {response.status_code}: {response.text}"
+            )
+
+    def post_ticket_message(
+        self, ticket_id: int, message: str, private: bool = True, verbose: bool = False
+    ) -> Dict:
+        """
+        Post a message (note) to a ticket.
+        :param ticket_id: The ID of the ticket.
+        :param message: The content of the message.
+        :param private: Whether the message should be private or visible to the customer.
+        :param verbose: Whether to print a success message.
+        :return: Details of the posted message.
+        """
+
+        html_message = (
+            message.replace("\x85", "<br>")
+            .replace("\u2028", "<br>")
+            .replace("\u2029", "<br>")
+            .replace("\r\n", "<br>")
+            .replace("\r", "<br>")
+            .replace("\n", "<br>")
+            .replace("\\n", "<br>")
+        )
+
+        endpoint = f"/tickets/{ticket_id}/notes"
+        payload = {"body": html_message, "private": private}
+
+        print("Payload being sent:", json.dumps(payload, indent=2))
+
+        response = self._make_request("POST", endpoint, payload)
+
+        if "id" in response:
+            if verbose:
+                print(
+                    f"Message successfully posted to ticket {ticket_id} (Note ID: {response['id']})."
+                )
+        else:
+            raise Exception(
+                "Message posting was successful but the response is missing expected fields."
+            )
+
+        return {"body": response.get("body")}
+
+    def get_ticket(self, ticket_id: int, verbose: bool = False) -> Dict:
+        """Retrieve ticket details by ID"""
+        response = self._make_request("GET", f"/tickets/{ticket_id}")
+        if verbose:
+            print(f"Ticket {ticket_id} details retrieved.")
+        return response
+
+    def update_ticket(self, ticket_id: int, updates: Dict) -> Dict:
+        """Update ticket details (e.g., status, priority)"""
+        return self._make_request("PUT", f"/tickets/{ticket_id}", updates)
+
+    def get_ticket_tags(self, ticket_id: int) -> List[str]:
+        """
+        Retrieve the tags associated with a ticket.
+        :param ticket_id: The ID of the ticket.
+        :return: List of tags.
+        """
+        ticket = self.get_ticket(ticket_id)
+        return ticket.get("tags", [])
+
+    def set_ticket_status(self, ticket_id: int, status: int) -> Dict:
+        """
+        Update the status of a ticket.
+        :param ticket_id: The ID of the ticket.
+        :param status: The new status code.
+        :return: Updated ticket details.
+        """
+        updates = {"status": status}
+        return self.update_ticket(ticket_id, updates)
+
+    def get_ticket_group(self, ticket_id: int) -> List[str]:
+        """
+        Retrieve the groups associated with a ticket.
+        :param ticket_id:
+        :return: List of groups.
+        """
+        ticket = self.get_ticket(ticket_id)
+        return ticket.get("group_id", [])
+
+    def set_ticket_group(self, ticket_id: int, group_id: int) -> int:
+        """
+        Update the group associated with a ticket.
+        :param ticket_id: ID of the ticket to update.
+        :param group_id: ID of the group to assign.
+        :return: The group_id assigned to the ticket.
+        """
+        updates = {"group_id": group_id}
+        self.update_ticket(ticket_id, updates)
+        return self.get_ticket(ticket_id)["group_id"]
+
+    def add_ticket_tag(self, ticket_id: int, tag: str, verbose: bool = False) -> Dict:
+        """
+        Add a tag to a ticket.
+        :param ticket_id: The ID of the ticket.
+        :param tag: The tag to add.
+        :return: Updated ticket details.
+        """
+        ticket = self.get_ticket(ticket_id)
+        tags = ticket.get("tags", [])
+        if tag not in tags:
+            tags.append(tag)
+        if verbose:
+            print(f"Tag '{tag}' added to ticket {ticket_id}.")
+        return self.update_ticket(ticket_id, {"tags": tags})
+
+    def remove_ticket_tag(self, ticket_id: int, tag: str, verbose: bool = False) -> Dict:
+        """
+        Remove a tag from a ticket.
+        :param ticket_id: The ID of the ticket.
+        :param tag: The tag to remove.
+        :return: Updated ticket details.
+        """
+        ticket = self.get_ticket(ticket_id)
+        tags = ticket.get("tags", [])
+        if tag in tags:
+            tags.remove(tag)
+        else:
+            if verbose:
+                print(f"Tag '{tag}' not found in ticket {ticket_id}. No changes made.")
+            return ticket
+        response = self.update_ticket(ticket_id, {"tags": tags})
+        if verbose:
+            print(f"Tag '{tag}' removed from ticket {ticket_id}.")
+        return response
+
+    def get_ticket_status(self, ticket_id: int, verbose: bool = False) -> str:
+        """
+        Retrieve the status of a ticket.
+        :param ticket_id: The ID of the ticket.
+        :param verbose: Whether to return a human-readable status.
+        :return: Status code or status string.
+        """
+        ticket = self.get_ticket(ticket_id)
+        status = ticket.get("status")
+        if verbose:
+            status_mapping = {2: "Open", 3: "Pending", 4: "Resolved", 5: "Closed"}
+            return status_mapping.get(status, "Unknown")
+        return status

cg/cli/base.py

@@ -4,8 +4,8 @@
 import sys
 from pathlib import Path
 
-import rich_click as click
 import coloredlogs
+import rich_click as click
 from sqlalchemy.orm import scoped_session
 
 import cg
@@ -18,6 +18,7 @@
 from cg.cli.deliver.base import deliver as deliver_cmd
 from cg.cli.demultiplex.base import demultiplex_cmd_group as demultiplex_cmd
 from cg.cli.downsample import downsample
+from cg.cli.freshdesk.base import freshdesk_cli as freshdesk_cmd
 from cg.cli.generate.base import generate as generate_cmd
 from cg.cli.get import get
 from cg.cli.post_process.post_process import post_process_group as post_processing
@@ -141,6 +142,7 @@ def search(query):
 base.add_command(compress)
 base.add_command(decompress)
 base.add_command(delete)
+base.add_command(freshdesk_cmd)
 base.add_command(get)
 base.add_command(set_cmd)
 base.add_command(transfer_group)

cg/cli/freshdesk/base.py

@@ -0,0 +1,181 @@
+import click
+
+from cg.apps.freshdesk.freshdesk_api import Freshdesk
+
+# TODO: Add Keys to CG config
+# from freshdesk.keys import FRESHDESK_API_KEY, FRESHDESK_DOMAIN
+
+
+API_KEY = "YOUR_API_KEY"
+DOMAIN = "scilifelab.freshdesk.com"
+freshdesk = Freshdesk(api_key=API_KEY, domain=DOMAIN)
+
+
+@click.group()
+def freshdesk_cli():
+    """CLI for interacting with Freshdesk."""
+    pass
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def get_ticket(ticket_id, verbose):
+    """Retrieve details of a ticket."""
+    try:
+        ticket = Freshdesk.get_ticket(ticket_id, verbose=verbose)
+        click.echo(ticket)
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.argument("message", type=str)
+@click.option(
+    "--private/--public", default=True, help="Set the message visibility (default: private)."
+)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def post_message(ticket_id, message, private, verbose):
+    """Post a html message to a ticket."""
+    try:
+        response = Freshdesk.post_ticket_message(ticket_id, message, private, verbose=verbose)
+        click.echo(f"Message posted: {response}")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.argument("status", type=int)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def set_status(ticket_id, status, verbose):
+    """
+    Update the status of a ticket.
+    \n
+                2: "Open",\n
+                3: "Pending",\n
+                4: "Resolved",\n
+                5: "Closed"\n
+    """
+    try:
+        # TODO: Use response
+        Freshdesk.set_ticket_status(ticket_id, status, verbose=verbose)
+        click.echo(f"Ticket {ticket_id} status updated to {status}.")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def get_tags(ticket_id, verbose):
+    """Retrieve tags of a ticket."""
+    try:
+        tags = Freshdesk.get_ticket_tags(ticket_id)
+        click.echo(f"Tags: {tags}")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def get_group(ticket_id, verbose):
+    """Retrieve group assigned to a ticket."""
+    try:
+        group_id = Freshdesk.get_ticket_group(ticket_id)
+        if verbose:
+            # TODO: Add this to a utils function
+            group_id_mapping = {
+                None: "Unassigned",
+                202000118168: "Production Bioinformatics",
+                202000118167: "Production Lab",
+                202000118169: "Production Managers",
+            }
+            click.echo(f"group_id: {group_id_mapping[group_id]}")
+        else:
+            click.echo(f"group_id: {group_id}")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int, required=True)
+@click.argument("group_id", type=int, required=True)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def set_group(ticket_id, group_id, verbose):
+    """Assign a group to a ticket.\n
+    202000118168: "Production Bioinformatics",\n
+    202000118167: "Production Lab",\n
+    202000118169: "Production Managers"\n
+
+    """
+    try:
+        group_assigned = Freshdesk.set_ticket_group(ticket_id, group_id)
+        if verbose:
+            # TODO: Add this to a utils function
+            group_id_mapping = {
+                None: "Unassigned",
+                202000118168: "Production Bioinformatics",
+                202000118167: "Production Lab",
+                202000118169: "Production Managers",
+            }
+            click.echo(f"group_id: {group_id_mapping[group_assigned]}")
+        else:
+            click.echo(f"group_id: {group_assigned}")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.argument("tag", type=str)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def add_tag(ticket_id, tag, verbose):
+    """Add a tag to a ticket."""
+    try:
+        # TODO: Use the response
+        Freshdesk.add_ticket_tag(ticket_id, tag, verbose=verbose)
+        click.echo(f"Tag '{tag}' added to ticket {ticket_id}.")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.argument("tag", type=str)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def remove_tag(ticket_id, tag, verbose):
+    """Remove a tag from a ticket."""
+    try:
+        # TODO: Use the response
+        Freshdesk.remove_ticket_tag(ticket_id, tag, verbose=verbose)
+        click.echo(f"Tag '{tag}' removed from ticket {ticket_id}.")
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)
+
+
+@freshdesk_cli.command()
+@click.argument("ticket_id", type=int)
+@click.option("--verbose", is_flag=True, help="Show detailed output.")
+def get_status(ticket_id, verbose):
+    """Retrieve the status of a ticket.
+    \n
+            2: "Open",\n
+            3: "Pending",\n
+            4: "Resolved",\n
+            5: "Closed"\n
+    """
+    try:
+        status = Freshdesk.get_ticket_status(ticket_id)
+        if verbose:
+            # TODO: Add this to a utils function
+            status_mapping = {2: "Open", 3: "Pending", 4: "Resolved", 5: "Closed"}
+            click.echo(
+                f"Ticket {ticket_id} status: {status_mapping.get(status, 'Unknown')} ({status})"
+            )
+        else:
+            click.echo(status)
+    except Exception as e:
+        click.echo(f"Error: {e}", err=True)