Merge pull request #81 from chilli-axe/desktop-client-improvements

Desktop client improvements

Author: ndepaola

autofill/src/constants.py

@@ -107,6 +107,9 @@ class GoogleScriptsAPIs(str, Enum):
         "https://script.google.com/macros/s/AKfycbzzCWc2x3tfQU1Zp45LB1P19FNZE-4njwzfKT5_Rx399h-5dELZWyvf/exec"
     )
 
+    def __str__(self) -> str:
+        return str(self.value)
+
 
 BRACKETS = [18, 36, 55, 72, 90, 108, 126, 144, 162, 180, 198, 216, 234, 396, 504, 612]
 THREADS = 5  # shared between CardImageCollections

autofill/src/driver.py

@@ -26,34 +26,7 @@
 
 
 @attr.s
-class OrderStatusBarBaseClass:
-    order: CardOrder = attr.ib(default=attr.Factory(CardOrder.from_xml_in_folder))
-    state: str = attr.ib(init=False, default=States.initialising)
-
-    manager: enlighten.Manager = attr.ib(init=False, default=attr.Factory(enlighten.get_manager))
-    status_bar: enlighten.StatusBar = attr.ib(init=False, default=False)
-    download_bar: enlighten.Counter = attr.ib(init=False, default=None)
-    upload_bar: enlighten.Counter = attr.ib(init=False, default=None)
-
-    def configure_bars(self) -> None:
-        num_images = len(self.order.fronts.cards) + len(self.order.backs.cards)
-        status_format = "State: {state}, Action: {action}"
-        self.status_bar = self.manager.status_bar(
-            status_format=status_format,
-            state=f"{TEXT_BOLD}{self.state}{TEXT_END}",
-            action=f"{TEXT_BOLD}N/A{TEXT_END}",
-            position=1,
-        )
-        self.download_bar = self.manager.counter(total=num_images, desc="Images Downloaded", position=2)
-        self.upload_bar = self.manager.counter(total=num_images, desc="Images Uploaded", position=3)
-
-        self.status_bar.refresh()
-        self.download_bar.refresh()
-        self.upload_bar.refresh()
-
-
-@attr.s
-class AutofillDriver(OrderStatusBarBaseClass):
+class AutofillDriver:
     driver: webdriver.remote.webdriver.WebDriver = attr.ib(
         default=None
     )  # delay initialisation until XML is selected and parsed
@@ -63,7 +36,13 @@ class AutofillDriver(OrderStatusBarBaseClass):
         init=False,
         default="https://www.makeplayingcards.com/design/custom-blank-card.html",
     )
+    order: CardOrder = attr.ib(default=attr.Factory(CardOrder.from_xml_in_folder))
+    state: str = attr.ib(init=False, default=States.initialising)
     action: Optional[str] = attr.ib(init=False, default=None)
+    manager: enlighten.Manager = attr.ib(init=False, default=attr.Factory(enlighten.get_manager))
+    status_bar: enlighten.StatusBar = attr.ib(init=False, default=False)
+    download_bar: enlighten.Counter = attr.ib(init=False, default=None)
+    upload_bar: enlighten.Counter = attr.ib(init=False, default=None)
 
     # region initialisation
     def initialise_driver(self) -> None:
@@ -80,6 +59,22 @@ def initialise_driver(self) -> None:
 
         self.driver = driver
 
+    def configure_bars(self) -> None:
+        num_images = len(self.order.fronts.cards) + len(self.order.backs.cards)
+        status_format = "State: {state}, Action: {action}"
+        self.status_bar = self.manager.status_bar(
+            status_format=status_format,
+            state=f"{TEXT_BOLD}{self.state}{TEXT_END}",
+            action=f"{TEXT_BOLD}N/A{TEXT_END}",
+            position=1,
+        )
+        self.download_bar = self.manager.counter(total=num_images, desc="Images Downloaded", position=2)
+        self.upload_bar = self.manager.counter(total=num_images, desc="Images Uploaded", position=3)
+
+        self.status_bar.refresh()
+        self.download_bar.refresh()
+        self.upload_bar.refresh()
+
     def __attrs_post_init__(self) -> None:
         self.configure_bars()
         self.order.print_order_overview()
@@ -277,7 +272,7 @@ def upload_and_insert_image(self, image: CardImage) -> None:
             self.insert_image(pid, image, slots=unfilled_slot_numbers)
 
     def upload_and_insert_images(self, images: CardImageCollection) -> None:
-        for i in range(len(images.cards)):
+        for _ in range(len(images.cards)):
             image: CardImage = images.queue.get()
             if image.downloaded:
                 self.upload_and_insert_image(image)

autofill/src/order.py

@@ -117,7 +117,7 @@ def from_element(cls, element: Element) -> "CardImage":
         return card_image
 
     def download_image(self, queue: Queue["CardImage"], download_bar: enlighten.Counter) -> None:
