From ee5c824eb730dd882f8595fd527bf321f9c75059 Mon Sep 17 00:00:00 2001 From: yanyishuai <1093994647@qq.com> Date: Fri, 3 Jul 2026 10:05:11 +0800 Subject: [PATCH] fix: normalize LF line endings (Windows CRLF cleanup) --- scripts/check_bounty_issue_states.py | 50 ++---------- scripts/check_live_bounty_closing_refs.py | 50 +----------- scripts/gh_cli.py | 95 ++++++++++++++++++++++ scripts/public_api_json.py | 96 +++++++++++++++++++++++ 4 files changed, 200 insertions(+), 91 deletions(-) create mode 100644 scripts/gh_cli.py create mode 100644 scripts/public_api_json.py diff --git a/scripts/check_bounty_issue_states.py b/scripts/check_bounty_issue_states.py index 23a0a80d..7cc0953d 100644 --- a/scripts/check_bounty_issue_states.py +++ b/scripts/check_bounty_issue_states.py @@ -4,8 +4,6 @@ import json import subprocess import sys -import urllib.error -import urllib.request from pathlib import Path from typing import Any @@ -13,9 +11,11 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.api_host_args import public_api_host +from scripts.gh_cli import DEFAULT_GH_TIMEOUT_SECONDS as GH_TIMEOUT_SECONDS +from scripts.gh_cli import run_gh_json as _run_gh_json +from scripts.public_api_json import load_public_bounty_list DEFAULT_API_HOST = "https://api.mrwk.online" -GH_TIMEOUT_SECONDS = 30 def _int_or_none(value: Any) -> int | None: @@ -119,31 +119,8 @@ def format_text_report(report: dict[str, Any]) -> str: return "\n".join(lines) -def _run_gh_json(args: list[str]) -> Any: - command = " ".join(args) - try: - completed = subprocess.run( - args, - check=True, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=GH_TIMEOUT_SECONDS, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError(f"gh command timed out after {GH_TIMEOUT_SECONDS}s: {command}") from exc - except subprocess.CalledProcessError as exc: - raise RuntimeError( - "gh command failed " - f"(exit {exc.returncode}): {command}\n" - f"stdout:\n{exc.stdout or exc.output or ''}\n" - f"stderr:\n{exc.stderr or ''}" - ) from exc - return json.loads(completed.stdout) - - def _run_gh(args: list[str]) -> None: + """Run gh for maintainer --fix paths (may mutate GitHub state).""" command = " ".join(args) try: subprocess.run( @@ -166,23 +143,6 @@ def _run_gh(args: list[str]) -> None: ) from exc -def _fetch_json(url: str) -> Any: - request = urllib.request.Request(url, headers={"Accept": "application/json"}) - try: - with urllib.request.urlopen(request, timeout=GH_TIMEOUT_SECONDS) as response: - return json.loads(response.read().decode("utf-8")) - except (TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc: - raise RuntimeError(f"failed to fetch JSON from {url}: {exc}") from exc - - -def _load_public_bounties(api_host: str) -> list[dict[str, Any]]: - url = f"{api_host.rstrip('/')}/api/v1/bounties?status=open&limit=200" - data = _fetch_json(url) - if not isinstance(data, list): - raise RuntimeError(f"expected a JSON list from {url}") - return [item for item in data if isinstance(item, dict)] - - def _load_issue(repo: str, issue_number: int) -> dict[str, Any]: return _run_gh_json( [ @@ -199,7 +159,7 @@ def _load_issue(repo: str, issue_number: int) -> dict[str, Any]: def load_live_data(repo: str, api_host: str) -> dict[str, Any]: - bounties = _load_public_bounties(api_host) + bounties = load_public_bounty_list(api_host, query="status=open&limit=200") issue_numbers = sorted( number for number in (_issue_number(bounty) for bounty in bounties) if number is not None ) diff --git a/scripts/check_live_bounty_closing_refs.py b/scripts/check_live_bounty_closing_refs.py index 227ded65..f3615935 100644 --- a/scripts/check_live_bounty_closing_refs.py +++ b/scripts/check_live_bounty_closing_refs.py @@ -2,10 +2,8 @@ import argparse import json -import subprocess +import subprocess # noqa: F401 - monkeypatched in tests import sys -import urllib.error -import urllib.request from pathlib import Path from typing import Any @@ -14,9 +12,10 @@ from scripts.api_host_args import public_api_host from scripts.bounty_refs import GITHUB_CLOSING_ISSUE_RE +from scripts.gh_cli import run_gh_json as _run_gh_json +from scripts.public_api_json import load_public_bounty_list DEFAULT_API_HOST = "https://api.mrwk.online" -GH_TIMEOUT_SECONDS = 30 GH_PR_SAFETY_CAP = 200 MAX_BOUNTY_REF = 2**63 - 1 @@ -122,47 +121,6 @@ def format_text_report(report: dict[str, Any]) -> str: return "\n".join(lines) -def _run_gh_json(args: list[str]) -> Any: - command = " ".join(args) - try: - completed = subprocess.run( - args, - check=True, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=GH_TIMEOUT_SECONDS, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError(f"gh command timed out after {GH_TIMEOUT_SECONDS}s: {command}") from exc - except subprocess.CalledProcessError as exc: - raise RuntimeError( - "gh command failed " - f"(exit {exc.returncode}): {command}\n" - f"stdout:\n{exc.stdout or exc.output or ''}\n" - f"stderr:\n{exc.stderr or ''}" - ) from exc - return json.loads(completed.stdout) - - -def _fetch_json(url: str) -> Any: - request = urllib.request.Request(url, headers={"Accept": "application/json"}) - try: - with urllib.request.urlopen(request, timeout=GH_TIMEOUT_SECONDS) as response: - return json.loads(response.read().decode("utf-8")) - except (TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc: - raise RuntimeError(f"failed to fetch JSON from {url}: {exc}") from exc - - -def _load_public_bounties(api_host: str) -> list[dict[str, Any]]: - url = f"{api_host.rstrip('/')}/api/v1/bounties?status=open&limit=200" - data = _fetch_json(url) - if not isinstance(data, list): - raise RuntimeError(f"expected a JSON list from {url}") - return [item for item in data if isinstance(item, dict)] - - def _load_pull_requests(repo: str, state: str, pr_numbers: list[int]) -> list[dict[str, Any]]: if pr_numbers: return [ @@ -205,7 +163,7 @@ def _load_pull_requests(repo: str, state: str, pr_numbers: list[int]) -> list[di def load_live_data(repo: str, api_host: str, state: str, pr_numbers: list[int]) -> dict[str, Any]: return { - "bounties": _load_public_bounties(api_host), + "bounties": load_public_bounty_list(api_host, query="status=open&limit=200"), "pull_requests": _load_pull_requests(repo, state, pr_numbers), } diff --git a/scripts/gh_cli.py b/scripts/gh_cli.py new file mode 100644 index 00000000..74e38999 --- /dev/null +++ b/scripts/gh_cli.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import json +import subprocess +from typing import Any + +DEFAULT_GH_TIMEOUT_SECONDS = 30 + +_NON_READ_ONLY_GH_API = frozenset({"POST", "PUT", "PATCH", "DELETE"}) +_NON_READ_ONLY_GH_SUBCOMMANDS = frozenset( + {"comment", "edit", "close", "reopen", "merge", "review", "delete"} +) + + +def _command_text(args: list[str]) -> str: + return " ".join(args) + + +def assert_read_only_gh_command(args: list[str]) -> None: + """Reject gh invocations that could mutate GitHub state.""" + if args[:2] == ["gh", "api"]: + for flag in ("--method", "-X"): + if flag not in args: + continue + index = args.index(flag) + if index + 1 >= len(args): + continue + if args[index + 1].upper() in _NON_READ_ONLY_GH_API: + raise RuntimeError(f"refusing non-read-only gh api command: {_command_text(args)}") + if any(arg in {"issue", "pr"} for arg in args) and any( + arg in _NON_READ_ONLY_GH_SUBCOMMANDS for arg in args + ): + raise RuntimeError(f"refusing non-read-only gh command: {_command_text(args)}") + + +def run_gh(args: list[str], *, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS) -> str: + """Run a read-only gh command and return stdout text.""" + assert_read_only_gh_command(args) + command = _command_text(args) + try: + completed = subprocess.run( + args, + check=True, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError(f"gh command timed out after {timeout_seconds}s: {command}") from exc + except FileNotFoundError as exc: + raise RuntimeError( + "GitHub CLI executable 'gh' was not found; install gh and ensure it is on PATH " + "before using live --repo mode" + ) from exc + except subprocess.CalledProcessError as exc: + raise RuntimeError( + "gh command failed " + f"(exit {exc.returncode}): {command}\n" + f"stdout:\n{exc.stdout or exc.output or ''}\n" + f"stderr:\n{exc.stderr or ''}" + ) from exc + return completed.stdout + + +def run_gh_json(args: list[str], *, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS) -> Any: + """Run a read-only gh command and parse JSON stdout.""" + return json.loads(run_gh(args, timeout_seconds=timeout_seconds)) + + +def ensure_json_list(data: Any, *, label: str) -> list[Any]: + if not isinstance(data, list): + raise RuntimeError(f"{label} returned non-list JSON ({type(data).__name__})") + return data + + +def ensure_json_object(data: Any, *, label: str) -> dict[str, Any]: + if not isinstance(data, dict): + raise RuntimeError(f"{label} returned non-object JSON ({type(data).__name__})") + return data + + +def run_gh_json_list( + args: list[str], *, label: str | None = None, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS +) -> list[Any]: + context = label or _command_text(args) + return ensure_json_list(run_gh_json(args, timeout_seconds=timeout_seconds), label=context) + + +def run_gh_json_object( + args: list[str], *, label: str | None = None, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS +) -> dict[str, Any]: + context = label or _command_text(args) + return ensure_json_object(run_gh_json(args, timeout_seconds=timeout_seconds), label=context) diff --git a/scripts/public_api_json.py b/scripts/public_api_json.py new file mode 100644 index 00000000..b546e215 --- /dev/null +++ b/scripts/public_api_json.py @@ -0,0 +1,96 @@ +"""Shared JSON fetch + shape validation for MergeWork public API reads.""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from typing import Any + +DEFAULT_TIMEOUT_SECONDS = 30 + + +def fetch_public_json(url: str, *, timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS) -> Any: + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: + return json.loads(response.read().decode("utf-8")) + except (TimeoutError, urllib.error.URLError) as exc: + raise RuntimeError(f"public API request failed: {url}") from exc + except json.JSONDecodeError as exc: + raise RuntimeError(f"public API returned invalid JSON from {url}") from exc + + +def ensure_json_list(data: Any, *, url: str, label: str = "response") -> list[Any]: + if not isinstance(data, list): + raise RuntimeError( + f"expected a JSON list for {label} from {url}, got {type(data).__name__}" + ) + return data + + +def ensure_json_object(data: Any, *, url: str, label: str = "response") -> dict[str, Any]: + if not isinstance(data, dict): + raise RuntimeError( + f"expected a JSON object for {label} from {url}, got {type(data).__name__}" + ) + return data + + +def dict_rows(data: list[Any], *, url: str) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for item in ensure_json_list(data, url=url): + if isinstance(item, dict): + rows.append(item) + return rows + + +def load_public_bounty_list( + api_host: str, *, query: str = "status=open&limit=200" +) -> list[dict[str, Any]]: + url = f"{api_host.rstrip('/')}/api/v1/bounties?{query}" + return dict_rows(fetch_public_json(url), url=url) + + +def validate_public_activity(activity: dict[str, Any], *, url: str) -> dict[str, Any]: + contributors = activity.get("contributors") + recent = activity.get("recent") + if contributors is not None and not isinstance(contributors, list): + raise RuntimeError( + f"expected contributors list from {url}, got {type(contributors).__name__}" + ) + if recent is not None and not isinstance(recent, list): + raise RuntimeError(f"expected recent list from {url}, got {type(recent).__name__}") + return activity + + +def load_public_activity(api_host: str, *, limit: int = 200) -> dict[str, Any]: + url = f"{api_host.rstrip('/')}/api/v1/activity?limit={limit}" + activity = ensure_json_object(fetch_public_json(url), url=url, label="activity") + return validate_public_activity(activity, url=url) + + +def extract_public_api_state(bounties: Any, activity: Any) -> dict[str, Any]: + """Best-effort shape extraction used by claim inventory live loads.""" + data: dict[str, Any] = {} + if isinstance(bounties, list): + data["bounties"] = bounties + if isinstance(activity, dict): + contributors = activity.get("contributors") + if isinstance(contributors, list): + data["contributors"] = contributors + recent = activity.get("recent") + if isinstance(recent, list): + data["recent"] = recent + return data + + +def load_public_api_state(api_host: str, *, limit: int = 200) -> dict[str, Any]: + host = api_host.rstrip("/") + bounties_url = f"{host}/api/v1/bounties?limit={limit}" + activity_url = f"{host}/api/v1/activity?limit={limit}" + bounties = fetch_public_json(bounties_url) + activity = fetch_public_json(activity_url) + if isinstance(activity, dict): + validate_public_activity(activity, url=activity_url) + return extract_public_api_state(bounties, activity)