write progress to terminal, even if output is redirected
authorJacob Lifshay <programmerjake@gmail.com>
Tue, 30 Aug 2022 03:53:55 +0000 (20:53 -0700)
committerJacob Lifshay <programmerjake@gmail.com>
Tue, 30 Aug 2022 03:53:55 +0000 (20:53 -0700)
src/budget_sync/util.py

index f26c1a0aef849b209d817a332ae25139f9f0c584..7a4170c24105afbeb37156eb5db74d9c014f0b24 100644 (file)
@@ -5,6 +5,7 @@ from bugzilla.bug import Bug
 from typing import Any, Callable, Dict, Iterator, List, Type, Union
 from enum import Enum
 from io import StringIO
+import os
 
 
 class BugStatus(Enum):
@@ -37,13 +38,31 @@ class BugStatus(Enum):
 def all_bugs(bz: Bugzilla) -> Iterator[Bug]:
     chunk_start = 1
     chunk_size = 100
-    while True:
-        bugs = bz.getbugs(list(range(chunk_start, chunk_start + chunk_size)))
-        chunk_start += chunk_size
-        print("bugs loaded", len(bugs), chunk_start, flush=True)
-        if len(bugs) == 0:
-            return
-        yield from bugs
+    try:
+        if hasattr(os, "ctermid"):
+            term = open(os.ctermid(), "wt", encoding="utf-8")
+        elif os.name == "nt":
+            term = open("CONOUT$", "wt", encoding="utf-8")
+        else:
+            term = None
+    except OSError:
+        # no terminal available
+        term = None
+    try:  # can't use `with` since it doesn't work with None
+        while True:
+            bugs = list(range(chunk_start, chunk_start + chunk_size))
+            bugs = bz.getbugs(bugs)
+            chunk_start += chunk_size
+            # progress indicator, should go to terminal
+            if term is not None:
+                print("bugs loaded", len(bugs), chunk_start, flush=True,
+                      file=term)
+            if len(bugs) == 0:
+                return
+            yield from bugs
+    finally:
+        if term is not None:
+            term.close()
 
 
 class SequencePrettyPrinter: