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

use all release xpis cache

Browse Source
This commit is contained in:
ytshen 2025-12-23 17:34:57 +08:00
parent dc2d1bd336
commit e3e5025d84
9 changed files with 1256 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
.github/workflows/main.yml vendored
View File

@ -23,6 +23,23 @@ jobs:
- name: Checkout Repository
uses: actions/checkout@v4
- name: Checkout publish branch (for release_cache)
uses: actions/checkout@v4
with:
ref: publish
path: publish_branch
continue-on-error: true
- name: Restore release_cache from publish branch
run: |
if [ -d "publish_branch/release_cache" ]; then
cp -r publish_branch/release_cache ./release_cache
echo "Restored release_cache from publish branch"
ls -la release_cache | head -20
else
echo "No release_cache found in publish branch, starting fresh"
fi
- name: Set up Python
uses: actions/setup-python@v5
with:
@ -32,68 +49,30 @@ jobs:
run: |
pip install -r requirements.txt
- name: Store Cache Folder
uses: actions/cache/restore@v4
id: restore-cache
with:
path: |
caches
key: ${{ runner.os }}-files-${{ hashFiles('caches/caches_lockfile') }}
restore-keys: |
${{ runner.os }}-files-
- name: Get Cache Hash
id: get-cache-hash
run: |
echo "CACHE_HASH=$CACHE_HASH" >> "$GITHUB_OUTPUT"
env:
CACHE_HASH: ${{ hashFiles('caches/caches_lockfile') }}
- name: Run Script
- name: Run Scraper
run: |
python main.py \
-i addons \
-o published/addon_infos.json \
--github_repository ${GITHUB_REPOSITORY} \
--github_token ${GITHUB_TOKEN} \
--cache_directory caches \
--runtime_xpi_directory xpis \
--cache_lockfile caches_lockfile \
--create_release True \
# --previous_info_urls \
# https://raw.githubusercontent.com/${GITHUB_REPOSITORY}/publish/addon_infos.json \
# https://github.com/${GITHUB_REPOSITORY}/releases/latest/download/addon_infos.json
--release-cache-dir release_cache \
--github_repository "${GITHUB_REPOSITORY}" \
--github_token "${GITHUB_TOKEN}" \
--create_release True
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Publish in Repository
- name: Copy release_cache to published folder
run: |
cp -r release_cache published/release_cache
echo "Contents of published folder:"
ls -la published
- name: Publish to publish branch
uses: peaceiris/actions-gh-pages@v4
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
# deploy_key: ${{ secrets.ACTIONS_DEPLOY_KEY }}
publish_branch: publish
user_name: 'github-actions[bot]'
user_email: 'github-actions[bot]@users.noreply.github.com'
publish_dir: ./published
- name: Check Cache Need Update
id: check-cache-update
run: |
if [[ "$PRE_CACHE_HASH" != "$CUR_CACHE_HASH" ]]; then
echo "SAVE_CACHE=true" >> "$GITHUB_OUTPUT"
else
echo "SAVE_CACHE=false" >> "$GITHUB_OUTPUT"
fi
env:
PRE_CACHE_HASH: ${{ steps.get-cache-hash.outputs.CACHE_HASH }}
CUR_CACHE_HASH: ${{ hashFiles('caches/caches_lockfile') }}
- name: Save Cache Folder
if: steps.check-cache-update.outputs.SAVE_CACHE == 'true'
uses: actions/cache/save@v4
id: save-cache
with:
path: |
caches
key: ${{ runner.os }}-files-${{ hashFiles('caches/caches_lockfile') }}

189
scripts/migrate_addons.py Normal file
View File