-        if not self.file_exists() and not self.errored:
+        if not self.file_exists() and not self.errored and self.file_path is not None:
             self.errored = not download_google_drive_file(self.drive_id, self.file_path)
 
         if self.file_exists() and not self.errored:

autofill/src/pdf_maker.py

@@ -2,14 +2,18 @@
 from concurrent.futures import ThreadPoolExecutor
 
 import attr
+import enlighten
 import InquirerPy
 from fpdf import FPDF
-from src.constants import THREADS
-from src.driver import OrderStatusBarBaseClass
+from src.constants import THREADS, States
+from src.order import CardOrder
+from src.utils import TEXT_BOLD, TEXT_END
 
 
 @attr.s
-class PdfExporter(OrderStatusBarBaseClass):
+class PdfExporter:
+    order: CardOrder = attr.ib(default=attr.Factory(CardOrder.from_xml_in_folder))
+    state: str = attr.ib(init=False, default=States.initialising)
     pdf: FPDF = attr.ib(default=None)
     card_width_in_inches: float = attr.ib(default=2.73)
     card_height_in_inches: float = attr.ib(default=3.71)
@@ -19,6 +23,29 @@ class PdfExporter(OrderStatusBarBaseClass):
     save_path: str = attr.ib(default="")
     separate_faces: bool = attr.ib(default=False)
     current_face: str = attr.ib(default="all")
+    manager: enlighten.Manager = attr.ib(init=False, default=attr.Factory(enlighten.get_manager))
+    status_bar: enlighten.StatusBar = attr.ib(init=False, default=False)
+    download_bar: enlighten.Counter = attr.ib(init=False, default=None)
+    processed_bar: enlighten.Counter = attr.ib(init=False, default=None)
+
+    def configure_bars(self) -> None:
+        num_images = len(self.order.fronts.cards) + len(self.order.backs.cards)
+        status_format = "State: {state}"
+        self.status_bar = self.manager.status_bar(
+            status_format=status_format,
+            state=f"{TEXT_BOLD}{self.state}{TEXT_END}",
+            position=1,
+        )
+        self.download_bar = self.manager.counter(total=num_images, desc="Images Downloaded", position=2)
+        self.processed_bar = self.manager.counter(total=num_images, desc="Images Processed", position=3)
+
+        self.download_bar.refresh()
+        self.processed_bar.refresh()
+
+    def set_state(self, state: str) -> None:
+        self.state = state
+        self.status_bar.update(state=f"{TEXT_BOLD}{self.state}{TEXT_END}")
+        self.status_bar.refresh()
 
     def __attrs_post_init__(self) -> None:
         self.ask_questions()
@@ -45,22 +72,24 @@ def ask_questions(self) -> None:
                 + "the longer the processing will take and the larger the file size will be.",
                 "default": 60,
                 "when": lambda result: result["split_faces"] is False,
-                "transformer": lambda result: 1 if int(result) < 1 else int(result),
+                "transformer": lambda result: 1 if (int_result := int(result)) < 1 else int_result,
             },
         ]
         answers = InquirerPy.prompt(questions)
         if answers["split_faces"]:
             self.separate_faces = True
             self.number_of_cards_per_file = 1
         else:
-            self.number_of_cards_per_file = 1 if int(answers["cards_per_file"]) < 1 else int(answers["cards_per_file"])
+            self.number_of_cards_per_file = (
+                1 if (int_cards_per_file := int(answers["cards_per_file"])) < 1 else int_cards_per_file
+            )
 
     def generate_file_path(self) -> None:
         basename = os.path.basename(str(self.order.name))
         if not basename:
             basename = "cards.xml"
         file_name = os.path.splitext(basename)[0]
-        self.save_path = "export/%s/" % file_name
+        self.save_path = f"export/{file_name}/"
         os.makedirs(self.save_path, exist_ok=True)
         if self.separate_faces:
             for face in ["backs", "fronts"]:
@@ -77,8 +106,8 @@ def add_image(self, image_path: str) -> None:
     def save_file(self) -> None:
         extra = ""
         if self.separate_faces:
-            extra = "%s/" % self.current_face
-        self.pdf.output(self.save_path + extra + str(self.file_num) + ".pdf")
+            extra = f"{self.current_face}/"
+        self.pdf.output(f"{self.save_path}{extra}{self.file_num}.pdf")
 
     def download_and_collect_images(self) -> None:
         with ThreadPoolExecutor(max_workers=THREADS) as pool:
@@ -107,37 +136,38 @@ def execute(self) -> None:
         else:
             self.export()
 
-        print("Finished exporting files! They should be accessible at %s" % self.save_path)
+        print(f"Finished exporting files! They should be accessible at {self.save_path}.")
 
     def export(self) -> None:
         for slot in sorted(self.paths_by_slot.keys()):
             (back_path, front_path) = self.paths_by_slot[slot]
