forked from Rockachopa/Timmy-time-dashboard
Claude (Opus 4.6)
003e3883fb
Co-authored-by: Claude (Opus 4.6) <claude@hermes.local> Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
"""Gitea REST client — thin wrapper for PR creation and issue commenting.
|
|
|
|
Uses ``settings.gitea_url``, ``settings.gitea_token``, and
|
|
``settings.gitea_repo`` (owner/repo) from config. Degrades gracefully
|
|
when the token is absent or the server is unreachable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class PullRequest:
|
|
"""Minimal representation of a created pull request."""
|
|
|
|
number: int
|
|
title: str
|
|
html_url: str
|
|
|
|
|
|
class GiteaClient:
|
|
"""HTTP client for Gitea's REST API v1.
|
|
|
|
All methods return structured results and never raise — errors are
|
|
logged at WARNING level and indicated via return value.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str | None = None,
|
|
token: str | None = None,
|
|
repo: str | None = None,
|
|
) -> None:
|
|
from config import settings
|
|
|
|
self._base_url = (base_url or settings.gitea_url).rstrip("/")
|
|
self._token = token or settings.gitea_token
|
|
self._repo = repo or settings.gitea_repo
|
|
|
|
# ── internal ────────────────────────────────────────────────────────────
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {
|
|
"Authorization": f"token {self._token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
def _api(self, path: str) -> str:
|
|
return f"{self._base_url}/api/v1/{path.lstrip('/')}"
|
|
|
|
# ── public API ───────────────────────────────────────────────────────────
|
|
|
|
def create_pull_request(
|
|
self,
|
|
title: str,
|
|
body: str,
|
|
head: str,
|
|
base: str = "main",
|
|
) -> PullRequest | None:
|
|
"""Open a pull request.
|
|
|
|
Args:
|
|
title: PR title (keep under 70 chars).
|
|
body: PR body in markdown.
|
|
head: Source branch (e.g. ``self-modify/issue-983``).
|
|
base: Target branch (default ``main``).
|
|
|
|
Returns:
|
|
A ``PullRequest`` dataclass on success, ``None`` on failure.
|
|
"""
|
|
if not self._token:
|
|
logger.warning("Gitea token not configured — skipping PR creation")
|
|
return None
|
|
|
|
try:
|
|
import requests as _requests
|
|
|
|
resp = _requests.post(
|
|
self._api(f"repos/{self._repo}/pulls"),
|
|
headers=self._headers(),
|
|
json={"title": title, "body": body, "head": head, "base": base},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
pr = PullRequest(
|
|
number=data["number"],
|
|
title=data["title"],
|
|
html_url=data["html_url"],
|
|
)
|
|
logger.info("PR #%d created: %s", pr.number, pr.html_url)
|
|
return pr
|
|
except Exception as exc:
|
|
logger.warning("Failed to create PR: %s", exc)
|
|
return None
|
|
|
|
def add_issue_comment(self, issue_number: int, body: str) -> bool:
|
|
"""Post a comment on an issue or PR.
|
|
|
|
Returns:
|
|
True on success, False on failure.
|
|
"""
|
|
if not self._token:
|
|
logger.warning("Gitea token not configured — skipping issue comment")
|
|
return False
|
|
|
|
try:
|
|
import requests as _requests
|
|
|
|
resp = _requests.post(
|
|
self._api(f"repos/{self._repo}/issues/{issue_number}/comments"),
|
|
headers=self._headers(),
|
|
json={"body": body},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
logger.info("Comment posted on issue #%d", issue_number)
|
|
return True
|
|
except Exception as exc:
|
|
logger.warning("Failed to post comment on issue #%d: %s", issue_number, exc)
|
|
return False
|
|
|
|
|
|
# Module-level singleton
|
|
gitea_client = GiteaClient()
|