#! /usr/bin/env python3
import argparse
import sys
import logging
import requests
import subprocess
from textwrap import dedent
from pathlib import Path
from datetime import datetime
from typing import Optional
try:
import gitlab
except ImportError:
print("Please install the 'python-gitlab' package from pip", file=sys.stderr)
sys.exit(1)
TRELLO_KEY = "fba640a85f15c91e93e6b3f88e59489c"
TRELLO_BOARD_ID = "tw3Cn3L6"
TRELLO_LIST_ID = "6593166e9dd338621ed6848d"
TRELLO_TAGS = {
"company": "64e03ac77d27032282436d28", # Tag used to sort by company
"waiting_to_merge": "65524315edf2d2edb0cc5d09", # Tag used to indicate a MR is waiting to merge
"draft": "65fdd81c83e5d6e00f1b9721", # Tag used to indicate a draft MR
}
GITLAB_ENDPOINT = "http://gitlab.guru-domain.gurustudio.com"
MY_ID = 64
logger = logging.getLogger(__name__)
def main() -> int:
# Handle program arguments
ap = argparse.ArgumentParser(
prog="guru-sync-issues", description="Sync issues from GitLab to Trello"
)
ap.add_argument(
"-v", "--verbose", help="Enable verbose logging", action="store_true"
)
args = ap.parse_args()
# Configure logging
logging.basicConfig(
level=logging.DEBUG if args.verbose else logging.INFO,
format="%(levelname)s: %(message)s",
)
# Call `ewp-secrets` to obtain the GitLab token
secrets_proc = subprocess.run(
[
(Path(__file__).parent / "ewp-secrets").as_posix(),
"load",
"-n",
"gurustudio",
"-k",
"gitlab-pat",
],
capture_output=True,
)
# If the secret manager failed, exit
if secrets_proc.returncode != 0:
print("Failed to load GitLab PAT", file=sys.stderr)
print(
"Please run `ewp-secrets store -n gurustudio -k gitlab-pat` to set a token",
file=sys.stderr,
)
return 1
# Extract the GitLab token
gitlab_token = secrets_proc.stdout.decode().strip()
# Call `ewp-secrets` to obtain the Trello API token
secrets_proc = subprocess.run(
[
(Path(__file__).parent / "ewp-secrets").as_posix(),
"load",
"-n",
"ewpratten",
"-k",
"trello-api-token",
],
capture_output=True,
)
# If the secret manager failed, exit
if secrets_proc.returncode != 0:
print("Failed to load Trello API token", file=sys.stderr)
print(
"Please run `ewp-secrets store -n ewpratten -k trello-api-token` to set a token",
file=sys.stderr,
)
return 1
# Extract the Trello API token
trello_api_token = secrets_proc.stdout.decode().strip()
# Try to ping the gitlab server, and exit if it fails
try:
logger.debug("Attempting to connect to GitLab server")
requests.get(GITLAB_ENDPOINT, timeout=2)
except requests.exceptions.ConnectionError:
logger.error("Could not connect to GitLab server")
return 1
# Authenticate with GitLab
gitlab_client = gitlab.Gitlab(
GITLAB_ENDPOINT, private_token=gitlab_token, user_agent="guru-sync-issues"
)
# Access the main group
pp_group = gitlab_client.groups.get("pipeline-planning")
# Find all open issues I'm assigned to
open_issues = []
open_issues.extend(
gitlab_client.issues.list(state="opened", get_all=True, assignee_id=MY_ID)
)
open_issues.extend(
pp_group.issues.list(state="opened", get_all=True, assignee_id=MY_ID)
)
# De-dupe the issues
open_issues = list(set(open_issues))
# Find all open MRs I'm assigned to
open_mrs = []
open_mrs.extend(
gitlab_client.mergerequests.list(
state="opened", get_all=True, assignee_id=MY_ID
)
)
open_mrs.extend(
pp_group.mergerequests.list(state="opened", get_all=True, assignee_id=MY_ID)
)
# De-dupe the MRs
open_mrs = list(set(open_mrs))
# Log findings
logger.info(f"Found {len(open_issues)} open issues")
logger.info(f"Found {len(open_mrs)} open merge requests")
# Get all cards in Trello
response = requests.get(
f"https://api.trello.com/1/boards/{TRELLO_BOARD_ID}/cards",
params={
"key": TRELLO_KEY,
"token": trello_api_token,
},
)
response.raise_for_status()
trello_cards = response.json()
# Iterate over each GitLab actionable
actionables = [("issue", issue) for issue in open_issues] + [
("merge request", mr) for mr in open_mrs
]
for ty, issue in actionables:
# Get the URL of the issue
issue_url = issue.web_url
# Check if there is a card that references this URL
card = next((c for c in trello_cards if issue_url in c["desc"]), None)
# If none exists, make a new card
if not card:
logger.info(f"Creating card for issue: {issue.title}")
# Build params
card_params = {
"idList": TRELLO_LIST_ID,
"name": issue.title,
"key": TRELLO_KEY,
"token": trello_api_token,
"pos": "top",
"desc": dedent(
f"""
## Linked GitLab Issues
- {issue_url}
---
"""
),
"idLabels": TRELLO_TAGS["company"],
}
# Make the card
response = requests.post(
"https://api.trello.com/1/cards",
params=card_params,
)
response.raise_for_status()
# Capture the card for later
card = response.json()
logger.info(f"Created card: {card['id']}")
# Apply lables to MRs
if ty == "merge request":
if issue.title.startswith("Draft:") or issue.title.startswith("WIP:"):
# Check if the card already has the 'Draft' label
if any(
label["id"] == TRELLO_TAGS["draft"] for label in card["labels"]
):
continue
logger.info(f"Adding 'Draft' label to card {card['id']}")
response = requests.post(
f"https://api.trello.com/1/cards/{card['id']}/idLabels",
params={
"key": TRELLO_KEY,
"token": trello_api_token,
"value": TRELLO_TAGS["draft"],
},
)
response.raise_for_status()
else:
# Check if the card already has the 'Waiting to Merge' label
if any(
label["id"] == TRELLO_TAGS["waiting_to_merge"] for label in card["labels"]
):
continue
logger.info(f"Adding 'Waiting to Merge' label to card {card['id']}")
response = requests.post(
f"https://api.trello.com/1/cards/{card['id']}/idLabels",
params={
"key": TRELLO_KEY,
"token": trello_api_token,
"value": TRELLO_TAGS["waiting_to_merge"],
},
)
response.raise_for_status()
# Determine the due date assigned in GitLab
if ty != "merge request":
due_date: Optional[str] = issue.due_date
if due_date:
# Check if the card already has a due date
card_due_date = card.get("due", None)
if card_due_date:
card_due_date = card_due_date.split("T")[0]
# Convert to a useful format
due_date = datetime.strptime(due_date, "%Y-%m-%d")
card_due_date = datetime.strptime(card_due_date, "%Y-%m-%d") if card_due_date else None
# If the card has an earlier due date, skip
if card_due_date and due_date < card_due_date:
logger.debug(f"Skipping due date update for {card['id']} because it has an earlier due date already")
continue
# Update the due date
logger.info(f"Updating due date for card {card['id']} to {due_date}")
response = requests.put(
f"https://api.trello.com/1/cards/{card['id']}",
params={
"key": TRELLO_KEY,
"token": trello_api_token,
"due": due_date.isoformat(),
},
)
return 0
if __name__ == "__main__":
sys.exit(main())