#!/usr/bin/env python3 """Small Jira REST helper for Investibook agent skills.""" from __future__ import annotations import argparse import base64 import json import os import sys import urllib.error import urllib.parse import urllib.request from typing import Any DEFAULT_BASE_URL = "https://tidb.atlassian.net" class JiraConfigError(RuntimeError): pass def _env_first(*names: str, default: str = "") -> str: for name in names: value = os.environ.get(name) if value: return value.strip() return default def load_config() -> tuple[str, str, str]: base_url = _env_first( "INVESTIBOOK_JIRA_BASE_URL", "JIRA_BASE_URL", default=DEFAULT_BASE_URL ).rstrip("/") username = _env_first("INVESTIBOOK_JIRA_USERNAME", "JIRA_USERNAME") token = _env_first("INVESTIBOOK_JIRA_API_TOKEN", "JIRA_PASSWORD") if not username or not token: raise JiraConfigError( "Jira credentials are missing: set JIRA_USERNAME and JIRA_PASSWORD " "or INVESTIBOOK_JIRA_USERNAME and INVESTIBOOK_JIRA_API_TOKEN." ) return base_url, username, token def jira_request( method: str, path: str, *, params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, ) -> dict[str, Any]: base_url, username, token = load_config() query = urllib.parse.urlencode(params or {}, doseq=True) url = f"{base_url}{path}" if query: url = f"{url}?{query}" body = None headers = {"Accept": "application/json"} if payload is not None: body = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" auth = base64.b64encode(f"{username}:{token}".encode("utf-8")).decode("ascii") headers["Authorization"] = f"Basic {auth}" request = urllib.request.Request( url=url, data=body, headers=headers, method=method, ) with urllib.request.urlopen(request, timeout=30) as response: raw = response.read().decode("utf-8") if not raw: return {} return json.loads(raw) def extract_alert_info(issue: dict[str, Any]) -> dict[str, Any]: fields = issue.get("fields", {}) org_id = fields.get("customfield_10086") cluster_id = fields.get("customfield_10088") raw_alert_str = fields.get("customfield_10160") tenant_id = None alert_name = None if raw_alert_str: try: raw_alert = json.loads(raw_alert_str) labels = raw_alert.get("labels", {}) tenant_id = labels.get("o11y_tenant_id") alert_name = labels.get("alertname") except (json.JSONDecodeError, AttributeError): pass return { "tenant_id": tenant_id, "cluster_id": cluster_id, "alert_name": alert_name, "org_id": org_id, } def get_issue(args: argparse.Namespace) -> dict[str, Any]: return jira_request("GET", f"/rest/api/2/issue/{args.issue_key}") def get_alert_info(args: argparse.Namespace) -> dict[str, Any]: return extract_alert_info(get_issue(args)) def get_comments(args: argparse.Namespace) -> dict[str, Any]: return jira_request("GET", f"/rest/api/2/issue/{args.issue_key}/comment") def _search_jql( jql: str, fields: list[str], max_results: int, next_page_token: str | None = None, ) -> dict[str, Any]: """Search via Jira v3 POST /rest/api/3/search/jql. Jira Cloud's enhanced search endpoint uses cursor pagination. The old search APIs accepted ``startAt``, but sending that field here returns HTTP 400 "Invalid request payload". """ payload: dict[str, Any] = { "jql": jql, "fields": fields, "maxResults": max_results, } if next_page_token: payload["nextPageToken"] = next_page_token return jira_request( "POST", "/rest/api/3/search/jql", payload=payload, ) def search_by_cluster(args: argparse.Namespace) -> dict[str, Any]: project = args.project cluster_id = args.cluster_id jql = f'project = {project} AND "Cluster ID" = "{cluster_id}" ORDER BY created DESC' return _search_jql( jql, ["key", "summary", "status", "created", "updated", "customfield_10149"], args.max_results, args.next_page_token, ) def search_by_tenant(args: argparse.Namespace) -> dict[str, Any]: project = args.project tenant_id = args.tenant_id jql = f'project = {project} AND "Tenant ID" = "{tenant_id}" ORDER BY created DESC' return _search_jql( jql, ["key", "summary", "status", "created", "updated", "customfield_10157"], args.max_results, args.next_page_token, ) def search_by_alert_name(args: argparse.Namespace) -> dict[str, Any]: project = args.project alert_name = args.alert_name jql = f'project = {project} AND summary ~ "{alert_name}" ORDER BY created DESC' return _search_jql( jql, ["key", "summary", "status", "created", "updated"], args.max_results, args.next_page_token, ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Jira REST helper") subparsers = parser.add_subparsers(dest="command", required=True) get_issue_parser = subparsers.add_parser("get-issue") get_issue_parser.add_argument("--issue-key", required=True) get_issue_parser.set_defaults(func=get_issue) alert_parser = subparsers.add_parser("get-alert-info") alert_parser.add_argument("--issue-key", required=True) alert_parser.set_defaults(func=get_alert_info) comments_parser = subparsers.add_parser("get-comments") comments_parser.add_argument("--issue-key", required=True) comments_parser.set_defaults(func=get_comments) search_parser = subparsers.add_parser("search-by-cluster") search_parser.add_argument("--cluster-id", required=True) search_parser.add_argument("--project", default="O11Y", help="Jira project key (default: O11Y)") search_parser.add_argument("--max-results", type=int, default=50) search_parser.add_argument( "--next-page-token", default=None, help="Cursor token from the previous /search/jql response", ) search_parser.add_argument( "--start-at", type=int, default=0, help="Deprecated and ignored; Jira /search/jql now uses --next-page-token", ) search_parser.set_defaults(func=search_by_cluster) tenant_parser = subparsers.add_parser("search-by-tenant") tenant_parser.add_argument("--tenant-id", required=True) tenant_parser.add_argument("--project", default="O11Y", help="Jira project key (default: O11Y)") tenant_parser.add_argument("--max-results", type=int, default=50) tenant_parser.add_argument( "--next-page-token", default=None, help="Cursor token from the previous /search/jql response", ) tenant_parser.add_argument( "--start-at", type=int, default=0, help="Deprecated and ignored; Jira /search/jql now uses --next-page-token", ) tenant_parser.set_defaults(func=search_by_tenant) alert_name_parser = subparsers.add_parser("search-by-alert-name") alert_name_parser.add_argument("--alert-name", required=True) alert_name_parser.add_argument("--project", default="O11Y", help="Jira project key (default: O11Y)") alert_name_parser.add_argument("--max-results", type=int, default=50) alert_name_parser.add_argument( "--next-page-token", default=None, help="Cursor token from the previous /search/jql response", ) alert_name_parser.add_argument( "--start-at", type=int, default=0, help="Deprecated and ignored; Jira /search/jql now uses --next-page-token", ) alert_name_parser.set_defaults(func=search_by_alert_name) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: result = args.func(args) except JiraConfigError as exc: print(str(exc), file=sys.stderr) return 2 except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") print(f"Jira API error: HTTP {exc.code}: {detail}", file=sys.stderr) return 1 except urllib.error.URLError as exc: print(f"Jira API error: {exc.reason}", file=sys.stderr) return 1 print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())