@ -0,0 +1,189 @@
#!/usr/bin/env python3
"""
Migrate addons/*.json from old format to simplified format.
Old format:
{
"repo": "owner/name",
"releases": [
{"targetZoteroVersion": "7", "tagName": "latest"},
{"targetZoteroVersion": "6", "tagName": "v1.0.0"}
]
}
New format:
{
"repo": "owner/name"
}
The release information is now managed by release_cache.json instead.
Usage:
python scripts/migrate_addons.py --input addons --output addons_new
python scripts/migrate_addons.py --input addons --in-place
"""
import argparse
import json
import shutil
from pathlib import Path
def migrate_addon_file(input_path: Path, output_path: Path) -> bool:
"""Migrate a single addon config file.
Args:
input_path: Path to input file.
output_path: Path to output file.
Returns:
True if migration was successful.
"""
try:
with open(input_path, "r", encoding="utf-8") as f:
data = json.load(f)
# Extract only the repo field
repo = data.get("repo")
if not repo:
print(f" Warning: {input_path.name} has no 'repo' field, skipping")
return False
new_data = {"repo": repo}
# Write output
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
json.dump(new_data, f, ensure_ascii=False, indent=2)
f.write("\n")
return True
except Exception as e:
print(f" Error processing {input_path.name}: {e}")
return False
def migrate_all(input_dir: Path, output_dir: Path, in_place: bool = False) -> dict:
"""Migrate all addon config files.
Args:
input_dir: Input directory containing addon configs.
output_dir: Output directory for migrated configs.
in_place: If True, overwrite original files.
Returns:
Statistics dict.
"""
stats = {"total": 0, "migrated": 0, "skipped": 0, "errors": 0}
if not input_dir.exists():
print(f"Error: Input directory not found: {input_dir}")
return stats
json_files = list(input_dir.glob("*.json"))
stats["total"] = len(json_files)
print(f"Found {len(json_files)} addon config files")
for json_file in sorted(json_files):
if in_place:
output_path = json_file
else:
output_path = output_dir / json_file.name
print(f"Processing {json_file.name}...")
# Read and check if already migrated
with open(json_file, "r", encoding="utf-8") as f:
data = json.load(f)
if "releases" not in data:
print(f" Already in new format, skipping")
stats["skipped"] += 1
if not in_place:
shutil.copy(json_file, output_path)
continue
if migrate_addon_file(json_file, output_path):
stats["migrated"] += 1
print(f" Migrated successfully")
else:
stats["errors"] += 1
return stats
def main():
parser = argparse.ArgumentParser(
description="Migrate addons/*.json to simplified format"
)
parser.add_argument(
"--input",
type=str,
default="addons",
help="Input directory (default: addons)",
)
parser.add_argument(
"--output",
type=str,
default="addons_migrated",
help="Output directory (default: addons_migrated)",
)
parser.add_argument(
"--in-place",
action="store_true",
help="Modify files in place (overwrites original files)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be done without making changes",
)
args = parser.parse_args()
input_dir = Path(args.input)
output_dir = Path(args.output)
if args.dry_run:
print("Dry run mode - no changes will be made")
print(f"Input directory: {input_dir}")
print(f"Output directory: {output_dir}")
json_files = list(input_dir.glob("*.json"))
print(f"\nWould process {len(json_files)} files:")
for json_file in sorted(json_files):
with open(json_file, "r", encoding="utf-8") as f:
data = json.load(f)
if "releases" in data:
releases = data.get("releases", [])
print(f" {json_file.name}: would remove {len(releases)} releases")
else:
print(f" {json_file.name}: already migrated")
return
if args.in_place:
confirm = input(
f"This will modify files in {input_dir} in place. Continue? [y/N] "
)
if confirm.lower() != "y":
print("Aborted")
return
stats = migrate_all(input_dir, output_dir, in_place=args.in_place)
print(f"\nMigration complete:")
print(f" Total files: {stats['total']}")
print(f" Migrated: {stats['migrated']}")
print(f" Skipped (already new format): {stats['skipped']}")
print(f" Errors: {stats['errors']}")
if not args.in_place:
print(f"\nMigrated files are in: {output_dir}")
print("To use them, run:")
print(f" rm -rf {input_dir} && mv {output_dir} {input_dir}")
if __name__ == "__main__":
main()

147
src/zotero_scraper/__main__.py
View File

