hurui200320
0408c27e6e
Includes agents, skills, scripts, container launch scripts, and project configuration files (opencode.json, crush.json, Dockerfile). Add repo reference for skill container Add workspace for persist container, used for testing cli aa
290 lines
9.0 KiB
Python
Executable File
290 lines
9.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Search Forgejo issues and pull requests by keyword.
|
|
|
|
Fetches issues from the Forgejo REST API page by page and matches
|
|
keywords locally against title and body text. This avoids the
|
|
unreliable server-side ``q`` search parameter.
|
|
|
|
Requires only Python stdlib (no pip dependencies).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
PAGE_SIZE = 50 # Forgejo API max per page
|
|
MAX_RETRIES = 3
|
|
RETRY_DELAY = 2 # seconds, doubled each retry
|
|
|
|
|
|
def build_url(server: str, repo: str, params: dict) -> str:
|
|
owner, name = repo.split("/", 1)
|
|
base = f"{server}/api/v1/repos/{owner}/{name}/issues"
|
|
query = urllib.parse.urlencode(params)
|
|
return f"{base}?{query}"
|
|
|
|
|
|
def fetch_page(
|
|
server: str, repo: str, token: str, params: dict
|
|
) -> tuple[list[dict], int]:
|
|
"""Fetch one page of issues with retry. Returns (issues, total_count)."""
|
|
url = build_url(server, repo, params)
|
|
req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
|
|
|
|
last_err: Exception | None = None
|
|
delay = RETRY_DELAY
|
|
for attempt in range(1, MAX_RETRIES + 1):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
total = int(resp.headers.get("X-Total-Count", 0))
|
|
data = json.loads(resp.read().decode())
|
|
return data, total
|
|
except (urllib.error.URLError, OSError, ValueError) as e:
|
|
last_err = e
|
|
if attempt < MAX_RETRIES:
|
|
time.sleep(delay)
|
|
delay *= 2
|
|
|
|
# All retries exhausted
|
|
raise last_err # type: ignore[misc]
|
|
|
|
|
|
def matches_query(issue: dict, keywords: list[str], mode: str) -> bool:
|
|
"""Check if issue title+body contains the keywords per the match mode.
|
|
|
|
Each keyword is matched as a case-insensitive substring. Keywords
|
|
are expected to be pre-lowered by the caller.
|
|
|
|
Args:
|
|
issue: A Forgejo issue dict (must have "title" and optionally "body").
|
|
keywords: Lowercased keyword strings to search for.
|
|
mode: "and" (all keywords must match) or "or" (any keyword matches).
|
|
"""
|
|
text = (issue.get("title", "") + " " + (issue.get("body", "") or "")).lower()
|
|
if mode == "and":
|
|
return all(kw in text for kw in keywords)
|
|
return any(kw in text for kw in keywords)
|
|
|
|
|
|
def search_issues(args: argparse.Namespace) -> dict:
|
|
"""Run the search and return structured results."""
|
|
server = args.server.rstrip("/")
|
|
repo = args.repo
|
|
|
|
# Validate repo format
|
|
if "/" not in repo:
|
|
print(
|
|
f"Error: --repo must be in 'owner/repo' format, got: {repo}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
keywords = [kw.lower() for kw in args.keywords]
|
|
match_mode = args.match_mode
|
|
|
|
matches: list[dict] = []
|
|
page = 1
|
|
scanned = 0
|
|
total_on_server = 0
|
|
|
|
while len(matches) < args.limit:
|
|
params: dict[str, str | int] = {
|
|
"type": args.type,
|
|
"state": args.state,
|
|
"limit": PAGE_SIZE,
|
|
"page": page,
|
|
}
|
|
if args.labels:
|
|
params["labels"] = args.labels
|
|
if args.milestone:
|
|
params["milestones"] = args.milestone
|
|
if args.assignee:
|
|
params["owner"] = args.assignee
|
|
|
|
try:
|
|
issues, total = fetch_page(server, repo, args.token, params)
|
|
except urllib.error.HTTPError as e:
|
|
body = ""
|
|
try:
|
|
body = e.read().decode()
|
|
except Exception:
|
|
pass
|
|
print(f"Error: HTTP {e.code} from {server}: {body}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except (urllib.error.URLError, OSError) as e:
|
|
reason = getattr(e, "reason", None) or str(e)
|
|
print(
|
|
f"Error: Could not connect to {server} after {MAX_RETRIES} "
|
|
f"attempts: {reason}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
if page == 1:
|
|
total_on_server = total
|
|
|
|
scanned += len(issues)
|
|
|
|
for issue in issues:
|
|
if matches_query(issue, keywords, match_mode):
|
|
matches.append(issue)
|
|
if len(matches) >= args.limit:
|
|
break
|
|
|
|
# Stop if this page returned fewer than PAGE_SIZE (no more results)
|
|
if len(issues) < PAGE_SIZE:
|
|
break
|
|
|
|
page += 1
|
|
|
|
# Format results
|
|
results = []
|
|
for issue in matches:
|
|
labels = [lbl["name"] for lbl in issue.get("labels", []) if lbl.get("name")]
|
|
assignee = issue.get("assignee")
|
|
milestone = issue.get("milestone")
|
|
|
|
entry: dict = {
|
|
"number": issue["number"],
|
|
"title": issue["title"],
|
|
"state": issue["state"],
|
|
"labels": labels,
|
|
"assignee": assignee["login"] if assignee else None,
|
|
"milestone": milestone["title"] if milestone else None,
|
|
"url": issue["html_url"],
|
|
}
|
|
|
|
if args.body:
|
|
body_text = issue.get("body", "") or ""
|
|
# Collapse whitespace and truncate
|
|
snippet = " ".join(body_text.split())
|
|
if len(snippet) > 200:
|
|
snippet = snippet[:200] + "..."
|
|
entry["body_snippet"] = snippet
|
|
|
|
results.append(entry)
|
|
|
|
return {
|
|
"total_on_server": total_on_server,
|
|
"scanned": scanned,
|
|
"showing": len(results),
|
|
"issues": results,
|
|
}
|
|
|
|
|
|
def format_compact(data: dict) -> str:
|
|
"""Format results as compact text, one line per issue."""
|
|
lines = [
|
|
f"Scanned {data['scanned']} of {data['total_on_server']} issues, "
|
|
f"found {data['showing']} matches"
|
|
]
|
|
|
|
for issue in data["issues"]:
|
|
labels_str = ", ".join(issue["labels"]) if issue["labels"] else ""
|
|
labels_part = f" ({labels_str})" if labels_str else ""
|
|
body_part = ""
|
|
if "body_snippet" in issue and issue["body_snippet"]:
|
|
body_part = f"\n {issue['body_snippet']}"
|
|
lines.append(
|
|
f"#{issue['number']} [{issue['state']}]{labels_part} "
|
|
f"{issue['title']}{body_part}"
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Search Forgejo issues and pull requests by keyword.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""\
|
|
examples:
|
|
%(prog)s sandbox validate_path --server https://git.example.com --token TOKEN --repo org/repo
|
|
%(prog)s "foo bar" baz --server https://git.example.com --token TOKEN --repo org/repo
|
|
%(prog)s sandbox --match-mode or --server https://git.example.com --token TOKEN --repo org/repo --state all
|
|
%(prog)s "bug" --server https://git.example.com --token TOKEN --repo org/repo --labels Priority/Critical --json
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"keywords",
|
|
metavar="KEYWORD",
|
|
nargs="+",
|
|
help=(
|
|
"Search keywords. Each is matched as a case-insensitive "
|
|
"substring against issue title and body. Use shell quotes "
|
|
'for multi-word keywords, e.g. "foo bar".'
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--server",
|
|
required=True,
|
|
help="Forgejo server URL (e.g. https://git.example.com)",
|
|
)
|
|
parser.add_argument("--token", required=True, help="Forgejo API token")
|
|
parser.add_argument("--repo", required=True, help="Repository in owner/repo format")
|
|
parser.add_argument(
|
|
"--match-mode",
|
|
choices=["and", "or"],
|
|
default="and",
|
|
help="Keyword matching logic: 'and' requires all keywords, "
|
|
"'or' requires any keyword (default: and)",
|
|
)
|
|
parser.add_argument(
|
|
"--state",
|
|
choices=["open", "closed", "all"],
|
|
default="open",
|
|
help="Issue state filter (default: open)",
|
|
)
|
|
parser.add_argument(
|
|
"--type",
|
|
choices=["issues", "pulls"],
|
|
default="issues",
|
|
help="Type filter (default: issues)",
|
|
)
|
|
parser.add_argument(
|
|
"--labels", default=None, help="Comma-separated label names to filter by"
|
|
)
|
|
parser.add_argument("--milestone", default=None, help="Milestone name to filter by")
|
|
parser.add_argument(
|
|
"--assignee", default=None, help="Assignee username to filter by"
|
|
)
|
|
parser.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=50,
|
|
help="Max total results to return (default: 50)",
|
|
)
|
|
parser.add_argument(
|
|
"--body", action="store_true", help="Include first ~200 chars of issue body"
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
dest="output_json",
|
|
action="store_true",
|
|
help="Output as JSON instead of compact text",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.limit < 1:
|
|
print("Error: --limit must be at least 1", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
data = search_issues(args)
|
|
|
|
if args.output_json:
|
|
print(json.dumps(data, indent=2))
|
|
else:
|
|
print(format_compact(data))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|