-            print("Working on slot %s" % str(slot))
+            self.set_state(f"Working on slot {slot}")
             if slot == 0:
                 self.generate_pdf()
             elif slot % self.number_of_cards_per_file == 0:
-                print("Saving PDF #%s" % str(self.file_num))
+                self.set_state(f"Saving PDF #{slot}")
                 self.save_file()
                 self.file_num = self.file_num + 1
                 self.generate_pdf()
-            print("Adding images for slot %s" % str(slot))
+            self.set_state(f"Adding images for slot {slot}")
             self.add_image(back_path)
             self.add_image(front_path)
+            self.processed_bar.update()
 
-        print("Saving PDF #%s" % str(self.file_num))
+        self.set_state(f"Saving PDF #{self.file_num}")
         self.save_file()
 
     def export_separate_faces(self) -> None:
         all_faces = ["backs", "fronts"]
         for slot in sorted(self.paths_by_slot.keys()):
             image_paths_tuple = self.paths_by_slot[slot]
-            print("Working on slot " + str(slot))
+            self.set_state(f"Working on slot {slot}")
             for face in all_faces:
                 face_index = all_faces.index(face)
                 self.current_face = face
                 self.generate_pdf()
                 self.add_image(image_paths_tuple[face_index])
-                print("Saving %s PDF for slot %s" % (face, str(slot)))
+                self.set_state(f"Saving {face} PDF for slot {slot}")
                 self.save_file()
                 if face_index == 1:
                     self.file_num = self.file_num + 1

autofill/src/utils.py

@@ -41,46 +41,62 @@ class ValidationException(Exception):
 
 @ratelimit.sleep_and_retry  # type: ignore  # `ratelimit` does not implement decorator typing correctly
 @ratelimit.limits(calls=1, period=0.1)  # type: ignore  # `ratelimit` does not implement decorator typing correctly
+def rate_limit_post_api_call(url: str, data: dict[str, Any], timeout: Optional[int] = None) -> requests.Response:
+    with requests.post(url, data=data, timeout=timeout) as r_info:
+        return r_info
+
+
+def safe_post_api_call(
+    url: str, data: dict[str, Any], expected_keys: set[str], max_tries: int = 3, timeout: Optional[int] = None
+) -> Optional[dict[str, Any]]:
+    tries = 0
+    while True:
+        try:
+            r_info = rate_limit_post_api_call(url=url, data=data, timeout=timeout)
+            r_json = r_info.json()
+            # validate contents of response
+            if (
+                r_info.status_code != 500
+                and len(expected_keys - r_json.keys()) == 0
+                and not any([bool(r_json[x]) is False for x in expected_keys])
+            ):
+                return r_json
+        except (requests.exceptions.RequestException, TimeoutError):
+            pass
+
+        tries += 1
+        if tries >= max_tries:
+            return None
+
+
 def get_google_drive_file_name(drive_id: str) -> Optional[str]:
     """
     Retrieve the name for the Google Drive file identified by `drive_id`.
     """
 
     if not drive_id:
         return None
-    try:
-        with requests.post(
-            constants.GoogleScriptsAPIs.image_name.value,
-            data={"id": drive_id},
-            timeout=30,
-        ) as r_info:
-            if r_info.status_code == 500:
-                return None
-            return r_info.json()["name"]
-    except requests.exceptions.Timeout:
-        return None
+    response = safe_post_api_call(
+        url=constants.GoogleScriptsAPIs.image_name, data={"id": drive_id}, timeout=30, expected_keys={"name"}
+    )
+    return response["name"] if response is not None else None
 
 
-@ratelimit.sleep_and_retry  # type: ignore  # `ratelimit` does not implement decorator typing correctly
-@ratelimit.limits(calls=1, period=0.1)  # type: ignore  # `ratelimit` does not implement decorator typing correctly
 def download_google_drive_file(drive_id: str, file_path: str) -> bool:
     """
     Download the Google Drive file identified by `drive_id` to the specified `file_path`.
     Returns whether the request was successful or not.
     """
 
-    with requests.post(
-        constants.GoogleScriptsAPIs.image_content.value,
-        data={"id": drive_id},
-    ) as r_contents:
-        if "<title>Error</title>" in r_contents.text:
-            # error occurred while attempting to retrieve from Google API
-            return False
-        filecontents = r_contents.json()["result"]
-        if len(filecontents) > 0:
+    response = safe_post_api_call(
+        url=constants.GoogleScriptsAPIs.image_content, data={"id": drive_id}, timeout=5 * 60, expected_keys={"result"}
+    )
+    if response is not None:
+        file_contents = response["result"]
+        if len(file_contents) > 0:
             # Download the image
             with open(file_path, "wb") as f:
-                f.write(np.array(filecontents, dtype=np.uint8))
+                f.write(np.array(file_contents, dtype=np.uint8))
                 return True
     return False