fime/task_completer.py

import os
import sys
import traceback
from functools import reduce
from queue import Queue, Empty
from urllib.parse import urlparse, parse_qs

from PySide2 import QtCore
from PySide2.QtCore import QTimer
from PySide2.QtWidgets import QCompleter
from requests_futures.sessions import FuturesSession

from config import Config


class TaskCompleter(QCompleter):
    def __init__(self, parent=None, *args, **kwargs):
        super().__init__([], parent, *args, **kwargs)
        self.setFilterMode(QtCore.Qt.MatchFlag.MatchContains)
        self.setCaseSensitivity(QtCore.Qt.CaseInsensitive)
        self.session = FuturesSession()
        self.config = Config()
        self.picker_url = os.path.join(self.config.jira_url, "rest/api/2/issue/picker")
        self.search_url = os.path.join(self.config.jira_url, "rest/api/2/search")
        self.text = ""
        self.response_text = ""
        self.model_data = set()
        self.escalate = False
        self.update_timer = QTimer(self)
        self.update_timer.timeout.connect(self.process_response)
        self.update_timer.setInterval(250)
        self.queue = Queue()

    @QtCore.Slot()
    def process_response(self):
        try:
            while not self.queue.empty():
                result_dict = self.queue.get_nowait()
                if result_dict["response_text"] == self.text:
                    if self.text == self.response_text:
                        self.response_text = result_dict["response_text"]
                        self.model_data = set(result_dict["result"])
                    else:
                        self.model_data = self.model_data.union(set(result_dict["result"]))
                    self.model().setStringList(self.model_data)
                    self.complete()
        except Empty:
            return

    @QtCore.Slot(str)
    def update_picker(self, text):
        self.text = text
        if self.text == self.currentCompletion():
            # do not update, after auto completion was used
            return
        if self.escalate:
            self.update_search()
        if not self.update_timer.isActive():
            self.update_timer.start()
        future = self.session.get(
            url=self.picker_url,
            params={
                "query": self.text
            },
            headers={
                "Authorization": f"Bearer {self.config.jira_token}",
                "Accept": "application/json",
            },
        )
        future.add_done_callback(self.picker_response_callback)

    def picker_response_callback(self, future):
        try:
            result = future.result()
            parsed = urlparse(result.request.url)
            response_text = parse_qs(parsed.query)["query"][0]
            issues = reduce(lambda x, y: x + (y["issues"]), result.json()["sections"], [])
            extracted = list(map(lambda x: f"{x['key']} {x['summaryText']}", issues))
            if extracted:
                self.queue.put({
                    "response_text": response_text,
                    "result": extracted,
                })
            else:
                if not self.escalate:
                    print("No picker results. Escalating")
                    self.escalate = True
                    self.update_search()
        except Exception:
            print("Ignoring exception, as it only breaks autocompletion:", file=sys.stderr)
            print(traceback.format_exc(), file=sys.stderr)
            return

    def update_search(self):
        for jql in [f"text = {self.text}", f"key = {self.text}"]:
            future = self.session.get(
                url=self.search_url,
                params={
                    "jql": jql,
                    "maxResults": 10,
                    "fields": "key,summary",
                },
                headers={
                    "Authorization": f"Bearer {self.config.jira_token}",
                    "Accept": "application/json",
                },
            )
            future.add_done_callback(self.search_response_callback)

    def search_response_callback(self, future):
        try:
            result = future.result()
            json_result = result.json()
            parsed = urlparse(result.request.url)
            response_text = parse_qs(parsed.query)["jql"][0].split()[2]
            if "issues" in json_result:
                extracted = list(map(lambda x: f"{x['key']} {x['fields']['summary']}", json_result["issues"]))
                self.queue.put({
                    "response_text": response_text,
                    "result": extracted,
                })
        except Exception:
            print("Ignoring exception, as it only breaks autocompletion:", file=sys.stderr)
            print(traceback.format_exc(), file=sys.stderr)
            return