parvej/zotero-addons-scraper
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix: add pagination for releases API and simplify update_url check

Browse Source 
1. get_releases() now fetches all pages (up to 1000 releases)
2. Remove redundant _check_for_updates() that ran on every release
3. Only check update_url for the latest cached release via
   _check_latest_release_update_url()

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
ytshen 2025-12-24 10:14:56 +08:00
parent c700b3b03a
commit 9d489247f2
2 changed files with 28 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
src/zotero_scraper/clients/github.py
View File

@ -123,23 +123,46 @@ class GitHubClient(BaseHTTPClient):
logger.warning(f"Failed to get repo {owner}/{repo}: {e}")
return None
def get_releases(self, owner: str, repo: str) -> list[dict[str, Any]]:
"""Get all releases for a repository.
def get_releases(
self, owner: str, repo: str, per_page: int = 100, max_pages: int = 10
) -> list[dict[str, Any]]:
"""Get all releases for a repository with pagination.
Args:
owner: Repository owner.
repo: Repository name.
per_page: Number of releases per page (max 100).
max_pages: Maximum number of pages to fetch.
Returns:
List of release data dictionaries.
"""
url = GitHubAPI.releases(owner, repo)
all_releases: list[dict[str, Any]] = []
try:
response = self.get(url, headers=self._headers)
return response.json()
for page in range(1, max_pages + 1):
response = self.get(
url,
headers=self._headers,
params={"per_page": per_page, "page": page},
)
releases = response.json()
if not releases:
break
all_releases.extend(releases)
# If we got fewer than per_page, we've reached the end
if len(releases) < per_page:
break
logger.debug(f"Fetched {len(all_releases)} releases for {owner}/{repo}")
return all_releases
except Exception as e:
logger.warning(f"Failed to get releases for {owner}/{repo}: {e}")
return []
return all_releases if all_releases else []
def get_release(
self, owner: str, repo: str, tag: str

66
src/zotero_scraper/services/cache_builder.py
View File

@ -331,12 +331,6 @@ class ReleaseCacheBuilder:
parse_error="Parse failed or no addon ID",
)
# Check for updates via update_url
updated = self._check_for_updates(repo, details)
if updated:
details = updated["details"]
xpi_url = updated["xpi_url"]
return CachedRelease(
tag=tag,
published_at=published_at,
@ -377,66 +371,6 @@ class ReleaseCacheBuilder:
return None
def _check_for_updates(
self, repo: str, details: Any
) -> Optional[dict[str, Any]]:
"""Check for updates via update_url.
Args:
repo: Repository in "owner/name" format.
details: Parsed XPI details with update_url.
Returns:
Dict with 'details' and 'xpi_url' if update found, None otherwise.
"""
if not details.update_url or not details.id or not details.version:
return None
try:
response = requests.get(details.update_url, timeout=30)
if response.status_code != 200:
return None
update_info = response.json()
updates = (
update_info.get("addons", {})
.get(details.id, {})
.get("updates", [])
)
# Find newer versions
newer_versions = [
u
for u in updates
if compare_versions(u.get("version", "0"), details.version) > 0
]
for update in newer_versions:
xpi_url = update.get("update_link")
if not xpi_url:
continue
owner, name = repo.split("/")
update_filename = f"{owner}#{name}+update{details.version}.xpi"
update_path = self.downloader.download(xpi_url, update_filename)
if update_path:
update_details = self.xpi_parser.parse(update_path)
if update_details and update_details.id:
logger.info(
f"{repo}: Found update via update_url: "
f"{details.version} -> {update_details.version}"
)
return {
"details": update_details,
"xpi_url": xpi_url,
}
except Exception as e:
logger.debug(f"Failed to check updates for {repo}: {e}")
return None
def _check_latest_release_update_url(self, repo: str) -> bool:
"""Check latest cached release's update_url for newer versions.