|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
3 | 3 | import base64 |
| 4 | +import binascii |
4 | 5 | import json |
5 | 6 | import re |
6 | 7 | import subprocess |
@@ -1391,23 +1392,151 @@ def fetch_file_lines(self, ref: PullRequestRef, *, path: str, revision: str) -> |
1391 | 1392 | self._file_lines_cache[cache_key] = None |
1392 | 1393 | return None |
1393 | 1394 |
|
1394 | | - encoding = _as_optional_str(payload.get("encoding")) |
1395 | | - content = _as_optional_str(payload.get("content")) |
1396 | | - if encoding != "base64" or content is None: |
1397 | | - self._file_lines_cache[cache_key] = None |
1398 | | - return None |
1399 | | - |
1400 | | - normalized = content.replace("\n", "") |
1401 | | - try: |
1402 | | - decoded = base64.b64decode(normalized, validate=False).decode("utf-8", errors="replace") |
1403 | | - except (ValueError, UnicodeDecodeError): |
| 1395 | + decoded = _decode_repository_contents_text(payload) |
| 1396 | + if decoded is None: |
1404 | 1397 | self._file_lines_cache[cache_key] = None |
1405 | 1398 | return None |
1406 | 1399 |
|
1407 | 1400 | lines = tuple(decoded.splitlines()) |
1408 | 1401 | self._file_lines_cache[cache_key] = lines |
1409 | 1402 | return lines |
1410 | 1403 |
|
| 1404 | + def fetch_pull_request_template(self, repo: str) -> tuple[str | None, str | None]: |
| 1405 | + owner, name = _parse_repo_full_name(repo) |
| 1406 | + self._assert_repository_accessible(owner=owner, name=name) |
| 1407 | + for candidate in _iter_direct_pull_request_template_candidate_paths(): |
| 1408 | + text = self._fetch_repository_text_file(owner=owner, name=name, path=candidate) |
| 1409 | + if text is not None: |
| 1410 | + return candidate, text |
| 1411 | + |
| 1412 | + for parent_path in _PULL_REQUEST_TEMPLATE_PARENT_PATHS: |
| 1413 | + candidate = self._find_direct_pull_request_template_via_listing( |
| 1414 | + owner=owner, |
| 1415 | + name=name, |
| 1416 | + parent_path=parent_path, |
| 1417 | + ) |
| 1418 | + if candidate is None: |
| 1419 | + continue |
| 1420 | + text = self._fetch_repository_text_file(owner=owner, name=name, path=candidate) |
| 1421 | + if text is not None: |
| 1422 | + return candidate, text |
| 1423 | + |
| 1424 | + seen_directories: set[str] = set() |
| 1425 | + for parent_path in _PULL_REQUEST_TEMPLATE_PARENT_PATHS: |
| 1426 | + for directory in self._list_pull_request_template_directories( |
| 1427 | + owner=owner, |
| 1428 | + name=name, |
| 1429 | + parent_path=parent_path, |
| 1430 | + ): |
| 1431 | + if directory in seen_directories: |
| 1432 | + continue |
| 1433 | + seen_directories.add(directory) |
| 1434 | + for candidate in self._list_repository_template_files(owner=owner, name=name, path=directory): |
| 1435 | + text = self._fetch_repository_text_file(owner=owner, name=name, path=candidate) |
| 1436 | + if text is not None: |
| 1437 | + return candidate, text |
| 1438 | + |
| 1439 | + return None, None |
| 1440 | + |
| 1441 | + def _assert_repository_accessible(self, *, owner: str, name: str) -> None: |
| 1442 | + _run_command_json( |
| 1443 | + ["gh", "api", f"repos/{owner}/{name}"], |
| 1444 | + max_attempts=GRAPHQL_MAX_ATTEMPTS, |
| 1445 | + backoff_base_seconds=GRAPHQL_BACKOFF_BASE_SECONDS, |
| 1446 | + backoff_max_seconds=GRAPHQL_BACKOFF_MAX_SECONDS, |
| 1447 | + ) |
| 1448 | + |
| 1449 | + def _fetch_repository_text_file(self, *, owner: str, name: str, path: str) -> str | None: |
| 1450 | + api_path = _build_repository_contents_api_path(owner=owner, name=name, path=path) |
| 1451 | + try: |
| 1452 | + payload = _run_command_json( |
| 1453 | + ["gh", "api", api_path], |
| 1454 | + max_attempts=GRAPHQL_MAX_ATTEMPTS, |
| 1455 | + backoff_base_seconds=GRAPHQL_BACKOFF_BASE_SECONDS, |
| 1456 | + backoff_max_seconds=GRAPHQL_BACKOFF_MAX_SECONDS, |
| 1457 | + ) |
| 1458 | + except RuntimeError as error: |
| 1459 | + if _is_gh_api_not_found_error(str(error)): |
| 1460 | + return None |
| 1461 | + raise |
| 1462 | + |
| 1463 | + if (_as_optional_str(payload.get("type")) or "") != "file": |
| 1464 | + return None |
| 1465 | + return _decode_repository_contents_text(payload) |
| 1466 | + |
| 1467 | + def _find_direct_pull_request_template_via_listing( |
| 1468 | + self, |
| 1469 | + *, |
| 1470 | + owner: str, |
| 1471 | + name: str, |
| 1472 | + parent_path: str, |
| 1473 | + ) -> str | None: |
| 1474 | + candidates: list[str] = [] |
| 1475 | + for entry in self._list_repository_contents(owner=owner, name=name, path=parent_path): |
| 1476 | + if (_as_optional_str(entry.get("type")) or "") != "file": |
| 1477 | + continue |
| 1478 | + entry_name = _as_optional_str(entry.get("name")) or "" |
| 1479 | + if not _is_direct_pull_request_template_name(entry_name): |
| 1480 | + continue |
| 1481 | + candidate_path = _as_optional_str(entry.get("path")) or "" |
| 1482 | + if candidate_path: |
| 1483 | + candidates.append(candidate_path) |
| 1484 | + if not candidates: |
| 1485 | + return None |
| 1486 | + return sorted(candidates, key=str.casefold)[0] |
| 1487 | + |
| 1488 | + def _list_pull_request_template_directories( |
| 1489 | + self, |
| 1490 | + *, |
| 1491 | + owner: str, |
| 1492 | + name: str, |
| 1493 | + parent_path: str, |
| 1494 | + ) -> tuple[str, ...]: |
| 1495 | + candidates: list[str] = [] |
| 1496 | + for entry in self._list_repository_contents(owner=owner, name=name, path=parent_path): |
| 1497 | + if (_as_optional_str(entry.get("type")) or "") != "dir": |
| 1498 | + continue |
| 1499 | + entry_name = _as_optional_str(entry.get("name")) or "" |
| 1500 | + if not _is_pull_request_template_directory_name(entry_name): |
| 1501 | + continue |
| 1502 | + candidate_path = _as_optional_str(entry.get("path")) or "" |
| 1503 | + if candidate_path: |
| 1504 | + candidates.append(candidate_path) |
| 1505 | + return tuple(sorted(candidates, key=str.casefold)) |
| 1506 | + |
| 1507 | + def _list_repository_contents(self, *, owner: str, name: str, path: str) -> tuple[dict[str, object], ...]: |
| 1508 | + api_path = _build_repository_contents_api_path(owner=owner, name=name, path=path) |
| 1509 | + try: |
| 1510 | + payload = _run_command_json_any( |
| 1511 | + ["gh", "api", api_path], |
| 1512 | + max_attempts=GRAPHQL_MAX_ATTEMPTS, |
| 1513 | + backoff_base_seconds=GRAPHQL_BACKOFF_BASE_SECONDS, |
| 1514 | + backoff_max_seconds=GRAPHQL_BACKOFF_MAX_SECONDS, |
| 1515 | + ) |
| 1516 | + except RuntimeError as error: |
| 1517 | + if _is_gh_api_not_found_error(str(error)): |
| 1518 | + return () |
| 1519 | + raise |
| 1520 | + |
| 1521 | + entries: list[dict[str, object]] = [] |
| 1522 | + for raw_entry in _as_list(payload): |
| 1523 | + entry = _as_dict_optional(raw_entry) |
| 1524 | + if entry is None: |
| 1525 | + continue |
| 1526 | + entries.append(entry) |
| 1527 | + return tuple(entries) |
| 1528 | + |
| 1529 | + def _list_repository_template_files(self, *, owner: str, name: str, path: str) -> tuple[str, ...]: |
| 1530 | + candidates: list[str] = [] |
| 1531 | + for entry in self._list_repository_contents(owner=owner, name=name, path=path): |
| 1532 | + if (_as_optional_str(entry.get("type")) or "") != "file": |
| 1533 | + continue |
| 1534 | + candidate_path = _as_optional_str(entry.get("path")) or "" |
| 1535 | + if not _is_pull_request_template_path(candidate_path): |
| 1536 | + continue |
| 1537 | + candidates.append(candidate_path) |
| 1538 | + return tuple(sorted(candidates, key=str.casefold)) |
| 1539 | + |
1411 | 1540 | def submit_pull_request_review( |
1412 | 1541 | self, |
1413 | 1542 | *, |
@@ -2257,6 +2386,76 @@ def _parse_owner_repo(pr_url: str) -> tuple[str, str]: |
2257 | 2386 | return parts[0], parts[1] |
2258 | 2387 |
|
2259 | 2388 |
|
| 2389 | +def _parse_repo_full_name(repo: str) -> tuple[str, str]: |
| 2390 | + owner, separator, name = repo.strip().partition("/") |
| 2391 | + if not owner or not separator or not name: |
| 2392 | + raise RuntimeError(f"invalid repo format: {repo}. Expected OWNER/REPO") |
| 2393 | + return owner, name |
| 2394 | + |
| 2395 | + |
| 2396 | +_PULL_REQUEST_TEMPLATE_PARENT_PATHS = (".github", "", "docs") |
| 2397 | +_DIRECT_PULL_REQUEST_TEMPLATE_BASENAME_VARIANTS = ("PULL_REQUEST_TEMPLATE", "pull_request_template") |
| 2398 | +_PULL_REQUEST_TEMPLATE_FILE_SUFFIXES = (".md", ".txt", ".markdown", ".mdown") |
| 2399 | +_DIRECT_PULL_REQUEST_TEMPLATE_FILENAMES = frozenset( |
| 2400 | + f"{basename}{suffix}".casefold() |
| 2401 | + for basename in _DIRECT_PULL_REQUEST_TEMPLATE_BASENAME_VARIANTS |
| 2402 | + for suffix in _PULL_REQUEST_TEMPLATE_FILE_SUFFIXES |
| 2403 | +) |
| 2404 | + |
| 2405 | + |
| 2406 | +def _build_repository_contents_api_path(*, owner: str, name: str, path: str) -> str: |
| 2407 | + base = f"repos/{owner}/{name}/contents" |
| 2408 | + if not path: |
| 2409 | + return base |
| 2410 | + return f"{base}/{quote(path, safe='/')}" |
| 2411 | + |
| 2412 | + |
| 2413 | +def _build_repository_relative_path(*, parent_path: str, child_name: str) -> str: |
| 2414 | + if not parent_path: |
| 2415 | + return child_name |
| 2416 | + return f"{parent_path}/{child_name}" |
| 2417 | + |
| 2418 | + |
| 2419 | +def _iter_direct_pull_request_template_candidate_paths() -> tuple[str, ...]: |
| 2420 | + return tuple( |
| 2421 | + _build_repository_relative_path(parent_path=parent_path, child_name=f"{basename}{suffix}") |
| 2422 | + for parent_path in _PULL_REQUEST_TEMPLATE_PARENT_PATHS |
| 2423 | + for basename in _DIRECT_PULL_REQUEST_TEMPLATE_BASENAME_VARIANTS |
| 2424 | + for suffix in _PULL_REQUEST_TEMPLATE_FILE_SUFFIXES |
| 2425 | + ) |
| 2426 | + |
| 2427 | + |
| 2428 | +def _decode_repository_contents_text(payload: dict[str, object]) -> str | None: |
| 2429 | + encoding = _as_optional_str(payload.get("encoding")) |
| 2430 | + content = _as_optional_str(payload.get("content")) |
| 2431 | + if encoding != "base64" or content is None: |
| 2432 | + return None |
| 2433 | + |
| 2434 | + normalized = content.replace("\n", "") |
| 2435 | + try: |
| 2436 | + return base64.b64decode(normalized, validate=False).decode("utf-8", errors="replace") |
| 2437 | + except (binascii.Error, ValueError): |
| 2438 | + return None |
| 2439 | + |
| 2440 | + |
| 2441 | +def _is_direct_pull_request_template_name(name: str) -> bool: |
| 2442 | + return name.casefold() in _DIRECT_PULL_REQUEST_TEMPLATE_FILENAMES |
| 2443 | + |
| 2444 | + |
| 2445 | +def _is_pull_request_template_directory_name(name: str) -> bool: |
| 2446 | + return name.casefold() == "pull_request_template" |
| 2447 | + |
| 2448 | + |
| 2449 | +def _is_pull_request_template_path(path: str) -> bool: |
| 2450 | + lowered = path.casefold() |
| 2451 | + return lowered.endswith((".md", ".markdown", ".mdown", ".txt")) |
| 2452 | + |
| 2453 | + |
| 2454 | +def _is_gh_api_not_found_error(message: str) -> bool: |
| 2455 | + lowered = message.casefold() |
| 2456 | + return "404" in lowered and "not found" in lowered |
| 2457 | + |
| 2458 | + |
2260 | 2459 | def _clip_text(text: str | None, fallback: str, limit: int = MAX_INLINE_TEXT) -> tuple[str, bool]: |
2261 | 2460 | if not text: |
2262 | 2461 | return fallback, False |
|
0 commit comments