300 lines
10 KiB
Python
300 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared helpers for Gitea maintenance scripts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
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) -> None:
|
|
self.target = target
|
|
self.token = token
|
|
|
|
def request_json(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
body: dict[str, Any] | None = None,
|
|
query: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
url = f"{self.target.api_base}{path}"
|
|
if query:
|
|
url = f"{url}?{urllib.parse.urlencode(query, doseq=True)}"
|
|
|
|
data = None
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"User-Agent": "codex-gitea-maintenance",
|
|
}
|
|
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"
|
|
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response: # noqa: S310 - validated Gitea http(s) API URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
|
|
payload = response.read().decode("utf-8")
|
|
if not payload:
|
|
return None
|
|
return json.loads(payload)
|
|
except urllib.error.HTTPError as exc:
|
|
message = exc.read().decode("utf-8", errors="replace")
|
|
raise GiteaError(f"{method} {url} failed with HTTP {exc.code}: {message}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise GiteaError(f"{method} {url} failed: {exc.reason}") from exc
|
|
|
|
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
|