f3nris
725e6921f4
Given up to 50 keys and an instant, returns the issues updated since with their status, new comments (author id, plain-text body, own comments flagged) and status transitions, plus the instance's own account. JQL dates are converted to the account's time zone; malformed keys are never queried; a deleted key falls back to per-key reads instead of failing the batch. get_comments gains order, start_at and author_id, and decodes HTML entities. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
214 lines
8.4 KiB
Python
214 lines
8.4 KiB
Python
import base64, html, json, os, re, sys, urllib.parse, urllib.request, urllib.error
|
|
import re
|
|
|
|
|
|
def _cfg():
|
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
|
|
|
|
def _is_cloud():
|
|
return bool(str(_cfg().get("cloud_id") or "").strip())
|
|
|
|
|
|
def _api():
|
|
return "3" if _is_cloud() else "2"
|
|
|
|
|
|
def _base():
|
|
cfg = _cfg()
|
|
root = str(cfg.get("server_url") or "https://api.atlassian.com/ex/jira").rstrip("/")
|
|
cloud_id = str(cfg.get("cloud_id") or "").strip()
|
|
if not cloud_id and "api.atlassian.com" in root:
|
|
# The Cloud gateway addresses a site by its cloud id; without one every
|
|
# path below it is a dead link, and Jira answers with an HTML 404 that
|
|
# says nothing about the cause.
|
|
raise Exception(
|
|
"server_url is the Jira Cloud gateway (" + root + ") but cloud_id is empty. "
|
|
"Set cloud_id, found at admin.atlassian.com/s/<cloud_id>/users, or point "
|
|
"server_url at your own Jira URL for On-Prem/Data Center."
|
|
)
|
|
if cloud_id and "api.atlassian.com" not in root:
|
|
# The reverse mistake, and the harder one to see: against any other host
|
|
# a cloud id is just a path segment, and every call 404s.
|
|
raise Exception(
|
|
"cloud_id is set but server_url (" + root + ") is not the Jira Cloud gateway. "
|
|
"A cloud id addresses a site through https://api.atlassian.com/ex/jira only. "
|
|
"Clear cloud_id for On-Prem/Data Center, or restore the gateway URL."
|
|
)
|
|
return root + "/" + cloud_id if cloud_id else root
|
|
|
|
|
|
def _creds():
|
|
# A credential pasted from a password manager often carries a trailing
|
|
# newline; unstripped it corrupts the header and reads back as a wrong
|
|
# password. Basic wins whenever the pair is complete: a token left behind
|
|
# from another deployment must not outrank credentials just filled in.
|
|
# A lone username, or a lone api_key, is not a pair.
|
|
cfg = _cfg()
|
|
user = str(cfg.get("username") or "").strip()
|
|
key = str(cfg.get("api_key") or "").strip()
|
|
pat = str(cfg.get("pat") or "").strip()
|
|
mode = "basic" if (user and key) else ("pat" if pat else "")
|
|
return mode, user, key, pat
|
|
|
|
|
|
def _auth_method():
|
|
# Named in a 401/403, where the useful question is which credential was sent.
|
|
mode, user, key, pat = _creds()
|
|
if mode == "basic":
|
|
return "Basic authentication as " + user + (
|
|
" (a personal access token is also configured and was ignored)" if pat else ""
|
|
)
|
|
if mode == "pat":
|
|
half = ""
|
|
if user or key:
|
|
half = " (" + ("username" if user else "api_key") + " is set but its pair is "
|
|
half += "empty, so Basic authentication was not used)"
|
|
return "a personal access token" + half
|
|
return "no credential"
|
|
|
|
|
|
def _auth():
|
|
mode, user, key, pat = _creds()
|
|
if mode == "basic":
|
|
cred = (user + ":" + key).encode("utf-8")
|
|
return "Basic " + base64.b64encode(cred).decode("ascii")
|
|
if mode == "pat":
|
|
return "Bearer " + pat
|
|
raise Exception(
|
|
"No usable credential. Jira Cloud needs username (account email) together with "
|
|
"api_key (API token); On-Prem/Data Center needs pat. "
|
|
+ ("Only " + ("username" if user else "api_key") + " is set."
|
|
if (user or key) else "None of them is set.")
|
|
)
|
|
|
|
|
|
def _summarize(raw, ctype):
|
|
# An error page is a wall of markup around one sentence. Keep the <title>
|
|
# and the readable text: the rest only makes a log impossible to read.
|
|
text = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw)
|
|
if "json" in (ctype or "").lower():
|
|
return text.strip()[:500]
|
|
title = re.search(r"<title[^>]*>(.*?)</title>", text, re.S | re.I)
|
|
body = re.sub(r"<(script|style|head)[^>]*>.*?</\1>", " ", text, flags=re.S | re.I)
|
|
body = re.sub(r"<[^>]+>", " ", body)
|
|
out = (title.group(1) + " \u2014 " if title else "") + body
|
|
return re.sub(r"\s+", " ", html.unescape(out)).strip()[:500]
|
|
|
|
|
|
def request(method, path, params=None, body=None, full_url=None, raw=False):
|
|
url = full_url or (_base() + path)
|
|
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
|
if q:
|
|
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
|
headers = {"Accept": "application/json", "Authorization": _auth()}
|
|
if data is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
with urllib.request.urlopen(req, timeout=90) as r:
|
|
content = r.read()
|
|
if raw:
|
|
return content
|
|
if not content:
|
|
return {}
|
|
try:
|
|
return json.loads(content)
|
|
except ValueError:
|
|
# A 2xx that is not JSON is a login page, an SSO redirect or a proxy
|
|
# answering in place of Jira. The URL the redirects landed on is
|
|
# usually the whole diagnosis.
|
|
landed = urllib.parse.urlsplit(r.geturl())
|
|
ctype = r.headers.get("Content-Type") or "unknown"
|
|
raise Exception(
|
|
"Jira answered " + str(r.getcode()) + " with a non-JSON body at "
|
|
+ landed.scheme + "://" + landed.netloc + landed.path
|
|
+ " (content-type: " + ctype + "): " + _summarize(content, ctype)
|
|
)
|
|
|
|
|
|
def issue_ref(inputs):
|
|
ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")
|
|
if not ref:
|
|
raise Exception("issue_id or issue_key is required")
|
|
return urllib.parse.quote(ref, safe="")
|
|
|
|
|
|
def adf(text):
|
|
# Jira Cloud (API v3) expects Atlassian Document Format for rich text; On-Prem (v2) takes plain text.
|
|
if not _is_cloud():
|
|
return text
|
|
return {"type": "doc", "version": 1,
|
|
"content": [{"type": "paragraph", "content": [{"type": "text", "text": text}]}]}
|
|
|
|
|
|
def _html_text(rendered):
|
|
t = re.sub(r"(?i)<br\s*/?>", "\n", str(rendered))
|
|
t = re.sub(r"(?i)</(p|div|li|h[1-6]|tr|blockquote|pre)>", "\n", t)
|
|
t = re.sub(r"<[^>]+>", "", t)
|
|
t = html.unescape(t)
|
|
t = re.sub(r"[ \t\r\f\v]+\n", "\n", t)
|
|
return re.sub(r"\n{3,}", "\n\n", t).strip()
|
|
|
|
|
|
def _body_text(comment):
|
|
rendered = comment.get("renderedBody")
|
|
if rendered:
|
|
return _html_text(rendered)
|
|
raw_body = comment.get("body")
|
|
if isinstance(raw_body, str):
|
|
return raw_body
|
|
return ""
|
|
|
|
|
|
def _user_id(u):
|
|
# Cloud identifies a user by accountId; On-Prem by key, then name.
|
|
u = u or {}
|
|
return str(u.get("accountId") or u.get("key") or u.get("name") or "")
|
|
|
|
|
|
def main():
|
|
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
raw_ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")
|
|
ref = issue_ref(inputs)
|
|
limit = inputs.get("limit") or 50
|
|
order = str(inputs.get("order") or "oldest").strip().lower()
|
|
if order not in ("oldest", "newest"):
|
|
raise Exception("order is oldest or newest, not " + order)
|
|
start_at = inputs.get("start_at")
|
|
start_at = int(start_at) if str(start_at or "").strip() != "" else 0
|
|
|
|
res = request("GET", "/rest/api/" + _api() + "/issue/" + ref + "/comment",
|
|
params={"maxResults": limit, "startAt": start_at,
|
|
"orderBy": "-created" if order == "newest" else "created",
|
|
"expand": "renderedBody"})
|
|
|
|
comments = []
|
|
for c in res.get("comments", []) or []:
|
|
comments.append({
|
|
"id": c.get("id", ""),
|
|
"body": _body_text(c),
|
|
"author": c.get("author", {}).get("displayName", ""),
|
|
"author_id": _user_id(c.get("author")),
|
|
"created": c.get("created", ""),
|
|
"updated": c.get("updated", ""),
|
|
"update_author": c.get("updateAuthor", {}).get("displayName", ""),
|
|
})
|
|
|
|
total = int(res.get("total") or 0)
|
|
print(json.dumps({"issue": raw_ref, "comments": comments, "total": total,
|
|
"next_start_at": start_at + len(comments) if start_at + len(comments) < total else None}))
|
|
|
|
try:
|
|
main()
|
|
except urllib.error.HTTPError as e:
|
|
ctype = e.headers.get("Content-Type", "") if e.headers else ""
|
|
detail = _summarize(e.read(), ctype)
|
|
if e.code in (401, 403):
|
|
detail += " \u2014 the request was sent using " + _auth_method()
|
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": detail}))
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(json.dumps({"error": str(e)}))
|
|
sys.exit(1)
|