feat(jira): new Atlassian Jira integration
Jira Cloud and On-Prem/Data Center (27 commands): JQL ingestion with an OCSF mapper and JQL query, full issue lifecycle (create/get/edit/delete, transitions, assign), comments, remote web links and issue links, attachments (upload/download/delete), field and user lookups, and agile boards/sprints/epics. Basic (email + API token) or Personal Access Token authentication; API v3 + ADF bodies on Cloud, v2 on On-Prem. Stdlib-only, no extra Python dependencies. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,88 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
comment = str(inputs.get("comment") or "")
|
||||
if not comment:
|
||||
raise Exception("comment is required")
|
||||
visibility = str(inputs.get("visibility") or "")
|
||||
|
||||
body = {"body": adf(comment)}
|
||||
if visibility:
|
||||
body["visibility"] = {"type": "role", "value": visibility}
|
||||
|
||||
res = request("POST", "/rest/api/" + _api() + "/issue/" + ref + "/comment",
|
||||
params={"expand": "renderedBody"}, body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,94 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
url = str(inputs.get("url") or "")
|
||||
if not url:
|
||||
raise Exception("url is required")
|
||||
title = str(inputs.get("title") or "")
|
||||
if not title:
|
||||
raise Exception("title is required")
|
||||
summary = str(inputs.get("summary") or "")
|
||||
relationship = str(inputs.get("relationship") or "")
|
||||
|
||||
obj = {"url": url, "title": title}
|
||||
if summary:
|
||||
obj["summary"] = summary
|
||||
body = {"object": obj}
|
||||
if relationship:
|
||||
body["relationship"] = relationship
|
||||
|
||||
res = request("POST", "/rest/api/" + _api() + "/issue/" + ref + "/remotelink", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,91 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
raw_ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")
|
||||
|
||||
assignee_id = str(inputs.get("assignee_id") or "").strip()
|
||||
assignee = str(inputs.get("assignee") or "").strip()
|
||||
if assignee_id:
|
||||
body = {"accountId": assignee_id}
|
||||
who = assignee_id
|
||||
elif assignee:
|
||||
body = {"name": assignee}
|
||||
who = assignee
|
||||
else:
|
||||
raise Exception("assignee_id (Cloud) or assignee (On-Prem) is required")
|
||||
|
||||
request("PUT", "/rest/api/" + _api() + "/issue/" + ref + "/assignee", body=body)
|
||||
print(json.dumps({"ok": True, "issue": raw_ref, "assignee": who}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,119 @@
|
||||
import base64, json, os, 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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def flatten_issue(issue):
|
||||
f = issue.get("fields") or {}
|
||||
rendered = issue.get("renderedFields") or {}
|
||||
|
||||
def user(u):
|
||||
u = u or {}
|
||||
return u.get("displayName") or u.get("name") or u.get("emailAddress") or ""
|
||||
|
||||
desc = rendered.get("description") or ""
|
||||
if desc:
|
||||
desc = re.sub(r"<[^>]+>", "", desc)
|
||||
else:
|
||||
d = f.get("description")
|
||||
desc = d if isinstance(d, str) else ""
|
||||
return {
|
||||
"id": issue.get("id") or "",
|
||||
"key": issue.get("key") or "",
|
||||
"summary": f.get("summary") or "",
|
||||
"description": desc,
|
||||
"status": (f.get("status") or {}).get("name") or "",
|
||||
"resolution": (f.get("resolution") or {}).get("name") or "",
|
||||
"priority": (f.get("priority") or {}).get("name") or "",
|
||||
"project": (f.get("project") or {}).get("name") or "",
|
||||
"issue_type": (f.get("issuetype") or {}).get("name") or "",
|
||||
"created": f.get("created") or "",
|
||||
"updated": f.get("updated") or "",
|
||||
"due_date": f.get("duedate") or "",
|
||||
"labels": f.get("labels") or [],
|
||||
"assignee": user(f.get("assignee")),
|
||||
"reporter": user(f.get("reporter")),
|
||||
"creator": user(f.get("creator")),
|
||||
"attachments": [{"id": a.get("id"), "filename": a.get("filename"), "size": a.get("size")}
|
||||
for a in (f.get("attachment") or [])],
|
||||
"url": issue.get("self") or "",
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
board_id = str(args.get("board_id") or "").strip()
|
||||
if not board_id:
|
||||
raise Exception("board_id is required")
|
||||
jql_query = str(args.get("jql_query") or "").strip()
|
||||
limit = int(args.get("limit") or 50)
|
||||
offset = int(args.get("offset") or 0)
|
||||
|
||||
res = request("GET",
|
||||
"/rest/agile/1.0/board/" + urllib.parse.quote(board_id, safe="") + "/issue",
|
||||
params={
|
||||
"jql": jql_query,
|
||||
"maxResults": limit,
|
||||
"startAt": offset,
|
||||
"expand": "renderedFields",
|
||||
})
|
||||
|
||||
print(json.dumps({
|
||||
"board_id": board_id,
|
||||
"result": [flatten_issue(i) for i in res.get("issues", []) or []],
|
||||
"total": res.get("total", 0),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,92 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
board_id = str(args.get("board_id") or "").strip()
|
||||
board_name = str(args.get("board_name") or "").strip()
|
||||
project_key_id = str(args.get("project_key_id") or "").strip()
|
||||
board_type = str(args.get("type") or "").strip()
|
||||
limit = int(args.get("limit") or 50)
|
||||
offset = int(args.get("offset") or 0)
|
||||
|
||||
if board_id:
|
||||
res = request("GET", "/rest/agile/1.0/board/" + urllib.parse.quote(board_id, safe=""))
|
||||
boards = [res]
|
||||
else:
|
||||
res = request("GET", "/rest/agile/1.0/board", params={
|
||||
"name": board_name,
|
||||
"projectKeyOrId": project_key_id,
|
||||
"type": board_type,
|
||||
"maxResults": limit,
|
||||
"startAt": offset,
|
||||
})
|
||||
boards = res.get("values", []) or []
|
||||
|
||||
out = []
|
||||
for b in boards:
|
||||
loc = b.get("location") or {}
|
||||
out.append({
|
||||
"id": b.get("id") or "",
|
||||
"name": b.get("name") or "",
|
||||
"type": b.get("type") or "",
|
||||
"project_id": loc.get("projectId", ""),
|
||||
"project_name": loc.get("projectName", ""),
|
||||
})
|
||||
|
||||
print(json.dumps({"boards": out}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,82 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
board_id = str(args.get("board_id") or "").strip()
|
||||
if not board_id:
|
||||
raise Exception("board_id is required")
|
||||
limit = int(args.get("limit") or 50)
|
||||
offset = int(args.get("offset") or 0)
|
||||
|
||||
res = request("GET",
|
||||
"/rest/agile/1.0/board/" + urllib.parse.quote(board_id, safe="") + "/sprint",
|
||||
params={"maxResults": limit, "startAt": offset})
|
||||
|
||||
sprints = []
|
||||
for s in res.get("values", []) or []:
|
||||
sprints.append({
|
||||
"id": s.get("id") or "",
|
||||
"name": s.get("name") or "",
|
||||
"state": s.get("state") or "",
|
||||
"start_date": s.get("startDate", ""),
|
||||
"end_date": s.get("endDate", ""),
|
||||
"goal": s.get("goal", ""),
|
||||
})
|
||||
|
||||
print(json.dumps({"board_id": board_id, "sprints": sprints}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,151 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 resolve_project_by_name(name):
|
||||
if _is_cloud():
|
||||
res = request("GET", "/rest/api/3/project/search", params={"query": name})
|
||||
candidates = res.get("values") or []
|
||||
else:
|
||||
res = request("GET", "/rest/api/2/project")
|
||||
candidates = [p for p in (res or []) if str(p.get("name") or "").lower() == name.lower()]
|
||||
if len(candidates) == 0:
|
||||
raise Exception("no project found with name " + name)
|
||||
if len(candidates) > 1:
|
||||
raise Exception("multiple projects match name " + name)
|
||||
return candidates[0]
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
fields = {}
|
||||
|
||||
summary = str(inputs.get("summary") or "").strip()
|
||||
if not summary:
|
||||
raise Exception("summary is required")
|
||||
fields["summary"] = summary
|
||||
|
||||
project_key = str(inputs.get("project_key") or "").strip()
|
||||
project_name = str(inputs.get("project_name") or "").strip()
|
||||
if project_key:
|
||||
fields["project"] = {"key": project_key}
|
||||
elif project_name:
|
||||
proj = resolve_project_by_name(project_name)
|
||||
fields["project"] = {"key": proj.get("key")}
|
||||
else:
|
||||
raise Exception("project_key or project_name is required")
|
||||
|
||||
issue_type_id = str(inputs.get("issue_type_id") or "").strip()
|
||||
issue_type_name = str(inputs.get("issue_type_name") or "").strip()
|
||||
if issue_type_id:
|
||||
fields["issuetype"] = {"id": issue_type_id}
|
||||
elif issue_type_name:
|
||||
fields["issuetype"] = {"name": issue_type_name}
|
||||
else:
|
||||
raise Exception("issue_type_name or issue_type_id is required")
|
||||
|
||||
description = str(inputs.get("description") or "").strip()
|
||||
if description:
|
||||
fields["description"] = adf(description)
|
||||
|
||||
priority = str(inputs.get("priority") or "").strip()
|
||||
if priority:
|
||||
fields["priority"] = {"name": priority}
|
||||
|
||||
labels = str(inputs.get("labels") or "").strip()
|
||||
if labels:
|
||||
fields["labels"] = [x.strip() for x in labels.split(",") if x.strip()]
|
||||
|
||||
components = str(inputs.get("components") or "").strip()
|
||||
if components:
|
||||
fields["components"] = [{"name": c.strip()} for c in components.split(",") if c.strip()]
|
||||
|
||||
due_date = str(inputs.get("due_date") or "").strip()
|
||||
if due_date:
|
||||
fields["duedate"] = due_date
|
||||
|
||||
assignee_id = str(inputs.get("assignee_id") or "").strip()
|
||||
assignee = str(inputs.get("assignee") or "").strip()
|
||||
if assignee_id:
|
||||
fields["assignee"] = {"accountId": assignee_id}
|
||||
elif assignee:
|
||||
fields["assignee"] = {"name": assignee}
|
||||
|
||||
fields_json = str(inputs.get("fields_json") or "").strip()
|
||||
if fields_json:
|
||||
extra = json.loads(fields_json)
|
||||
if not isinstance(extra, dict):
|
||||
raise Exception("fields_json must be a JSON object")
|
||||
fields.update(extra)
|
||||
|
||||
res = request("POST", "/rest/api/" + _api() + "/issue", body={"fields": fields})
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,84 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
attachment_id = str(inputs.get("attachment_id") or "")
|
||||
if not attachment_id:
|
||||
raise Exception("attachment_id is required")
|
||||
|
||||
path = "/rest/api/" + _api() + "/attachment/" + urllib.parse.quote(attachment_id, safe="")
|
||||
res = request("DELETE", path)
|
||||
if not res:
|
||||
res = {"ok": True, "attachment_id": attachment_id}
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,86 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
comment_id = str(inputs.get("comment_id") or "")
|
||||
if not comment_id:
|
||||
raise Exception("comment_id is required")
|
||||
|
||||
path = ("/rest/api/" + _api() + "/issue/" + ref + "/comment/"
|
||||
+ urllib.parse.quote(comment_id, safe=""))
|
||||
res = request("DELETE", path)
|
||||
if not res:
|
||||
res = {"ok": True, "comment_id": comment_id}
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,80 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
raw_ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")
|
||||
|
||||
request("DELETE", "/rest/api/" + _api() + "/issue/" + ref, params={"deleteSubtasks": "true"})
|
||||
print(json.dumps({"ok": True, "issue": raw_ref}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,92 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
comment_id = str(inputs.get("comment_id") or "")
|
||||
if not comment_id:
|
||||
raise Exception("comment_id is required")
|
||||
comment = str(inputs.get("comment") or "")
|
||||
if not comment:
|
||||
raise Exception("comment is required")
|
||||
visibility = str(inputs.get("visibility") or "")
|
||||
|
||||
body = {"body": adf(comment)}
|
||||
if visibility:
|
||||
body["visibility"] = {"type": "role", "value": visibility}
|
||||
|
||||
path = ("/rest/api/" + _api() + "/issue/" + ref + "/comment/"
|
||||
+ urllib.parse.quote(comment_id, safe=""))
|
||||
res = request("PUT", path, params={"expand": "renderedBody"}, body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,159 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 build_fields(inputs):
|
||||
fields = {}
|
||||
|
||||
summary = str(inputs.get("summary") or "").strip()
|
||||
if summary:
|
||||
fields["summary"] = summary
|
||||
|
||||
description = str(inputs.get("description") or "").strip()
|
||||
if description:
|
||||
fields["description"] = adf(description)
|
||||
|
||||
priority = str(inputs.get("priority") or "").strip()
|
||||
if priority:
|
||||
fields["priority"] = {"name": priority}
|
||||
|
||||
labels = str(inputs.get("labels") or "").strip()
|
||||
if labels:
|
||||
fields["labels"] = [x.strip() for x in labels.split(",") if x.strip()]
|
||||
|
||||
components = str(inputs.get("components") or "").strip()
|
||||
if components:
|
||||
fields["components"] = [{"name": c.strip()} for c in components.split(",") if c.strip()]
|
||||
|
||||
due_date = str(inputs.get("due_date") or "").strip()
|
||||
if due_date:
|
||||
fields["duedate"] = due_date
|
||||
|
||||
assignee_id = str(inputs.get("assignee_id") or "").strip()
|
||||
assignee = str(inputs.get("assignee") or "").strip()
|
||||
if assignee_id:
|
||||
fields["assignee"] = {"accountId": assignee_id}
|
||||
elif assignee:
|
||||
fields["assignee"] = {"name": assignee}
|
||||
|
||||
fields_json = str(inputs.get("fields_json") or "").strip()
|
||||
if fields_json:
|
||||
extra = json.loads(fields_json)
|
||||
if not isinstance(extra, dict):
|
||||
raise Exception("fields_json must be a JSON object")
|
||||
fields.update(extra)
|
||||
|
||||
return fields
|
||||
|
||||
|
||||
def find_transition(ref, status, transition):
|
||||
res = request("GET", "/rest/api/" + _api() + "/issue/" + ref + "/transitions")
|
||||
transitions = res.get("transitions") or []
|
||||
if transition:
|
||||
wanted = transition.lower()
|
||||
for t in transitions:
|
||||
if str(t.get("name") or "").lower() == wanted:
|
||||
return t.get("id")
|
||||
names = [str(t.get("name") or "") for t in transitions]
|
||||
raise Exception("transition not found: " + transition + " — valid transitions: " + ", ".join(names))
|
||||
wanted = status.lower()
|
||||
for t in transitions:
|
||||
if str((t.get("to") or {}).get("name") or "").lower() == wanted:
|
||||
return t.get("id")
|
||||
names = [str((t.get("to") or {}).get("name") or "") for t in transitions]
|
||||
raise Exception("status not reachable: " + status + " — valid statuses: " + ", ".join(names))
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
|
||||
status = str(inputs.get("status") or "").strip()
|
||||
transition = str(inputs.get("transition") or "").strip()
|
||||
if status and transition:
|
||||
raise Exception("provide only one of status or transition")
|
||||
|
||||
fields = build_fields(inputs)
|
||||
|
||||
if status or transition:
|
||||
tid = find_transition(ref, status, transition)
|
||||
body = {"transition": {"id": str(tid)}}
|
||||
if fields:
|
||||
body["fields"] = fields
|
||||
request("POST", "/rest/api/" + _api() + "/issue/" + ref + "/transitions", body=body)
|
||||
elif fields:
|
||||
request("PUT", "/rest/api/" + _api() + "/issue/" + ref, body={"fields": fields})
|
||||
else:
|
||||
raise Exception("nothing to update")
|
||||
|
||||
res = request("GET", "/rest/api/" + _api() + "/issue/" + ref)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,121 @@
|
||||
import base64, json, os, 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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def flatten_issue(issue):
|
||||
f = issue.get("fields") or {}
|
||||
rendered = issue.get("renderedFields") or {}
|
||||
|
||||
def user(u):
|
||||
u = u or {}
|
||||
return u.get("displayName") or u.get("name") or u.get("emailAddress") or ""
|
||||
|
||||
desc = rendered.get("description") or ""
|
||||
if desc:
|
||||
desc = re.sub(r"<[^>]+>", "", desc)
|
||||
else:
|
||||
d = f.get("description")
|
||||
desc = d if isinstance(d, str) else ""
|
||||
return {
|
||||
"id": issue.get("id") or "",
|
||||
"key": issue.get("key") or "",
|
||||
"summary": f.get("summary") or "",
|
||||
"description": desc,
|
||||
"status": (f.get("status") or {}).get("name") or "",
|
||||
"resolution": (f.get("resolution") or {}).get("name") or "",
|
||||
"priority": (f.get("priority") or {}).get("name") or "",
|
||||
"project": (f.get("project") or {}).get("name") or "",
|
||||
"issue_type": (f.get("issuetype") or {}).get("name") or "",
|
||||
"created": f.get("created") or "",
|
||||
"updated": f.get("updated") or "",
|
||||
"due_date": f.get("duedate") or "",
|
||||
"labels": f.get("labels") or [],
|
||||
"assignee": user(f.get("assignee")),
|
||||
"reporter": user(f.get("reporter")),
|
||||
"creator": user(f.get("creator")),
|
||||
"attachments": [{"id": a.get("id"), "filename": a.get("filename"), "size": a.get("size")}
|
||||
for a in (f.get("attachment") or [])],
|
||||
"url": issue.get("self") or "",
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
epic_id = str(args.get("epic_id") or "").strip()
|
||||
epic_key = str(args.get("epic_key") or "").strip()
|
||||
ref = epic_id or epic_key
|
||||
if not ref:
|
||||
raise Exception("epic_id or epic_key is required")
|
||||
jql_query = str(args.get("jql_query") or "").strip()
|
||||
limit = int(args.get("limit") or 50)
|
||||
offset = int(args.get("offset") or 0)
|
||||
|
||||
res = request("GET",
|
||||
"/rest/agile/1.0/epic/" + urllib.parse.quote(ref, safe="") + "/issue",
|
||||
params={
|
||||
"jql": jql_query,
|
||||
"maxResults": limit,
|
||||
"startAt": offset,
|
||||
"expand": "renderedFields",
|
||||
})
|
||||
|
||||
print(json.dumps({
|
||||
"epic": ref,
|
||||
"result": [flatten_issue(i) for i in res.get("issues", []) or []],
|
||||
"total": res.get("total", 0),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,98 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
attachment_id = str(inputs.get("attachment_id") or "")
|
||||
if not attachment_id:
|
||||
raise Exception("attachment_id is required")
|
||||
quoted = urllib.parse.quote(attachment_id, safe="")
|
||||
|
||||
meta = request("GET", "/rest/api/" + _api() + "/attachment/" + quoted)
|
||||
|
||||
if _is_cloud():
|
||||
data = request("GET", "/rest/api/3/attachment/content/" + quoted, raw=True)
|
||||
else:
|
||||
# On-Prem serves the bytes from the metadata "content" URL (urllib follows the redirect).
|
||||
content_url = meta.get("content", "")
|
||||
if not content_url:
|
||||
raise Exception("attachment content URL is missing")
|
||||
data = request("GET", "", full_url=content_url, raw=True)
|
||||
|
||||
print(json.dumps({
|
||||
"id": meta.get("id", attachment_id),
|
||||
"filename": meta.get("filename", ""),
|
||||
"mime_type": meta.get("mimeType", ""),
|
||||
"size": meta.get("size", 0),
|
||||
"content_base64": base64.b64encode(data).decode("ascii"),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,105 @@
|
||||
import base64, json, os, 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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 _body_text(comment):
|
||||
rendered = comment.get("renderedBody")
|
||||
if rendered:
|
||||
return re.sub(r"<[^>]+>", "", str(rendered))
|
||||
raw_body = comment.get("body")
|
||||
if isinstance(raw_body, str):
|
||||
return raw_body
|
||||
return ""
|
||||
|
||||
|
||||
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
|
||||
|
||||
res = request("GET", "/rest/api/" + _api() + "/issue/" + ref + "/comment",
|
||||
params={"maxResults": limit, "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", ""),
|
||||
"created": c.get("created", ""),
|
||||
"updated": c.get("updated", ""),
|
||||
"update_author": c.get("updateAuthor", {}).get("displayName", ""),
|
||||
})
|
||||
|
||||
print(json.dumps({"issue": raw_ref, "comments": comments}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,104 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
attribute = str(args.get("attribute") or "").strip()
|
||||
if not attribute:
|
||||
raise Exception("attribute is required")
|
||||
max_results = int(args.get("max_results") or 50)
|
||||
|
||||
cloud = _is_cloud()
|
||||
if cloud:
|
||||
res = request("GET", "/rest/api/3/user/search",
|
||||
params={"query": attribute, "maxResults": max_results})
|
||||
else:
|
||||
res = request("GET", "/rest/api/2/user/search",
|
||||
params={"username": attribute, "maxResults": max_results})
|
||||
|
||||
users = res if isinstance(res, list) else (res.get("values") or [])
|
||||
|
||||
if len(users) == 1:
|
||||
u = users[0]
|
||||
account_id = u.get("accountId") or u.get("name") or ""
|
||||
print(json.dumps({"attribute": attribute, "account_id": account_id}))
|
||||
return
|
||||
|
||||
needle = attribute.lower()
|
||||
candidates = []
|
||||
for u in users:
|
||||
values = [str(u.get("displayName") or ""), str(u.get("emailAddress") or "")]
|
||||
if not cloud:
|
||||
values.append(str(u.get("name") or ""))
|
||||
if any(v.lower() == needle for v in values if v):
|
||||
candidates.append(u.get("accountId") or "" if cloud else u.get("name") or "")
|
||||
|
||||
if len(candidates) == 1:
|
||||
print(json.dumps({"attribute": attribute, "account_id": candidates[0]}))
|
||||
elif len(candidates) == 0:
|
||||
print(json.dumps({
|
||||
"attribute": attribute,
|
||||
"account_id": "",
|
||||
"message": "no unambiguous match — try the exact display name or email",
|
||||
}))
|
||||
else:
|
||||
print(json.dumps({
|
||||
"attribute": attribute,
|
||||
"account_id": "",
|
||||
"message": "multiple matches — provide a more specific attribute",
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,156 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 flatten_issue(issue):
|
||||
f = issue.get("fields") or {}
|
||||
rendered = issue.get("renderedFields") or {}
|
||||
|
||||
def user(u):
|
||||
u = u or {}
|
||||
return u.get("displayName") or u.get("name") or u.get("emailAddress") or ""
|
||||
|
||||
desc = rendered.get("description") or ""
|
||||
if desc:
|
||||
desc = re.sub(r"<[^>]+>", "", desc)
|
||||
else:
|
||||
d = f.get("description")
|
||||
desc = d if isinstance(d, str) else ""
|
||||
return {
|
||||
"id": issue.get("id") or "",
|
||||
"key": issue.get("key") or "",
|
||||
"summary": f.get("summary") or "",
|
||||
"description": desc,
|
||||
"status": (f.get("status") or {}).get("name") or "",
|
||||
"resolution": (f.get("resolution") or {}).get("name") or "",
|
||||
"priority": (f.get("priority") or {}).get("name") or "",
|
||||
"project": (f.get("project") or {}).get("name") or "",
|
||||
"issue_type": (f.get("issuetype") or {}).get("name") or "",
|
||||
"created": f.get("created") or "",
|
||||
"updated": f.get("updated") or "",
|
||||
"due_date": f.get("duedate") or "",
|
||||
"labels": f.get("labels") or [],
|
||||
"assignee": user(f.get("assignee")),
|
||||
"reporter": user(f.get("reporter")),
|
||||
"creator": user(f.get("creator")),
|
||||
"attachments": [{"id": a.get("id"), "filename": a.get("filename"), "size": a.get("size")}
|
||||
for a in (f.get("attachment") or [])],
|
||||
"url": issue.get("self") or "",
|
||||
}
|
||||
|
||||
|
||||
def search_issues(jql, max_results, next_page_token=None, start_at=None):
|
||||
params = {"jql": jql, "maxResults": max_results, "fields": "*all", "expand": "renderedFields,names"}
|
||||
if _is_cloud() and not start_at:
|
||||
if next_page_token:
|
||||
params["nextPageToken"] = next_page_token
|
||||
return request("GET", "/rest/api/3/search/jql", params=params)
|
||||
if start_at:
|
||||
params["startAt"] = start_at
|
||||
return request("GET", "/rest/api/" + _api() + "/search", params=params)
|
||||
|
||||
|
||||
def to_jira_time(v):
|
||||
s = str(v or "").strip()
|
||||
if not s:
|
||||
return None
|
||||
if s.isdigit():
|
||||
ts = int(s)
|
||||
if ts > 10 ** 12:
|
||||
ts = ts // 1000
|
||||
return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
|
||||
try:
|
||||
dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
|
||||
if dt.tzinfo is not None:
|
||||
dt = dt.astimezone(timezone.utc)
|
||||
return dt.strftime("%Y-%m-%d %H:%M")
|
||||
except Exception:
|
||||
return s
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
query = str(inputs.get("query") or "").strip()
|
||||
max_results = int(inputs.get("max") or 50)
|
||||
|
||||
watermark = to_jira_time(inputs.get("after"))
|
||||
|
||||
jql = query
|
||||
if watermark:
|
||||
clause = 'created >= "' + watermark + '"'
|
||||
jql = (jql + " AND " + clause) if jql else clause
|
||||
jql = jql + " ORDER BY created ASC"
|
||||
|
||||
res = search_issues(jql, max_results)
|
||||
print(json.dumps({"result": [flatten_issue(i) for i in res.get("issues", [])]}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,117 @@
|
||||
import base64, json, os, 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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 flatten_issue(issue):
|
||||
f = issue.get("fields") or {}
|
||||
rendered = issue.get("renderedFields") or {}
|
||||
|
||||
def user(u):
|
||||
u = u or {}
|
||||
return u.get("displayName") or u.get("name") or u.get("emailAddress") or ""
|
||||
|
||||
desc = rendered.get("description") or ""
|
||||
if desc:
|
||||
desc = re.sub(r"<[^>]+>", "", desc)
|
||||
else:
|
||||
d = f.get("description")
|
||||
desc = d if isinstance(d, str) else ""
|
||||
return {
|
||||
"id": issue.get("id") or "",
|
||||
"key": issue.get("key") or "",
|
||||
"summary": f.get("summary") or "",
|
||||
"description": desc,
|
||||
"status": (f.get("status") or {}).get("name") or "",
|
||||
"resolution": (f.get("resolution") or {}).get("name") or "",
|
||||
"priority": (f.get("priority") or {}).get("name") or "",
|
||||
"project": (f.get("project") or {}).get("name") or "",
|
||||
"issue_type": (f.get("issuetype") or {}).get("name") or "",
|
||||
"created": f.get("created") or "",
|
||||
"updated": f.get("updated") or "",
|
||||
"due_date": f.get("duedate") or "",
|
||||
"labels": f.get("labels") or [],
|
||||
"assignee": user(f.get("assignee")),
|
||||
"reporter": user(f.get("reporter")),
|
||||
"creator": user(f.get("creator")),
|
||||
"attachments": [{"id": a.get("id"), "filename": a.get("filename"), "size": a.get("size")}
|
||||
for a in (f.get("attachment") or [])],
|
||||
"url": issue.get("self") or "",
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
res = request("GET", "/rest/api/" + _api() + "/issue/" + ref,
|
||||
params={"expand": "renderedFields,names,transitions"})
|
||||
print(json.dumps(flatten_issue(res)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,84 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
res = request("GET", "/rest/api/" + _api() + "/issueLinkType")
|
||||
link_types = []
|
||||
for t in res.get("issueLinkTypes", []) or []:
|
||||
link_types.append({
|
||||
"id": t.get("id", ""),
|
||||
"name": t.get("name", ""),
|
||||
"inward": t.get("inward", ""),
|
||||
"outward": t.get("outward", ""),
|
||||
})
|
||||
print(json.dumps({"link_types": link_types}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,88 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
account_id = str(args.get("account_id") or "").strip()
|
||||
key = str(args.get("key") or "").strip()
|
||||
username = str(args.get("username") or "").strip()
|
||||
|
||||
params = {}
|
||||
if _is_cloud():
|
||||
if account_id:
|
||||
params["accountId"] = account_id
|
||||
else:
|
||||
if key:
|
||||
params["key"] = key
|
||||
elif username:
|
||||
params["username"] = username
|
||||
|
||||
if params:
|
||||
res = request("GET", "/rest/api/" + _api() + "/user", params=params)
|
||||
else:
|
||||
res = request("GET", "/rest/api/" + _api() + "/myself")
|
||||
|
||||
print(json.dumps({
|
||||
"account_id": res.get("accountId", ""),
|
||||
"key": res.get("key", ""),
|
||||
"name": res.get("name", ""),
|
||||
"display_name": res.get("displayName", ""),
|
||||
"email": res.get("emailAddress", ""),
|
||||
"active": res.get("active"),
|
||||
"timezone": res.get("timeZone", ""),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,110 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 _issue_selector(value):
|
||||
# A numeric value is an issue id; anything else is an issue key.
|
||||
if str(value).isdigit():
|
||||
return {"id": str(value)}
|
||||
return {"key": str(value)}
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
inward_issue = str(inputs.get("inward_issue") or "")
|
||||
if not inward_issue:
|
||||
raise Exception("inward_issue is required")
|
||||
outward_issue = str(inputs.get("outward_issue") or "")
|
||||
if not outward_issue:
|
||||
raise Exception("outward_issue is required")
|
||||
link_type = str(inputs.get("link_type") or "")
|
||||
if not link_type:
|
||||
raise Exception("link_type is required")
|
||||
comment = str(inputs.get("comment") or "")
|
||||
|
||||
body = {
|
||||
"type": {"name": link_type},
|
||||
"inwardIssue": _issue_selector(inward_issue),
|
||||
"outwardIssue": _issue_selector(outward_issue),
|
||||
}
|
||||
if comment:
|
||||
body["comment"] = {"body": adf(comment)}
|
||||
|
||||
res = request("POST", "/rest/api/" + _api() + "/issueLink", body=body)
|
||||
if not res:
|
||||
res = {
|
||||
"ok": True,
|
||||
"inward_issue": inward_issue,
|
||||
"outward_issue": outward_issue,
|
||||
"link_type": link_type,
|
||||
}
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,79 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
limit = int(args.get("limit") or 50)
|
||||
offset = int(args.get("offset") or 0)
|
||||
|
||||
res = request("GET", "/rest/api/" + _api() + "/field")
|
||||
rows = res if isinstance(res, list) else []
|
||||
sliced = rows[offset:offset + limit]
|
||||
|
||||
fields = []
|
||||
for f in sliced:
|
||||
fields.append({
|
||||
"id": f.get("id") or "",
|
||||
"key": f.get("key") or "",
|
||||
"name": f.get("name") or "",
|
||||
"custom": f.get("custom"),
|
||||
"searchable": f.get("searchable"),
|
||||
"type": (f.get("schema") or {}).get("type", ""),
|
||||
})
|
||||
|
||||
print(json.dumps({"total": len(rows), "fields": fields}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,84 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
raw_ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")
|
||||
|
||||
res = request("GET", "/rest/api/" + _api() + "/issue/" + ref + "/transitions")
|
||||
transitions = [
|
||||
{"id": t.get("id"), "name": t.get("name"), "to": (t.get("to") or {}).get("name", "")}
|
||||
for t in (res.get("transitions") or [])
|
||||
]
|
||||
print(json.dumps({"issue": raw_ref, "transitions": transitions}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,138 @@
|
||||
import base64, json, os, 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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 flatten_issue(issue):
|
||||
f = issue.get("fields") or {}
|
||||
rendered = issue.get("renderedFields") or {}
|
||||
|
||||
def user(u):
|
||||
u = u or {}
|
||||
return u.get("displayName") or u.get("name") or u.get("emailAddress") or ""
|
||||
|
||||
desc = rendered.get("description") or ""
|
||||
if desc:
|
||||
desc = re.sub(r"<[^>]+>", "", desc)
|
||||
else:
|
||||
d = f.get("description")
|
||||
desc = d if isinstance(d, str) else ""
|
||||
return {
|
||||
"id": issue.get("id") or "",
|
||||
"key": issue.get("key") or "",
|
||||
"summary": f.get("summary") or "",
|
||||
"description": desc,
|
||||
"status": (f.get("status") or {}).get("name") or "",
|
||||
"resolution": (f.get("resolution") or {}).get("name") or "",
|
||||
"priority": (f.get("priority") or {}).get("name") or "",
|
||||
"project": (f.get("project") or {}).get("name") or "",
|
||||
"issue_type": (f.get("issuetype") or {}).get("name") or "",
|
||||
"created": f.get("created") or "",
|
||||
"updated": f.get("updated") or "",
|
||||
"due_date": f.get("duedate") or "",
|
||||
"labels": f.get("labels") or [],
|
||||
"assignee": user(f.get("assignee")),
|
||||
"reporter": user(f.get("reporter")),
|
||||
"creator": user(f.get("creator")),
|
||||
"attachments": [{"id": a.get("id"), "filename": a.get("filename"), "size": a.get("size")}
|
||||
for a in (f.get("attachment") or [])],
|
||||
"url": issue.get("self") or "",
|
||||
}
|
||||
|
||||
|
||||
def search_issues(jql, max_results, next_page_token=None, start_at=None):
|
||||
params = {"jql": jql, "maxResults": max_results, "fields": "*all", "expand": "renderedFields,names"}
|
||||
if _is_cloud() and not start_at:
|
||||
if next_page_token:
|
||||
params["nextPageToken"] = next_page_token
|
||||
return request("GET", "/rest/api/3/search/jql", params=params)
|
||||
if start_at:
|
||||
params["startAt"] = start_at
|
||||
return request("GET", "/rest/api/" + _api() + "/search", params=params)
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
query = str(inputs.get("query") or "").strip()
|
||||
if not query:
|
||||
raise Exception("query is required")
|
||||
max_results = int(inputs.get("max_results") or 50)
|
||||
next_page_token = str(inputs.get("next_page_token") or "") or None
|
||||
start_at = inputs.get("start_at")
|
||||
start_at = int(start_at) if str(start_at or "").strip() != "" else None
|
||||
|
||||
res = search_issues(query, max_results, next_page_token=next_page_token, start_at=start_at)
|
||||
print(json.dumps({
|
||||
"result": [flatten_issue(i) for i in res.get("issues", [])],
|
||||
"next_page_token": res.get("nextPageToken", ""),
|
||||
"total": res.get("total", 0),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,119 @@
|
||||
import base64, json, os, 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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
def flatten_issue(issue):
|
||||
f = issue.get("fields") or {}
|
||||
rendered = issue.get("renderedFields") or {}
|
||||
|
||||
def user(u):
|
||||
u = u or {}
|
||||
return u.get("displayName") or u.get("name") or u.get("emailAddress") or ""
|
||||
|
||||
desc = rendered.get("description") or ""
|
||||
if desc:
|
||||
desc = re.sub(r"<[^>]+>", "", desc)
|
||||
else:
|
||||
d = f.get("description")
|
||||
desc = d if isinstance(d, str) else ""
|
||||
return {
|
||||
"id": issue.get("id") or "",
|
||||
"key": issue.get("key") or "",
|
||||
"summary": f.get("summary") or "",
|
||||
"description": desc,
|
||||
"status": (f.get("status") or {}).get("name") or "",
|
||||
"resolution": (f.get("resolution") or {}).get("name") or "",
|
||||
"priority": (f.get("priority") or {}).get("name") or "",
|
||||
"project": (f.get("project") or {}).get("name") or "",
|
||||
"issue_type": (f.get("issuetype") or {}).get("name") or "",
|
||||
"created": f.get("created") or "",
|
||||
"updated": f.get("updated") or "",
|
||||
"due_date": f.get("duedate") or "",
|
||||
"labels": f.get("labels") or [],
|
||||
"assignee": user(f.get("assignee")),
|
||||
"reporter": user(f.get("reporter")),
|
||||
"creator": user(f.get("creator")),
|
||||
"attachments": [{"id": a.get("id"), "filename": a.get("filename"), "size": a.get("size")}
|
||||
for a in (f.get("attachment") or [])],
|
||||
"url": issue.get("self") or "",
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
sprint_id = str(args.get("sprint_id") or "").strip()
|
||||
if not sprint_id:
|
||||
raise Exception("sprint_id is required")
|
||||
jql_query = str(args.get("jql_query") or "").strip()
|
||||
limit = int(args.get("limit") or 50)
|
||||
offset = int(args.get("offset") or 0)
|
||||
|
||||
res = request("GET",
|
||||
"/rest/agile/1.0/sprint/" + urllib.parse.quote(sprint_id, safe="") + "/issue",
|
||||
params={
|
||||
"jql": jql_query,
|
||||
"maxResults": limit,
|
||||
"startAt": offset,
|
||||
"expand": "renderedFields",
|
||||
})
|
||||
|
||||
print(json.dumps({
|
||||
"sprint_id": sprint_id,
|
||||
"result": [flatten_issue(i) for i in res.get("issues", []) or []],
|
||||
"total": res.get("total", 0),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,82 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
res = request("GET", "/rest/api/" + _api() + "/myself")
|
||||
if not res.get("accountId") and not res.get("name"):
|
||||
raise Exception("Unexpected response from Jira: no user identity returned")
|
||||
print(json.dumps({
|
||||
"ok": True,
|
||||
"user": res.get("displayName") or res.get("name") or "",
|
||||
"cloud": _is_cloud(),
|
||||
}))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,111 @@
|
||||
import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
import mimetypes
|
||||
import uuid
|
||||
|
||||
|
||||
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()
|
||||
return root + "/" + cloud_id if cloud_id else root
|
||||
|
||||
|
||||
def _auth():
|
||||
cfg = _cfg()
|
||||
pat = str(cfg.get("pat") or "")
|
||||
if pat:
|
||||
return "Bearer " + pat
|
||||
cred = (str(cfg.get("username") or "") + ":" + str(cfg.get("api_key") or "")).encode("utf-8")
|
||||
return "Basic " + base64.b64encode(cred).decode("ascii")
|
||||
|
||||
|
||||
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
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ref = issue_ref(inputs)
|
||||
file_name = str(inputs.get("file_name") or "")
|
||||
if not file_name:
|
||||
raise Exception("file_name is required")
|
||||
content_base64 = str(inputs.get("content_base64") or "")
|
||||
if not content_base64:
|
||||
raise Exception("content_base64 is required")
|
||||
|
||||
boundary = "----riposte" + uuid.uuid4().hex
|
||||
content = base64.b64decode(content_base64)
|
||||
mime = mimetypes.guess_type(file_name)[0] or "application/octet-stream"
|
||||
|
||||
body = (
|
||||
("--" + boundary + "\r\n"
|
||||
+ 'Content-Disposition: form-data; name="file"; filename="' + file_name + '"\r\n'
|
||||
+ "Content-Type: " + mime + "\r\n\r\n").encode("utf-8")
|
||||
+ content
|
||||
+ ("\r\n--" + boundary + "--\r\n").encode("utf-8")
|
||||
)
|
||||
|
||||
headers = {
|
||||
"Accept": "application/json",
|
||||
"Authorization": _auth(),
|
||||
# Jira requires this header for attachment uploads (XSRF check bypass).
|
||||
"X-Atlassian-Token": "no-check",
|
||||
"Content-Type": "multipart/form-data; boundary=" + boundary,
|
||||
}
|
||||
|
||||
url = _base() + "/rest/api/" + _api() + "/issue/" + ref + "/attachments"
|
||||
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=120) as r:
|
||||
raw = r.read()
|
||||
res = json.loads(raw) if raw else {}
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user