Jira Automation

jira-auto-process.py

#   pyTstProj - jira-auto-process.py
#   Copyright (C) 2023 Patmeow Limited
#
#   This program is free software: you can redistribute it and/or modify
#   it under the terms of the GNU Affero General Public License as published by
#   the Free Software Foundation, either version 3 of the License, or
#   (at your option) any later version.
#
#   This program is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU Affero General Public License for more details.
#
#   You should have received a copy of the GNU Affero General Public License
#   along with this program.  If not, see <http://www.gnu.org/licenses/>.

import time

import requests
import urllib3
import logging
import sqlite3
import mycred

logging.basicConfig(format="%(asctime)s - %(module)s : %(funcName)s - L%(lineno)d - %(message)s",
                    level=logging.INFO, handlers=[
                        logging.FileHandler("jira-auto-process-1.log"),
                        logging.StreamHandler()
                    ])

JIRA_BASEURL = ""
JIRA_TLS_VERIFY = False
# all keys here, must be in string.
JIRA_MONITORED_FILTERS = {"10411": "",
                          "10413": ""}
JIRA_SLA_BREACHED_ALERT = ""
JIRA_AUTO_PROCESSED_FP_FILTERS = {"ID_XXXX_2": "10411", "ID_XXXX_2": "10413"}
SOFTWARE_VERSION = "JiraAutoProcessor-Py-v230331/1.0"

JIRA_AUTOREPLY_TMPL = "【工单处置自动化】[Automated Reply] \n\n *信息仅供参考* \n\n " + "From " + SOFTWARE_VERSION + " By xxx \n\n {msg}"

urllib3.disable_warnings()

ID_XXXX_1 = "Reply Message 1"
ID_XXXX_2 = "Reply Message 2"

def compare_list(a: list, b: list) -> bool:
    if len(a) != len(b):
        return True
    else:
        a.sort()
        b.sort()
        for i in range(0, len(a)):
            if a[i] != b[i]:
                return True
        return False


def checkNetworkConn():
    r = requests.get("https://connect.rom.miui.com/generate_204", timeout=5, headers={"User-Agent": "MIUI/11.1.0"})
    if r.status_code == 204:
        return
    else:
        raise ConnectionError("Internet is not connected.")


class JiraInstance(object):
    def __init__(self, sqliteDb: sqlite3.Connection):
        self.filters = JIRA_MONITORED_FILTERS
        self.username = mycred.JIRA_USERNAME
        self.patoken = mycred.JIRA_PAT
        self.baseurl = JIRA_BASEURL
        self.req_headers = {"User-Agent": SOFTWARE_VERSION, "Authorization": "Bearer " + self.patoken,
                            "X-ExperimentalApi": "true"}
        self.session = requests.Session()
        self.session.headers = self.req_headers
        self.session.verify = JIRA_TLS_VERIFY
        self.ap_filters = JIRA_AUTO_PROCESSED_FP_FILTERS
        self.mon_filters = JIRA_MONITORED_FILTERS
        self.current_assignee = ""
        self.dbconn = sqliteDb

    def login(self):
        r0 = self.session.get(self.baseurl)
        if r0.status_code != 200:
            raise SystemError("Cannot access website.")
        r1 = self.session.get(self.baseurl + "/rest/api/2/myself")
        if r1.status_code != 200:
            raise SystemError("Authentication failed.")
        resp1 = r1.json()
        self.current_assignee = resp1["name"]
        logging.info("Successfully logged-in as " + self.current_assignee)

    def getIssuesFromSavedFilter(self, filter_id) -> list[any]:
        r1 = self.session.get(self.baseurl + "/rest/api/2/filter/" + filter_id,
                              params={"_": str(int(time.time() * 1000))})
        if r1.status_code != 200:
            raise SystemError("Cannot access specific filter.")
        resp1 = r1.json()
        r2 = self.session.get(resp1["searchUrl"]+"&maxResults=500")
        if r2.status_code != 200:
            raise SystemError("Cannot access filter search results")
        resp2 = r2.json()
        if resp2["total"] == 0:
            return []
        logging.info("Successfully got the issue from filter, id: " + filter_id)
        return resp2["issues"]

    def postAutomatedComments(self, issueKey, commentMsg, isPublic=False) -> bool:
        cmtUrl = self.baseurl + "/rest/servicedeskapi/request/" + issueKey + "/comment"
        r5 = self.session.post(cmtUrl, json={"body": JIRA_AUTOREPLY_TMPL.format(msg=commentMsg), "public": isPublic})
        if r5.status_code == 201:
            logging.info("Comment added for " + issueKey)
            return True
        else:
            logging.error("Failed to add comments for " + issueKey, " Error " + str(r5.status_code))
            return False

    def issueTransit(self, issueKey, currentStatus, toStatus, solutionName="", shouldAssign=False) -> int:
        # Transition ID=a Create Issue
        # Transition ID=b Resolved Issue
        # Transition ID=c For L2 Review
        # Transition ID=d Back To Support
        # To prevent from error, we do not allow auto response to client:
        #       Transition ID=e Respond to customer
        # Status ID=g Work in progress
        # Status ID=h OPEN
        # Status ID=i Resolved
        transitMap = {"OPEN": {"Work in progress": "x"}, "Work in progress": {"L2 REVIEW": "x", "Resolved": "x"},
                      "L2 REVIEW": {"Resolved": "x", "Work in progress": "x"}}
        issueTransitUrl = self.baseurl + "/rest/api/2/issue/" + issueKey + "/transitions"
        data = {"transition": {"id": ""}, "fields": {}}
        if solutionName and currentStatus != "OPEN":
            data["fields"]["resolution"] = {"name": solutionName}
        if shouldAssign and currentStatus != "OPEN":
            data["fields"]["assignee"] = {"name": self.current_assignee}
        elif shouldAssign and currentStatus == "OPEN":
            logging.warning("Ticket is not created, cannot assign to user, assign request skipped.")
        try:
            data["transition"]["id"] = transitMap[currentStatus][toStatus]
            r0 = self.session.post(issueTransitUrl, json=data)
            return r0.status_code
        except Exception as e:
            return -1