feat(g3fo-commit-jira): 新增无 MCP 的 Jira CLI 并增强 AI 执行流程约束
- 新增 `jira_cli.py` 脚本，支持在不依赖 Atlassian MCP 的环境下通过同一 Jira Token 进行搜索、创建 Issue 和评论 - 新增共享配置模块 `jira_env.py`，统一加载 `jira_upload.env` 并支持在技能根目录查找配置文件 - 更新 `SKILL.md` 和参考文档，增加「AI 执行契约」和详细的前置条件说明，强制 Step 1→5 线性执行顺序 - 新增 `agent_execution_checklist.md` 检查清单，防止 AI 跳步、误用工具或未等用户确认就绑定 Issue - 改进 `jira_cli.py` 的评论和描述生成，将常见 Markdown 语法转换为 Jira ADF 格式以获得更好的渲染效果
This commit is contained in:
@@ -0,0 +1,558 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Jira REST API CLI — same credentials as upload_attachment.py (jira_upload.env).
|
||||
Use when Atlassian MCP is unavailable (other IDEs).
|
||||
|
||||
Commands:
|
||||
myself Current user (accountId for assignee)
|
||||
search JQL search
|
||||
issue Get one issue by key
|
||||
create Create Task (G3SF by default), assign to self
|
||||
comment Add comment (ADF) to issue
|
||||
|
||||
Examples:
|
||||
python jira_cli.py --config ../jira_upload.env search --limit 5
|
||||
python jira_cli.py --config ../jira_upload.env issue G3SF-123
|
||||
python jira_cli.py --config ../jira_upload.env create --summary "[G3SF] fix foo" --description-file report.md
|
||||
python jira_cli.py --config ../jira_upload.env comment --issue G3SF-123 --body-file impact.md
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
print("ERROR: pip install requests", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
from jira_env import CONFIG_FILENAME, load_config, require_credentials, session_headers
|
||||
|
||||
DEFAULT_JQL = (
|
||||
"project = G3SF AND assignee = currentUser() AND statusCategory != Done "
|
||||
"ORDER BY updated DESC"
|
||||
)
|
||||
|
||||
|
||||
def _empty_doc() -> Dict[str, Any]:
|
||||
return {
|
||||
"type": "doc",
|
||||
"version": 1,
|
||||
"content": [{"type": "paragraph", "content": [{"type": "text", "text": " "}]}],
|
||||
}
|
||||
|
||||
|
||||
def _text_nodes(s: str) -> List[Dict[str, Any]]:
|
||||
"""Single plain text node (non-empty)."""
|
||||
if not s:
|
||||
return []
|
||||
return [{"type": "text", "text": s}]
|
||||
|
||||
|
||||
def parse_inline_adf(s: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Parse inline **bold**, `code`, and [text](url) into ADF text nodes with marks.
|
||||
"""
|
||||
if not s:
|
||||
return []
|
||||
# Order: links, bold, code (non-greedy). Avoid ``` by not matching double backtick start.
|
||||
pattern = re.compile(
|
||||
r"\[([^\]]+)\]\(([^)]+)\)" # [label](url)
|
||||
r"|\*\*(.+?)\*\*" # **bold**
|
||||
r"|(?<!`)`([^`\n]+)`(?!`)" # `code`
|
||||
)
|
||||
nodes: List[Dict[str, Any]] = []
|
||||
last = 0
|
||||
for m in pattern.finditer(s):
|
||||
if m.start() > last:
|
||||
chunk = s[last : m.start()]
|
||||
if chunk:
|
||||
nodes.append({"type": "text", "text": chunk})
|
||||
if m.group(1) is not None and m.group(2) is not None:
|
||||
nodes.append(
|
||||
{
|
||||
"type": "text",
|
||||
"text": m.group(1),
|
||||
"marks": [{"type": "link", "attrs": {"href": m.group(2).strip()}}],
|
||||
}
|
||||
)
|
||||
elif m.group(3) is not None:
|
||||
nodes.append(
|
||||
{"type": "text", "text": m.group(3), "marks": [{"type": "strong"}]}
|
||||
)
|
||||
elif m.group(4) is not None:
|
||||
nodes.append(
|
||||
{"type": "text", "text": m.group(4), "marks": [{"type": "code"}]}
|
||||
)
|
||||
last = m.end()
|
||||
if last < len(s):
|
||||
tail = s[last:]
|
||||
if tail:
|
||||
nodes.append({"type": "text", "text": tail})
|
||||
return nodes
|
||||
|
||||
|
||||
def _paragraph_from_buffer(lines: List[str]) -> Optional[Dict[str, Any]]:
|
||||
if not lines:
|
||||
return None
|
||||
inner: List[Dict[str, Any]] = []
|
||||
for i, line in enumerate(lines):
|
||||
if i > 0:
|
||||
inner.append({"type": "hardBreak"})
|
||||
inner.extend(parse_inline_adf(line) or _text_nodes(line))
|
||||
if not inner:
|
||||
return None
|
||||
return {"type": "paragraph", "content": inner}
|
||||
|
||||
|
||||
def markdown_to_adf(text: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Convert common Markdown to Atlassian Document Format (headings, lists, bold, code, links).
|
||||
Jira REST API stores comments/descriptions as ADF; plain paragraphs showed ### literally.
|
||||
"""
|
||||
text = text or ""
|
||||
if not text.strip():
|
||||
return _empty_doc()
|
||||
|
||||
lines = text.split("\n")
|
||||
content: List[Dict[str, Any]] = []
|
||||
n = len(lines)
|
||||
i = 0
|
||||
para_buf: List[str] = []
|
||||
|
||||
def flush_paragraph() -> None:
|
||||
nonlocal para_buf
|
||||
if not para_buf:
|
||||
return
|
||||
p = _paragraph_from_buffer(para_buf)
|
||||
para_buf = []
|
||||
if p:
|
||||
content.append(p)
|
||||
|
||||
while i < n:
|
||||
raw = lines[i]
|
||||
stripped = raw.strip()
|
||||
|
||||
if not stripped:
|
||||
flush_paragraph()
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# ATX heading # .. ######
|
||||
hm = re.match(r"^(#{1,6})\s+(.+)$", stripped)
|
||||
if hm and len(hm.group(1)) <= 6:
|
||||
flush_paragraph()
|
||||
level = len(hm.group(1))
|
||||
title = hm.group(2).strip()
|
||||
title_nodes = parse_inline_adf(title) or _text_nodes(title)
|
||||
content.append(
|
||||
{"type": "heading", "attrs": {"level": level}, "content": title_nodes}
|
||||
)
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Horizontal rule
|
||||
if re.fullmatch(r"[-*_]{3,}", stripped):
|
||||
flush_paragraph()
|
||||
content.append({"type": "rule"})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Fenced code block
|
||||
if stripped.startswith("```"):
|
||||
flush_paragraph()
|
||||
lang = stripped[3:].strip() or "plaintext"
|
||||
code_lines: List[str] = []
|
||||
i += 1
|
||||
while i < n:
|
||||
if lines[i].strip().startswith("```"):
|
||||
i += 1
|
||||
break
|
||||
code_lines.append(lines[i])
|
||||
i += 1
|
||||
code_text = "\n".join(code_lines)
|
||||
content.append(
|
||||
{
|
||||
"type": "codeBlock",
|
||||
"attrs": {"language": lang},
|
||||
"content": [{"type": "text", "text": code_text or " "}],
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
# Bullet list (consecutive - or *)
|
||||
if re.match(r"^[-*]\s+", stripped):
|
||||
flush_paragraph()
|
||||
items: List[str] = []
|
||||
while i < n:
|
||||
s = lines[i].strip()
|
||||
if not s:
|
||||
break
|
||||
bm = re.match(r"^[-*]\s+(.*)$", s)
|
||||
if not bm:
|
||||
break
|
||||
items.append(bm.group(1))
|
||||
i += 1
|
||||
if items:
|
||||
content.append(
|
||||
{
|
||||
"type": "bulletList",
|
||||
"content": [
|
||||
{
|
||||
"type": "listItem",
|
||||
"content": [
|
||||
{
|
||||
"type": "paragraph",
|
||||
"content": parse_inline_adf(it)
|
||||
or _text_nodes(it),
|
||||
}
|
||||
],
|
||||
}
|
||||
for it in items
|
||||
],
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
# Ordered list
|
||||
om = re.match(r"^(\d+)\.\s+(.*)$", stripped)
|
||||
if om:
|
||||
flush_paragraph()
|
||||
start_order = int(om.group(1))
|
||||
items = [om.group(2)]
|
||||
i += 1
|
||||
while i < n:
|
||||
s = lines[i].strip()
|
||||
if not s:
|
||||
break
|
||||
m = re.match(r"^\d+\.\s+(.*)$", s)
|
||||
if not m:
|
||||
break
|
||||
items.append(m.group(1))
|
||||
i += 1
|
||||
content.append(
|
||||
{
|
||||
"type": "orderedList",
|
||||
"attrs": {"order": start_order},
|
||||
"content": [
|
||||
{
|
||||
"type": "listItem",
|
||||
"content": [
|
||||
{
|
||||
"type": "paragraph",
|
||||
"content": parse_inline_adf(it)
|
||||
or _text_nodes(it),
|
||||
}
|
||||
],
|
||||
}
|
||||
for it in items
|
||||
],
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
# Blockquote: single line > text
|
||||
if stripped.startswith("> "):
|
||||
flush_paragraph()
|
||||
quote_lines: List[str] = [stripped[2:].strip()]
|
||||
i += 1
|
||||
while i < n and lines[i].strip().startswith("> "):
|
||||
quote_lines.append(lines[i].strip()[2:].strip())
|
||||
i += 1
|
||||
q_inner: List[Dict[str, Any]] = []
|
||||
for j, ql in enumerate(quote_lines):
|
||||
if j > 0:
|
||||
q_inner.append({"type": "hardBreak"})
|
||||
q_inner.extend(parse_inline_adf(ql) or _text_nodes(ql))
|
||||
content.append(
|
||||
{
|
||||
"type": "blockquote",
|
||||
"content": [{"type": "paragraph", "content": q_inner}],
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
para_buf.append(raw)
|
||||
i += 1
|
||||
|
||||
flush_paragraph()
|
||||
|
||||
if not content:
|
||||
return _empty_doc()
|
||||
return {"type": "doc", "version": 1, "content": content}
|
||||
|
||||
|
||||
def plain_to_adf(text: str) -> Dict[str, Any]:
|
||||
"""Backward-compatible name: Markdown-aware conversion for Jira ADF."""
|
||||
return markdown_to_adf(text)
|
||||
|
||||
|
||||
def _out(data: Any, fmt: str) -> None:
|
||||
if fmt == "json":
|
||||
print(json.dumps(data, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
print(data)
|
||||
|
||||
|
||||
def cmd_myself(base: str, email: str, token: str, fmt: str) -> int:
|
||||
url = f"{base}/rest/api/3/myself"
|
||||
r = requests.get(url, headers=session_headers(email, token, json_body=False), timeout=60)
|
||||
if r.status_code != 200:
|
||||
print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
|
||||
print(r.text[:800], file=sys.stderr)
|
||||
return 1
|
||||
j = r.json()
|
||||
if fmt == "json":
|
||||
_out(j, "json")
|
||||
else:
|
||||
print(f"accountId: {j.get('accountId')}")
|
||||
print(f"displayName: {j.get('displayName')}")
|
||||
print(f"email: {j.get('emailAddress')}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_search(base: str, email: str, token: str, jql: str, limit: int, fmt: str) -> int:
|
||||
url = f"{base}/rest/api/3/search"
|
||||
params = {
|
||||
"jql": jql,
|
||||
"maxResults": limit,
|
||||
"fields": "key,summary,status,assignee,updated",
|
||||
}
|
||||
r = requests.get(
|
||||
url,
|
||||
headers=session_headers(email, token, json_body=False),
|
||||
params=params,
|
||||
timeout=60,
|
||||
)
|
||||
if r.status_code != 200:
|
||||
print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
|
||||
print(r.text[:800], file=sys.stderr)
|
||||
return 1
|
||||
data = r.json()
|
||||
issues = data.get("issues") or []
|
||||
if fmt == "json":
|
||||
_out(data, "json")
|
||||
return 0
|
||||
if not issues:
|
||||
print("(no issues)")
|
||||
return 0
|
||||
for i, iss in enumerate(issues, 1):
|
||||
f = iss.get("fields") or {}
|
||||
st = (f.get("status") or {}).get("name") or "?"
|
||||
summ = (f.get("summary") or "")[:80]
|
||||
print(f"{i}. {iss.get('key')} [{summ}] status: {st}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_issue(base: str, email: str, token: str, key: str, fmt: str) -> int:
|
||||
url = f"{base}/rest/api/3/issue/{key}"
|
||||
params = {"fields": "summary,status,assignee,project,description,created,updated"}
|
||||
r = requests.get(
|
||||
url,
|
||||
headers=session_headers(email, token, json_body=False),
|
||||
params=params,
|
||||
timeout=60,
|
||||
)
|
||||
if r.status_code != 200:
|
||||
print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
|
||||
print(r.text[:800], file=sys.stderr)
|
||||
return 1
|
||||
j = r.json()
|
||||
if fmt == "json":
|
||||
_out(j, "json")
|
||||
return 0
|
||||
f = j.get("fields") or {}
|
||||
print(f"key: {j.get('key')}")
|
||||
print(f"summary: {f.get('summary')}")
|
||||
print(f"status: {(f.get('status') or {}).get('name')}")
|
||||
proj = f.get("project") or {}
|
||||
print(f"project: {proj.get('key')} {proj.get('name')}")
|
||||
asn = f.get("assignee")
|
||||
print(f"assignee: {(asn or {}).get('displayName') or asn}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_create(
|
||||
base: str,
|
||||
email: str,
|
||||
token: str,
|
||||
project: str,
|
||||
summary: str,
|
||||
description: str,
|
||||
issuetype: str,
|
||||
assign_self: bool,
|
||||
fmt: str,
|
||||
) -> int:
|
||||
fields: Dict[str, Any] = {
|
||||
"project": {"key": project},
|
||||
"summary": summary.strip(),
|
||||
"description": plain_to_adf(description),
|
||||
"issuetype": {"name": issuetype},
|
||||
}
|
||||
if assign_self:
|
||||
mr = requests.get(
|
||||
f"{base}/rest/api/3/myself",
|
||||
headers=session_headers(email, token, json_body=False),
|
||||
timeout=30,
|
||||
)
|
||||
if mr.status_code == 200:
|
||||
aid = mr.json().get("accountId")
|
||||
if aid:
|
||||
fields["assignee"] = {"accountId": aid}
|
||||
payload = {"fields": fields}
|
||||
url = f"{base}/rest/api/3/issue"
|
||||
r = requests.post(url, headers=session_headers(email, token), json=payload, timeout=60)
|
||||
if r.status_code not in (200, 201):
|
||||
# Retry without assignee if permission error
|
||||
if r.status_code == 400 and assign_self and "assignee" in fields:
|
||||
del fields["assignee"]
|
||||
r2 = requests.post(
|
||||
url, headers=session_headers(email, token), json={"fields": fields}, timeout=60
|
||||
)
|
||||
if r2.status_code in (200, 201):
|
||||
key = r2.json().get("key")
|
||||
print(f"WARN: created without assignee (API rejected assignee). key={key}", file=sys.stderr)
|
||||
if fmt == "json":
|
||||
_out(r2.json(), "json")
|
||||
else:
|
||||
print(key)
|
||||
return 0
|
||||
print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
|
||||
print(r.text[:1200], file=sys.stderr)
|
||||
return 1
|
||||
j = r.json()
|
||||
if fmt == "json":
|
||||
_out(j, "json")
|
||||
else:
|
||||
print(j.get("key", j))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_comment(base: str, email: str, token: str, issue: str, body: str, fmt: str) -> int:
|
||||
url = f"{base}/rest/api/3/issue/{issue.strip()}/comment"
|
||||
payload = {"body": plain_to_adf(body)}
|
||||
r = requests.post(url, headers=session_headers(email, token), json=payload, timeout=60)
|
||||
if r.status_code not in (200, 201):
|
||||
print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
|
||||
print(r.text[:800], file=sys.stderr)
|
||||
return 1
|
||||
j = r.json()
|
||||
if fmt == "json":
|
||||
_out(j, "json")
|
||||
else:
|
||||
print(f"OK comment id={j.get('id')}")
|
||||
return 0
|
||||
|
||||
|
||||
def read_body(description: Optional[str], path: Optional[str]) -> str:
|
||||
if path:
|
||||
p = Path(path)
|
||||
if not p.is_file():
|
||||
raise FileNotFoundError(f"File not found: {p}")
|
||||
return p.read_text(encoding="utf-8")
|
||||
return description or ""
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Jira REST CLI (same token as upload_attachment)")
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
default=None,
|
||||
help=f"path to {CONFIG_FILENAME} (default: cwd / script dir / skill dir)",
|
||||
)
|
||||
parser.add_argument("--format", choices=("text", "json"), default="text", dest="fmt")
|
||||
|
||||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p_my = sub.add_parser("myself", help="GET /myself (accountId, email)")
|
||||
p_my.set_defaults(func="myself")
|
||||
|
||||
p_se = sub.add_parser("search", help="JQL search")
|
||||
p_se.add_argument("--jql", default=DEFAULT_JQL, help="JQL (default: G3SF in-progress for current user)")
|
||||
p_se.add_argument("--limit", type=int, default=5)
|
||||
p_se.set_defaults(func="search")
|
||||
|
||||
p_is = sub.add_parser("issue", help="Get issue by key")
|
||||
p_is.add_argument("key", help="e.g. G3SF-123")
|
||||
p_is.set_defaults(func="issue")
|
||||
|
||||
p_cr = sub.add_parser("create", help="Create issue (Task)")
|
||||
p_cr.add_argument("--project", default="G3SF")
|
||||
p_cr.add_argument("--summary", required=True)
|
||||
p_cr.add_argument(
|
||||
"--description",
|
||||
default="",
|
||||
help="Markdown → ADF (headings, lists, bold, code, links)",
|
||||
)
|
||||
p_cr.add_argument("--description-file", dest="description_file", default=None)
|
||||
p_cr.add_argument("--issuetype", default="Task", dest="issuetype")
|
||||
p_cr.add_argument(
|
||||
"--no-assign-self",
|
||||
action="store_true",
|
||||
help="do not set assignee to API token user",
|
||||
)
|
||||
p_cr.set_defaults(func="create")
|
||||
|
||||
p_co = sub.add_parser("comment", help="Add comment to issue")
|
||||
p_co.add_argument("--issue", required=True)
|
||||
p_co.add_argument("--body", default="")
|
||||
p_co.add_argument("--body-file", dest="body_file", default=None)
|
||||
p_co.set_defaults(func="comment")
|
||||
|
||||
args = parser.parse_args()
|
||||
script_dir = Path(__file__).resolve().parent
|
||||
try:
|
||||
cfg = load_config(args.config, script_dir)
|
||||
base, email, token = require_credentials(cfg)
|
||||
except ValueError as e:
|
||||
print(f"ERROR: {e}", file=sys.stderr)
|
||||
return 1
|
||||
except FileNotFoundError as e:
|
||||
print(f"ERROR: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
fmt = args.fmt
|
||||
if args.func == "myself":
|
||||
return cmd_myself(base, email, token, fmt)
|
||||
if args.func == "search":
|
||||
return cmd_search(base, email, token, args.jql, args.limit, fmt)
|
||||
if args.func == "issue":
|
||||
return cmd_issue(base, email, token, args.key.strip(), fmt)
|
||||
if args.func == "create":
|
||||
try:
|
||||
desc = read_body(args.description, args.description_file)
|
||||
except FileNotFoundError as e:
|
||||
print(f"ERROR: {e}", file=sys.stderr)
|
||||
return 1
|
||||
return cmd_create(
|
||||
base,
|
||||
email,
|
||||
token,
|
||||
args.project,
|
||||
args.summary,
|
||||
desc,
|
||||
args.issuetype,
|
||||
assign_self=not args.no_assign_self,
|
||||
fmt=fmt,
|
||||
)
|
||||
if args.func == "comment":
|
||||
try:
|
||||
body = read_body(args.body, args.body_file)
|
||||
except FileNotFoundError as e:
|
||||
print(f"ERROR: {e}", file=sys.stderr)
|
||||
return 1
|
||||
if not body.strip():
|
||||
print("ERROR: --body or --body-file required", file=sys.stderr)
|
||||
return 1
|
||||
return cmd_comment(base, email, token, args.issue, body, fmt)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,75 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Shared Jira config loading for g3fo-commit-jira scripts.
|
||||
Uses the same jira_upload.env / env vars as upload_attachment.py.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional
|
||||
|
||||
CONFIG_FILENAME = "jira_upload.env"
|
||||
|
||||
|
||||
def find_config_file(explicit_path: Optional[str], script_dir: Path) -> Optional[Path]:
|
||||
if explicit_path:
|
||||
p = Path(explicit_path)
|
||||
return p if p.exists() and p.is_file() else None
|
||||
cwd_file = Path.cwd() / CONFIG_FILENAME
|
||||
if cwd_file.exists():
|
||||
return cwd_file
|
||||
script_dir_file = script_dir / CONFIG_FILENAME
|
||||
if script_dir_file.exists():
|
||||
return script_dir_file
|
||||
parent_skill = script_dir.parent / CONFIG_FILENAME
|
||||
if parent_skill.exists():
|
||||
return parent_skill
|
||||
return None
|
||||
|
||||
|
||||
def load_config(config_path: Optional[str] = None, script_dir: Optional[Path] = None) -> Dict[str, str]:
|
||||
"""Load KEY=VALUE from jira_upload.env. Env vars override file values."""
|
||||
if script_dir is None:
|
||||
script_dir = Path(__file__).resolve().parent
|
||||
path = find_config_file(config_path, script_dir)
|
||||
out: Dict[str, str] = {}
|
||||
if path:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if "=" in line:
|
||||
k, _, v = line.partition("=")
|
||||
out[k.strip()] = v.strip().strip('"').strip("'")
|
||||
for key in ("JIRA_BASE_URL", "JIRA_EMAIL", "JIRA_API_TOKEN"):
|
||||
if key in os.environ:
|
||||
out[key] = os.environ[key]
|
||||
return out
|
||||
|
||||
|
||||
def require_credentials(cfg: Dict[str, str]) -> tuple:
|
||||
base_url = (cfg.get("JIRA_BASE_URL") or "").rstrip("/")
|
||||
email = cfg.get("JIRA_EMAIL")
|
||||
token = cfg.get("JIRA_API_TOKEN")
|
||||
if not base_url or not email or not token:
|
||||
raise ValueError(
|
||||
"Missing JIRA_BASE_URL, JIRA_EMAIL, or JIRA_API_TOKEN. "
|
||||
f"Set in {CONFIG_FILENAME} or environment (see references/env_config.md)."
|
||||
)
|
||||
return base_url, email, token
|
||||
|
||||
|
||||
def auth_header(email: str, token: str) -> str:
|
||||
return base64.b64encode(f"{email}:{token}".encode()).decode()
|
||||
|
||||
|
||||
def session_headers(email: str, token: str, json_body: bool = True) -> Dict[str, str]:
|
||||
h = {
|
||||
"Authorization": f"Basic {auth_header(email, token)}",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
if json_body:
|
||||
h["Content-Type"] = "application/json"
|
||||
return h
|
||||
@@ -2,18 +2,15 @@
|
||||
"""
|
||||
Upload one or more files as attachments to a Jira Cloud issue via REST API v3.
|
||||
|
||||
Config: from jira_upload.env (current dir or script dir) or env vars.
|
||||
Config: from jira_upload.env (current dir, script dir, or skill parent dir) or env vars.
|
||||
Usage:
|
||||
python upload_attachment.py --issue ISSUE_KEY --file path1 [--file path2 ...]
|
||||
python upload_attachment.py --config /path/to/jira_upload.env --issue KEY --file path1
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional
|
||||
|
||||
try:
|
||||
import requests
|
||||
@@ -21,57 +18,34 @@ except ImportError:
|
||||
print("ERROR: 'requests' is required. Run: pip install requests", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
CONFIG_FILENAME = "jira_upload.env"
|
||||
from jira_env import CONFIG_FILENAME, auth_header, load_config, require_credentials
|
||||
|
||||
|
||||
def _find_config_file(explicit_path: Optional[str], script_dir: Path) -> Optional[Path]:
|
||||
if explicit_path:
|
||||
p = Path(explicit_path)
|
||||
return p if p.exists() and p.is_file() else None
|
||||
cwd_file = Path.cwd() / CONFIG_FILENAME
|
||||
if cwd_file.exists():
|
||||
return cwd_file
|
||||
script_dir_file = script_dir / CONFIG_FILENAME
|
||||
if script_dir_file.exists():
|
||||
return script_dir_file
|
||||
return None
|
||||
|
||||
|
||||
def load_config(config_path: Optional[str]) -> Dict[str, str]:
|
||||
"""Load KEY=VALUE from jira_upload.env. Env vars override file values."""
|
||||
script_dir = Path(__file__).resolve().parent
|
||||
path = _find_config_file(config_path, script_dir)
|
||||
out = {}
|
||||
if path:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if "=" in line:
|
||||
k, _, v = line.partition("=")
|
||||
out[k.strip()] = v.strip().strip('"').strip("'")
|
||||
for key in ("JIRA_BASE_URL", "JIRA_EMAIL", "JIRA_API_TOKEN"):
|
||||
if key in os.environ:
|
||||
out[key] = os.environ[key]
|
||||
return out
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Upload files as attachments to a Jira issue")
|
||||
parser.add_argument("--issue", required=True, help="Jira issue key (e.g. G3SF-123)")
|
||||
parser.add_argument("--file", action="append", required=True, dest="files", help="Path to file to upload (can be repeated)")
|
||||
parser.add_argument("--config", default=None, help=f"Path to config file (default: {CONFIG_FILENAME} in cwd or script dir)")
|
||||
parser.add_argument(
|
||||
"--file",
|
||||
action="append",
|
||||
required=True,
|
||||
dest="files",
|
||||
help="Path to file to upload (can be repeated)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
default=None,
|
||||
help=f"Path to config file (default: {CONFIG_FILENAME} in cwd, script dir, or skill dir)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
cfg = load_config(args.config)
|
||||
base_url = (cfg.get("JIRA_BASE_URL") or "").rstrip("/")
|
||||
email = cfg.get("JIRA_EMAIL")
|
||||
token = cfg.get("JIRA_API_TOKEN")
|
||||
|
||||
if not base_url or not email or not token:
|
||||
print("ERROR: Set JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN in config file or environment.", file=sys.stderr)
|
||||
print(f" Config file: {CONFIG_FILENAME} (in current dir or script dir), or use --config PATH", file=sys.stderr)
|
||||
try:
|
||||
cfg = load_config(args.config, SCRIPT_DIR)
|
||||
base_url, email, token = require_credentials(cfg)
|
||||
except ValueError as e:
|
||||
print(f"ERROR: {e}", file=sys.stderr)
|
||||
print(f" Use --config PATH or create {CONFIG_FILENAME}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
issue_key = args.issue.strip()
|
||||
@@ -87,9 +61,8 @@ def main() -> int:
|
||||
files_to_upload.append(path)
|
||||
|
||||
url = f"{base_url}/rest/api/3/issue/{issue_key}/attachments"
|
||||
auth_str = base64.b64encode(f"{email}:{token}".encode()).decode()
|
||||
headers = {
|
||||
"Authorization": f"Basic {auth_str}",
|
||||
"Authorization": f"Basic {auth_header(email, token)}",
|
||||
"X-Atlassian-Token": "no-check",
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user