105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib import error, parse, request
|
|
|
|
|
|
def _required_env(name: str) -> str:
|
|
value = os.getenv(name, '').strip()
|
|
if not value:
|
|
raise SystemExit(f'Missing required environment variable: {name}')
|
|
return value
|
|
|
|
|
|
def _headers(token: str) -> dict[str, str]:
|
|
return {
|
|
'Authorization': f'token {token}',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
}
|
|
|
|
|
|
def _call_json(method: str, url: str, token: str, payload: dict) -> dict:
|
|
data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
|
|
req = request.Request(url, data=data, headers=_headers(token), method=method)
|
|
try:
|
|
with request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode('utf-8'))
|
|
except error.HTTPError as exc:
|
|
body = exc.read().decode('utf-8', errors='replace')
|
|
raise SystemExit(f'Gitea API error {exc.code}: {body}') from exc
|
|
except error.URLError as exc:
|
|
raise SystemExit(f'Failed to reach Gitea API: {exc}') from exc
|
|
|
|
|
|
def _load_text(path: Path) -> str:
|
|
return path.read_text(encoding='utf-8')
|
|
|
|
|
|
def _repo_api_base(base_url: str, repo: str) -> str:
|
|
owner, name = repo.split('/', 1)
|
|
return f"{base_url.rstrip('/')}/api/v1/repos/{parse.quote(owner)}/{parse.quote(name)}"
|
|
|
|
|
|
def create_issue(base_url: str, token: str, repo: str, title: str, body: str) -> dict:
|
|
url = f'{_repo_api_base(base_url, repo)}/issues'
|
|
return _call_json('POST', url, token, {'title': title, 'body': body})
|
|
|
|
|
|
def create_comment(base_url: str, token: str, repo: str, issue_number: int, body: str) -> dict:
|
|
url = f'{_repo_api_base(base_url, repo)}/issues/{issue_number}/comments'
|
|
return _call_json('POST', url, token, {'body': body})
|
|
|
|
|
|
def update_issue(base_url: str, token: str, repo: str, issue_number: int, title: str | None, body: str | None) -> dict:
|
|
payload: dict[str, str] = {}
|
|
if title is not None:
|
|
payload['title'] = title
|
|
if body is not None:
|
|
payload['body'] = body
|
|
url = f'{_repo_api_base(base_url, repo)}/issues/{issue_number}'
|
|
return _call_json('PATCH', url, token, payload)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description='Create or update Gitea issues/comments.')
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
|
|
common = argparse.ArgumentParser(add_help=False)
|
|
common.add_argument('--repo', required=True, help='owner/repo')
|
|
common.add_argument('--body-file', required=True, help='UTF-8 markdown file path')
|
|
|
|
p_issue = sub.add_parser('create-issue', parents=[common])
|
|
p_issue.add_argument('--title', required=True)
|
|
|
|
p_comment = sub.add_parser('create-comment', parents=[common])
|
|
p_comment.add_argument('--issue-number', type=int, required=True)
|
|
|
|
p_update = sub.add_parser('update-issue', parents=[common])
|
|
p_update.add_argument('--issue-number', type=int, required=True)
|
|
p_update.add_argument('--title')
|
|
|
|
args = parser.parse_args()
|
|
|
|
base_url = _required_env('GITEA_URL')
|
|
token = _required_env('GITEA_TOKEN')
|
|
body = _load_text(Path(args.body_file))
|
|
|
|
if args.command == 'create-issue':
|
|
result = create_issue(base_url, token, args.repo, args.title, body)
|
|
elif args.command == 'create-comment':
|
|
result = create_comment(base_url, token, args.repo, args.issue_number, body)
|
|
else:
|
|
result = update_issue(base_url, token, args.repo, args.issue_number, args.title, body)
|
|
|
|
json.dump(result, sys.stdout, ensure_ascii=False, indent=2)
|
|
sys.stdout.write('\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|