329 lines
13 KiB
Plaintext
329 lines
13 KiB
Plaintext
# Git Board — Starlark Backend
|
|
#
|
|
# Entry points:
|
|
# on_request(req) → surface API routes
|
|
# on_tool_call(tool_name, params) → LLM tool execution
|
|
#
|
|
# Modules:
|
|
# http — outbound HTTP to git platform (requires api.http permission)
|
|
# settings — admin: platform, base_url / user: api_token, default_owner, default_repo
|
|
# json — json.encode() / json.decode() (universal, no permission needed)
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Surface API routes (/s/git-board/api/*)
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def on_request(req):
|
|
path = req["path"]
|
|
method = req["method"]
|
|
|
|
if method == "GET" and path == "/repos":
|
|
return _handle_repos(req)
|
|
elif method == "GET" and path == "/board":
|
|
return _handle_board(req)
|
|
elif method == "GET" and path.startswith("/ci/"):
|
|
return _handle_ci(req, path[len("/ci/"):])
|
|
|
|
return _resp(404, {"error": "not found"})
|
|
|
|
|
|
def _handle_repos(req):
|
|
"""List repositories accessible to the user."""
|
|
platform = _platform()
|
|
data = _git_get(_repos_path(platform))
|
|
if data == None:
|
|
return _resp(502, {"error": "git platform request failed"})
|
|
|
|
# Normalize — Gitea returns list, GitHub returns {items: [...]}
|
|
repos = data if type(data) == "list" else data.get("items", data.get("repositories", []))
|
|
out = []
|
|
for r in repos:
|
|
out.append({
|
|
"full_name": r.get("full_name", ""),
|
|
"name": r.get("name", ""),
|
|
"owner": r.get("owner", {}).get("login", ""),
|
|
"description": r.get("description", ""),
|
|
"html_url": r.get("html_url", ""),
|
|
"open_issues": r.get("open_issues_count", 0),
|
|
})
|
|
return _resp(200, {"data": out})
|
|
|
|
|
|
def _handle_board(req):
|
|
"""Combined issues + PRs for kanban board."""
|
|
q = req.get("query", {})
|
|
owner = _str(q.get("owner", "")) or settings.get("default_owner") or ""
|
|
repo = _str(q.get("repo", "")) or settings.get("default_repo") or ""
|
|
if not owner or not repo:
|
|
return _resp(400, {"error": "owner and repo required (query params or user settings)"})
|
|
|
|
platform = _platform()
|
|
|
|
issues = _git_get(_issues_path(platform, owner, repo, "open")) or []
|
|
prs = _git_get(_prs_path(platform, owner, repo, "open")) or []
|
|
|
|
board = {"issues": [], "pull_requests": []}
|
|
for i in issues:
|
|
labels = [l.get("name", "") for l in (i.get("labels") or [])]
|
|
board["issues"].append({
|
|
"number": i.get("number", 0),
|
|
"title": i.get("title", ""),
|
|
"state": i.get("state", ""),
|
|
"labels": labels,
|
|
"assignee": (i.get("assignee") or {}).get("login", ""),
|
|
"created_at": i.get("created_at", ""),
|
|
"updated_at": i.get("updated_at", ""),
|
|
"html_url": i.get("html_url", ""),
|
|
})
|
|
for p in prs:
|
|
board["pull_requests"].append({
|
|
"number": p.get("number", 0),
|
|
"title": p.get("title", ""),
|
|
"state": p.get("state", ""),
|
|
"head": p.get("head", {}).get("ref", ""),
|
|
"base": p.get("base", {}).get("ref", ""),
|
|
"mergeable": p.get("mergeable", None),
|
|
"user": (p.get("user") or {}).get("login", ""),
|
|
"created_at": p.get("created_at", ""),
|
|
"html_url": p.get("html_url", ""),
|
|
})
|
|
|
|
return _resp(200, board)
|
|
|
|
|
|
def _handle_ci(req, ref_path):
|
|
"""CI status for owner/repo/ref."""
|
|
parts = ref_path.split("/", 2)
|
|
if len(parts) < 3:
|
|
return _resp(400, {"error": "path must be /ci/:owner/:repo/:ref"})
|
|
owner, repo, ref = parts[0], parts[1], parts[2]
|
|
platform = _platform()
|
|
data = _git_get(_ci_path(platform, owner, repo, ref))
|
|
if data == None:
|
|
return _resp(502, {"error": "CI status request failed"})
|
|
return _resp(200, data)
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# LLM Tools (called by AI during chat)
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def on_tool_call(tool_name, params):
|
|
platform = _platform()
|
|
owner = params.get("owner", "")
|
|
repo = params.get("repo", "")
|
|
|
|
if tool_name == "list_issues":
|
|
state = params.get("state", "open")
|
|
data = _git_get(_issues_path(platform, owner, repo, state))
|
|
if data == None:
|
|
return {"error": "failed to list issues"}
|
|
items = []
|
|
for i in data:
|
|
labels = [l.get("name", "") for l in (i.get("labels") or [])]
|
|
items.append({
|
|
"number": i.get("number", 0),
|
|
"title": i.get("title", ""),
|
|
"state": i.get("state", ""),
|
|
"labels": labels,
|
|
"assignee": (i.get("assignee") or {}).get("login", ""),
|
|
})
|
|
return {"issues": items, "count": len(items)}
|
|
|
|
elif tool_name == "get_issue":
|
|
num = int(params.get("number", 0))
|
|
data = _git_get(_issue_path(platform, owner, repo, num))
|
|
if data == None:
|
|
return {"error": "issue not found"}
|
|
# Also fetch comments
|
|
comments_data = _git_get(_issue_path(platform, owner, repo, num) + "/comments") or []
|
|
comments = []
|
|
for c in comments_data:
|
|
comments.append({
|
|
"user": (c.get("user") or {}).get("login", ""),
|
|
"body": c.get("body", ""),
|
|
"created_at": c.get("created_at", ""),
|
|
})
|
|
return {
|
|
"number": data.get("number", 0),
|
|
"title": data.get("title", ""),
|
|
"body": data.get("body", ""),
|
|
"state": data.get("state", ""),
|
|
"labels": [l.get("name", "") for l in (data.get("labels") or [])],
|
|
"assignee": (data.get("assignee") or {}).get("login", ""),
|
|
"created_at": data.get("created_at", ""),
|
|
"comments": comments,
|
|
}
|
|
|
|
elif tool_name == "create_issue":
|
|
body = {"title": params.get("title", "")}
|
|
if params.get("body", ""):
|
|
body["body"] = params["body"]
|
|
if params.get("labels", ""):
|
|
body["labels"] = [l.strip() for l in params["labels"].split(",") if l.strip()]
|
|
data = _git_post(_issues_path(platform, owner, repo, ""), body)
|
|
if data == None:
|
|
return {"error": "failed to create issue"}
|
|
return {"number": data.get("number", 0), "title": data.get("title", ""), "html_url": data.get("html_url", "")}
|
|
|
|
elif tool_name == "update_issue":
|
|
num = int(params.get("number", 0))
|
|
patch = {}
|
|
if params.get("title", ""):
|
|
patch["title"] = params["title"]
|
|
if params.get("body", ""):
|
|
patch["body"] = params["body"]
|
|
if params.get("state", ""):
|
|
patch["state"] = params["state"]
|
|
data = _git_patch(_issue_path(platform, owner, repo, num), patch)
|
|
if data == None:
|
|
return {"error": "failed to update issue"}
|
|
return {"number": data.get("number", 0), "state": data.get("state", ""), "title": data.get("title", "")}
|
|
|
|
elif tool_name == "add_comment":
|
|
num = int(params.get("number", 0))
|
|
data = _git_post(_issue_path(platform, owner, repo, num) + "/comments", {"body": params.get("body", "")})
|
|
if data == None:
|
|
return {"error": "failed to add comment"}
|
|
return {"id": data.get("id", 0), "body": data.get("body", "")}
|
|
|
|
elif tool_name == "list_pull_requests":
|
|
state = params.get("state", "open")
|
|
data = _git_get(_prs_path(platform, owner, repo, state))
|
|
if data == None:
|
|
return {"error": "failed to list PRs"}
|
|
items = []
|
|
for p in data:
|
|
items.append({
|
|
"number": p.get("number", 0),
|
|
"title": p.get("title", ""),
|
|
"state": p.get("state", ""),
|
|
"head": p.get("head", {}).get("ref", ""),
|
|
"base": p.get("base", {}).get("ref", ""),
|
|
"user": (p.get("user") or {}).get("login", ""),
|
|
})
|
|
return {"pull_requests": items, "count": len(items)}
|
|
|
|
elif tool_name == "get_ci_status":
|
|
ref = params.get("ref", "")
|
|
data = _git_get(_ci_path(platform, owner, repo, ref))
|
|
if data == None:
|
|
return {"error": "failed to get CI status"}
|
|
# Normalize: Gitea returns {statuses: [...]}, GitHub returns {state, statuses: [...]}
|
|
statuses = data.get("statuses", [])
|
|
items = []
|
|
for s in statuses:
|
|
items.append({
|
|
"context": s.get("context", s.get("name", "")),
|
|
"state": s.get("state", s.get("status", "")),
|
|
"description": s.get("description", ""),
|
|
"target_url": s.get("target_url", ""),
|
|
})
|
|
return {"ref": ref, "state": data.get("state", ""), "statuses": items, "count": len(items)}
|
|
|
|
return {"error": "unknown tool: " + tool_name}
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Git platform abstraction
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _platform():
|
|
return settings.get("platform") or "gitea"
|
|
|
|
def _base_url():
|
|
url = settings.get("base_url") or ""
|
|
if url.endswith("/"):
|
|
url = url[:-1]
|
|
return url
|
|
|
|
def _token():
|
|
return settings.get("api_token") or ""
|
|
|
|
def _api_prefix(platform):
|
|
# Gitea: /api/v1, GitHub: "" (api.github.com already has /repos etc)
|
|
if platform == "github":
|
|
return ""
|
|
return "/api/v1"
|
|
|
|
def _repos_path(platform):
|
|
if platform == "github":
|
|
return "/user/repos?per_page=50&sort=updated"
|
|
return "/repos/search?limit=50&sort=updated"
|
|
|
|
def _issues_path(platform, owner, repo, state):
|
|
if platform == "github":
|
|
return "/repos/" + owner + "/" + repo + "/issues?state=" + state + "&per_page=50"
|
|
return "/repos/" + owner + "/" + repo + "/issues?state=" + state + "&limit=50&type=issues"
|
|
|
|
def _issue_path(platform, owner, repo, number):
|
|
return "/repos/" + owner + "/" + repo + "/issues/" + str(number)
|
|
|
|
def _prs_path(platform, owner, repo, state):
|
|
if platform == "github":
|
|
return "/repos/" + owner + "/" + repo + "/pulls?state=" + state + "&per_page=50"
|
|
return "/repos/" + owner + "/" + repo + "/pulls?state=" + state + "&limit=50"
|
|
|
|
def _ci_path(platform, owner, repo, ref):
|
|
if platform == "github":
|
|
return "/repos/" + owner + "/" + repo + "/commits/" + ref + "/status"
|
|
return "/repos/" + owner + "/" + repo + "/commits/" + ref + "/status"
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# HTTP helpers
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _auth_headers():
|
|
token = _token()
|
|
if not token:
|
|
return {}
|
|
platform = _platform()
|
|
if platform == "github":
|
|
return {"Authorization": "Bearer " + token, "Accept": "application/vnd.github.v3+json"}
|
|
return {"Authorization": "token " + token}
|
|
|
|
def _git_get(path):
|
|
url = _base_url() + _api_prefix(_platform()) + path
|
|
resp = http.get(url=url, headers=_auth_headers())
|
|
if int(resp["status"]) >= 400:
|
|
return None
|
|
body = resp.get("body", "")
|
|
if not body:
|
|
return None
|
|
return json.decode(body)
|
|
|
|
def _git_post(path, data):
|
|
url = _base_url() + _api_prefix(_platform()) + path
|
|
resp = http.post(url=url, body=json.encode(data), headers=dict(_auth_headers(), **{"Content-Type": "application/json"}))
|
|
if int(resp["status"]) >= 400:
|
|
return None
|
|
body = resp.get("body", "")
|
|
if not body:
|
|
return None
|
|
return json.decode(body)
|
|
|
|
def _git_patch(path, data):
|
|
url = _base_url() + _api_prefix(_platform()) + path
|
|
resp = http.post(url=url, body=json.encode(data), headers=dict(_auth_headers(), **{"Content-Type": "application/json", "X-HTTP-Method-Override": "PATCH"}))
|
|
if int(resp["status"]) >= 400:
|
|
return None
|
|
body = resp.get("body", "")
|
|
if not body:
|
|
return None
|
|
return json.decode(body)
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Response helpers
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _resp(status, data):
|
|
return {"status": status, "body": json.encode(data), "headers": {"Content-Type": "application/json"}}
|
|
|
|
def _str(v):
|
|
"""Coerce Starlark query param to string."""
|
|
if v == None:
|
|
return ""
|
|
return str(v)
|