Release v0.1.6
This commit is contained in:
249
scripts/gitea_common.py
Normal file
249
scripts/gitea_common.py
Normal file
@@ -0,0 +1,249 @@
|
||||
#!/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"{self.base_url.rstrip('/')}/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="")
|
||||
|
||||
|
||||
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:
|
||||
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 _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
|
||||
Reference in New Issue
Block a user