155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
|
|
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
|
|
|
|
|
|
def _auth() -> dict[str, str]:
|
|
headers: dict[str, str] = {"Accept": "application/json"}
|
|
if GITEA_TOKEN:
|
|
headers["Authorization"] = f"token {GITEA_TOKEN}"
|
|
return headers
|
|
|
|
|
|
async def fetch(path: str) -> Any:
|
|
async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client:
|
|
r = await client.get(f"/api/v1/{path}", headers=_auth())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def current_user() -> dict:
|
|
return await fetch("user")
|
|
|
|
|
|
async def repos() -> list[dict]:
|
|
return await fetch("user/repos?limit=50")
|
|
|
|
|
|
async def issues() -> list[dict]:
|
|
return await fetch(
|
|
"repos/issues/search?state=open&assigned=true&type=issues&limit=50"
|
|
)
|
|
|
|
|
|
async def pull_requests() -> list[dict]:
|
|
assigned = await fetch(
|
|
"repos/issues/search?state=open&assigned=true&type=pulls&limit=50"
|
|
)
|
|
review_requested = await fetch(
|
|
"repos/issues/search?state=open&review_requested=true&type=pulls&limit=50"
|
|
)
|
|
merged: dict[int, dict] = {}
|
|
for reason, pulls in (
|
|
("assigned_to_me", assigned or []),
|
|
("review_requested", review_requested or []),
|
|
):
|
|
for pull in pulls:
|
|
identity = pull.get("id")
|
|
if identity not in merged:
|
|
merged[identity] = {**pull, "work_reasons": []}
|
|
merged[identity]["work_reasons"].append(reason)
|
|
return list(merged.values())
|
|
|
|
|
|
async def is_requested_review(repository: str, number: int) -> bool:
|
|
pulls = await pull_requests()
|
|
return any(
|
|
isinstance(pull, dict)
|
|
and pull.get("number") == number
|
|
and "review_requested" in (pull.get("work_reasons") or [])
|
|
and isinstance(pull.get("repository"), dict)
|
|
and pull["repository"].get("full_name") == repository
|
|
for pull in pulls
|
|
)
|
|
|
|
|
|
async def pull_review_detail(repository: str, number: int) -> dict:
|
|
base = f"repos/{repository}/pulls/{number}"
|
|
pull = await fetch(base)
|
|
if not isinstance(pull, dict):
|
|
raise ValueError("Gitea pull request response was not an object")
|
|
files = await fetch(f"{base}/files")
|
|
head_value = pull.get("head")
|
|
head: dict = head_value if isinstance(head_value, dict) else {}
|
|
sha_value = head.get("sha")
|
|
sha = sha_value if isinstance(sha_value, str) else ""
|
|
status = await fetch(f"repos/{repository}/commits/{sha}/status")
|
|
reviews = await fetch(f"{base}/reviews")
|
|
user_value = pull.get("user")
|
|
user: dict = user_value if isinstance(user_value, dict) else {}
|
|
normalized_files = [
|
|
{
|
|
"filename": file["filename"],
|
|
"status": file.get("status") or "changed",
|
|
"additions": file.get("additions") or 0,
|
|
"deletions": file.get("deletions") or 0,
|
|
}
|
|
for file in (files if isinstance(files, list) else [])[:100]
|
|
if isinstance(file, dict) and isinstance(file.get("filename"), str)
|
|
]
|
|
normalized_reviews = [
|
|
{
|
|
"user": {
|
|
"login": (
|
|
review["user"].get("login", "")
|
|
if isinstance(review.get("user"), dict)
|
|
else ""
|
|
)
|
|
},
|
|
"state": review.get("state") or "COMMENT",
|
|
"body": review.get("body") or "",
|
|
}
|
|
for review in (reviews if isinstance(reviews, list) else [])[:50]
|
|
if isinstance(review, dict)
|
|
]
|
|
return {
|
|
"repository": repository,
|
|
"number": number,
|
|
"title": pull.get("title", ""),
|
|
"body": pull.get("body") or "",
|
|
"url": pull.get("html_url", ""),
|
|
"author": user.get("login", ""),
|
|
"ci_state": status.get("state", "unknown") if isinstance(status, dict) else "unknown",
|
|
"files": normalized_files,
|
|
"reviews": normalized_reviews,
|
|
}
|
|
|
|
|
|
async def activity_events() -> list[dict]:
|
|
user = await current_user()
|
|
events = await fetch(f"users/{user['login']}/activities/feeds?limit=20")
|
|
if events is None:
|
|
events = []
|
|
elif not isinstance(events, list):
|
|
raise ValueError("Gitea activity feed response was not a list")
|
|
return [
|
|
{
|
|
"type": (
|
|
event.get("op_type")
|
|
if isinstance(event.get("op_type"), str) and event.get("op_type")
|
|
else "activity"
|
|
),
|
|
"actor": (
|
|
event.get("act_user")
|
|
if isinstance(event.get("act_user"), dict)
|
|
else {}
|
|
),
|
|
"repo": (
|
|
event.get("repo")
|
|
if isinstance(event.get("repo"), dict)
|
|
else {}
|
|
),
|
|
"created_at": (
|
|
event.get("created")
|
|
if isinstance(event.get("created"), str)
|
|
else ""
|
|
),
|
|
}
|
|
for event in (events or [])
|
|
if isinstance(event, dict)
|
|
]
|