368 lines
15 KiB
Python
368 lines
15 KiB
Python
# /// script
|
|
# requires-python = ">=3.12"
|
|
# dependencies = ["copier>=9.10", "questionary>=2.1"]
|
|
# ///
|
|
import argparse
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import questionary
|
|
from copier import run_copy, run_update
|
|
|
|
SOURCE = os.environ.get("TEMPLATES_DIR", "https://git.kotikot.com/templates")
|
|
REF = os.environ.get("TEMPLATES_REF")
|
|
BLANK = "blank"
|
|
BACKENDS = {"python": "backend-python", "other": BLANK}
|
|
FRONTENDS = {"svelte": "frontend-svelte", "other": BLANK}
|
|
SERVE = ["bun", "static", "api"]
|
|
MCP = ["context7", "svelte"]
|
|
|
|
|
|
def template(name: str) -> str:
|
|
local = Path(SOURCE) / name
|
|
return str(local) if local.is_dir() else f"{SOURCE}/{name}.git"
|
|
|
|
|
|
def git_config(key: str) -> str:
|
|
result = subprocess.run(["git", "config", key], capture_output=True, text=True, check=False)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def interactive() -> bool:
|
|
return sys.stdin.isatty()
|
|
|
|
|
|
def ask_text(prompt: str, default: str) -> str:
|
|
return questionary.text(prompt, default=default).unsafe_ask() if interactive() else default
|
|
|
|
|
|
def ask_select(prompt: str, choices: list[str], default: str) -> str:
|
|
if not interactive():
|
|
return default
|
|
return questionary.select(prompt, choices=choices, default=default).unsafe_ask()
|
|
|
|
|
|
def ask_checkbox(prompt: str, choices: list[str], default: list[str]) -> list[str]:
|
|
if not interactive():
|
|
return default
|
|
items = [questionary.Choice(c, checked=c in default) for c in choices]
|
|
return questionary.checkbox(prompt, choices=items).unsafe_ask()
|
|
|
|
|
|
def slugify(name: str) -> str:
|
|
return re.sub(r"[^a-z0-9]", "_", name.lower())
|
|
|
|
|
|
def new(args: argparse.Namespace) -> None:
|
|
dest = Path(args.dest).resolve()
|
|
name = args.name or ask_text("Project name", dest.name)
|
|
slug = args.slug or ask_text("Slug", slugify(name))
|
|
description = args.description if args.description is not None else ask_text("Description", name)
|
|
parts = args.parts.split(",") if args.parts else ask_checkbox("Parts", ["backend", "frontend", "caddy"], ["backend", "frontend", "caddy"])
|
|
backend = frontend = None
|
|
if "backend" in parts:
|
|
backend = args.backend or ask_select("Backend", list(BACKENDS), "python")
|
|
if "frontend" in parts:
|
|
frontend = args.frontend or ask_select("Frontend", list(FRONTENDS), "svelte")
|
|
serve = "bun"
|
|
if frontend and FRONTENDS[frontend] != BLANK:
|
|
options = SERVE if backend else SERVE[:-1]
|
|
serve = args.serve or ask_select("Frontend served by", options, "bun")
|
|
mcp_default = ["svelte"] if frontend and FRONTENDS[frontend] == "frontend-svelte" else []
|
|
mcp = args.mcp.split(",") if args.mcp is not None else ask_checkbox("MCP servers", MCP, mcp_default)
|
|
mcp = [m for m in mcp if m]
|
|
stack = [p for p in ("backend", "frontend") if p in parts]
|
|
single = len(stack) == 1
|
|
routes = [r for r, ok in (("api", backend), ("frontend", frontend and serve != "api")) if ok]
|
|
|
|
def copy(name: str, data: dict, answers_file: str | None = None) -> None:
|
|
run_copy(template(name), dest, data=data, answers_file=answers_file, unsafe=True,
|
|
defaults=args.defaults, vcs_ref=REF)
|
|
|
|
def copy_part(part: str, name: str, data: dict) -> None:
|
|
if name == BLANK:
|
|
copy(name, {**identity, "layout": layout, "part": part}, f".copier/{part}.yml")
|
|
else:
|
|
copy(name, {**identity, "layout": layout, **data})
|
|
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
identity = {"project_name": name, "project_slug": slug}
|
|
copy("infra", {"project_name": name, "project_description": description, "layout": "single" if single else "parts",
|
|
"parts": [] if single else stack, "mcp_servers": mcp})
|
|
layout = "root" if single else "part"
|
|
if backend:
|
|
copy_part("backend", BACKENDS[backend], {"project_description": description, "serve_spa": serve == "api",
|
|
"author_name": git_config("user.name"),
|
|
"author_email": git_config("user.email")})
|
|
if frontend:
|
|
copy_part("frontend", FRONTENDS[frontend], {"serve": serve})
|
|
if "caddy" in parts:
|
|
copy("caddy", {"project_slug": slug, "routes": routes})
|
|
if (dest / "Makefile").is_file():
|
|
subprocess.run(["make", "env"], cwd=dest, check=True)
|
|
if args.repo or (args.repo is None and interactive()):
|
|
call = gitea_api()
|
|
repo = choose_repo(dest, args.repo, call)
|
|
if repo:
|
|
setup_repo(dest, repo, call)
|
|
|
|
|
|
def git_dirty() -> bool:
|
|
result = subprocess.run(["git", "status", "--porcelain"], capture_output=True, text=True, check=False)
|
|
return bool(result.stdout.strip())
|
|
|
|
|
|
def update_message(answers: Path) -> str:
|
|
"""Conventional commit for a template update, with the version it moved to."""
|
|
ref = ""
|
|
for line in answers.read_text(encoding="utf-8").splitlines():
|
|
if line.startswith("_commit:"):
|
|
ref = line.split(":", 1)[1].strip().strip("'\"")
|
|
break
|
|
scope = answers.stem
|
|
return f"chore({scope}): update template to {ref}" if ref else f"chore({scope}): update template"
|
|
|
|
|
|
def refresh_locks(part: str) -> None:
|
|
root = Path(part) if Path(part, "Makefile").is_file() else Path()
|
|
if (root / "uv.lock").is_file():
|
|
subprocess.run(["uv", "lock", "-q"], cwd=root, check=True)
|
|
if (root / "bun.lock").is_file():
|
|
subprocess.run(["bun", "install", "--silent"], cwd=root, check=True)
|
|
|
|
|
|
def update(args: argparse.Namespace) -> None:
|
|
answers = sorted(Path(".copier").glob("*.yml"))
|
|
if args.parts:
|
|
answers = [Path(".copier") / f"{p}.yml" for p in args.parts]
|
|
if not answers:
|
|
sys.exit("no .copier/*.yml here")
|
|
if git_dirty():
|
|
sys.exit("working tree is dirty, commit or stash first")
|
|
for file in answers:
|
|
run_update(".", answers_file=str(file), unsafe=True, overwrite=True,
|
|
skip_answered=not args.ask, defaults=args.defaults, vcs_ref=REF)
|
|
conflicts = subprocess.run(["git", "diff", "--name-only", "--diff-filter=U"],
|
|
capture_output=True, text=True, check=False).stdout.split()
|
|
if conflicts:
|
|
sys.exit(f"{file.stem}: resolve conflicts, then commit and rerun:\n " + "\n ".join(conflicts))
|
|
refresh_locks(file.stem)
|
|
if git_dirty():
|
|
subprocess.run(["git", "add", "-A"], check=True)
|
|
subprocess.run(["git", "commit", "-qm", update_message(file)], check=True)
|
|
print(f"committed {file.stem} update")
|
|
|
|
|
|
CONFIG_FILE = Path.home() / ".config/templates/config.toml"
|
|
|
|
|
|
def config() -> dict:
|
|
import tomllib
|
|
|
|
return tomllib.loads(CONFIG_FILE.read_text(encoding="utf-8")) if CONFIG_FILE.is_file() else {}
|
|
|
|
|
|
GITEA = os.environ.get("GITEA_URL", config().get("gitea_url", "https://git.kotikot.com")).rstrip("/")
|
|
RENOVATE_FILE = "renovate.jsonc"
|
|
RENOVATE_STUB = """{
|
|
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
|
|
"extends": ["local>templates/renovate"]
|
|
}
|
|
"""
|
|
RENOVATE_LABELS = [
|
|
("renovate", "1f6feb", "Pull request from the Renovate bot"),
|
|
("wait-3d", "d29922", "Waits three days: the registry has no release timestamp, the age gate merges it"),
|
|
("deps-code", "7d8590", "Code dependencies: pull requests are collected, nothing merges itself"),
|
|
]
|
|
|
|
|
|
def gitea_token() -> str:
|
|
token = os.environ.get("GITEA_TOKEN", "")
|
|
if token:
|
|
return token
|
|
path = Path.home() / ".config/gitea/token"
|
|
if path.exists():
|
|
return path.read_text(encoding="utf-8").strip()
|
|
sys.exit("no Gitea token: set GITEA_TOKEN or write ~/.config/gitea/token")
|
|
|
|
|
|
def gitea_api():
|
|
import json
|
|
import urllib.request
|
|
|
|
token = gitea_token()
|
|
|
|
def call(method: str, path: str, body: dict | None = None):
|
|
request = urllib.request.Request(
|
|
f"{GITEA}/api/v1{path}",
|
|
data=json.dumps(body).encode() if body is not None else None,
|
|
method=method,
|
|
headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
raw = response.read()
|
|
return json.loads(raw) if raw else None
|
|
|
|
return call
|
|
|
|
|
|
def git(*args: str, cwd: Path) -> str:
|
|
result = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, check=False)
|
|
return result.stdout.strip() if result.returncode == 0 else ""
|
|
|
|
|
|
def project_description(cwd: Path) -> str:
|
|
answers = cwd / ".copier/infra.yml"
|
|
if not answers.is_file():
|
|
return ""
|
|
for line in answers.read_text(encoding="utf-8").splitlines():
|
|
if line.startswith("project_description:"):
|
|
return line.split(":", 1)[1].strip().strip("'\"")
|
|
return ""
|
|
|
|
|
|
def choose_repo(cwd: Path, repo: str | None, call) -> str | None:
|
|
if repo:
|
|
return repo
|
|
orgs = sorted(o["username"] for o in call("GET", "/user/orgs?limit=100") or [])
|
|
default = config().get("org") or (cwd.parent.name if cwd.parent.name in orgs else None)
|
|
if not interactive():
|
|
return f"{default}/{cwd.name}" if default else None
|
|
skip = "(no repo)"
|
|
org = questionary.select("Gitea organization", choices=[skip, *orgs], default=default or skip).unsafe_ask()
|
|
if org == skip:
|
|
return None
|
|
return f"{org}/{ask_text('Repo name', cwd.name)}"
|
|
|
|
|
|
def setup_repo(cwd: Path, repo: str, call) -> None:
|
|
import base64
|
|
import urllib.error
|
|
|
|
org, name = repo.split("/", 1)
|
|
try:
|
|
info = call("GET", f"/repos/{repo}")
|
|
print(f"repo {repo}: already there")
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 404:
|
|
sys.exit(f"repo {repo}: {error.code} {error.read()[:200]!r}")
|
|
info = call("POST", f"/orgs/{org}/repos", {
|
|
"name": name,
|
|
"description": project_description(cwd),
|
|
"private": config().get("private", True),
|
|
"default_branch": "main",
|
|
})
|
|
print(f"repo {repo}: created")
|
|
|
|
url = info["clone_url"] if config().get("remote") == "https" else info["ssh_url"]
|
|
if git("rev-parse", "--git-dir", cwd=cwd):
|
|
current = git("remote", "get-url", "origin", cwd=cwd)
|
|
if not current:
|
|
git("remote", "add", "origin", url, cwd=cwd)
|
|
print(f"origin: {url}")
|
|
elif current != url:
|
|
print(f"origin stays {current}, the repo is {url}")
|
|
if info.get("empty", True) and git("rev-parse", "--verify", "-q", "HEAD", cwd=cwd):
|
|
if not git("rev-parse", "--verify", "-q", "refs/heads/main", cwd=cwd):
|
|
git("branch", "-M", "main", cwd=cwd)
|
|
if not git("rev-parse", "--verify", "-q", "refs/heads/dev", cwd=cwd):
|
|
git("branch", "dev", "main", cwd=cwd)
|
|
subprocess.run(["git", "push", "-q", "-u", "origin", "main", "dev"], cwd=cwd, check=True)
|
|
git("switch", "-q", "dev", cwd=cwd)
|
|
print("pushed main and dev, on dev now")
|
|
elif info.get("empty", True):
|
|
print("nothing committed yet: commit, then run giteaproject again to push main and dev")
|
|
|
|
existing = {label["name"] for label in call("GET", f"/repos/{repo}/labels?limit=100") or []}
|
|
for label, color, description in RENOVATE_LABELS:
|
|
if label in existing:
|
|
continue
|
|
call("POST", f"/repos/{repo}/labels", {"name": label, "color": color, "description": description})
|
|
print(f"label {label}: created")
|
|
|
|
remote_empty = call("GET", f"/repos/{repo}").get("empty", True)
|
|
if not remote_empty and not (cwd / RENOVATE_FILE).is_file():
|
|
try:
|
|
call("GET", f"/repos/{repo}/contents/{RENOVATE_FILE}")
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 404:
|
|
sys.exit(f"{RENOVATE_FILE}: {error.code} {error.read()[:200]!r}")
|
|
call("POST", f"/repos/{repo}/contents/{RENOVATE_FILE}", {
|
|
"content": base64.b64encode(RENOVATE_STUB.encode()).decode(),
|
|
"message": "chore(renovate): extend the shared preset",
|
|
})
|
|
print(f"{RENOVATE_FILE}: committed on the server, pull it")
|
|
|
|
teams = call("GET", f"/orgs/{org}/teams?limit=50") or []
|
|
team = next((t for t in teams if t["name"] == "renovate"), None)
|
|
if team is None:
|
|
team = call("POST", f"/orgs/{org}/teams", {
|
|
"name": "renovate",
|
|
"description": "Renovate bot: write on listed repos only, not the whole org",
|
|
"includes_all_repositories": False,
|
|
"permission": "none",
|
|
"units": ["repo.code", "repo.issues", "repo.pulls"],
|
|
"units_map": {"repo.code": "write", "repo.issues": "write", "repo.pulls": "write"},
|
|
"can_create_org_repo": False,
|
|
})
|
|
print(f"team renovate in {org}: created")
|
|
members = {m["login"] for m in call("GET", f"/teams/{team['id']}/members?limit=50") or []}
|
|
if "renovate" not in members:
|
|
call("PUT", f"/teams/{team['id']}/members/renovate")
|
|
print("bot renovate: added to the team")
|
|
repos = {r["name"] for r in call("GET", f"/teams/{team['id']}/repos?limit=50") or []}
|
|
if name not in repos:
|
|
call("PUT", f"/teams/{team['id']}/repos/{org}/{name}")
|
|
print(f"team renovate: access to {name}")
|
|
|
|
print(
|
|
f"\nStill manual: add {repo} to `repositories` in\n"
|
|
"projects/personal/renovate/renovate-config.js of infra/komodo,\n"
|
|
"the bot only visits repos named in that list."
|
|
)
|
|
|
|
|
|
def gitea(args: argparse.Namespace) -> None:
|
|
"""Create the Gitea repo if needed, point origin at it and prepare it for Renovate."""
|
|
cwd = Path.cwd()
|
|
call = gitea_api()
|
|
repo = choose_repo(cwd, args.repo, call)
|
|
if repo is None:
|
|
sys.exit("no repo chosen")
|
|
setup_repo(cwd, repo, call)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(prog="templates")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
n = sub.add_parser("new", help="Compose a project from templates")
|
|
n.add_argument("dest")
|
|
n.add_argument("--name")
|
|
n.add_argument("--slug")
|
|
n.add_argument("--description")
|
|
n.add_argument("--parts", help="comma-separated: backend,frontend,caddy")
|
|
n.add_argument("--backend", choices=list(BACKENDS))
|
|
n.add_argument("--frontend", choices=list(FRONTENDS))
|
|
n.add_argument("--serve", choices=SERVE)
|
|
n.add_argument("--mcp", help="comma-separated: context7,svelte; empty for none")
|
|
n.add_argument("--repo", help="org/name on Gitea to create and use as origin; empty for none")
|
|
n.add_argument("--defaults", action="store_true", help="accept template defaults without asking")
|
|
n.set_defaults(func=new)
|
|
u = sub.add_parser("update", help="Update every part from its template")
|
|
u.add_argument("parts", nargs="*", help="parts to update, default all")
|
|
u.add_argument("--ask", action="store_true", help="ask every question again")
|
|
u.add_argument("--defaults", action="store_true")
|
|
u.set_defaults(func=update)
|
|
g = sub.add_parser("gitea", help="Create the Gitea repo, set origin, prepare it for Renovate")
|
|
g.add_argument("repo", nargs="?", help="org/name on Gitea, asked when omitted")
|
|
g.set_defaults(func=gitea)
|
|
args = parser.parse_args()
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|