feat: check existence of labels

use `hut graphql` instead

1 files changed, 52 insertions, 11 deletions

diff --git a/import_issues.py b/import_issues.py
index 83823a2..fca08a8 100755
--- a/import_issues.py
+++ b/import_issues.py
@@ -123,8 +123,8 @@
 
 import argparse
 import csv
-import logging
 import json
+import logging
 import os
 import re
 import subprocess
@@ -137,14 +137,44 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Dict, List, Optional
 
-
-ID_RE = re.compile(r"^[0-9]+$")
-
 logging.basicConfig(
     format="%(levelname)s:%(funcName)s:%(message)s",
     level=logging.DEBUG,
     stream=sys.stdout,
 )
+
+ID_RE = re.compile(r"^[0-9]+$")
+
+
+def get_labels(tracker: str) -> list[dict[str, str]]:
+    """
+    collects labels for your named tracker
+
+    param: tracker: name of the tracker
+    return: list of all labels in the tracker
+    """
+    query = (
+        'query { me { tracker(name: "'
+        + tracker
+        + '") { labels { results { id, name, foregroundColor, backgroundColor, created } } } }}'
+    )
+
+    try:
+        ret = subprocess.run(
+            ["hut", "graphql", "todo", "--stdin"],
+            input=query,
+            text=True,
+            check=True,
+            capture_output=True,
+        )
+    except subprocess.CalledProcessError as ex:
+        raise RuntimeError(
+            f"hut failed with excitcode {ex.returncode} and stderr:\n{ex.stderr}"
+        ) from ex
+    data = json.loads(ret.stdout)
+    return data["me"]["tracker"]["labels"]["results"]
+
+
 log = logging.getLogger()
 
 email_count = 0
@@ -448,6 +478,17 @@ def close_ticket(
         )
 
 
+def ensure_label(tracker: str, name: str, bg_color: str, fg_color: str = "#FFFFFF"):
+    labels = get_labels(tracker.split("/", 1)[1])
+    if not ([x for x in labels if x["name"] == name]):
+        run_hut(
+            ["label", "create"],
+            tracker,
+            None,
+            ["--background", bg_color, "--foreground", fg_color, name],
+        )
+
+
 def run(
     *,
     smtp,
@@ -823,7 +864,7 @@ def main():
     args = vars(parser.parse_args())
 
     export_dir = args["export_dir"]
-    assert export_dir, f"Must have a exported project directory."
+    assert export_dir, "Must have a exported project directory."
     export_dir_path = Path(export_dir)
     assert (
         export_dir_path.is_dir()
@@ -837,26 +878,26 @@ def main():
     skip_unknown_labels = args["skip_unknown_labels"]
     assert (
         labels_file or skip_labels
-    ), f"One of --labels-file or --skip-labels must be provided."
+    ), "One of --labels-file or --skip-labels must be provided."
 
     users_file = args["users_file"]
     skip_users = args["skip_users"]
     skip_unknown_users = args["skip_unknown_users"]
     assert (
         skip_users or users_file
-    ), f"One of --users-file or --skip-users must be provided."
+    ), "One of --users-file or --skip-users must be provided."
 
     skip_missing_issues = args["skip_missing_issues"]
     create_missing_issues = args["create_missing_issues"]
     assert not (
         skip_missing_issues and create_missing_issues
-    ), f"Can accept at most one of --skip-missing-issues and --create-missing-issues."
+    ), "Can accept at most one of --skip-missing-issues and --create-missing-issues."
 
     include_confidential = args["include_confidential"]
     skip_confidential = args["skip_confidential"]
     assert not (
         include_confidential and skip_confidential
-    ), f"Can accept at most one of --include-confidential and --skip-confidential."
+    ), "Can accept at most one of --include-confidential and --skip-confidential."
 
     srht_owner = args["srht_owner"]
     srht_tracker = args["srht_tracker"]
@@ -874,8 +915,8 @@ def main():
         smtp_user = args["smtp_user"] or os.environ.get("SMTP_USER", None)
         smtp_password = args["smtp_password"] or os.environ.get("SMTP_PASSWORD", None)
 
-        assert smtp_user, f"No SMTP user given."
-        assert smtp_password, f"No SMTP password given."
+        assert smtp_user, "No SMTP user given."
+        assert smtp_password, "No SMTP password given."
 
         log.info("Connecting to %s:%d, user %r.", smtp_host, smtp_port, smtp_user)