Add direct Gitea issue helper and refresh run-001 outputs

scripts/gitea_issue_sync.py

@@ -0,0 +1,104 @@
+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()