@ -4,9 +4,10 @@ import argparse
import sys
from pathlib import Path
from .cache.file_cache import FileCache
from .cache.release_cache import ReleaseCache
from .config.settings import ScraperConfig
from .services.scraper import AddonScraper
from .services.cache_builder import ReleaseCacheBuilder
from .services.cache_scraper import CacheScraper
from .utils.logging import setup_logging
@ -16,21 +17,57 @@ def parse_args() -> argparse.Namespace:
description="Scrape Zotero addon information from GitHub"
)
# Required arguments
# Mode options
parser.add_argument(
"--build-cache-only",
action="store_true",
help="Only build/update release cache, don't generate output",
)
parser.add_argument(
"--skip-build-cache",
action="store_true",
help="Skip building cache, only generate output from existing cache",
)
# Release cache options
parser.add_argument(
"--release-cache-dir",
type=str,
default="release_cache",
help="Release cache directory (default: release_cache)",
)
parser.add_argument(
"--full-rebuild",
action="store_true",
help="Rebuild cache from scratch, ignoring existing cache",
)
parser.add_argument(
"--parallel",
action="store_true",
help="Build cache in parallel (faster but uses more resources)",
)
parser.add_argument(
"--max-workers",
type=int,
default=4,
help="Max parallel workers for cache building (default: 4)",
)
# GitHub options
parser.add_argument(
"--github_repository",
type=str,
required=True,
help="GitHub repository (owner/repo)",
default=None,
help="GitHub repository (owner/repo) for publishing",
)
# Optional arguments
parser.add_argument(
"--github_token",
type=str,
default=None,
help="GitHub API token",
)
# Input/Output options
parser.add_argument(
"-i",
"--input",
@ -45,17 +82,13 @@ def parse_args() -> argparse.Namespace:
default="addon_infos.json",
help="Output JSON file path (default: addon_infos.json)",
)
# Cache options
parser.add_argument(
"--cache_directory",
type=str,
default="caches",
help="Cache directory (default: caches)",
)
parser.add_argument(
"--cache_lockfile",
type=str,
default="caches_lockfile",
help="Cache lockfile name (default: caches_lockfile)",
help="XPI download cache directory (default: caches)",
)
parser.add_argument(
"--runtime_xpi_directory",
@ -63,18 +96,24 @@ def parse_args() -> argparse.Namespace:
default="xpis",
help="Runtime XPI directory (default: xpis)",
)
# Fallback options
parser.add_argument(
"--previous_info_urls",
nargs="+",
default=[],
help="URLs to previous addon_infos.json for fallback",
)
# Publish options
parser.add_argument(
"--create_release",
type=lambda x: x.lower() in ("true", "1", "yes"),
default=True,
help="Create GitHub release (default: True)",
)
# Logging
parser.add_argument(
"--log_level",
type=str,
@ -87,7 +126,13 @@ def parse_args() -> argparse.Namespace:
def main() -> int:
"""Main entry point."""
"""Main entry point.
Default flow:
1. Build/update release cache (parse new releases)
2. Generate addon_infos.json from cache
3. Publish to GitHub release (if --create_release)
"""
args = parse_args()
# Setup logging
@ -99,8 +144,8 @@ def main() -> int:
logger = logging.getLogger("zotero_scraper")
# Validate arguments
if not args.github_repository:
logger.error("--github_repository is required")
if args.create_release and not args.github_repository:
logger.error("--github_repository is required when --create_release is True")
return 1
# Create configuration
@ -110,35 +155,61 @@ def main() -> int:
config.cache.cache_dir.mkdir(parents=True, exist_ok=True)
config.cache.runtime_xpi_dir.mkdir(parents=True, exist_ok=True)
# Initialize scraper
scraper = AddonScraper(config)
# Initialize release cache
release_cache_dir = Path(args.release_cache_dir)
release_cache = ReleaseCache(release_cache_dir)
# Step 1: Build/update cache (unless skipped)
if not args.skip_build_cache:
logger.info("Step 1: Building/updating release cache...")
builder = ReleaseCacheBuilder(config, release_cache)
if config.github.token:
builder.github.get_rate_limit()
if args.parallel:
stats = builder.build_cache_parallel(
full_rebuild=args.full_rebuild,
max_workers=args.max_workers,
)
else:
stats = builder.build_cache(
full_rebuild=args.full_rebuild,
)
logger.info(f"Cache build stats: {stats}")
if args.build_cache_only:
logger.info("Done! (build-cache-only mode)")
return 0
else:
if not release_cache_dir.exists():
logger.error(
f"Release cache directory not found: {release_cache_dir}. "
"Cannot use --skip-build-cache without existing cache."
)
return 1
logger.info("Step 1: Skipped (using existing cache)")
# Step 2: Generate addon_infos.json from cache
logger.info("Step 2: Generating addon_infos.json from cache...")
scraper = CacheScraper(config, release_cache)
# Check rate limit
if config.github.token:
scraper.github.get_rate_limit()
# Run scraper
logger.info("Starting addon scraper...")
addon_infos = scraper.scrape_all()
logger.info(f"Scraped {len(addon_infos)} addons")
logger.info(f"Generated info for {len(addon_infos)} addons")
# Publish if requested
if args.create_release:
logger.info("Publishing to GitHub release...")
# Step 3: Publish to GitHub release (if requested)
if args.create_release and args.github_repository:
logger.info("Step 3: Publishing to GitHub release...")
scraper.publisher.publish(config.output_file)
# Update cache
logger.info("Updating cache...")
cache = FileCache(
cache_dir=config.cache.cache_dir,
runtime_dir=config.cache.runtime_xpi_dir,
lockfile_name=config.cache.lockfile_name,
)
cache.update_cache()
# Clean up old caches
if config.github.token:
scraper.publisher.cleanup_caches(keep_count=1)
else:
logger.info("Step 3: Skipped (--create_release is False or no repository)")
logger.info("Done!")
return 0

8
src/zotero_scraper/cache/__init__.py vendored
View File

@ -1,5 +1,11 @@
"""Cache module."""
from .file_cache import FileCache
from .release_cache import CachedRelease, ReleaseCache, RepoCache
__all__ = ["FileCache"]
__all__ = [
"CachedRelease",
"FileCache",
"ReleaseCache",
"RepoCache",
]

346
src/zotero_scraper/cache/release_cache.py vendored Normal file
View File

@ -0,0 +1,346 @@
"""Release cache service for storing parsed release information."""
import json
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Optional
from ..utils.logging import get_logger
logger = get_logger("cache.release_cache")
@dataclass
class CachedRelease:
"""Cached release information parsed from XPI."""
tag: str
published_at: str
xpi_asset_id: int
xpi_name: str
xpi_download_url: str
# Parsed from XPI manifest
addon_id: Optional[str] = None
addon_name: Optional[str] = None
addon_version: Optional[str] = None
addon_description: Optional[str] = None
min_zotero_version: Optional[str] = None
max_zotero_version: Optional[str] = None
# Parse status
parse_success: bool = True
parse_error: Optional[str] = None
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary."""
return {k: v for k, v in asdict(self).items() if v is not None}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CachedRelease":
"""Create from dictionary."""
return cls(
tag=data.get("tag", ""),
published_at=data.get("published_at", ""),
xpi_asset_id=data.get("xpi_asset_id", 0),
xpi_name=data.get("xpi_name", ""),
xpi_download_url=data.get("xpi_download_url", ""),
addon_id=data.get("addon_id"),
addon_name=data.get("addon_name"),
addon_version=data.get("addon_version"),
addon_description=data.get("addon_description"),
min_zotero_version=data.get("min_zotero_version"),
max_zotero_version=data.get("max_zotero_version"),
parse_success=data.get("parse_success", True),
parse_error=data.get("parse_error"),
)
def is_compatible_with(self, zotero_version: str) -> bool:
"""Check if this release is compatible with a Zotero version.
Args:
zotero_version: Target version like "6" or "7".
Returns:
True if compatible.
"""
if not self.parse_success or not self.min_zotero_version:
return False
# Normalize versions
min_ver = self.min_zotero_version.replace("*", "0")
max_ver = (self.max_zotero_version or "999").replace("*", "999")
# Check version range
target_min = float(zotero_version)
target_max = target_min + 0.999
try:
min_float = float(min_ver.split(".")[0])
max_float = float(max_ver.split(".")[0]) if max_ver != "999" else 999
except ValueError:
return False
return min_float <= target_max and max_float >= target_min
@dataclass
class RepoCache:
"""Cache for a single repository."""
last_checked: str = ""
checked_releases: list[CachedRelease] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary."""
return {
"last_checked": self.last_checked,
"checked_releases": [r.to_dict() for r in self.checked_releases],
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "RepoCache":
"""Create from dictionary."""
releases = [
CachedRelease.from_dict(r)
for r in data.get("checked_releases", [])
]
return cls(
last_checked=data.get("last_checked", ""),
checked_releases=releases,
)
def get_checked_tags(self) -> set[str]:
"""Get set of already checked tags."""
return {r.tag for r in self.checked_releases}
def get_release_by_tag(self, tag: str) -> Optional[CachedRelease]:
"""Get cached release by tag."""
for release in self.checked_releases:
if release.tag == tag:
return release
return None
def add_release(self, release: CachedRelease) -> None:
"""Add or update a release in cache."""
# Remove existing release with same tag
self.checked_releases = [
r for r in self.checked_releases if r.tag != release.tag
]
self.checked_releases.append(release)
def remove_deleted_releases(self, current_tags: set[str]) -> list[str]:
"""Remove releases that no longer exist on GitHub.
Args:
current_tags: Set of tags that currently exist on GitHub.
Returns:
List of removed tags.
"""
removed = []
new_releases = []
for release in self.checked_releases:
if release.tag in current_tags:
new_releases.append(release)
else:
removed.append(release.tag)
self.checked_releases = new_releases
return removed
def get_best_release_for_zotero(
self, zotero_version: str
) -> Optional[CachedRelease]:
"""Get the best (latest compatible) release for a Zotero version.
Args:
zotero_version: Target version like "6" or "7".
Returns:
Best compatible release or None.
"""
compatible = [
r for r in self.checked_releases
if r.is_compatible_with(zotero_version)
]
if not compatible:
return None
# Sort by published_at descending to get latest
compatible.sort(key=lambda r: r.published_at, reverse=True)
return compatible[0]
class ReleaseCache:
"""Service for managing release cache.
Uses a directory structure where each repo has its own JSON file:
release_cache/
├── windingwind#zotero-pdf-translate.json
├── MuiseDestiny#zotero-gpt.json
└── ...
"""
def __init__(self, cache_dir: Path):
"""Initialize release cache.
Args:
cache_dir: Directory to store cache files.
"""
self.cache_dir = cache_dir
self._cache: dict[str, RepoCache] = {}
self._dirty: set[str] = set() # Track which repos need saving
def _repo_to_filename(self, repo: str) -> str:
"""Convert repo name to filename."""
# owner/name -> owner#name.json
return repo.replace("/", "#") + ".json"
def _filename_to_repo(self, filename: str) -> str:
"""Convert filename to repo name."""
# owner#name.json -> owner/name
return filename.replace(".json", "").replace("#", "/")
def _get_repo_file(self, repo: str) -> Path:
"""Get the cache file path for a repo."""
return self.cache_dir / self._repo_to_filename(repo)
def _load_repo(self, repo: str) -> RepoCache:
"""Load cache for a single repo."""
cache_file = self._get_repo_file(repo)
if not cache_file.exists():
return RepoCache()
try:
with open(cache_file, "r", encoding="utf-8") as f:
data = json.load(f)
return RepoCache.from_dict(data)
except Exception as e:
logger.warning(f"Failed to load cache for {repo}: {e}")
return RepoCache()
def _save_repo(self, repo: str) -> None:
"""Save cache for a single repo."""
if repo not in self._cache:
return
try:
self.cache_dir.mkdir(parents=True, exist_ok=True)
cache_file = self._get_repo_file(repo)
with open(cache_file, "w", encoding="utf-8") as f:
json.dump(self._cache[repo].to_dict(), f, ensure_ascii=False, indent=2)
self._dirty.discard(repo)
except Exception as e:
logger.error(f"Failed to save cache for {repo}: {e}")
def save(self) -> None:
"""Save all dirty (modified) repos."""
saved_count = 0
for repo in list(self._dirty):
self._save_repo(repo)
saved_count += 1
if saved_count > 0:
logger.info(f"Saved cache for {saved_count} repos")
def save_all(self) -> None:
"""Save all repos in memory (regardless of dirty state)."""
for repo in self._cache:
self._save_repo(repo)
logger.info(f"Saved cache for {len(self._cache)} repos")
def get_repo_cache(self, repo: str) -> RepoCache:
"""Get cache for a repository.
Args:
repo: Repository in "owner/name" format.
Returns:
RepoCache instance (loads from file or creates empty one).
"""
if repo not in self._cache:
self._cache[repo] = self._load_repo(repo)
return self._cache[repo]
def update_repo_checked_time(self, repo: str) -> None:
"""Update the last checked time for a repository.
Args:
repo: Repository in "owner/name" format.
"""
cache = self.get_repo_cache(repo)
cache.last_checked = datetime.now(UTC).isoformat()
self._dirty.add(repo)
def get_unchecked_tags(
self, repo: str, all_tags: list[str]
) -> list[str]:
"""Get tags that haven't been checked yet.
Args:
repo: Repository in "owner/name" format.
all_tags: List of all tags from GitHub.
Returns:
List of unchecked tags.
"""
cache = self.get_repo_cache(repo)
checked = cache.get_checked_tags()
return [tag for tag in all_tags if tag not in checked]
def add_release(self, repo: str, release: CachedRelease) -> None:
"""Add a release to cache.
Args:
repo: Repository in "owner/name" format.
release: CachedRelease instance.
"""
cache = self.get_repo_cache(repo)
cache.add_release(release)
self._dirty.add(repo)
def sync_with_github(
self, repo: str, current_tags: set[str]
) -> list[str]:
"""Sync cache with GitHub, removing deleted releases.
Args:
repo: Repository in "owner/name" format.
current_tags: Set of tags that currently exist on GitHub.
Returns:
List of removed tags.
"""
cache = self.get_repo_cache(repo)
removed = cache.remove_deleted_releases(current_tags)
if removed:
self._dirty.add(repo)
return removed
def get_repos(self) -> list[str]:
"""Get all cached repository names from directory."""
if not self.cache_dir.exists():
return []
repos = []
for cache_file in self.cache_dir.glob("*.json"):
repo = self._filename_to_repo(cache_file.name)
repos.append(repo)
return repos
def get_stats(self) -> dict[str, Any]:
"""Get cache statistics."""
repos = self.get_repos()
total_releases = 0
for repo in repos:
cache = self.get_repo_cache(repo)
total_releases += len(cache.checked_releases)
return {
"repos": len(repos),
"total_releases": total_releases,
}

2
src/zotero_scraper/config/settings.py
View File

@ -35,7 +35,6 @@ class CacheConfig:
cache_dir: Path = field(default_factory=lambda: Path("caches"))
runtime_xpi_dir: Path = field(default_factory=lambda: Path("xpis"))
lockfile_name: str = "caches_lockfile"
@dataclass(frozen=True)
@ -70,7 +69,6 @@ class ScraperConfig:
cache=CacheConfig(
cache_dir=Path(args.cache_directory),
runtime_xpi_dir=Path(args.runtime_xpi_directory),
lockfile_name=args.cache_lockfile,
),
previous_info_urls=args.previous_info_urls or [],
create_release=args.create_release,

4
src/zotero_scraper/services/__init__.py
View File

@ -1,11 +1,15 @@
"""Business services module."""
from .cache_builder import ReleaseCacheBuilder
from .cache_scraper import CacheScraper
from .fallback import FallbackService
from .publisher import PublisherService
from .scraper import AddonScraper
__all__ = [
"AddonScraper",
"CacheScraper",
"FallbackService",
"PublisherService",
"ReleaseCacheBuilder",
]

359
src/zotero_scraper/services/cache_builder.py Normal file
View File

@ -0,0 +1,359 @@
"""Service for building and updating release cache."""
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Optional
from ..cache.release_cache import CachedRelease, ReleaseCache
from ..clients.downloader import XPIDownloader
from ..clients.github import GitHubClient, ReleaseAsset
from ..config.constants import ContentTypes
from ..config.settings import ScraperConfig
from ..parsers.xpi_parser import XPIParser
from ..utils.logging import get_logger
logger = get_logger("services.cache_builder")
class ReleaseCacheBuilder:
"""Service for building release cache by parsing all releases."""
def __init__(
self,
config: ScraperConfig,
release_cache: ReleaseCache,
):
"""Initialize cache builder.
Args:
config: Scraper configuration.
release_cache: Release cache instance.
"""
self.config = config
self.cache = release_cache
self.github = GitHubClient(config.github)
self.downloader = XPIDownloader(config.cache)
self.xpi_parser = XPIParser()
def build_cache(
self,
repos: Optional[list[str]] = None,
full_rebuild: bool = False,
max_releases_per_repo: Optional[int] = None,
) -> dict[str, Any]:
"""Build or update release cache for all repositories.
Args:
repos: Optional list of repos to process. If None, loads from input dir.
full_rebuild: If True, ignore existing cache and rebuild everything.
max_releases_per_repo: Limit releases to process per repo (for testing).
Returns:
Statistics about the build process.
"""
if repos is None:
repos = self._load_repos_from_input()
logger.info(f"Building cache for {len(repos)} repositories")
stats = {
"repos_processed": 0,
"repos_failed": 0,
"new_releases_parsed": 0,
"releases_deleted": 0,
"errors": [],
}
for i, repo in enumerate(repos, 1):
logger.info(f"[{i}/{len(repos)}] Processing {repo}")
try:
result = self._process_repo(
repo,
full_rebuild=full_rebuild,
max_releases=max_releases_per_repo,
)
stats["repos_processed"] += 1
stats["new_releases_parsed"] += result.get("new_releases", 0)
stats["releases_deleted"] += result.get("deleted_releases", 0)
except Exception as e:
logger.error(f"Failed to process {repo}: {e}")
stats["repos_failed"] += 1
stats["errors"].append({"repo": repo, "error": str(e)})
# Save cache after each repo (immediate persistence)
self.cache.save()
# Final save
self.cache.save()
logger.info(
f"Cache build complete: {stats['repos_processed']} repos, "
f"{stats['new_releases_parsed']} new releases, "
f"{stats['releases_deleted']} deleted releases"
)
return stats
def build_cache_parallel(
self,
repos: Optional[list[str]] = None,
full_rebuild: bool = False,
max_releases_per_repo: Optional[int] = None,
max_workers: int = 4,
) -> dict[str, Any]:
"""Build cache in parallel (faster but uses more resources).
Args:
repos: Optional list of repos to process.
full_rebuild: If True, ignore existing cache.
max_releases_per_repo: Limit releases per repo.
max_workers: Number of parallel workers.
Returns:
Statistics about the build process.
"""
if repos is None:
repos = self._load_repos_from_input()
logger.info(f"Building cache for {len(repos)} repos with {max_workers} workers")
stats = {
"repos_processed": 0,
"repos_failed": 0,
"new_releases_parsed": 0,
"releases_deleted": 0,
"errors": [],
}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(
self._process_repo,
repo,
full_rebuild=full_rebuild,
max_releases=max_releases_per_repo,
): repo
for repo in repos
}
for future in as_completed(futures):
repo = futures[future]
try:
result = future.result()
stats["repos_processed"] += 1
stats["new_releases_parsed"] += result.get("new_releases", 0)
stats["releases_deleted"] += result.get("deleted_releases", 0)
logger.info(
f"[{stats['repos_processed']}/{len(repos)}] "
f"Completed {repo}: {result.get('new_releases', 0)} new releases"
)
except Exception as e:
logger.error(f"Failed to process {repo}: {e}")
stats["repos_failed"] += 1
stats["errors"].append({"repo": repo, "error": str(e)})
self.cache.save()
logger.info(
f"Cache build complete: {stats['repos_processed']} repos, "
f"{stats['new_releases_parsed']} new releases"
)
return stats
def _load_repos_from_input(self) -> list[str]:
"""Load repository list from input directory."""
repos = []
if not self.config.input_dir.exists():
logger.error(f"Input directory not found: {self.config.input_dir}")
return repos
for config_file in self.config.input_dir.glob("*.json"):
try:
with open(config_file, "r", encoding="utf-8") as f:
data = json.load(f)
if repo := data.get("repo"):
repos.append(repo)
except Exception as e:
logger.error(f"Failed to load {config_file}: {e}")
return repos
def _process_repo(
self,
repo: str,
full_rebuild: bool = False,
max_releases: Optional[int] = None,
) -> dict[str, int]:
"""Process a single repository.
Args:
repo: Repository in "owner/name" format.
full_rebuild: If True, reprocess all releases.
max_releases: Limit number of releases to process.
Returns:
Statistics dict with new_releases and deleted_releases counts.
"""
parts = repo.split("/")
if len(parts) != 2:
raise ValueError(f"Invalid repo format: {repo}")
owner, name = parts
# Get all releases from GitHub
releases = self.github.get_releases(owner, name)
if not releases:
logger.warning(f"No releases found for {repo}")
return {"new_releases": 0, "deleted_releases": 0}
# Get current tags from GitHub
current_tags = {r.get("tag_name") for r in releases if r.get("tag_name")}
# Sync cache - remove deleted releases
deleted_tags = self.cache.sync_with_github(repo, current_tags)
if deleted_tags:
logger.info(f"{repo}: Removed {len(deleted_tags)} deleted releases: {deleted_tags}")
# Get tags to process
if full_rebuild:
tags_to_process = list(current_tags)
else:
tags_to_process = self.cache.get_unchecked_tags(repo, list(current_tags))
if max_releases:
tags_to_process = tags_to_process[:max_releases]
if not tags_to_process:
logger.debug(f"{repo}: No new releases to process")
self.cache.update_repo_checked_time(repo)
return {"new_releases": 0, "deleted_releases": len(deleted_tags)}
logger.info(f"{repo}: Processing {len(tags_to_process)} new releases")
# Process each new release
new_count = 0
for release_data in releases:
tag = release_data.get("tag_name")
if tag not in tags_to_process:
continue
cached = self._parse_release(repo, release_data)
if cached:
self.cache.add_release(repo, cached)
new_count += 1
logger.debug(f"{repo}@{tag}: Parsed successfully")
else:
# Still add to cache with parse_success=False to avoid re-parsing
cached = CachedRelease(
tag=tag,
published_at=release_data.get("published_at", ""),
xpi_asset_id=0,
xpi_name="",
xpi_download_url="",
parse_success=False,
parse_error="No XPI found or parse failed",
)
self.cache.add_release(repo, cached)
logger.debug(f"{repo}@{tag}: Parse failed, cached as failed")
self.cache.update_repo_checked_time(repo)
return {"new_releases": new_count, "deleted_releases": len(deleted_tags)}
def _parse_release(
self, repo: str, release_data: dict[str, Any]
) -> Optional[CachedRelease]:
"""Parse a single release and return CachedRelease.
Args:
repo: Repository in "owner/name" format.
release_data: Release data from GitHub API.
Returns:
CachedRelease or None if no XPI found.
"""
tag = release_data.get("tag_name", "")
published_at = release_data.get("published_at", "")
assets = release_data.get("assets", [])
# Find XPI asset
xpi_asset = self._find_xpi_asset(assets)
if not xpi_asset:
return None
xpi_url = xpi_asset.get("browser_download_url", "")
xpi_name = xpi_asset.get("name", "")
xpi_id = xpi_asset.get("id", 0)
# Download XPI
owner, name = repo.split("/")
filename = f"{owner}#{name}+{tag}@{xpi_id}.xpi"
xpi_path = self.downloader.download(xpi_url, filename)
if not xpi_path:
return CachedRelease(
tag=tag,
published_at=published_at,
xpi_asset_id=xpi_id,
xpi_name=xpi_name,
xpi_download_url=xpi_url,
parse_success=False,
parse_error="Download failed",
)
# Parse XPI
details = self.xpi_parser.parse(xpi_path)
if not details or not details.id:
return CachedRelease(
tag=tag,
published_at=published_at,
xpi_asset_id=xpi_id,
xpi_name=xpi_name,
xpi_download_url=xpi_url,
parse_success=False,
parse_error="Parse failed or no addon ID",
)
return CachedRelease(
tag=tag,
published_at=published_at,
xpi_asset_id=xpi_id,
xpi_name=xpi_name,
xpi_download_url=xpi_url,
addon_id=details.id,
addon_name=details.name,
addon_version=details.version,
addon_description=details.description,
min_zotero_version=details.min_version,
max_zotero_version=details.max_version,
parse_success=True,
)
def _find_xpi_asset(self, assets: list[dict]) -> Optional[dict]:
"""Find XPI asset from release assets."""
# Sort by updated_at descending
sorted_assets = sorted(
assets, key=lambda a: a.get("updated_at", ""), reverse=True
)
# First try XPI content type
for asset in sorted_assets:
if asset.get("content_type") == ContentTypes.XPI:
return asset
# Fallback to ZIP content type
for asset in sorted_assets:
if asset.get("content_type") == ContentTypes.ZIP:
return asset
# Fallback to .xpi extension
for asset in sorted_assets:
if asset.get("name", "").endswith(".xpi"):
return asset
return None

213
src/zotero_scraper/services/cache_scraper.py Normal file
View File

@ -0,0 +1,213 @@
"""Scraper service that uses release cache to generate output."""
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Optional
from ..cache.release_cache import CachedRelease, ReleaseCache
from ..clients.github import GitHubClient
from ..config.constants import XPIProxy
from ..config.settings import ScraperConfig
from ..models.addon import AddonInfo, AddonRelease, Author, XpiDownloadUrls
from ..utils.logging import get_logger
from .fallback import FallbackService
from .publisher import PublisherService
logger = get_logger("services.cache_scraper")
# Target Zotero versions to generate releases for
TARGET_ZOTERO_VERSIONS = ["7", "6"]
class CacheScraper:
"""Scraper that uses release cache to generate addon information."""
def __init__(
self,
config: ScraperConfig,
release_cache: ReleaseCache,
):
"""Initialize cache-based scraper.
Args:
config: Scraper configuration.
release_cache: Release cache instance.
"""
self.config = config
self.cache = release_cache
self.github = GitHubClient(config.github)
self.fallback = FallbackService()
self.publisher = PublisherService(self.github, config.github)
def scrape_all(self) -> list[dict[str, Any]]:
"""Generate addon information from cache.
Returns:
List of addon info dictionaries.
"""
repos = self._load_repos_from_input()
logger.info(f"Generating addon info for {len(repos)} repositories")
addon_infos: list[AddonInfo] = []
with ThreadPoolExecutor(max_workers=self.config.max_workers) as executor:
futures = {
executor.submit(self._process_repo, repo): repo
for repo in repos
}
for future in as_completed(futures):
repo = futures[future]
try:
if result := future.result():
addon_infos.append(result)
except Exception as e:
logger.error(f"Failed to process {repo}: {e}")
# Convert to dict
result = [info.to_dict() for info in addon_infos]
# Apply fallback from previous versions
for url in self.config.previous_info_urls:
result = self.fallback.apply_fallback(result, url)
# Sort by stars (descending)
result.sort(key=lambda x: x.get("stars") or 0, reverse=True)
# Save output
self._save_output(result)
return result
def _load_repos_from_input(self) -> list[str]:
"""Load repository list from input directory."""
repos = []
if not self.config.input_dir.exists():
logger.error(f"Input directory not found: {self.config.input_dir}")
return repos
for config_file in self.config.input_dir.glob("*.json"):
try:
with open(config_file, "r", encoding="utf-8") as f:
data = json.load(f)
if repo := data.get("repo"):
repos.append(repo)
except Exception as e:
logger.error(f"Failed to load {config_file}: {e}")
return repos
def _process_repo(self, repo: str) -> Optional[AddonInfo]:
"""Process a single repository using cache.
Args:
repo: Repository in "owner/name" format.
Returns:
AddonInfo or None if no valid releases.
"""
parts = repo.split("/")
if len(parts) != 2:
logger.warning(f"Invalid repo format: {repo}")
return None
owner, name = parts
repo_cache = self.cache.get_repo_cache(repo)
# Find best release for each Zotero version
releases: list[AddonRelease] = []
for zotero_version in TARGET_ZOTERO_VERSIONS:
cached_release = repo_cache.get_best_release_for_zotero(zotero_version)
if cached_release:
release = self._cached_to_addon_release(cached_release, zotero_version)
if release:
releases.append(release)
if not releases:
logger.debug(f"{repo}: No compatible releases found")
return None
# Create AddonInfo
addon_info = AddonInfo(
repo=repo,
releases=releases,
name=name,
)
# Fetch author info
self._fetch_author_info(addon_info, owner)
# Fetch repository info
self._fetch_repo_info(addon_info, owner, name)
# Use addon info from first release
first_release = releases[0]
if first_release.name:
addon_info.name = first_release.name
if first_release.description and not addon_info.description:
addon_info.description = first_release.description
return addon_info
def _cached_to_addon_release(
self, cached: CachedRelease, zotero_version: str
) -> Optional[AddonRelease]:
"""Convert CachedRelease to AddonRelease.
Args:
cached: Cached release data.
zotero_version: Target Zotero version.
Returns:
AddonRelease or None.
"""
if not cached.parse_success:
return None
xpi_download_url = XpiDownloadUrls(
github=cached.xpi_download_url,
ghProxy=XPIProxy.ghproxy_url(cached.xpi_download_url),
kgithub=XPIProxy.kkgithub_url(cached.xpi_download_url),
)
return AddonRelease(
targetZoteroVersion=zotero_version,
tagName=cached.tag,
xpiDownloadUrl=xpi_download_url,
releaseDate=cached.published_at,
id=cached.addon_id,
xpiVersion=cached.addon_version,
name=cached.addon_name,
description=cached.addon_description,
minZoteroVersion=cached.min_zotero_version,
maxZoteroVersion=cached.max_zotero_version,
)
def _fetch_author_info(self, addon_info: AddonInfo, owner: str) -> None:
"""Fetch author info from GitHub."""
if user_info := self.github.get_user(owner):
addon_info.author = Author(
name=user_info.name,
url=user_info.html_url,
avatar=user_info.avatar_url,
)
def _fetch_repo_info(
self, addon_info: AddonInfo, owner: str, name: str
) -> None:
"""Fetch repository info from GitHub."""
if repo_info := self.github.get_repo(owner, name):
if not addon_info.description and repo_info.description:
addon_info.description = repo_info.description
addon_info.stars = repo_info.stargazers_count
def _save_output(self, data: list[dict[str, Any]]) -> None:
"""Save output to file."""
self.config.output_file.parent.mkdir(parents=True, exist_ok=True)
with open(self.config.output_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
logger.info(f"Saved {len(data)} addons to {self.config.output_file}")