def __init__()

in server/app/plugins/jirastats.py [0:0]

• 85 lines of code
• 38 McCabe index (conditional complexity)

    def __init__(self, data):
        self._data = data
        self.assignee = data["fields"]["assignee"]["name"] if data["fields"]["assignee"] else None
        self.status = data["fields"]["status"]["name"]
        self.closed = self.status == "Closed"
        self.reopened = False
        self.key = data["key"]
        self.project = self.key.split("-")[0]
        self.url = config.reporting.jira["ticket_url"].format(**data)
        self.summary = data["fields"]["summary"]
        self.created_at = self.get_time(data["fields"]["created"])
        self.updated_at = self.get_time(data["fields"]["updated"])
        self.priority = data["fields"]["priority"]["name"]
        self.author = data["fields"]["creator"] and data["fields"]["creator"]["name"] or "(nobody)"  # May not exist!
        self.issuetype = data["fields"]["issuetype"]["name"]
        self.sla = config.reporting.jira["slas"].get(self.priority, DEFAULT_SLA)

        # SLA stuff
        self.first_response = 0
        self.response_time = 0
        self.resolve_time = 0
        self.closed_at = 0
        self.sla_met_respond = None  # True/False if responded to at all
        self.sla_met_resolve = None
        self.sla_time_counted = 0
        self.statuses = []
        self.changelog = []
        self.paused = self.issuetype in config.reporting.jira.get("no_slas", [])

        # Scan all changelog entries
        for changelog_entry in data.get("changelog", {}).get("histories", []):
            changelog_author = (
                "author" in changelog_entry and changelog_entry["author"]["name"] or "nobody"
            )  # May have been deleted
            changelog_epoch = self.get_time(changelog_entry["created"])
            self.changelog.append((changelog_author, changelog_epoch))
            for item in changelog_entry.get("items", []):
                field = item["field"]
                if field == "assignee":  # Ticket (re)assigned
                    #  self.set_fr(changelog_epoch)
                    pass  # Should not count as a response
                elif field == "resolution":  # Ticket resolved
                    self.set_fr(changelog_epoch)
                    self.closed_at = changelog_epoch
                elif field == "status":  # Status change
                    if (
                        self.closed_at
                    ):  # if we already logged a close, but there are new status changes, it's been reopened
                        self.reopened = True
                    if not self.statuses:  # First status change, log initial status from this
                        self.statuses.append((item["fromString"].lower(), self.created_at))
                    self.statuses.append((item["toString"].lower(), changelog_epoch))  # Note change to status at time

        # Scan all comments, looking for a response earlier than changelog entries
        for comment in data["fields"].get("comment", {}).get("comments", []):
            comment_author = comment["author"]["name"]
            comment_epoch = self.get_time(comment["created"])
            self.changelog.append((comment_author, comment_epoch))
            if comment_author != self.author:  # Comment by someone other than the ticket author
                self.set_fr(comment_epoch)
                break  # Only need to find the first (earliest) occurrence

        # Calculate time spent in WFI
        times_in_wfi = []

        if not self.statuses:  # No status changes, WFI is assumed to be entire duration
            if self.closed:
                times_in_wfi.append((self.created_at, self.closed_at))  # Ticket is closed, use closed_at
            else:
                times_in_wfi.append((self.created_at, int(time.time())))  # Ticket is open, use $now
        else:
            sla_statuses_lower = [x.lower() for x in config.reporting.jira.get("sla_apply_statuses")]
            previous_ts = 0
            previous_is_sla = False
            for status in self.statuses:
                if previous_ts and previous_is_sla:
                    times_in_wfi.append((previous_ts, status[1]))  # From previous TS to this one
                previous_ts = status[1]
                previous_is_sla = status[0] in sla_statuses_lower

            # Not in WFI mode? pause if not paused
            if not self.closed and self.statuses[-1][0] not in sla_statuses_lower:
                self.paused = True

        for spans in times_in_wfi:
            self.sla_time_counted += self.calc_sla_duration(*spans)
        if self.first_response:
            self.response_time = self.calc_sla_duration(self.created_at, self.first_response)
        if self.closed_at:
            self.resolve_time = self.calc_sla_duration(self.created_at, self.closed_at)

        # If closed or responded to, check if the duration met the SLA guides
        # If not closed or responded to, check if time spent in WFI surpasses SLA guides
        if self.closed:
            self.sla_met_resolve = self.resolve_time <= (self.sla["resolve"] * 3600)
        elif self.sla_time_counted > (self.sla["resolve"] * 3600):
            self.sla_met_resolve = False
        if self.first_response:
            self.sla_met_respond = self.response_time <= (self.sla["respond"] * 3600)
        elif self.sla_time_counted > (self.sla["respond"] * 3600):
            self.sla_met_respond = False