365 lines
12 KiB
Python
365 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared helpers for Gitea maintenance scripts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import http.client
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import subprocess
|
|
import urllib.parse
|
|
from typing import Any
|
|
|
|
|
|
class GiteaError(RuntimeError):
|
|
"""Raised when a Gitea API request fails."""
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class RepoTarget:
|
|
base_url: str
|
|
owner: str
|
|
repo: str
|
|
|
|
@property
|
|
def display(self) -> str:
|
|
return f"{self.base_url.rstrip('/')}/{self.owner}/{self.repo}"
|
|
|
|
@property
|
|
def api_base(self) -> str:
|
|
return f"{validated_http_base_url(self.base_url)}/api/v1"
|
|
|
|
|
|
def repo_root(start: pathlib.Path) -> pathlib.Path:
|
|
result = subprocess.run(
|
|
["git", "-C", str(start), "rev-parse", "--show-toplevel"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
return start.resolve()
|
|
return pathlib.Path(result.stdout.strip()).resolve()
|
|
|
|
|
|
def infer_target(root: pathlib.Path, remote_name: str = "origin") -> RepoTarget:
|
|
env_url = os.environ.get("GITEA_URL")
|
|
env_owner = os.environ.get("GITEA_OWNER")
|
|
env_repo = os.environ.get("GITEA_REPO")
|
|
|
|
remote_url = _git_remote(root, remote_name)
|
|
inferred_url, inferred_owner, inferred_repo = _parse_remote(remote_url)
|
|
|
|
base_url = (env_url or inferred_url).rstrip("/")
|
|
owner = env_owner or inferred_owner
|
|
repo = env_repo or inferred_repo
|
|
|
|
missing = [
|
|
name
|
|
for name, value in (
|
|
("GITEA_URL", base_url),
|
|
("GITEA_OWNER", owner),
|
|
("GITEA_REPO", repo),
|
|
)
|
|
if not value
|
|
]
|
|
if missing:
|
|
joined = ", ".join(missing)
|
|
raise GiteaError(
|
|
f"Could not infer Gitea target. Set {joined}, or configure git remote {remote_name!r}."
|
|
)
|
|
return RepoTarget(base_url=base_url, owner=owner, repo=_strip_git_suffix(repo))
|
|
|
|
|
|
def quote_path(value: str) -> str:
|
|
return urllib.parse.quote(value, safe="")
|
|
|
|
|
|
def validated_http_base_url(value: str) -> str:
|
|
parsed = urllib.parse.urlparse(value.rstrip("/"))
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise GiteaError("Gitea URL must use http:// or https:// with a hostname.")
|
|
if parsed.username or parsed.password:
|
|
raise GiteaError("Gitea URL must not include embedded credentials.")
|
|
return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "", ""))
|
|
|
|
|
|
class GiteaClient:
|
|
def __init__(self, target: RepoTarget, token: str | None, *, timeout: float = 30.0) -> None:
|
|
self.target = target
|
|
self.token = token
|
|
self.timeout = timeout
|
|
self._api_url = urllib.parse.urlparse(target.api_base)
|
|
self._connection: http.client.HTTPConnection | None = None
|
|
|
|
def __enter__(self) -> GiteaClient:
|
|
return self
|
|
|
|
def __exit__(self, *_exc_info: object) -> None:
|
|
self.close()
|
|
|
|
def close(self) -> None:
|
|
if self._connection is not None:
|
|
self._connection.close()
|
|
self._connection = None
|
|
|
|
def request_json(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
body: dict[str, Any] | None = None,
|
|
query: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
request_target = self._request_target(path, query)
|
|
|
|
data = None
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"User-Agent": "codex-gitea-maintenance",
|
|
"Connection": "keep-alive",
|
|
}
|
|
if self.token:
|
|
headers["Authorization"] = f"token {self.token}"
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
method = method.upper()
|
|
safe_to_retry = method in {"GET", "HEAD", "OPTIONS"}
|
|
last_error: BaseException | None = None
|
|
for attempt in range(2 if safe_to_retry else 1):
|
|
try:
|
|
status, _reason, payload = self._request_once(method, request_target, data, headers)
|
|
break
|
|
except (OSError, http.client.HTTPException) as exc:
|
|
self.close()
|
|
last_error = exc
|
|
if attempt == 0 and safe_to_retry:
|
|
continue
|
|
raise GiteaError(f"{method} {self._display_url(request_target)} failed: {exc}") from exc
|
|
else:
|
|
raise GiteaError(f"{method} {self._display_url(request_target)} failed: {last_error}")
|
|
|
|
if status >= 400:
|
|
message = payload.decode("utf-8", errors="replace")
|
|
raise GiteaError(f"{method} {self._display_url(request_target)} failed with HTTP {status}: {message}")
|
|
if not payload:
|
|
return None
|
|
return json.loads(payload.decode("utf-8"))
|
|
|
|
def _connection_for_request(self) -> http.client.HTTPConnection:
|
|
if self._connection is not None:
|
|
return self._connection
|
|
host = self._api_url.hostname
|
|
if not host:
|
|
raise GiteaError("Gitea API URL is missing a hostname.")
|
|
if self._api_url.scheme == "https":
|
|
self._connection = http.client.HTTPSConnection(host, self._api_url.port, timeout=self.timeout) # nosemgrep: python.lang.security.audit.httpsconnection-detected.httpsconnection-detected
|
|
elif self._api_url.scheme == "http":
|
|
self._connection = http.client.HTTPConnection(host, self._api_url.port, timeout=self.timeout)
|
|
else:
|
|
raise GiteaError("Gitea API URL must use http:// or https://.")
|
|
return self._connection
|
|
|
|
def _request_once(
|
|
self,
|
|
method: str,
|
|
request_target: str,
|
|
data: bytes | None,
|
|
headers: dict[str, str],
|
|
) -> tuple[int, str, bytes]:
|
|
connection = self._connection_for_request()
|
|
connection.request(method, request_target, body=data, headers=headers)
|
|
response = connection.getresponse()
|
|
try:
|
|
payload = response.read()
|
|
return response.status, response.reason, payload
|
|
finally:
|
|
if response.will_close:
|
|
self.close()
|
|
|
|
def _request_target(self, path: str, query: dict[str, Any] | None) -> str:
|
|
base_path = self._api_url.path.rstrip("/")
|
|
suffix = path if path.startswith("/") else f"/{path}"
|
|
request_target = f"{base_path}{suffix}"
|
|
if query:
|
|
request_target = f"{request_target}?{urllib.parse.urlencode(query, doseq=True)}"
|
|
return request_target
|
|
|
|
def _display_url(self, request_target: str) -> str:
|
|
netloc = self._api_url.netloc
|
|
return urllib.parse.urlunparse((self._api_url.scheme, netloc, request_target, "", "", ""))
|
|
|
|
def paginate(
|
|
self,
|
|
path: str,
|
|
*,
|
|
query: dict[str, Any] | None = None,
|
|
limit: int = 100,
|
|
) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
page = 1
|
|
effective_limit = min(limit, 50)
|
|
seen_pages: set[tuple[Any, ...]] = set()
|
|
while True:
|
|
page_query = dict(query or {})
|
|
page_query.update({"page": page, "limit": effective_limit})
|
|
payload = self.request_json("GET", path, query=page_query)
|
|
if not isinstance(payload, list):
|
|
raise GiteaError(f"Expected a list from {path}, got {type(payload).__name__}")
|
|
if not payload:
|
|
return items
|
|
signature = tuple(
|
|
item.get("id") or item.get("number") or item.get("index") or item.get("name")
|
|
for item in payload
|
|
if isinstance(item, dict)
|
|
)
|
|
if signature in seen_pages:
|
|
return items
|
|
seen_pages.add(signature)
|
|
items.extend(payload)
|
|
if len(payload) < effective_limit:
|
|
return items
|
|
page += 1
|
|
|
|
|
|
def load_json(path: pathlib.Path) -> Any:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def load_dotenv(path: pathlib.Path | None) -> list[str]:
|
|
if path is None or not path.exists():
|
|
return []
|
|
|
|
loaded: list[str] = []
|
|
for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if stripped.startswith("export "):
|
|
stripped = stripped[len("export ") :].strip()
|
|
if "=" not in stripped:
|
|
raise GiteaError(f"{path}:{line_number}: expected KEY=value")
|
|
key, value = stripped.split("=", 1)
|
|
key = key.strip()
|
|
value = _strip_env_value(value.strip())
|
|
if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key):
|
|
raise GiteaError(f"{path}:{line_number}: invalid environment key {key!r}")
|
|
if key not in os.environ:
|
|
os.environ[key] = value
|
|
loaded.append(key)
|
|
return loaded
|
|
|
|
|
|
def require_token() -> str:
|
|
token = os.environ.get("GITEA_TOKEN")
|
|
if not token:
|
|
raise GiteaError(
|
|
"GITEA_TOKEN is required when using --apply. "
|
|
"Set it in the environment, the target .env, or --env-file."
|
|
)
|
|
return token
|
|
|
|
|
|
def repo_path(owner: str, repo: str, suffix: str) -> str:
|
|
return f"/repos/{quote_path(owner)}/{quote_path(repo)}{suffix}"
|
|
|
|
|
|
def org_path(owner: str, suffix: str) -> str:
|
|
return f"/orgs/{quote_path(owner)}{suffix}"
|
|
|
|
|
|
def labels_by_name(client: GiteaClient, owner: str, repo: str | None = None) -> dict[str, dict[str, Any]]:
|
|
"""Return org labels plus optional repository labels, keyed by label name.
|
|
|
|
Gitea stores organization labels separately from repository labels. Issue
|
|
APIs can use label ids from both scopes on supported instances, so issue
|
|
creation helpers should resolve both. Repository labels intentionally win
|
|
when a repo carries a local label with the same name.
|
|
"""
|
|
|
|
labels: dict[str, dict[str, Any]] = {}
|
|
try:
|
|
for label in client.paginate(org_path(owner, "/labels")):
|
|
name = str(label.get("name") or "")
|
|
if name:
|
|
labels[name] = label
|
|
except GiteaError:
|
|
# User-owned repositories or older instances may not expose org labels.
|
|
pass
|
|
|
|
if repo:
|
|
for label in client.paginate(repo_path(owner, repo, "/labels")):
|
|
name = str(label.get("name") or "")
|
|
if name:
|
|
labels[name] = label
|
|
return labels
|
|
|
|
|
|
def label_ids_by_name(client: GiteaClient, owner: str, repo: str | None = None) -> dict[str, int]:
|
|
ids: dict[str, int] = {}
|
|
for name, label in labels_by_name(client, owner, repo).items():
|
|
label_id = label.get("id") or label.get("index")
|
|
if label_id is None:
|
|
continue
|
|
ids[name] = int(label_id)
|
|
return ids
|
|
|
|
|
|
def _git_remote(root: pathlib.Path, remote_name: str) -> str:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "remote", "get-url", remote_name],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
return ""
|
|
return result.stdout.strip()
|
|
|
|
|
|
def _parse_remote(remote_url: str) -> tuple[str, str, str]:
|
|
if not remote_url:
|
|
return "", "", ""
|
|
|
|
parsed = urllib.parse.urlparse(remote_url)
|
|
if parsed.scheme in {"http", "https", "ssh"} and parsed.netloc:
|
|
path_parts = [part for part in parsed.path.strip("/").split("/") if part]
|
|
owner, repo = _owner_repo_from_parts(path_parts)
|
|
if parsed.scheme in {"http", "https"}:
|
|
prefix = "/".join(path_parts[:-2])
|
|
base_path = f"/{prefix}" if prefix else ""
|
|
return f"{parsed.scheme}://{parsed.netloc}{base_path}", owner, repo
|
|
return f"https://{parsed.hostname or parsed.netloc}", owner, repo
|
|
|
|
scp_like = re.match(r"^(?:[^@]+@)?(?P<host>[^:]+):(?P<path>.+)$", remote_url)
|
|
if scp_like:
|
|
path_parts = [part for part in scp_like.group("path").strip("/").split("/") if part]
|
|
owner, repo = _owner_repo_from_parts(path_parts)
|
|
return f"https://{scp_like.group('host')}", owner, repo
|
|
|
|
return "", "", ""
|
|
|
|
|
|
def _owner_repo_from_parts(path_parts: list[str]) -> tuple[str, str]:
|
|
if len(path_parts) < 2:
|
|
return "", ""
|
|
return path_parts[-2], _strip_git_suffix(path_parts[-1])
|
|
|
|
|
|
def _strip_git_suffix(value: str) -> str:
|
|
return value[:-4] if value.endswith(".git") else value
|
|
|
|
|
|
def _strip_env_value(value: str) -> str:
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
return value[1:-1]
|
|
return value
|