#!/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()