673 lines
28 KiB
Python
673 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Jira ↔ Jotty Bidirectional Sync Script
|
|
|
|
Syncs Jira stories to Jotty checklists and keeps status in sync both ways:
|
|
- Jira → Jotty: Every story becomes a checklist item (task type)
|
|
- Jotty → Jira: Checking/unchecking an item updates the mapped Jira issue status
|
|
|
|
Usage:
|
|
python jira_jotty_sync.py --direction=jira-to-jotty # Pull stories into Jotty
|
|
python jira_jotty_sync.py --direction=jotty-to-jira # Push Jotty check changes to Jira
|
|
python jira_jotty_sync.py --direction=both # Full two-way sync
|
|
|
|
Configuration is stored in sync_config.json (see template below).
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auto-venv setup: if not running inside a virtual environment, create one,
|
|
# install requirements, and re-execute this script with the venv Python.
|
|
# ---------------------------------------------------------------------------
|
|
def _ensure_venv():
|
|
"""Create .venv if missing, install requirements, then re-execute self."""
|
|
# Check if we're already in a venv
|
|
in_venv = getattr(sys, "prefix", None) != getattr(sys, "base_prefix", None)
|
|
if in_venv:
|
|
return
|
|
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
venv_path = os.path.join(script_dir, ".venv")
|
|
requirements_file = os.path.join(script_dir, "requirements.txt")
|
|
|
|
# Create virtual environment if it doesn't exist
|
|
if not os.path.isdir(venv_path):
|
|
print(f"[setup] Creating virtual environment in {venv_path} ...")
|
|
subprocess.run([sys.executable, "-m", "venv", venv_path], check=True)
|
|
|
|
# Determine the venv python executable (cross-platform)
|
|
if os.name == "nt": # Windows
|
|
venv_python = os.path.join(venv_path, "Scripts", "python.exe")
|
|
else:
|
|
venv_python = os.path.join(venv_path, "bin", "python")
|
|
|
|
# Install requirements if not already installed
|
|
print(f"[setup] Installing dependencies ...")
|
|
subprocess.run(
|
|
[venv_python, "-m", "pip", "install", "--upgrade", "-r", requirements_file],
|
|
check=True,
|
|
)
|
|
|
|
# Re-execute this script using the venv Python
|
|
print("[setup] Relaunching with virtual environment ...")
|
|
sys.exit(subprocess.call([venv_python, __file__] + sys.argv[1:]))
|
|
|
|
|
|
_ensure_venv()
|
|
del _ensure_venv # Clean up namespace
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
from dotenv import load_dotenv
|
|
import requests
|
|
from jira import JIRA
|
|
from jira.exceptions import JIRAError as JiraError
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load .env file from the same directory as this script (if it exists)
|
|
# ---------------------------------------------------------------------------
|
|
_env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
|
|
load_dotenv(_env_path)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Logging
|
|
# ---------------------------------------------------------------------------
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
)
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jira ↔ Jotty status mapping
|
|
# ---------------------------------------------------------------------------
|
|
# Maps Jira workflow statuses → Jira Apollo (Jotty Kanban) statuses.
|
|
# Based on your Jira Apollo board: To Do | In Progress | Completed | Paused
|
|
JIRA_TO_JOTTY_STATUS: dict[str, str] = {
|
|
# To Do column — items waiting to be worked on
|
|
"NEW": "todo",
|
|
"WAITING": "todo",
|
|
"READY": "todo",
|
|
"TO DO": "todo",
|
|
"IN REFINEMENT": "todo",
|
|
|
|
# In Progress column — active work in flight
|
|
"IN BUSINESS DESIGN": "in_progress",
|
|
"PREPARATION": "in_progress",
|
|
"IN PROGRESS": "in_progress",
|
|
"PO REVIEW": "in_progress",
|
|
"READY FOR QA": "in_progress",
|
|
"TEST": "in_progress",
|
|
"CONFIGURATION INTEGRATION": "in_progress",
|
|
|
|
# Completed column — finished work
|
|
"DONE": "completed",
|
|
"DEPLOYED": "completed",
|
|
"LIVE": "completed",
|
|
"TEST COMPLETED": "completed",
|
|
|
|
# Paused column — blocked, rejected, or on hold
|
|
"IN QA": "paused", # waiting for QA to pick up
|
|
"REJECTED": "paused",
|
|
}
|
|
|
|
# Maps Jira Apollo (Jotty Kanban) statuses → Jira workflow statuses.
|
|
# When you move an item in Jira Apollo, this determines the target Jira status.
|
|
JOTTY_TO_JIRA_STATUS: dict[str, str] = {
|
|
"todo": "TO DO",
|
|
"in_progress": "IN PROGRESS",
|
|
"completed": "DONE",
|
|
"paused": "WAITING", # paused → waiting (can resume)
|
|
}
|
|
|
|
# JQL query used to fetch stories from Jira. Adjust as needed.
|
|
DEFAULT_JIRA_QUERY = (
|
|
'project = "${PROJECT_KEY}" AND issuetype = Story'
|
|
' AND status NOT IN ("DONE", "DEPLOYED", "LIVE") ORDER BY created DESC'
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data classes
|
|
# ---------------------------------------------------------------------------
|
|
@dataclass
|
|
class SyncConfig:
|
|
"""All configuration needed for the sync."""
|
|
|
|
# Jira settings (self-hosted uses token_auth with PAT)
|
|
jira_base_url: str = "" # e.g. https://jira.your-domain.com
|
|
jira_api_token: str = "" # Personal Access Token (PAT) or API token
|
|
project_key: str = "" # e.g. PROJ
|
|
jql_query: str = DEFAULT_JIRA_QUERY
|
|
|
|
# Jotty settings
|
|
jotty_base_url: str = "" # e.g. https://jotty.your-domain.com
|
|
jotty_api_key: str = ""
|
|
checklist_title: str = "Jira Apollo" # name of the Kanban board in Jotty to sync against
|
|
checklist_category: str = "Work"
|
|
|
|
# Persistence
|
|
mapping_file: str = "sync_mapping.json" # stores Jira→Jotty item index map
|
|
|
|
|
|
@dataclass
|
|
class SyncMapping:
|
|
"""Maps a Jira issue key → (checklist UUID, item_index)."""
|
|
|
|
mappings: dict[str, dict] = field(default_factory=dict)
|
|
# Example entry:
|
|
# { "PROJ-42": {"checklist_id": "...", "item_index": 0} }
|
|
|
|
def save(self, path: str):
|
|
with open(path, "w") as f:
|
|
json.dump(self.mappings, f, indent=2)
|
|
|
|
@classmethod
|
|
def load(cls, path: str) -> "SyncMapping":
|
|
instance = cls()
|
|
if os.path.exists(path):
|
|
try:
|
|
with open(path) as f:
|
|
instance.mappings = json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
log.warning("Could not load mapping file; starting fresh.")
|
|
return instance
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jira API client — using the official jira library
|
|
# ---------------------------------------------------------------------------
|
|
class JiraClient:
|
|
"""Jira client wrapping the official jira library for self-hosted/Cloud."""
|
|
|
|
def __init__(self, base_url: str, api_token: str):
|
|
options = {"server": base_url.rstrip("/")}
|
|
# For self-hosted (Server/Data Center), use token_auth
|
|
# For Cloud, use basic_auth=("email", "api_token")
|
|
self.jira = JIRA(options=options, token_auth=api_token)
|
|
|
|
def search_stories(self, jql: str) -> list[dict]:
|
|
"""Run a JQL search and return every matching issue as a dict."""
|
|
log.info("Jira search JQL=%s", jql[:80])
|
|
issues = self.jira.search_issues(jql, maxResults=100)
|
|
results = [self._issue_to_dict(issue) for issue in issues]
|
|
log.info("Jira search returned %d issues total.", len(results))
|
|
return results
|
|
|
|
def get_issue(self, issue_key: str) -> dict:
|
|
"""Fetch a single issue by key."""
|
|
try:
|
|
issue = self.jira.issue(issue_key)
|
|
return self._issue_to_dict(issue)
|
|
except JiraError as exc:
|
|
log.error("Failed to fetch %s: %s", issue_key, exc)
|
|
raise
|
|
|
|
def update_status(self, issue_key: str, status_name: str) -> None:
|
|
"""Update the status of an issue using transitions."""
|
|
try:
|
|
issue = self.jira.issue(issue_key)
|
|
# Find available transitions
|
|
transitions = self.jira.transitions(issue)
|
|
target_lower = status_name.lower()
|
|
|
|
chosen = None
|
|
for t in transitions:
|
|
to_status = t["name"].lower() if isinstance(t, dict) else t["to"]["name"].lower()
|
|
if to_status == target_lower:
|
|
chosen = t["id"] if isinstance(t, dict) else t["id"]
|
|
break
|
|
|
|
if not chosen:
|
|
log.warning("No transition found for status '%s' on %s. Trying direct update.", status_name, issue_key)
|
|
# Fallback: try direct field update
|
|
issue.update(fields={"status": {"name": status_name}})
|
|
log.info("Updated %s → status '%s' (direct)", issue_key, status_name)
|
|
return
|
|
|
|
self.jira.transition_issue(issue, chosen)
|
|
log.info("Updated %s → status '%s' (via transition %s)", issue_key, status_name, chosen)
|
|
except JiraError as exc:
|
|
log.error("Failed to update status for %s: %s", issue_key, exc)
|
|
raise
|
|
|
|
@staticmethod
|
|
def _issue_to_dict(issue) -> dict:
|
|
"""Convert a jira.Issue object to a plain dict."""
|
|
return {
|
|
"key": issue.key,
|
|
"fields": {
|
|
"summary": getattr(issue.fields, "summary", ""),
|
|
"status": {"name": getattr(getattr(issue.fields, "status", None), "name", "")},
|
|
"description": getattr(issue.fields, "description", ""),
|
|
},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jotty API client
|
|
# ---------------------------------------------------------------------------
|
|
class JottyClient:
|
|
def __init__(self, base_url: str, api_key: str):
|
|
self.base = base_url.rstrip("/")
|
|
self.headers = {"x-api-key": api_key, "Content-Type": "application/json"}
|
|
|
|
def _get(self, endpoint: str) -> dict:
|
|
url = f"{self.base}{endpoint}" if endpoint.startswith("/") else f"{self.base}/{endpoint}"
|
|
log.debug("GET %s", url)
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def _post(self, endpoint: str, body: dict) -> dict:
|
|
url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint
|
|
log.debug("POST %s body=%s", url, json.dumps(body)[:200])
|
|
r = requests.post(url, headers=self.headers, json=body, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def _put(self, endpoint: str, body: dict | None = None) -> dict:
|
|
url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint
|
|
log.debug("PUT %s body=%s", url, json.dumps(body)[:200] if body else "None")
|
|
r = requests.put(url, headers=self.headers, json=body, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def _patch(self, endpoint: str, body: dict) -> dict:
|
|
url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint
|
|
log.debug("PATCH %s body=%s", url, json.dumps(body)[:200])
|
|
r = requests.patch(url, headers=self.headers, json=body, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def _delete(self, endpoint: str) -> dict:
|
|
url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint
|
|
log.debug("DELETE %s", url)
|
|
r = requests.delete(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
# -- checklists -------------------------------------------------------
|
|
def get_checklists(self) -> list[dict]:
|
|
data = self._get("/api/checklists")
|
|
return data.get("checklists", [])
|
|
|
|
def find_or_create_task_list(self, title: str, category: str) -> dict:
|
|
"""Find an existing task checklist by title or create a new one."""
|
|
all_checklists = self.get_checklists()
|
|
# Match by title; accept "task" or "kanban" type (Jotty uses kanban for boards like this)
|
|
valid_types = {"task", "kanban"}
|
|
existing = [cl for cl in all_checklists if cl["title"] == title and cl.get("type") in valid_types]
|
|
if existing:
|
|
log.info("Found existing task list '%s' (id=%s)", title, existing[0]["id"])
|
|
return existing[0]
|
|
|
|
log.info("Creating new task list '%s'", title)
|
|
body = {
|
|
"title": title,
|
|
"category": category,
|
|
"type": "task",
|
|
"statuses": [
|
|
{"id": "todo", "label": "To Do", "order": 0},
|
|
{"id": "in_progress", "label": "In Progress", "order": 1},
|
|
{"id": "completed", "label": "Completed", "order": 2},
|
|
],
|
|
}
|
|
resp = self._post("/api/tasks", body)
|
|
return resp.get("data", {})
|
|
|
|
# -- task items -------------------------------------------------------
|
|
def get_task(self, task_id: str) -> dict:
|
|
"""Get a task (checklist) by ID.
|
|
|
|
The API returns the task object directly at the top level.
|
|
"""
|
|
data = self._get(f"/api/tasks/{task_id}")
|
|
# Handle both wrapped and unwrapped responses
|
|
if "task" in data:
|
|
return data["task"]
|
|
return data
|
|
|
|
def create_item(self, task_id: str, text: str, status: str = "todo") -> dict:
|
|
body = {"text": text, "status": status}
|
|
resp = self._post(f"/api/tasks/{task_id}/items", body)
|
|
return resp.get("data", {})
|
|
|
|
def update_item_text(self, list_id: str, item_index: int | str, text: str) -> None:
|
|
"""Update the text/description of a checklist item.
|
|
|
|
Uses /api/checklists/{listId}/items/{itemIndex} (PATCH) which works for both
|
|
regular checklists and task-type checklists.
|
|
"""
|
|
self._patch(f"/api/checklists/{list_id}/items/{item_index}", {"text": text})
|
|
|
|
def update_item_status(self, task_id: str, item_index: int | str, new_status: str) -> None:
|
|
"""Update the Kanban status of a task checklist item."""
|
|
self._put(f"/api/tasks/{task_id}/items/{item_index}/status", {"status": new_status})
|
|
|
|
def check_item(self, task_id: str, item_index: int | str) -> None:
|
|
"""Mark an item as completed (check)."""
|
|
self._put(f"/api/tasks/{task_id}/items/{item_index}/check")
|
|
|
|
def uncheck_item(self, task_id: str, item_index: int | str) -> None:
|
|
"""Mark an item as incomplete (un-check)."""
|
|
self._put(f"/api/tasks/{task_id}/items/{item_index}/uncheck")
|
|
|
|
def delete_item(self, list_id: str, item_index: int | str) -> None:
|
|
"""Delete a checklist item.
|
|
|
|
Uses /api/checklists endpoint which works for both regular and task checklists.
|
|
"""
|
|
self._delete(f"/api/checklists/{list_id}/items/{item_index}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sync engine
|
|
# ---------------------------------------------------------------------------
|
|
class SyncEngine:
|
|
def __init__(self, config: SyncConfig):
|
|
self.config = config
|
|
self.jira = JiraClient(
|
|
config.jira_base_url, config.jira_api_token,
|
|
)
|
|
self.jotty = JottyClient(config.jotty_base_url, config.jotty_api_key)
|
|
self.mapping = SyncMapping.load(config.mapping_file)
|
|
|
|
# -- Jira → Jotty -----------------------------------------------------
|
|
def sync_jira_to_jotty(self) -> None:
|
|
"""Pull stories from Jira and create/update them as checklist items in Jotty.
|
|
|
|
Jira is the source of truth. Each run scans Jotty fresh and reconciles:
|
|
- Existing items are updated to match Jira status/text
|
|
- Missing items are created
|
|
- Extra/duplicate items are deleted
|
|
"""
|
|
jql = self.config.jql_query.replace("${PROJECT_KEY}", self.config.project_key)
|
|
issues = self.jira.search_stories(jql)
|
|
|
|
task_list = self.jotty.find_or_create_task_list(
|
|
self.config.checklist_title, self.config.checklist_category,
|
|
)
|
|
task_id = task_list["id"]
|
|
|
|
# Build the desired state from Jira: key -> (summary, status_raw, description)
|
|
jira_map: dict[str, tuple] = {}
|
|
for issue in issues:
|
|
key = issue["key"]
|
|
summary = issue["fields"].get("summary", "")
|
|
status_raw = (issue["fields"].get("status") or {}).get("name", "To Do")
|
|
description = issue["fields"].get("description") or ""
|
|
jira_map[key] = (summary, status_raw, description)
|
|
|
|
# Scan current Jotty items and build a map: jira_key -> first matching index
|
|
current_task = self.jotty.get_task(task_id)
|
|
current_items = current_task.get("items", [])
|
|
|
|
# Track which indices we've used so we can delete extras/duplicates later
|
|
key_to_idx: dict[str, int] = {} # jira_key -> single index to use
|
|
extra_indices: list[int] = [] # indices to delete (duplicates or orphaned)
|
|
|
|
for item in current_items:
|
|
text = item.get("text", "")
|
|
jira_key = self._extract_jira_key(text)
|
|
if not jira_key:
|
|
continue # skip items without a Jira key — they're orphaned
|
|
if jira_key in jira_map and jira_key not in key_to_idx:
|
|
# First match for this Jira issue — keep it
|
|
key_to_idx[jira_key] = item["index"]
|
|
else:
|
|
# Duplicate or orphaned — mark for deletion
|
|
extra_indices.append(item["index"])
|
|
|
|
# Update existing items (status + text) and delete extras
|
|
for jira_key, idx in key_to_idx.items():
|
|
summary, status_raw, description = jira_map[jira_key]
|
|
jotty_status = JIRA_TO_JOTTY_STATUS.get(status_raw, "todo")
|
|
|
|
try:
|
|
self.jotty.update_item_status(task_id, idx, jotty_status)
|
|
log.info("Updated %s → Jotty status '%s' (index %d)", jira_key, jotty_status, idx)
|
|
except Exception as exc:
|
|
log.warning("Could not update status for %s at index %d: %s", jira_key, idx, exc)
|
|
|
|
# Delete duplicate/orphaned items (reverse order to preserve indices)
|
|
for idx in sorted(extra_indices, reverse=True):
|
|
try:
|
|
self.jotty.delete_item(task_id, idx)
|
|
log.info("Deleted extra item at index %d", idx)
|
|
except Exception as exc:
|
|
log.warning("Could not delete item at index %d: %s", idx, exc)
|
|
|
|
# Create items for any Jira issues that don't have a corresponding Jotty item yet
|
|
for jira_key in jira_map:
|
|
if jira_key not in key_to_idx:
|
|
summary, status_raw, description = jira_map[jira_key]
|
|
jotty_status = JIRA_TO_JOTTY_STATUS.get(status_raw, "todo")
|
|
|
|
# Create item with just the title (Jotty truncates multi-line text)
|
|
create_text = f"[{jira_key}] {summary}"
|
|
self.jotty.create_item(task_id, create_text, jotty_status)
|
|
|
|
# Discover the newly created item's index
|
|
new_task = self.jotty.get_task(task_id)
|
|
for it in new_task.get("items", []):
|
|
if jira_key in it.get("text", ""):
|
|
idx = it["index"]
|
|
key_to_idx[jira_key] = idx
|
|
|
|
# Add Jira URL and description preview via PATCH on /api/checklists
|
|
issue_url = f"{self.config.jira_base_url}/browse/{jira_key}"
|
|
patch_body = {"description": issue_url}
|
|
if description and len(str(description)) > 0:
|
|
desc_preview = str(description).replace("\n", " ").strip()[:200]
|
|
patch_body["description"] += f"\n{desc_preview}"
|
|
|
|
try:
|
|
self.jotty._patch(f"/api/checklists/{task_id}/items/{idx}", patch_body)
|
|
log.info("Created new item for %s at index %d with status '%s' and description", jira_key, idx, jotty_status)
|
|
except Exception as exc:
|
|
log.warning("Could not set description for %s at index %d: %s", jira_key, idx, exc)
|
|
|
|
break
|
|
|
|
# Save the mapping (single index per key — Jira is source of truth)
|
|
self.mapping.mappings = {
|
|
k: {"checklist_id": task_id, "item_index": idx}
|
|
for k, idx in key_to_idx.items()
|
|
}
|
|
self.mapping.save(self.config.mapping_file)
|
|
log.info("Jira→Jotty sync complete. %d issues processed.", len(issues))
|
|
|
|
# -- Jotty → Jira -----------------------------------------------------
|
|
def sync_jotty_to_jira(self) -> None:
|
|
"""Read the current state of checklist items and push status changes to Jira."""
|
|
task_list = self.jotty.find_or_create_task_list(
|
|
self.config.checklist_title, self.config.checklist_category,
|
|
)
|
|
task_id = task_list["id"]
|
|
|
|
# Refresh mapping from disk in case another process changed it.
|
|
self.mapping = SyncMapping.load(self.config.mapping_file)
|
|
|
|
task_data = self.jotty.get_task(task_id)
|
|
items = task_data.get("items", [])
|
|
|
|
synced_count = 0
|
|
for item in items:
|
|
text = item.get("text", "")
|
|
jira_key = self._extract_jira_key(text)
|
|
if not jira_key or jira_key not in self.mapping.mappings:
|
|
continue
|
|
|
|
current_status = item.get("status", "todo")
|
|
completed = item.get("completed", False)
|
|
|
|
# Determine target Jira status from the Jotty item state.
|
|
if completed:
|
|
jira_status_name = JOTTY_TO_JIRA_STATUS.get("completed", "Done")
|
|
else:
|
|
jira_status_name = JOTTY_TO_JIRA_STATUS.get(current_status, "To Do")
|
|
|
|
# Check if a status change is needed
|
|
try:
|
|
issue_data = self.jira.get_issue(jira_key)
|
|
current_jira_status = (issue_data["fields"].get("status") or {}).get("name", "")
|
|
except Exception as exc:
|
|
log.warning("Could not fetch Jira issue %s: %s. Skipping.", jira_key, exc)
|
|
continue
|
|
|
|
if current_jira_status == jira_status_name:
|
|
log.debug("%s already at '%s' — no change needed.", jira_key, jira_status_name)
|
|
continue
|
|
|
|
self.jira.update_status(jira_key, jira_status_name)
|
|
log.info("Pushed %s → Jira status '%s' (was '%s')", jira_key, jira_status_name, current_jira_status)
|
|
synced_count += 1
|
|
|
|
log.info("Jotty→Jira sync complete. %d issues updated.", synced_count)
|
|
|
|
# -- helpers ----------------------------------------------------------
|
|
@staticmethod
|
|
def _extract_jira_key(text: str) -> Optional[str]:
|
|
"""Extract PROJ-123 from text like '[PROJ-123] Summary...'."""
|
|
import re
|
|
# Allow alphanumeric characters in the project key (e.g., S2R-10690)
|
|
m = re.search(r"\[([A-Za-z0-9]+-\d+)\]", text)
|
|
return m.group(1) if m else None
|
|
|
|
# -- full sync --------------------------------------------------------
|
|
def run(self, direction: str):
|
|
if direction in ("jira-to-jotty", "both"):
|
|
self.sync_jira_to_jotty()
|
|
if direction in ("jotty-to-jira", "both"):
|
|
self.sync_jotty_to_jira()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config loader (JSON file or environment variables)
|
|
# ---------------------------------------------------------------------------
|
|
def load_config(args: argparse.Namespace) -> SyncConfig:
|
|
cfg = SyncConfig()
|
|
|
|
# 1. Try to read from sync_config.json
|
|
config_path = args.config if hasattr(args, "config") and args.config else "sync_config.json"
|
|
if os.path.exists(config_path):
|
|
log.info("Loading configuration from %s", config_path)
|
|
with open(config_path) as f:
|
|
raw = json.load(f)
|
|
|
|
jira_cfg = raw.get("jira", {})
|
|
cfg.jira_base_url = args.jira_base or jira_cfg.get("base_url", os.environ.get("JIRA_BASE_URL", ""))
|
|
cfg.jira_api_token = args.jira_token or jira_cfg.get("api_token", os.environ.get("JIRA_API_TOKEN", ""))
|
|
cfg.project_key = args.project or jira_cfg.get("project_key", os.environ.get("PROJECT_KEY", ""))
|
|
if "jql_query" in raw.get("jira", {}):
|
|
cfg.jql_query = raw["jira"]["jql_query"]
|
|
|
|
jotty_cfg = raw.get("jotty", {})
|
|
cfg.jotty_base_url = args.jotty_base or jotty_cfg.get("base_url", os.environ.get("JOTTY_BASE_URL", ""))
|
|
cfg.jotty_api_key = args.jotty_key or jotty_cfg.get("api_key", os.environ.get("JOTTY_API_KEY", ""))
|
|
if "checklist_title" in raw.get("jotty", {}):
|
|
cfg.checklist_title = raw["jotty"]["checklist_title"]
|
|
if "checklist_category" in raw.get("jotty", {}):
|
|
cfg.checklist_category = raw["jotty"]["checklist_category"]
|
|
|
|
# 2. CLI args override everything
|
|
cfg.jira_base_url = args.jira_base or cfg.jira_base_url
|
|
cfg.jira_api_token = args.jira_token or cfg.jira_api_token
|
|
cfg.project_key = args.project or cfg.project_key
|
|
cfg.jotty_base_url = args.jotty_base or cfg.jotty_base_url
|
|
cfg.jotty_api_key = args.jotty_key or cfg.jotty_api_key
|
|
|
|
# 3. Environment variables are the last fallback for anything not set above
|
|
if not cfg.jira_base_url:
|
|
cfg.jira_base_url = os.environ.get("JIRA_BASE_URL", "")
|
|
if not cfg.jira_api_token:
|
|
cfg.jira_api_token = os.environ.get("JIRA_API_TOKEN", "")
|
|
if not cfg.project_key:
|
|
cfg.project_key = os.environ.get("PROJECT_KEY", "")
|
|
env_jql = os.environ.get("JQL_QUERY")
|
|
if env_jql:
|
|
cfg.jql_query = env_jql
|
|
if not cfg.jotty_base_url:
|
|
cfg.jotty_base_url = os.environ.get("JOTTY_BASE_URL", "")
|
|
if not cfg.jotty_api_key:
|
|
cfg.jotty_api_key = os.environ.get("JOTTY_API_KEY", "")
|
|
|
|
return cfg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Argument parser
|
|
# ---------------------------------------------------------------------------
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(description="Sync Jira stories ↔ Jotty checklists")
|
|
p.add_argument(
|
|
"--direction", "-d",
|
|
choices=["jira-to-jotty", "jotty-to-jira", "both"],
|
|
default="both",
|
|
help="Sync direction (default: both)",
|
|
)
|
|
p.add_argument("--config", "-c", default="sync_config.json", help="Path to config JSON file")
|
|
|
|
# CLI overrides for Jira
|
|
p.add_argument("--jira-base", dest="jira_base", help="Jira base URL (overrides config)")
|
|
p.add_argument("--jira-token", dest="jira_token", help="Jira API token / PAT")
|
|
p.add_argument("--project", "-p", help="Jira project key (e.g. PROJ)")
|
|
|
|
# CLI overrides for Jotty
|
|
p.add_argument("--jotty-base", dest="jotty_base", help="Jotty base URL (overrides config)")
|
|
p.add_argument("--jotty-key", dest="jotty_key", help="Jotty API key")
|
|
return p
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
def main():
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
config = load_config(args)
|
|
except Exception as exc:
|
|
log.error("Failed to load configuration: %s", exc)
|
|
sys.exit(1)
|
|
|
|
# Validate required fields
|
|
missing = []
|
|
if not config.jira_base_url:
|
|
missing.append("JIRA_BASE_URL")
|
|
if not config.jira_api_token:
|
|
missing.append("JIRA_API_TOKEN")
|
|
if not config.project_key:
|
|
missing.append("PROJECT_KEY")
|
|
if not config.jotty_base_url:
|
|
missing.append("JOTTY_BASE_URL")
|
|
if not config.jotty_api_key:
|
|
missing.append("JOTTY_API_KEY")
|
|
|
|
if missing:
|
|
log.error("Missing required configuration. Set these env vars or add them to sync_config.json:")
|
|
for m in missing:
|
|
log.error(" %s", m)
|
|
sys.exit(1)
|
|
|
|
engine = SyncEngine(config)
|
|
try:
|
|
engine.run(args.direction)
|
|
except Exception as exc:
|
|
log.exception("Sync failed: %s", exc)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|