290 lines
8.9 KiB
Python
Executable File
290 lines
8.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Search Gitea issues and pull requests by keyword.
|
|
|
|
Fetches issues from the Gitea 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 # Gitea 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 Gitea 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 Gitea 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="Gitea server URL (e.g. https://git.example.com)",
|
|
)
|
|
parser.add_argument("--token", required=True, help="Gitea 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()
|