diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 0000000..edb3850 --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,13 @@ +{ + "permissions": { + "allow": [ + "Bash(tree:*)", + "Bash(wc:*)", + "Bash(grep:*)", + "Bash(pip install:*)", + "Bash(python3 -m pip install:*)", + "Bash(python3 -m pytest:*)", + "Bash(python3:*)" + ] + } +} diff --git a/addon_info.py b/addon_info.py deleted file mode 100644 index 1262bae..0000000 --- a/addon_info.py +++ /dev/null @@ -1,118 +0,0 @@ -class AddonInfoAuthor: - def __init__(self, name: str = None, url: str = None, avatar: str = None, - **kwargs): - self.name = name - self.url = url - self.avatar = avatar - [setattr(self, key, value) for (key, value) in kwargs.items()] - - @property - def __dict__(self): - result = {} - if self.name: - result["name"] = self.name - if self.url: - result["url"] = self.url - if self.avatar: - result["avatar"] = self.avatar - return result - - -class AddonInfoRelease: - def __init__(self, targetZoteroVersion: str, tagName: str, - xpiDownloadUrl: dict = None, releaseDate: str = None, id: str = None, xpiVersion: str = None, - name: str = None, description: str = None, - minZoteroVersion: str = None, maxZoteroVersion: str = None, - **kwargs): - self.targetZoteroVersion = targetZoteroVersion - self.tagName = tagName - self.xpiDownloadUrl = xpiDownloadUrl - self.releaseDate = releaseDate - self.id = id - self.xpiVersion = xpiVersion - self.name = name - self.description = description - self.minZoteroVersion = minZoteroVersion - self.maxZoteroVersion = maxZoteroVersion - [setattr(self, key, value) for (key, value) in kwargs.items()] - - @property - def zotero_check_version(self) -> str: - if self.targetZoteroVersion == '6': - return '6.*' - elif self.targetZoteroVersion == '7': - return '7.*' - raise Exception(f'Invalid targetZoteroVersion({self.targetZoteroVersion})') - - @property - def __dict__(self): - result = {} - if self.targetZoteroVersion: - result["targetZoteroVersion"] = self.targetZoteroVersion - if self.tagName: - result["tagName"] = self.tagName - if self.xpiDownloadUrl: - result["xpiDownloadUrl"] = self.xpiDownloadUrl - if self.releaseDate: - result["releaseDate"] = self.releaseDate - if self.id: - result["id"] = self.id - if self.xpiVersion: - result["xpiVersion"] = self.xpiVersion - if self.name: - result["name"] = self.name - if self.description: - result["description"] = self.description - if self.minZoteroVersion: - result["minZoteroVersion"] = self.minZoteroVersion - if self.maxZoteroVersion: - result["maxZoteroVersion"] = self.maxZoteroVersion - return result - - -class AddonInfo: - def __init__(self, repo: str, - releases: list[dict] | list[AddonInfoRelease], - name: str = None, - description: str = None, - stars: int = None, - author: dict | AddonInfoAuthor = None, - **kwargs): - self.repo = repo - self.releases = [e if isinstance(e, AddonInfoRelease) else AddonInfoRelease(**(e if e else {})) for e in releases] - self.name = name - self.description = description - self.stars = stars - self.author = author if isinstance(author, AddonInfoAuthor) else AddonInfoAuthor(**(author if author else {})) - [setattr(self, key, value) for (key, value) in kwargs.items()] - - @property - def owner(self): - if len(self.repo.split('/')) != 2: - return None - return self.repo.split('/')[0] - - @property - def repository(self): - if len(self.repo.split('/')) != 2: - return None - return self.repo.split('/')[1] - - @property - def __dict__(self): - result = {} - if self.repo: - result["repo"] = self.repo - if self.releases: - result["releases"] = [release.__dict__ for release in self.releases] - if self.name: - result["name"] = self.name - if self.description: - result["description"] = self.description - if self.stars: - result["stars"] = self.stars - # todo: Deprecation, used for z7 < 1.5.3 and z6 < 0.6.7 - result["star"] = self.stars - if self.author: - result["author"] = self.author.__dict__ - return result diff --git a/convert_addon_from_zchinese.py b/convert_addon_from_zchinese.py deleted file mode 100644 index 94467b1..0000000 --- a/convert_addon_from_zchinese.py +++ /dev/null @@ -1,17 +0,0 @@ -import json -import os - -# convert addon from zotero-chinese - -repo = 'repo' -releases = 'releases' -targetZoteroVersion = 'targetZoteroVersion' -tagName = 'tagName' - -PluginInfoBase = [] - - -for addon in PluginInfoBase: - filename = addon[repo].replace('/', '#') + '.json' - with open(os.path.join('addons', filename), 'w') as f: - json.dump(addon, f, ensure_ascii=False, indent=' ') diff --git a/fallback_infos.py b/fallback_infos.py deleted file mode 100644 index 64aaf0d..0000000 --- a/fallback_infos.py +++ /dev/null @@ -1,40 +0,0 @@ -fallback_associated_unique_id = 'p_unique_associated_id##80502300-9DE6-4510-8768-EC42B0EF14E6' -fallback_associated_exclude_ids = 'p_exclude_associated_ids##073F5543-C70A-46AA-8529-E05168852D8F' - -fallback_associated_map = { - fallback_associated_unique_id: 'repo', - fallback_associated_exclude_ids: ['id'], - 'repo': { - 'releases': { - fallback_associated_unique_id: 'tagName', - fallback_associated_exclude_ids: ['currentVersion'], - } - } -} - - -def fallback_if_need(current, previous, associate_map=None): - if associate_map is None: - associate_map = fallback_associated_map - if previous and current and isinstance(previous, type(current)): - if isinstance(current, dict): - for key, value in previous.items(): - if key in associate_map.get(fallback_associated_exclude_ids, []): - continue - if key in current and (isinstance(value, (list, dict))): - if current[key]: - fallback_if_need(current[key], value, associate_map.get(key, {})) - else: - print(f'fallback {key}') - current[key] = value - elif key not in current: - print(f'fallback {key}') - current[key] = value - - elif isinstance(current, list) and (uid := associate_map.get(fallback_associated_unique_id)): - for value in previous: - if not isinstance(value, dict): - continue - if cur := next((x for x in current if isinstance(x, dict) and x[uid] == value[uid]), None): - fallback_if_need(cur, value, associate_map.get(uid, {})) - return current diff --git a/file_cache.py b/file_cache.py deleted file mode 100644 index 16d750b..0000000 --- a/file_cache.py +++ /dev/null @@ -1,27 +0,0 @@ -import os -import shutil -import hashlib - - -def folder_filename_hash(dir): - filenames = sorted(os.listdir(dir)) - folder_hash = hashlib.new('sha256') - for filename in filenames: - hash_obj = hashlib.new('sha256') - hash_obj.update(filename.encode('utf-8')) - folder_hash.update(hash_obj.hexdigest().encode('utf-8')) - return folder_hash.hexdigest() - - -def update_cache(cache_directory, runtime_xpi_directory, cache_hash_filename): - if not cache_directory or not runtime_xpi_directory or cache_directory == runtime_xpi_directory: - return - try: - shutil.rmtree(cache_directory) - shutil.move(runtime_xpi_directory, cache_directory) - folder_hash = folder_filename_hash(cache_directory) - with open(os.path.join(cache_directory, cache_hash_filename), 'w') as file: - file.write(folder_hash) - print(folder_hash) - except Exception as e: - print(f'update cache failed: {e}') diff --git a/github_operations.py b/github_operations.py deleted file mode 100644 index bfeab27..0000000 --- a/github_operations.py +++ /dev/null @@ -1,186 +0,0 @@ -import json -import time -import requests -from datetime import datetime, timedelta, UTC - - -def github_api_headers(**kwargs): - result = { - 'Accept': 'application/vnd.github+json', - 'X-GitHub-Api-Version': '2022-11-28', - } - if github_token := kwargs.get('github_token'): - result['Authorization'] = f'token {github_token}' - return result - - -def report_issue(github_repository: str, title: str, body: str, **kwargs): - if not github_repository: - print('report issue repository not found') - return - try: - if issue_check_id := kwargs.get('id'): - body = f'{body}\n----{issue_check_id}' - try: - response = requests.get(f'https://api.github.com/repos/{github_repository}/issues', - headers=github_api_headers(github_token=kwargs.get('github_token')), - params={ - 'state': 'open', - 'since': (datetime.now(UTC) - timedelta(days=10)).strftime('%Y-%m-%dT%H:%M:%S') + 'Z', - 'per_page': 100, - }) - exist_issues = response.json() - for issue in exist_issues: - if issue.get('body').endswith(f'----{issue_check_id}'): - return - except Exception as e: - print(f'fetch from exist issues failed: {e}') - - response = requests.post(f'https://api.github.com/repos/{github_repository}/issues', - headers=github_api_headers(github_token=kwargs.get('github_token')), - json={ - 'title': title, - 'body': body - }) - - if response.status_code == 201: - print('Issue created successfully.') - print('Issue URL:', response.json()['html_url']) - else: - print('Failed to create issue.') - print('Response:', response.content) - except Exception as e: - print(f'report issue failed: {e}') - - -def upload_json_to_release(github_repository, release_id, upload_file_name, upload_file, **kwargs): - headers = github_api_headers(github_token=kwargs.get('github_token')) - headers["Content-Type"] = "application/octet-stream" - upload_url = (f'https://uploads.github.com/repos/{github_repository}' - f'/releases/{release_id}/assets?name={upload_file_name}') - try: - with open(upload_file, "rb") as file: - upload_resp = requests.post(upload_url, data=file, headers=headers) - if upload_resp.status_code != 201: - print(f'upload release assets code: {upload_resp.status_code}') - else: - print('upload release assets succeed') - except Exception as e: - print(f'upload release assets failed: {e}') - - -def create_release(github_repository, **kwargs): - create_release_url = f'https://api.github.com/repos/{github_repository}/releases' - cur_time = int(time.time()) - param = { - 'tag_name': f'{cur_time}', - 'target_commitish': 'publish', - 'name': f'{cur_time}', - 'body': f'![](https://img.shields.io/github/downloads/{github_repository}/{cur_time}/total?label=downloads)\n' - f'publish addon_infos.json', - 'draft': False, - 'prerelease': False, - 'generate_release_notes': False, - } - try: - create_resp = requests.post(create_release_url, json=param, - headers=github_api_headers(github_token=kwargs.get('github_token'))) - if create_resp.status_code == 201: - create_release_info = json.loads(create_resp.content) - release_id = create_release_info['id'] - return release_id - else: - print(f'create release code: {create_resp.status_code}') - except Exception as e: - print(f'create release failed: {e}') - - -def delete_cache(github_repository, github_token, remain_count=2): - headers = github_api_headers(github_token=github_token) - get_caches_url = (f'https://api.github.com/repos/{github_repository}' - f'/actions/caches?per_page=100&sort=last_accessed_at&direction=desc') - try: - caches_resp = requests.get(get_caches_url, headers=headers) - caches = json.loads(caches_resp.content) - if caches.get('total_count', 0) < remain_count: - return - delete_cache_url = f'https://api.github.com/repos/{github_repository}/actions/caches/' - for cache in caches.get('actions_caches', [])[remain_count:]: - cache_key = cache.get('key') - if cache_id := cache.get('id'): - try: - delete_cache_resp = requests.delete(f'{delete_cache_url}{cache_id}', headers=headers) - if delete_cache_resp.status_code == 204: - print(f'delete {cache_key} succeed') - else: - print(f'delete {cache_key} failed: {delete_cache_resp.text}') - except Exception as e: - print(f'delete cache for {cache_key} failed: {e}') - except Exception as e: - print(f'get caches failed: {e}') - - -def delete_release(github_repository, github_token, remain_count=2): - headers = github_api_headers(github_token=github_token) - get_release_url = (f'https://api.github.com/repos/{github_repository}' - f'/releases?per_page=100&page=1') - try: - releases_resp = requests.get(get_release_url, headers=headers) - releases = json.loads(releases_resp.content) - if len(releases) < remain_count: - return - releases.sort(key=lambda release: release['tag_name'], reverse=True) - delete_release_url = f'https://api.github.com/repos/{github_repository}/releases/' - for release in releases[remain_count:]: - release_tag = release.get('tag_name') - if release_id := release.get('id'): - try: - delete_release_resp = requests.delete(f'{delete_release_url}{release_id}', headers=headers) - if delete_release_resp.status_code == 204: - print(f'delete release {release_tag} succeed') - else: - print(f'delete release {release_tag} failed: {delete_release_resp.text}') - except Exception as e: - print(f'delete release for {release_tag} failed: {e}') - except Exception as e: - print(f'get releases failed: {e}') - - -def delete_tag(github_repository, github_token, remain_count=2): - headers = github_api_headers(github_token=github_token) - get_tags_url = (f'https://api.github.com/repos/{github_repository}' - f'/git/refs/tags') - try: - tags_response = requests.get(get_tags_url, headers=headers) - tags = json.loads(tags_response.content) - if len(tags) < remain_count: - return - tags.sort(key=lambda tag: tag['ref'], reverse=True) - delete_tag_url = f'https://api.github.com/repos/{github_repository}/git/refs/tags/' - for tag in tags[remain_count:]: - if ref := tag.get('ref').replace('refs/tags/', ''): - if len(ref) < 10: - continue - try: # automatically tag name is timestamp - int(ref) - except ValueError: - continue - try: - delete_tag_resp = requests.delete(f'{delete_tag_url}{ref}', headers=headers) - if delete_tag_resp.status_code == 204: - print(f'delete tag {ref} succeed') - else: - print(f'delete tag {ref} failed: {delete_tag_resp.text}') - except Exception as e: - print(f'delete tag for {ref} failed: {e}') - except Exception as e: - print(f'get tags failed: {e}') - - -def rate_limit(github_token): - try: - resp = requests.get('https://api.github.com/rate_limit', headers=github_api_headers(github_token=github_token)) - rate = json.loads(resp.content) - print(f'token rate {rate.get("rate")}') - except Exception as e: - print(f'get rate limit failed: {e}') diff --git a/main.py b/main.py index bbc4d1c..2059498 100644 --- a/main.py +++ b/main.py @@ -1,320 +1,30 @@ -import concurrent.futures -import urllib.parse -import argparse -from addon_info import * -from fallback_infos import fallback_if_need -from moz_addons import addon_details, compare_versions -from github_operations import * -from file_cache import * +#!/usr/bin/env python3 +""" +Backward compatible entry point for Zotero addons scraper. +This file maintains backward compatibility with existing CI/CD pipelines +and command-line usage. It delegates to the new modular implementation. -def download_xpi(xpi_url: str, download_dir: str, unique_name: str, force_download: bool, **kwargs): - try: - download_filepath = os.path.join(download_dir, unique_name) - folder = os.path.dirname(download_filepath) - if folder and not os.path.exists(folder): - os.makedirs(folder) +Usage: + python main.py -i addons -o addon_infos.json \\ + --github_repository owner/repo \\ + --github_token TOKEN \\ + --cache_directory caches \\ + --runtime_xpi_directory xpis \\ + --previous_info_urls URL1 URL2 \\ + --create_release True +""" - if cache_dir := kwargs.get('cache_dir'): - cache_filepath = os.path.join(cache_dir, unique_name) - if not force_download and os.path.isfile(cache_filepath): - shutil.copy(cache_filepath, download_filepath) # copy to download filepath - return download_filepath +import sys - if not force_download and os.path.isfile(download_filepath): - return download_filepath +# Add src to path for development +from pathlib import Path - response = requests.get(xpi_url, stream=True) - if response.status_code == 200: - print(f'download {unique_name} from {xpi_url}') - with open(download_filepath, "wb") as file: - for chunk in response.iter_content(chunk_size=1024): - file.write(chunk) - return download_filepath - except Exception as e: - print(f'download {unique_name} from {xpi_url} failed: {e}') +src_path = Path(__file__).parent / "src" +if src_path.exists(): + sys.path.insert(0, str(src_path)) +from zotero_scraper.__main__ import main -# 输出格式参考 [zotero-chinese/zotero-plugins](https://github.com/zotero-chinese/zotero-plugins) -def parse(plugin: AddonInfo, **kwargs): - if not plugin.owner or not plugin.releases or len(plugin.releases) <= 0: - return - plugin.name = plugin.repository - headers = github_api_headers(github_token=kwargs.get('github_token')) - - # fetch author info - author_url = f"https://api.github.com/users/{plugin.owner}" - try: - author_resp = requests.get(author_url, headers=headers) - author_resp_info = json.loads(author_resp.content) - - plugin.author.name = author_resp_info['name'] if 'name' in author_resp_info and author_resp_info[ - 'name'] else plugin.owner - if 'html_url' in author_resp_info: - plugin.author.url = author_resp_info['html_url'] - if 'avatar_url' in author_resp_info: - plugin.author.avatar = author_resp_info['avatar_url'] - except Exception as e: - print(f'request {author_url} failed: {e}') - - # fetch repo info - repo_url = f'https://api.github.com/repos/{plugin.repo}' - try: - repos_resp = requests.get(repo_url, headers=headers) - repos_info = json.loads(repos_resp.content) - if 'description' in repos_info and repos_info['description'] and not plugin.description: - plugin.description = repos_info['description'] - if 'stargazers_count' in repos_info and repos_info['stargazers_count'] is not None: - plugin.stars = repos_info['stargazers_count'] - except Exception as e: - print(f'request {repo_url} failed: {e}') - - # fetch release info - invalid_releases = [] - for release in plugin.releases: - if not release.tagName: - continue - release_url = f'https://api.github.com/repos/{plugin.repo}/releases' - if release.tagName == 'latest': - release_url += '/latest' - elif release.tagName != 'pre': - release_url += f'/tags/{release.tagName}' - - try: - release_resp = requests.get(release_url, headers=headers) - release_info = json.loads(release_resp.content) - - if release.tagName == 'pre': - release_info = [info for info in release_info if info['prerelease']] - if release_info: - release_info = release_info[0] - else: - continue - if 'tag_name' in release_info: - release.tagName = release_info['tag_name'] - - if 'assets' not in release_info: - continue - all_release_assets = release_info['assets'] - all_release_assets.sort(key=lambda item: item['updated_at'] if 'updated_at' in item else '', reverse=True) - release_assets = [asset for asset in all_release_assets if asset['content_type'] == 'application/x-xpinstall'] - if not release_assets: - release_assets = [asset for asset in all_release_assets if asset['content_type'] == 'application/x-zip-compressed'] - if not release_assets: - continue - release_asset = release_assets[0] - if 'browser_download_url' not in release_asset: - continue - details = parse_xpi_detail( - plugin=plugin, release=release, release_asset=release_asset, - **kwargs, - ) - - if details: - if detail_id := details.id: - release.id = detail_id - if detail_name := details.name: - release.name = detail_name - if detail_version := details.version: - release.xpiVersion = detail_version - if detail_desc := details.description: - release.description = detail_desc - if zotero_min_version := details.min_version: - release.minZoteroVersion = zotero_min_version - if zotero_max_version := details.max_version: - release.maxZoteroVersion = zotero_max_version - - if not details.check_compatible_for_zotero_version(release.zotero_check_version): - invalid_releases.append(release) - - if ((not release.minZoteroVersion or release.minZoteroVersion == '*') and - (not release.maxZoteroVersion or release.maxZoteroVersion == '*')): - report_issue(kwargs.get('github_repository'), - title=f'Parse {plugin.repo} of zotero version failed', - body=f'xpi: https://github.com/{plugin.repo} @{release.tagName} on {release.targetZoteroVersion}\n', - github_token=kwargs.get('github_token'), - id=f'Parse min/max version failed: {plugin.repo}+{release.tagName}@{release.targetZoteroVersion}') - else: - report_issue(kwargs.get('github_repository'), - title=f'Parse {plugin.repo} addon details failed', - body=f'xpi: https://github.com/{plugin.repo} @{release.tagName} on {release.targetZoteroVersion}\n', - github_token=kwargs.get('github_token'), - id=f'Parse details failed: {plugin.repo}+{release.tagName}@{release.targetZoteroVersion}') - - - except Exception as e: - print(f'handle {plugin.repo} request {release_url} failed: {e}') - - for invalid_release in invalid_releases: - print(plugin.repo, 'invalid', invalid_release.zotero_check_version) - report_issue(kwargs.get('github_repository'), - title=f'Invalid {plugin.repo} xpi with zotero version {invalid_release.zotero_check_version}', - body=f'xpi: https://github.com/{plugin.repo} @{invalid_release.tagName}\n' - f'min zotero Version:{invalid_release.minZoteroVersion}\n' - f'max Zotero version:{invalid_release.maxZoteroVersion}' - f'expect Zotero version:{invalid_release.zotero_check_version}\n', - github_token=kwargs.get('github_token'), - id=f'Target zotero version not match: {plugin.repo}+{invalid_release.tagName}@{release.targetZoteroVersion}') - plugin.releases.remove(invalid_release) - return plugin - - -def parse_xpi_detail(plugin: AddonInfo, release: AddonInfoRelease, release_asset, **kwargs): - priority_sources = ['rdf', 'json'] - if release.targetZoteroVersion == '6': - priority_sources = ['json', 'rdf'] - - def xpi_detail(xpi_url, xpi_filename): - try: - xpi_filepath = download_xpi(xpi_url=xpi_url, - download_dir=kwargs.get('runtime_xpi_directory'), - unique_name=xpi_filename, - force_download=False, - cache_dir=kwargs.get('cache_directory')) - details = addon_details(xpi_filepath, priority_sources=priority_sources) - return details - except: - try: - xpi_filepath = download_xpi(xpi_url=xpi_url, - download_dir=kwargs.get('runtime_xpi_directory'), - unique_name=xpi_filename, - force_download=True) - details = addon_details(xpi_filepath, priority_sources=priority_sources) - return details - except Exception as e: - print(f'fetch addon detail of {plugin.repo} with {xpi_url} failed: {e}') - - github_xpi_url = release_asset['browser_download_url'] - release.releaseDate = release_asset['updated_at'] - release.xpiDownloadUrl = { - 'github': github_xpi_url, - 'ghProxy': 'https://gh-proxy.com/' + urllib.parse.quote(github_xpi_url), - 'kgithub': github_xpi_url.replace('github.com', 'kkgithub.com'), - } - details = xpi_detail(xpi_url=github_xpi_url, - xpi_filename=f'{plugin.owner}#{plugin.repository}+{release.tagName}@{release_asset["id"]}.xpi') - - if details and details.update_url and details.id and details.version: - try: - update_json_resp = requests.get(details.update_url, - headers=github_api_headers(github_token=kwargs.get('github_token'))) - update_json_info = json.loads(update_json_resp.content) - - updates = update_json_info.get('addons', {}).get(details.id, {}).get('updates', []) - xpi_urls = [e.get('update_link') for e in updates if compare_versions(e.get('version'), details.version) > 0] - - for xpi_url in xpi_urls: - update_details = xpi_detail(xpi_url=xpi_url, - xpi_filename=f'{plugin.owner}#{plugin.repository}+update{details.version}.xpi') - if update_details and update_details.id and update_details.check_compatible_for_zotero_version(release.zotero_check_version): - details = update_details - release.xpiDownloadUrl = { - 'github': xpi_url, - } - break - except: pass - return details - - -def fallback(addon_infos, previous_info_url, **kwargs): - try: - repos_resp = requests.get(previous_info_url, - headers=github_api_headers(github_token=kwargs.get('github_token'))) - repos_info = json.loads(repos_resp.content) - addon_infos = fallback_if_need(addon_infos, repos_info) - except Exception as e: - print(f'request previous addon_info.json to merge failed: {e}') - return addon_infos - - -def parse_addon_infos(input_dir, output_filepath, **kwargs): - plugins = [] - for addon_json_filename in os.listdir(input_dir): - if not addon_json_filename.endswith('.json'): - continue - addon_json_filepath = os.path.join(input_dir, addon_json_filename) - with open(addon_json_filepath, 'r') as file: - try: - plugins.append(AddonInfo(**json.load(file))) - except Exception as e: - print(f'parse initial addon info from {addon_json_filepath} failed: {e}') - - addon_infos = [] - with concurrent.futures.ThreadPoolExecutor() as executor: - futures = [executor.submit(parse, - plugin, - github_token=kwargs.get('github_token'), - cache_directory=kwargs.get('cache_directory'), - runtime_xpi_directory=kwargs.get('runtime_xpi_directory'), - github_repository=kwargs.get('github_repository')) - for plugin in plugins] - - for future in concurrent.futures.as_completed(futures): - if addon_info := future.result(): - addon_infos.append(addon_info) - - addon_infos = [info.__dict__ for info in addon_infos] - for previous_info_url in kwargs.get('previous_info_urls', []): - addon_infos = fallback(addon_infos, previous_info_url, github_token=kwargs.get('github_token')) - - addon_infos.sort(key=lambda item: item.get('stars') if item.get('stars') else 0, reverse=True) - - dir = os.path.dirname(output_filepath) - if dir and not os.path.exists(dir): - os.makedirs(dir) - with open(output_filepath, "w") as json_file: - json.dump(addon_infos, json_file, ensure_ascii=False) - - return addon_infos - - -if __name__ == '__main__': - parser = argparse.ArgumentParser(description='params') - parser.add_argument('--github_repository', nargs='?', type=str, required=True, help='github repository') - parser.add_argument('--github_token', nargs='?', type=str, help='github token') - parser.add_argument('-i', '--input', nargs='?', type=str, default="addons", help='input addon dir') - parser.add_argument('-o', '--output', nargs='?', type=str, default="addon_infos.json", help='output addon filepath') - - parser.add_argument('--cache_directory', nargs='?', default="caches", type=str, help='folder for caches') - parser.add_argument('--cache_lockfile', nargs='?', default="caches_lockfile", type=str, help='hashfile for caches') - parser.add_argument('--runtime_xpi_directory', nargs='?', default="xpis", type=str, help='folder for download xpi') - parser.add_argument('--previous_info_urls', nargs='+', default=[], help='previous published info json to fallback') - parser.add_argument('--create_release', nargs='?', default=True, type=bool, help='create release in github') - - args = parser.parse_args() - - if not args.github_repository: - raise 'Need specific github repository' - - if args.github_token: - rate_limit(args.github_token) - - try: - if not os.path.isdir(args.cache_directory): - os.makedirs(args.cache_directory, exist_ok=True) - if not os.path.isdir(args.runtime_xpi_directory): - os.makedirs(args.runtime_xpi_directory, exist_ok=True) - except Exception as e: - print(f'create cache_directory failed: {e}') - - parse_addon_infos(args.input, - args.output, - github_repository=args.github_repository, - github_token=args.github_token, - cache_directory=args.cache_directory, - runtime_xpi_directory=args.runtime_xpi_directory, - previous_info_urls=args.previous_info_urls) - - delete_release(args.github_repository, github_token=args.github_token) - delete_tag(args.github_repository, github_token=args.github_token) - if args.create_release and (release_id := create_release(args.github_repository, github_token=args.github_token)): - upload_json_to_release(args.github_repository, - release_id, - upload_file_name='addon_infos.json', - upload_file=args.output, - github_token=args.github_token) - - update_cache(args.cache_directory, args.runtime_xpi_directory, args.cache_lockfile) - - if args.github_token: - delete_cache(args.github_repository, args.github_token, remain_count=1) +if __name__ == "__main__": + sys.exit(main()) diff --git a/moz_addons.py b/moz_addons.py deleted file mode 100644 index ec49d63..0000000 --- a/moz_addons.py +++ /dev/null @@ -1,283 +0,0 @@ -# modify from `https://github.com/ -# mozilla/gecko-dev/blob/ac19e2c0d7c09a2deaedbe6afc4cdcf1a4561456/testing/mozbase/mozprofile/mozprofile/addons.py#L213` - -import zipfile -import commentjson as json -import os -from xml.dom import minidom -import re - - -def compare_versions(version1, version2): - parts1 = version1.replace('-', '.').split('.') - parts2 = version2.replace('-', '.').split('.') - - for i in range(max(len(parts1), len(parts2))): - v1 = parts1[i] if i < len(parts1) else '0' - v2 = parts2[i] if i < len(parts2) else '0' - try: - num1 = int(v1) - except ValueError: - num1 = v1 - try: - num2 = int(v2) - except ValueError: - num2 = v2 - - if isinstance(num1, int) and isinstance(num2, int): - if num1 < num2: - return -1 - if num1 > num2: - return 1 - else: - if v1 < v2: - return -1 - if v1 > v2: - return 1 - return 0 - - -class XpiDetail: - def __init__(self): - self.id = None - self.name = None - self.version = None - self.description = None - self.update_url = None - self.min_version = "*" - self.max_version = "*" - - def _append_info(self, details: dict): - if id := details.get('id'): - if self.id and self.id != id: - print(f'Xpi ID not match? {self.id} <==> {id}') - return - self.id = id - if name := details.get('name'): - self.name = name - if version := details.get('version'): - self.version = version - if description := details.get('description'): - self.description = description - if update_url := details.get('updateURL'): - self.update_url = update_url - if update_url := details.get('update_url'): - self.update_url = update_url - if (min_version := details.get('min_version')) and (max_version := details.get('max_version')): - if compare_versions(min_version.replace('*', '0'), self.min_version.replace('*', '999')) <= 0: - self.min_version = min_version - if compare_versions(max_version.replace('*', '999'), self.max_version.replace('*', '0')) >= 0: - self.max_version = max_version - - def check_compatible_for_zotero_version(self, version: str | int): - if isinstance(version, int): - version = str(version) + '.*' - if (min_version := self.min_version.replace('*', '0')) and ( - max_version := self.max_version.replace('*', '999')): - return (compare_versions(min_version, version.replace('*', '999')) <= 0 - <= compare_versions(max_version, version.replace('*', '0'))) - - -def get_namespace_id(doc, url): - attributes = doc.documentElement.attributes - for i in range(attributes.length): - if attributes.item(i).value == url and ":" in attributes.item(i).name: - return attributes.item(i).name.split(":")[1] + ":" - return "" - - -def get_text(element): - """Retrieve the text value of a given node""" - rc = [] - for node in element.childNodes: - if node.nodeType == node.TEXT_NODE: - rc.append(node.data) - return "".join(rc).strip() - - -def manifest_from_json(addon_path): - try: - if zipfile.is_zipfile(addon_path): - with zipfile.ZipFile(addon_path, "r") as compressed_file: - filenames = [f.filename for f in compressed_file.filelist] - if "manifest.json" in filenames: - manifest = compressed_file.read("manifest.json").decode() - manifest = json.loads(manifest) - return manifest - elif os.path.isdir(addon_path): - with open(os.path.join(addon_path, "manifest.json")) as f: - manifest = json.loads(f.read()) - return manifest - except Exception as e: - print(f'Invalid Addon Path {addon_path}: {e}') - - -def manifest_from_rdf(addon_path): - try: - if zipfile.is_zipfile(addon_path): - with zipfile.ZipFile(addon_path, "r") as compressed_file: - filenames = [f.filename for f in compressed_file.filelist] - if "install.rdf" in filenames: - manifest = compressed_file.read("install.rdf") - return manifest - elif os.path.isdir(addon_path): - with open(os.path.join(addon_path, "install.rdf")) as f: - manifest = json.loads(f.read()) - return manifest - except Exception as e: - print(f'Invalid Addon Path {addon_path}: {e}') - - -def detail_from_manifest_json(addon_path, manifest): - details = { - "name": manifest.get("name"), - "version": manifest.get("version"), - "description": manifest.get("description"), - "id": None, - "min_version": None, - "max_version": None, - "update_url": None, - } - for location in ("applications", "browser_specific_settings"): - if details["id"]: - break - for app in ("zotero", "gecko"): - try: - details["id"] = manifest[location][app].get("id") - details["min_version"] = manifest[location][app].get("strict_min_version") - details["max_version"] = manifest[location][app].get("strict_max_version") - details["update_url"] = manifest[location][app].get("update_url") - break - except KeyError: - continue - - # handler for __MSG_{}__ items - def extract_msg_placeholder(text): - if match := re.search(r'__MSG_(.*?)__', text): - return match.group(1) - - def load_locale_for_msg(): - default_locale = manifest.get('default_locale') - locale_filename = f"_locales/{default_locale}/messages.json" - try: - if zipfile.is_zipfile(addon_path): - with zipfile.ZipFile(addon_path, "r") as compressed_file: - filenames = [f.filename for f in compressed_file.filelist] - if locale_filename in filenames: - locale = compressed_file.read(locale_filename).decode() - return json.loads(locale) - elif os.path.isdir(addon_path): - with open(os.path.join(addon_path, locale_filename)) as f: - return json.loads(f.read()) - except Exception as e: - raise e - - locale_for_msg = None - for key in details: - if isinstance(details[key], str) and (placeholder := extract_msg_placeholder(details[key])): - if not locale_for_msg: - locale_for_msg = load_locale_for_msg() - if not locale_for_msg: - break - if value := locale_for_msg.get(placeholder, {}).get('message'): - details[key] = value - - return details - - -def detail_from_manifest_rdf(manifest): - details = { - "id": None, - "name": None, - "version": None, - "description": None, - "min_version": None, - "max_version": None, - "updateURL": None, - } - try: - doc = minidom.parseString(manifest) - - # Get the namespaces abbreviations - em = get_namespace_id(doc, "http://www.mozilla.org/2004/em-rdf#") - rdf = get_namespace_id(doc, "http://www.w3.org/1999/02/22-rdf-syntax-ns#") - - description = doc.getElementsByTagName(rdf + "Description").item(0) - try: - descriptions = [e for e in doc.getElementsByTagName(rdf + "Description") - if len(e.getElementsByTagName(em + "targetApplication")) > 0] - if descriptions: - description = descriptions[0] - except Exception as e: - pass - - def extract_info(node, result): - try: - for entry, value in node.attributes.items(): - entry = entry.replace(em, "") - if entry in result.keys(): - result.update({entry: value}) - for child_node in node.childNodes: - entry = child_node.nodeName.replace(em, "") - if entry in result.keys(): - result.update({entry: get_text(child_node)}) - except: - return - - extract_info(description, details) - if not details.get('id') or (details.get('id').startswith("__") and details.get('id').endswith("__")): - return - - def update_details(version_info): - if version_info['id'] != 'zotero@chnm.gmu.edu': - return - if (min_version := version_info['minVersion']) and (max_version := version_info['maxVersion']): - if exist_min_version := details['min_version']: - if compare_versions(min_version.replace('*', '0'), exist_min_version.replace('*', '999')) <= 0: - details['min_version'] = min_version - else: - details['min_version'] = min_version - - if compare_versions(max_version.replace('*', '999'), '6.*') > 0: - # rdf not support z7 - max_version = '6.*' - if exist_max_version := details['max_version']: - if compare_versions(max_version.replace('*', '999'), exist_max_version.replace('*', '0')) >= 0: - details['max_version'] = max_version - else: - details['max_version'] = max_version - - for targetApplication in description.getElementsByTagName(em + "targetApplication"): - version_info = {'id': None, 'minVersion': None, 'maxVersion': None} - extract_info(targetApplication, version_info) - update_details(version_info) - - for node in targetApplication.childNodes: - version_info = {'id': None, 'minVersion': None, 'maxVersion': None} - extract_info(node, version_info) - update_details(version_info) - - if update_url := details.get('updateURL'): - details['update_url'] = update_url - - return details - except Exception as e: - raise e - - -def addon_details(addon_path, priority_sources=None) -> XpiDetail: - if priority_sources is None: - priority_sources = ['json', 'rdf'] - if not os.path.exists(addon_path): - raise IOError(f"Add-on path does not exist: {addon_path}") - xpi_detail = XpiDetail() - for source in priority_sources: - if source == 'json': - if ((manifest := manifest_from_json(addon_path)) and - (details := detail_from_manifest_json(addon_path, manifest))): - xpi_detail._append_info(details) - if source == 'rdf': - if ((manifest := manifest_from_rdf(addon_path)) and - (details := detail_from_manifest_rdf(manifest))): - xpi_detail._append_info(details) - return xpi_detail diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..21c905a --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,48 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "zotero-addons-scraper" +version = "2.0.0" +description = "Scrape Zotero addon information from GitHub" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.10" +dependencies = [ + "requests>=2.28.0", + "commentjson>=0.9.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "pytest-mock>=3.10.0", + "requests-mock>=1.11.0", + "mypy>=1.0.0", + "ruff>=0.1.0", +] + +[project.scripts] +zotero-scraper = "zotero_scraper.__main__:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +addopts = "-v --cov=src/zotero_scraper --cov-report=term-missing" + +[tool.mypy] +python_version = "3.10" +strict = true +warn_return_any = true +warn_unused_configs = true +ignore_missing_imports = true + +[tool.ruff] +line-length = 88 +target-version = "py310" +select = ["E", "F", "I", "N", "W", "UP"] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..c667030 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,13 @@ +-r requirements.txt + +# Testing +pytest>=7.0.0 +pytest-cov>=4.0.0 +pytest-mock>=3.10.0 +requests-mock>=1.11.0 + +# Type checking +mypy>=1.0.0 + +# Linting +ruff>=0.1.0 diff --git a/requirements.txt b/requirements.txt index b267f2d..87f5b98 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,2 @@ -requests -argparse -commentjson \ No newline at end of file +requests>=2.28.0 +commentjson>=0.9.0 diff --git a/src/zotero_scraper/__init__.py b/src/zotero_scraper/__init__.py new file mode 100644 index 0000000..e7af357 --- /dev/null +++ b/src/zotero_scraper/__init__.py @@ -0,0 +1,3 @@ +"""Zotero Addons Scraper - Scrape Zotero addon information from GitHub.""" + +__version__ = "2.0.0" diff --git a/src/zotero_scraper/__main__.py b/src/zotero_scraper/__main__.py new file mode 100644 index 0000000..2f3dc2c --- /dev/null +++ b/src/zotero_scraper/__main__.py @@ -0,0 +1,148 @@ +"""Command-line interface for Zotero addons scraper.""" + +import argparse +import sys +from pathlib import Path + +from .cache.file_cache import FileCache +from .config.settings import ScraperConfig +from .services.scraper import AddonScraper +from .utils.logging import setup_logging + + +def parse_args() -> argparse.Namespace: + """Parse command line arguments.""" + parser = argparse.ArgumentParser( + description="Scrape Zotero addon information from GitHub" + ) + + # Required arguments + parser.add_argument( + "--github_repository", + type=str, + required=True, + help="GitHub repository (owner/repo)", + ) + + # Optional arguments + parser.add_argument( + "--github_token", + type=str, + default=None, + help="GitHub API token", + ) + parser.add_argument( + "-i", + "--input", + type=str, + default="addons", + help="Input addon config directory (default: addons)", + ) + parser.add_argument( + "-o", + "--output", + type=str, + default="addon_infos.json", + help="Output JSON file path (default: addon_infos.json)", + ) + 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)", + ) + parser.add_argument( + "--runtime_xpi_directory", + type=str, + default="xpis", + help="Runtime XPI directory (default: xpis)", + ) + parser.add_argument( + "--previous_info_urls", + nargs="+", + default=[], + help="URLs to previous addon_infos.json for fallback", + ) + parser.add_argument( + "--create_release", + type=lambda x: x.lower() in ("true", "1", "yes"), + default=True, + help="Create GitHub release (default: True)", + ) + parser.add_argument( + "--log_level", + type=str, + default="INFO", + choices=["DEBUG", "INFO", "WARNING", "ERROR"], + help="Logging level (default: INFO)", + ) + + return parser.parse_args() + + +def main() -> int: + """Main entry point.""" + args = parse_args() + + # Setup logging + import logging + + log_level = getattr(logging, args.log_level.upper()) + setup_logging(level=log_level) + + logger = logging.getLogger("zotero_scraper") + + # Validate arguments + if not args.github_repository: + logger.error("--github_repository is required") + return 1 + + # Create configuration + config = ScraperConfig.from_args(args) + + # Ensure directories exist + 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) + + # 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") + + # Publish if requested + if args.create_release: + logger.info("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) + + logger.info("Done!") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/zotero_scraper/cache/__init__.py b/src/zotero_scraper/cache/__init__.py new file mode 100644 index 0000000..5222cc5 --- /dev/null +++ b/src/zotero_scraper/cache/__init__.py @@ -0,0 +1,5 @@ +"""Cache module.""" + +from .file_cache import FileCache + +__all__ = ["FileCache"] diff --git a/src/zotero_scraper/cache/file_cache.py b/src/zotero_scraper/cache/file_cache.py new file mode 100644 index 0000000..0931cc7 --- /dev/null +++ b/src/zotero_scraper/cache/file_cache.py @@ -0,0 +1,129 @@ +"""File cache management.""" + +import hashlib +import shutil +from pathlib import Path +from typing import Optional + +from ..utils.logging import get_logger + +logger = get_logger("cache") + + +class FileCache: + """File cache manager for XPI files.""" + + def __init__( + self, + cache_dir: Path, + runtime_dir: Path, + lockfile_name: str = "caches_lockfile", + ): + """Initialize file cache. + + Args: + cache_dir: Directory for cached files. + runtime_dir: Directory for runtime files. + lockfile_name: Name of the hash lockfile. + """ + self.cache_dir = cache_dir + self.runtime_dir = runtime_dir + self.lockfile_name = lockfile_name + + def folder_hash(self, directory: Path) -> str: + """Calculate SHA256 hash of filenames in a directory. + + Args: + directory: Directory to hash. + + Returns: + Hex digest of the hash. + """ + if not directory.exists(): + return "" + + filenames = sorted(f.name for f in directory.iterdir() if f.is_file()) + folder_hash = hashlib.sha256() + + for filename in filenames: + file_hash = hashlib.sha256(filename.encode("utf-8")).hexdigest() + folder_hash.update(file_hash.encode("utf-8")) + + return folder_hash.hexdigest() + + def update_cache(self) -> Optional[str]: + """Update cache from runtime directory. + + Moves runtime directory to cache directory and generates hash. + + Returns: + Hash of the cached files, or None if failed. + """ + if not self.cache_dir or not self.runtime_dir: + return None + + if self.cache_dir == self.runtime_dir: + logger.warning("Cache and runtime directories are the same") + return None + + try: + # Remove old cache + if self.cache_dir.exists(): + shutil.rmtree(self.cache_dir) + + # Move runtime to cache + if self.runtime_dir.exists(): + shutil.move(str(self.runtime_dir), str(self.cache_dir)) + + # Calculate and save hash + folder_hash = self.folder_hash(self.cache_dir) + lockfile_path = self.cache_dir / self.lockfile_name + + with open(lockfile_path, "w", encoding="utf-8") as f: + f.write(folder_hash) + + logger.info(f"Cache updated: {folder_hash}") + return folder_hash + + except Exception as e: + logger.error(f"Update cache failed: {e}") + return None + + def restore_from_cache(self) -> bool: + """Restore runtime directory from cache. + + Returns: + True if successful, False otherwise. + """ + if not self.cache_dir.exists(): + return False + + try: + if self.runtime_dir.exists(): + shutil.rmtree(self.runtime_dir) + + shutil.copytree(self.cache_dir, self.runtime_dir) + logger.info("Restored from cache") + return True + + except Exception as e: + logger.error(f"Restore from cache failed: {e}") + return False + + def clear_runtime(self) -> None: + """Clear runtime directory.""" + if self.runtime_dir.exists(): + try: + shutil.rmtree(self.runtime_dir) + logger.info("Runtime directory cleared") + except Exception as e: + logger.error(f"Clear runtime failed: {e}") + + def clear_cache(self) -> None: + """Clear cache directory.""" + if self.cache_dir.exists(): + try: + shutil.rmtree(self.cache_dir) + logger.info("Cache directory cleared") + except Exception as e: + logger.error(f"Clear cache failed: {e}") diff --git a/src/zotero_scraper/clients/__init__.py b/src/zotero_scraper/clients/__init__.py new file mode 100644 index 0000000..ddde0a3 --- /dev/null +++ b/src/zotero_scraper/clients/__init__.py @@ -0,0 +1,11 @@ +"""HTTP clients module.""" + +from .base import BaseHTTPClient +from .downloader import XPIDownloader +from .github import GitHubClient + +__all__ = [ + "BaseHTTPClient", + "GitHubClient", + "XPIDownloader", +] diff --git a/src/zotero_scraper/clients/base.py b/src/zotero_scraper/clients/base.py new file mode 100644 index 0000000..cdfaded --- /dev/null +++ b/src/zotero_scraper/clients/base.py @@ -0,0 +1,142 @@ +"""Base HTTP client with retry and timeout support.""" + +from abc import ABC +from typing import Any, Optional + +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry + +from ..exceptions import HTTPError, RateLimitError +from ..utils.logging import get_logger + +logger = get_logger("clients.base") + + +class BaseHTTPClient(ABC): + """Base HTTP client with retry and timeout support.""" + + def __init__( + self, + timeout: int = 30, + max_retries: int = 3, + backoff_factor: float = 0.5, + ): + """Initialize HTTP client. + + Args: + timeout: Request timeout in seconds. + max_retries: Maximum number of retries for failed requests. + backoff_factor: Backoff factor for retry delays. + """ + self.timeout = timeout + self.max_retries = max_retries + self.backoff_factor = backoff_factor + self.session = self._create_session() + + def _create_session(self) -> requests.Session: + """Create a session with retry strategy.""" + session = requests.Session() + + retry_strategy = Retry( + total=self.max_retries, + backoff_factor=self.backoff_factor, + status_forcelist=[429, 500, 502, 503, 504], + allowed_methods=["HEAD", "GET", "POST", "DELETE"], + ) + + adapter = HTTPAdapter(max_retries=retry_strategy) + session.mount("http://", adapter) + session.mount("https://", adapter) + + return session + + def _request( + self, + method: str, + url: str, + headers: Optional[dict[str, str]] = None, + **kwargs: Any, + ) -> requests.Response: + """Execute HTTP request with error handling. + + Args: + method: HTTP method (GET, POST, DELETE, etc.). + url: Request URL. + headers: Optional request headers. + **kwargs: Additional arguments passed to requests. + + Returns: + Response object. + + Raises: + HTTPError: If request fails. + RateLimitError: If rate limit is exceeded. + """ + try: + response = self.session.request( + method=method, + url=url, + headers=headers, + timeout=self.timeout, + **kwargs, + ) + + # Check for rate limiting + if response.status_code == 403: + if "rate limit" in response.text.lower(): + logger.error(f"Rate limit exceeded for {url}") + raise RateLimitError(f"Rate limit exceeded for {url}") + + response.raise_for_status() + return response + + except requests.exceptions.Timeout: + logger.error(f"Request timeout: {url}") + raise HTTPError(f"Request timeout: {url}") from None + except requests.exceptions.ConnectionError as e: + logger.error(f"Connection error: {url} - {e}") + raise HTTPError(f"Connection error: {url}") from e + except requests.exceptions.HTTPError as e: + logger.error(f"HTTP error: {url} - {e}") + raise HTTPError(f"HTTP error: {url} - {e}") from e + except requests.exceptions.RequestException as e: + logger.error(f"Request failed: {url} - {e}") + raise HTTPError(f"Request failed: {url}") from e + + def get( + self, + url: str, + headers: Optional[dict[str, str]] = None, + **kwargs: Any, + ) -> requests.Response: + """Execute GET request.""" + return self._request("GET", url, headers=headers, **kwargs) + + def post( + self, + url: str, + headers: Optional[dict[str, str]] = None, + **kwargs: Any, + ) -> requests.Response: + """Execute POST request.""" + return self._request("POST", url, headers=headers, **kwargs) + + def delete( + self, + url: str, + headers: Optional[dict[str, str]] = None, + **kwargs: Any, + ) -> requests.Response: + """Execute DELETE request.""" + return self._request("DELETE", url, headers=headers, **kwargs) + + def close(self) -> None: + """Close the session.""" + self.session.close() + + def __enter__(self) -> "BaseHTTPClient": + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.close() diff --git a/src/zotero_scraper/clients/downloader.py b/src/zotero_scraper/clients/downloader.py new file mode 100644 index 0000000..1e83b20 --- /dev/null +++ b/src/zotero_scraper/clients/downloader.py @@ -0,0 +1,149 @@ +"""XPI file downloader.""" + +import os +import shutil +from pathlib import Path +from typing import Optional + +from ..config.settings import CacheConfig +from ..exceptions import DownloadError +from ..utils.logging import get_logger +from .base import BaseHTTPClient + +logger = get_logger("clients.downloader") + + +class XPIDownloader(BaseHTTPClient): + """XPI file downloader with caching support.""" + + def __init__( + self, + cache_config: CacheConfig, + timeout: int = 60, + max_retries: int = 3, + ): + """Initialize XPI downloader. + + Args: + cache_config: Cache configuration. + timeout: Download timeout in seconds. + max_retries: Maximum download retries. + """ + super().__init__( + timeout=timeout, + max_retries=max_retries, + ) + self.cache_dir = cache_config.cache_dir + self.runtime_dir = cache_config.runtime_xpi_dir + + # Ensure directories exist + self.cache_dir.mkdir(parents=True, exist_ok=True) + self.runtime_dir.mkdir(parents=True, exist_ok=True) + + def download( + self, + url: str, + filename: str, + force_download: bool = False, + ) -> Optional[Path]: + """Download XPI file with caching. + + First checks cache, then downloads if not found or force_download=True. + + Args: + url: Download URL. + filename: Target filename. + force_download: Force download even if cached. + + Returns: + Path to downloaded file or None if failed. + """ + runtime_path = self.runtime_dir / filename + cache_path = self.cache_dir / filename + + # Check if file exists in runtime directory + if not force_download and runtime_path.exists(): + logger.debug(f"Using existing file: {filename}") + return runtime_path + + # Check if file exists in cache + if not force_download and cache_path.exists(): + try: + shutil.copy2(cache_path, runtime_path) + logger.debug(f"Copied from cache: {filename}") + return runtime_path + except Exception as e: + logger.warning(f"Failed to copy from cache: {e}") + + # Download file + try: + return self._download_file(url, runtime_path, cache_path) + except Exception as e: + logger.error(f"Download failed for {filename}: {e}") + + # Try once more with force + if not force_download: + logger.info(f"Retrying download for {filename}") + try: + return self._download_file(url, runtime_path, cache_path) + except Exception as e2: + logger.error(f"Retry failed for {filename}: {e2}") + + return None + + def _download_file( + self, + url: str, + runtime_path: Path, + cache_path: Path, + ) -> Path: + """Download file from URL. + + Args: + url: Download URL. + runtime_path: Target path in runtime directory. + cache_path: Target path in cache directory. + + Returns: + Path to downloaded file. + + Raises: + DownloadError: If download fails. + """ + try: + response = self.get(url, stream=True) + response.raise_for_status() + + # Write to runtime directory + with open(runtime_path, "wb") as f: + for chunk in response.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + + # Copy to cache + try: + shutil.copy2(runtime_path, cache_path) + except Exception as e: + logger.warning(f"Failed to cache file: {e}") + + logger.info(f"Downloaded: {runtime_path.name}") + return runtime_path + + except Exception as e: + # Clean up partial file + if runtime_path.exists(): + try: + runtime_path.unlink() + except Exception: + pass + raise DownloadError(f"Download failed: {url}") from e + + def clear_runtime(self) -> None: + """Clear runtime directory.""" + if self.runtime_dir.exists(): + for file in self.runtime_dir.iterdir(): + if file.is_file(): + try: + file.unlink() + except Exception as e: + logger.warning(f"Failed to delete {file}: {e}") diff --git a/src/zotero_scraper/clients/github.py b/src/zotero_scraper/clients/github.py new file mode 100644 index 0000000..59f874f --- /dev/null +++ b/src/zotero_scraper/clients/github.py @@ -0,0 +1,476 @@ +"""GitHub API client.""" + +import time +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any, Optional + +from ..config.constants import GitHubAPI +from ..config.settings import GitHubConfig +from ..exceptions import GitHubError +from ..utils.logging import get_logger +from .base import BaseHTTPClient + +logger = get_logger("clients.github") + + +@dataclass +class UserInfo: + """GitHub user information.""" + + name: str + html_url: str + avatar_url: Optional[str] = None + + +@dataclass +class RepoInfo: + """GitHub repository information.""" + + description: Optional[str] + stargazers_count: int + + +@dataclass +class ReleaseAsset: + """GitHub release asset information.""" + + id: int + name: str + browser_download_url: str + content_type: str + updated_at: str + + +@dataclass +class ReleaseInfo: + """GitHub release information.""" + + tag_name: str + prerelease: bool + published_at: str + assets: list[ReleaseAsset] + + +class GitHubClient(BaseHTTPClient): + """GitHub API client with typed responses.""" + + def __init__(self, config: GitHubConfig): + """Initialize GitHub client. + + Args: + config: GitHub configuration. + """ + super().__init__( + timeout=config.timeout, + max_retries=config.max_retries, + backoff_factor=config.retry_delay, + ) + self.config = config + self._headers = self._build_headers() + + def _build_headers(self) -> dict[str, str]: + """Build request headers.""" + headers = { + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": self.config.api_version, + } + if self.config.token: + headers["Authorization"] = f"token {self.config.token}" + return headers + + def get_user(self, username: str) -> Optional[UserInfo]: + """Get user information. + + Args: + username: GitHub username. + + Returns: + UserInfo or None if failed. + """ + url = GitHubAPI.user(username) + try: + response = self.get(url, headers=self._headers) + data = response.json() + return UserInfo( + name=data.get("name") or username, + html_url=data.get("html_url", ""), + avatar_url=data.get("avatar_url"), + ) + except Exception as e: + logger.warning(f"Failed to get user {username}: {e}") + return None + + def get_repo(self, owner: str, repo: str) -> Optional[RepoInfo]: + """Get repository information. + + Args: + owner: Repository owner. + repo: Repository name. + + Returns: + RepoInfo or None if failed. + """ + url = GitHubAPI.repo(owner, repo) + try: + response = self.get(url, headers=self._headers) + data = response.json() + return RepoInfo( + description=data.get("description"), + stargazers_count=data.get("stargazers_count", 0), + ) + except Exception as e: + 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. + + Args: + owner: Repository owner. + repo: Repository name. + + Returns: + List of release data dictionaries. + """ + url = GitHubAPI.releases(owner, repo) + try: + response = self.get(url, headers=self._headers) + return response.json() + except Exception as e: + logger.warning(f"Failed to get releases for {owner}/{repo}: {e}") + return [] + + def get_release( + self, owner: str, repo: str, tag: str + ) -> Optional[ReleaseInfo]: + """Get release by tag. + + Args: + owner: Repository owner. + repo: Repository name. + tag: Tag name ("latest", "pre", or specific tag). + + Returns: + ReleaseInfo or None if not found. + """ + try: + if tag == "latest": + url = GitHubAPI.release_latest(owner, repo) + response = self.get(url, headers=self._headers) + data = response.json() + elif tag == "pre": + # Find first prerelease + releases = self.get_releases(owner, repo) + prereleases = [r for r in releases if r.get("prerelease")] + if not prereleases: + return None + data = prereleases[0] + else: + url = GitHubAPI.release_by_tag(owner, repo, tag) + response = self.get(url, headers=self._headers) + data = response.json() + + assets = [ + ReleaseAsset( + id=a["id"], + name=a["name"], + browser_download_url=a["browser_download_url"], + content_type=a.get("content_type", ""), + updated_at=a.get("updated_at", ""), + ) + for a in data.get("assets", []) + ] + + return ReleaseInfo( + tag_name=data.get("tag_name", tag), + prerelease=data.get("prerelease", False), + published_at=data.get("published_at", ""), + assets=assets, + ) + except Exception as e: + logger.warning(f"Failed to get release {owner}/{repo}@{tag}: {e}") + return None + + def create_release( + self, + owner: str, + repo: str, + tag_name: Optional[str] = None, + target_commitish: str = "publish", + ) -> Optional[int]: + """Create a new release. + + Args: + owner: Repository owner. + repo: Repository name. + tag_name: Tag name (defaults to current timestamp). + target_commitish: Target branch/commit. + + Returns: + Release ID or None if failed. + """ + url = GitHubAPI.releases(owner, repo) + cur_time = int(time.time()) + tag = tag_name or str(cur_time) + + payload = { + "tag_name": tag, + "target_commitish": target_commitish, + "name": tag, + "body": ( + f"![](https://img.shields.io/github/downloads/{owner}/{repo}" + f"/{tag}/total?label=downloads)\npublish addon_infos.json" + ), + "draft": False, + "prerelease": False, + "generate_release_notes": False, + } + + try: + response = self.post(url, headers=self._headers, json=payload) + if response.status_code == 201: + release_id = response.json().get("id") + logger.info(f"Created release {tag} with ID {release_id}") + return release_id + logger.error(f"Create release failed: {response.status_code}") + return None + except Exception as e: + logger.error(f"Create release failed: {e}") + return None + + def upload_release_asset( + self, + owner: str, + repo: str, + release_id: int, + filename: str, + filepath: str, + ) -> bool: + """Upload asset to release. + + Args: + owner: Repository owner. + repo: Repository name. + release_id: Release ID. + filename: Asset filename. + filepath: Local file path. + + Returns: + True if successful. + """ + url = GitHubAPI.upload_asset(owner, repo, release_id, filename) + headers = {**self._headers, "Content-Type": "application/octet-stream"} + + try: + with open(filepath, "rb") as f: + response = self.post(url, headers=headers, data=f) + if response.status_code == 201: + logger.info(f"Uploaded asset {filename}") + return True + logger.error(f"Upload asset failed: {response.status_code}") + return False + except Exception as e: + logger.error(f"Upload asset failed: {e}") + return False + + def create_issue( + self, + owner: str, + repo: str, + title: str, + body: str, + check_duplicate_id: Optional[str] = None, + ) -> bool: + """Create an issue with optional duplicate check. + + Args: + owner: Repository owner. + repo: Repository name. + title: Issue title. + body: Issue body. + check_duplicate_id: Optional ID to check for duplicates. + + Returns: + True if created (or skipped as duplicate). + """ + if check_duplicate_id: + body = f"{body}\n----{check_duplicate_id}" + if self._issue_exists(owner, repo, check_duplicate_id): + logger.debug(f"Issue already exists: {check_duplicate_id}") + return True + + url = GitHubAPI.issues(owner, repo) + try: + response = self.post( + url, + headers=self._headers, + json={"title": title, "body": body}, + ) + if response.status_code == 201: + issue_url = response.json().get("html_url") + logger.info(f"Issue created: {issue_url}") + return True + logger.error(f"Create issue failed: {response.status_code}") + return False + except Exception as e: + logger.error(f"Create issue failed: {e}") + return False + + def _issue_exists(self, owner: str, repo: str, check_id: str) -> bool: + """Check if an issue with given ID already exists.""" + url = GitHubAPI.issues(owner, repo) + try: + since = (datetime.now(UTC) - timedelta(days=10)).isoformat() + "Z" + response = self.get( + url, + headers=self._headers, + params={"state": "open", "since": since, "per_page": 100}, + ) + issues = response.json() + return any( + issue.get("body", "").endswith(f"----{check_id}") + for issue in issues + ) + except Exception: + return False + + def delete_old_releases( + self, owner: str, repo: str, keep_count: int = 2 + ) -> None: + """Delete old releases, keeping the most recent ones. + + Args: + owner: Repository owner. + repo: Repository name. + keep_count: Number of releases to keep. + """ + url = GitHubAPI.releases(owner, repo) + try: + response = self.get( + url, headers=self._headers, params={"per_page": 100} + ) + releases = response.json() + if len(releases) <= keep_count: + return + + # Sort by tag_name descending + releases.sort(key=lambda r: r.get("tag_name", ""), reverse=True) + + for release in releases[keep_count:]: + release_id = release.get("id") + tag = release.get("tag_name") + if release_id: + delete_url = f"{url}/{release_id}" + try: + resp = self.delete(delete_url, headers=self._headers) + if resp.status_code == 204: + logger.info(f"Deleted release: {tag}") + else: + logger.warning( + f"Failed to delete release {tag}: {resp.text}" + ) + except Exception as e: + logger.warning(f"Failed to delete release {tag}: {e}") + except Exception as e: + logger.error(f"Delete old releases failed: {e}") + + def delete_old_tags( + self, owner: str, repo: str, keep_count: int = 2 + ) -> None: + """Delete old timestamp-based tags. + + Args: + owner: Repository owner. + repo: Repository name. + keep_count: Number of tags to keep. + """ + url = GitHubAPI.tags(owner, repo) + try: + response = self.get(url, headers=self._headers) + tags = response.json() + if len(tags) <= keep_count: + return + + # Sort by ref descending + tags.sort(key=lambda t: t.get("ref", ""), reverse=True) + + for tag in tags[keep_count:]: + ref = tag.get("ref", "").replace("refs/tags/", "") + # Only delete timestamp tags (numeric, >= 10 digits) + if len(ref) < 10: + continue + try: + int(ref) # Verify it's a timestamp + except ValueError: + continue + + delete_url = f"{url}/{ref}" + try: + resp = self.delete(delete_url, headers=self._headers) + if resp.status_code == 204: + logger.info(f"Deleted tag: {ref}") + else: + logger.warning(f"Failed to delete tag {ref}: {resp.text}") + except Exception as e: + logger.warning(f"Failed to delete tag {ref}: {e}") + except Exception as e: + logger.error(f"Delete old tags failed: {e}") + + def delete_old_caches( + self, owner: str, repo: str, keep_count: int = 2 + ) -> None: + """Delete old action caches. + + Args: + owner: Repository owner. + repo: Repository name. + keep_count: Number of caches to keep. + """ + url = GitHubAPI.caches(owner, repo) + try: + response = self.get( + url, + headers=self._headers, + params={ + "per_page": 100, + "sort": "last_accessed_at", + "direction": "desc", + }, + ) + data = response.json() + if data.get("total_count", 0) <= keep_count: + return + + for cache in data.get("actions_caches", [])[keep_count:]: + cache_id = cache.get("id") + cache_key = cache.get("key") + if cache_id: + delete_url = f"{url}/{cache_id}" + try: + resp = self.delete(delete_url, headers=self._headers) + if resp.status_code == 204: + logger.info(f"Deleted cache: {cache_key}") + else: + logger.warning( + f"Failed to delete cache {cache_key}: {resp.text}" + ) + except Exception as e: + logger.warning(f"Failed to delete cache {cache_key}: {e}") + except Exception as e: + logger.error(f"Delete old caches failed: {e}") + + def get_rate_limit(self) -> Optional[dict[str, Any]]: + """Get current rate limit status. + + Returns: + Rate limit info or None if failed. + """ + try: + response = self.get(GitHubAPI.rate_limit(), headers=self._headers) + rate = response.json().get("rate") + logger.info(f"Rate limit: {rate}") + return rate + except Exception as e: + logger.error(f"Get rate limit failed: {e}") + return None diff --git a/src/zotero_scraper/config/__init__.py b/src/zotero_scraper/config/__init__.py new file mode 100644 index 0000000..93d73a0 --- /dev/null +++ b/src/zotero_scraper/config/__init__.py @@ -0,0 +1,13 @@ +"""Configuration module.""" + +from .constants import GitHubAPI, XPIProxy, ZoteroApp +from .settings import CacheConfig, GitHubConfig, ScraperConfig + +__all__ = [ + "GitHubAPI", + "XPIProxy", + "ZoteroApp", + "CacheConfig", + "GitHubConfig", + "ScraperConfig", +] diff --git a/src/zotero_scraper/config/constants.py b/src/zotero_scraper/config/constants.py new file mode 100644 index 0000000..5a28ddb --- /dev/null +++ b/src/zotero_scraper/config/constants.py @@ -0,0 +1,113 @@ +"""Constants and API endpoints for the Zotero addons scraper.""" + +from urllib.parse import quote + + +class GitHubAPI: + """GitHub API endpoints.""" + + BASE = "https://api.github.com" + UPLOADS_BASE = "https://uploads.github.com" + DEFAULT_API_VERSION = "2022-11-28" + + @staticmethod + def user(username: str) -> str: + """Get user info endpoint.""" + return f"{GitHubAPI.BASE}/users/{username}" + + @staticmethod + def repo(owner: str, repo: str) -> str: + """Get repository info endpoint.""" + return f"{GitHubAPI.BASE}/repos/{owner}/{repo}" + + @staticmethod + def releases(owner: str, repo: str) -> str: + """Get releases list endpoint.""" + return f"{GitHubAPI.BASE}/repos/{owner}/{repo}/releases" + + @staticmethod + def release_latest(owner: str, repo: str) -> str: + """Get latest release endpoint.""" + return f"{GitHubAPI.releases(owner, repo)}/latest" + + @staticmethod + def release_by_tag(owner: str, repo: str, tag: str) -> str: + """Get release by tag endpoint.""" + return f"{GitHubAPI.releases(owner, repo)}/tags/{tag}" + + @staticmethod + def issues(owner: str, repo: str) -> str: + """Get issues endpoint.""" + return f"{GitHubAPI.BASE}/repos/{owner}/{repo}/issues" + + @staticmethod + def caches(owner: str, repo: str) -> str: + """Get actions caches endpoint.""" + return f"{GitHubAPI.BASE}/repos/{owner}/{repo}/actions/caches" + + @staticmethod + def tags(owner: str, repo: str) -> str: + """Get git tags endpoint.""" + return f"{GitHubAPI.BASE}/repos/{owner}/{repo}/git/refs/tags" + + @staticmethod + def upload_asset(owner: str, repo: str, release_id: int, name: str) -> str: + """Get upload asset endpoint.""" + return ( + f"{GitHubAPI.UPLOADS_BASE}/repos/{owner}/{repo}" + f"/releases/{release_id}/assets?name={name}" + ) + + @staticmethod + def rate_limit() -> str: + """Get rate limit endpoint.""" + return f"{GitHubAPI.BASE}/rate_limit" + + +class XPIProxy: + """XPI download proxy URLs.""" + + GHPROXY_BASE = "https://gh-proxy.com/" + KKGITHUB_DOMAIN = "kkgithub.com" + + @staticmethod + def ghproxy_url(github_url: str) -> str: + """Get ghProxy URL for a GitHub download URL.""" + return f"{XPIProxy.GHPROXY_BASE}?q={quote(github_url, safe='')}" + + @staticmethod + def kkgithub_url(github_url: str) -> str: + """Get kkgithub URL for a GitHub download URL.""" + return github_url.replace("github.com", XPIProxy.KKGITHUB_DOMAIN) + + +class ZoteroApp: + """Zotero application constants.""" + + APP_ID = "zotero@chnm.gmu.edu" + SUPPORTED_VERSIONS = ("6", "7") + + @staticmethod + def check_version(target_version: str) -> str: + """Get Zotero version check string. + + Args: + target_version: Target Zotero version ("6" or "7"). + + Returns: + Version check string (e.g., "6.*" or "7.*"). + + Raises: + ValueError: If target_version is not supported. + """ + if target_version not in ZoteroApp.SUPPORTED_VERSIONS: + raise ValueError(f"Unsupported Zotero version: {target_version}") + return f"{target_version}.*" + + +class ContentTypes: + """MIME content types.""" + + XPI = "application/x-xpinstall" + ZIP = "application/x-zip-compressed" + OCTET_STREAM = "application/octet-stream" diff --git a/src/zotero_scraper/config/settings.py b/src/zotero_scraper/config/settings.py new file mode 100644 index 0000000..b2439bb --- /dev/null +++ b/src/zotero_scraper/config/settings.py @@ -0,0 +1,77 @@ +"""Configuration settings for the Zotero addons scraper.""" + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + + +@dataclass(frozen=True) +class GitHubConfig: + """GitHub API configuration.""" + + token: Optional[str] = None + repository: str = "" + api_version: str = "2022-11-28" + timeout: int = 30 + max_retries: int = 3 + retry_delay: float = 1.0 + + @property + def owner(self) -> Optional[str]: + """Get repository owner.""" + parts = self.repository.split("/") + return parts[0] if len(parts) == 2 else None + + @property + def repo_name(self) -> Optional[str]: + """Get repository name.""" + parts = self.repository.split("/") + return parts[1] if len(parts) == 2 else None + + +@dataclass(frozen=True) +class CacheConfig: + """Cache configuration.""" + + 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) +class ScraperConfig: + """Main scraper configuration.""" + + input_dir: Path = field(default_factory=lambda: Path("addons")) + output_file: Path = field(default_factory=lambda: Path("addon_infos.json")) + github: GitHubConfig = field(default_factory=GitHubConfig) + cache: CacheConfig = field(default_factory=CacheConfig) + previous_info_urls: list[str] = field(default_factory=list) + create_release: bool = True + max_workers: int = 10 + + @classmethod + def from_args(cls, args: Any) -> "ScraperConfig": + """Create configuration from argparse namespace. + + Args: + args: Parsed command line arguments. + + Returns: + ScraperConfig instance. + """ + return cls( + input_dir=Path(args.input), + output_file=Path(args.output), + github=GitHubConfig( + token=args.github_token, + repository=args.github_repository or "", + ), + 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, + ) diff --git a/src/zotero_scraper/exceptions.py b/src/zotero_scraper/exceptions.py new file mode 100644 index 0000000..531f9b4 --- /dev/null +++ b/src/zotero_scraper/exceptions.py @@ -0,0 +1,49 @@ +"""Custom exceptions for the Zotero addons scraper.""" + + +class ScraperError(Exception): + """Base exception for all scraper errors.""" + + pass + + +class ConfigError(ScraperError): + """Configuration error.""" + + pass + + +class HTTPError(ScraperError): + """HTTP request error.""" + + pass + + +class RateLimitError(HTTPError): + """API rate limit exceeded error.""" + + pass + + +class GitHubError(ScraperError): + """GitHub API error.""" + + pass + + +class XPIParseError(ScraperError): + """XPI file parsing error.""" + + pass + + +class DownloadError(ScraperError): + """File download error.""" + + pass + + +class CacheError(ScraperError): + """Cache operation error.""" + + pass diff --git a/src/zotero_scraper/models/__init__.py b/src/zotero_scraper/models/__init__.py new file mode 100644 index 0000000..a11142b --- /dev/null +++ b/src/zotero_scraper/models/__init__.py @@ -0,0 +1,12 @@ +"""Data models module.""" + +from .addon import AddonInfo, AddonRelease, Author, XpiDownloadUrls +from .xpi import XpiDetail + +__all__ = [ + "AddonInfo", + "AddonRelease", + "Author", + "XpiDownloadUrls", + "XpiDetail", +] diff --git a/src/zotero_scraper/models/addon.py b/src/zotero_scraper/models/addon.py new file mode 100644 index 0000000..6404c31 --- /dev/null +++ b/src/zotero_scraper/models/addon.py @@ -0,0 +1,218 @@ +"""Addon data models.""" + +from dataclasses import dataclass, field +from typing import Any, Optional, Union + + +@dataclass +class Author: + """Addon author information.""" + + name: Optional[str] = None + url: Optional[str] = None + avatar: Optional[str] = None + + def to_dict(self) -> dict[str, str]: + """Convert to dictionary, excluding None values.""" + result: dict[str, str] = {} + if self.name: + result["name"] = self.name + if self.url: + result["url"] = self.url + if self.avatar: + result["avatar"] = self.avatar + return result + + @classmethod + def from_dict(cls, data: Optional[dict[str, Any]]) -> "Author": + """Create Author from dictionary.""" + if not data: + return cls() + return cls( + name=data.get("name"), + url=data.get("url"), + avatar=data.get("avatar"), + ) + + +@dataclass +class XpiDownloadUrls: + """XPI download URLs from multiple sources.""" + + github: str + ghProxy: Optional[str] = None + kgithub: Optional[str] = None + + def to_dict(self) -> dict[str, str]: + """Convert to dictionary, excluding None values.""" + result: dict[str, str] = {"github": self.github} + if self.ghProxy: + result["ghProxy"] = self.ghProxy + if self.kgithub: + result["kgithub"] = self.kgithub + return result + + @classmethod + def from_dict(cls, data: Optional[dict[str, str]]) -> Optional["XpiDownloadUrls"]: + """Create XpiDownloadUrls from dictionary.""" + if not data or "github" not in data: + return None + return cls( + github=data["github"], + ghProxy=data.get("ghProxy"), + kgithub=data.get("kgithub"), + ) + + +@dataclass +class AddonRelease: + """Addon release version information.""" + + targetZoteroVersion: str + tagName: str + xpiDownloadUrl: Optional[XpiDownloadUrls] = None + releaseDate: Optional[str] = None + id: Optional[str] = None + xpiVersion: Optional[str] = None + name: Optional[str] = None + description: Optional[str] = None + minZoteroVersion: Optional[str] = None + maxZoteroVersion: Optional[str] = None + + @property + def zotero_check_version(self) -> str: + """Get Zotero version check string. + + Returns: + Version check string (e.g., "6.*" or "7.*"). + + Raises: + ValueError: If targetZoteroVersion is not valid. + """ + if self.targetZoteroVersion == "6": + return "6.*" + elif self.targetZoteroVersion == "7": + return "7.*" + raise ValueError( + f"Invalid targetZoteroVersion: {self.targetZoteroVersion}" + ) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary, excluding None values.""" + result: dict[str, Any] = {} + if self.targetZoteroVersion: + result["targetZoteroVersion"] = self.targetZoteroVersion + if self.tagName: + result["tagName"] = self.tagName + if self.xpiDownloadUrl: + result["xpiDownloadUrl"] = self.xpiDownloadUrl.to_dict() + if self.releaseDate: + result["releaseDate"] = self.releaseDate + if self.id: + result["id"] = self.id + if self.xpiVersion: + result["xpiVersion"] = self.xpiVersion + if self.name: + result["name"] = self.name + if self.description: + result["description"] = self.description + if self.minZoteroVersion: + result["minZoteroVersion"] = self.minZoteroVersion + if self.maxZoteroVersion: + result["maxZoteroVersion"] = self.maxZoteroVersion + return result + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "AddonRelease": + """Create AddonRelease from dictionary.""" + xpi_url_data = data.get("xpiDownloadUrl") + xpi_download_url = ( + XpiDownloadUrls.from_dict(xpi_url_data) + if isinstance(xpi_url_data, dict) + else None + ) + return cls( + targetZoteroVersion=data.get("targetZoteroVersion", ""), + tagName=data.get("tagName", ""), + xpiDownloadUrl=xpi_download_url, + releaseDate=data.get("releaseDate"), + id=data.get("id"), + xpiVersion=data.get("xpiVersion"), + name=data.get("name"), + description=data.get("description"), + minZoteroVersion=data.get("minZoteroVersion"), + maxZoteroVersion=data.get("maxZoteroVersion"), + ) + + +@dataclass +class AddonInfo: + """Complete addon information.""" + + repo: str + releases: list[AddonRelease] = field(default_factory=list) + name: Optional[str] = None + description: Optional[str] = None + stars: Optional[int] = None + author: Author = field(default_factory=Author) + + @property + def owner(self) -> Optional[str]: + """Get repository owner.""" + parts = self.repo.split("/") + return parts[0] if len(parts) == 2 else None + + @property + def repository(self) -> Optional[str]: + """Get repository name.""" + parts = self.repo.split("/") + return parts[1] if len(parts) == 2 else None + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary for JSON serialization. + + Maintains backward compatibility with existing output format. + """ + result: dict[str, Any] = {} + if self.repo: + result["repo"] = self.repo + if self.releases: + result["releases"] = [release.to_dict() for release in self.releases] + if self.name: + result["name"] = self.name + if self.description: + result["description"] = self.description + if self.stars: + result["stars"] = self.stars + # Backward compatibility: keep "star" field for older versions + result["star"] = self.stars + if self.author: + author_dict = self.author.to_dict() + if author_dict: + result["author"] = author_dict + return result + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "AddonInfo": + """Create AddonInfo from dictionary.""" + releases_data = data.get("releases", []) + releases = [ + AddonRelease.from_dict(r) if isinstance(r, dict) else r + for r in releases_data + ] + + author_data = data.get("author") + author = ( + Author.from_dict(author_data) + if isinstance(author_data, dict) + else Author() + ) + + return cls( + repo=data.get("repo", ""), + releases=releases, + name=data.get("name"), + description=data.get("description"), + stars=data.get("stars"), + author=author, + ) diff --git a/src/zotero_scraper/models/xpi.py b/src/zotero_scraper/models/xpi.py new file mode 100644 index 0000000..b855287 --- /dev/null +++ b/src/zotero_scraper/models/xpi.py @@ -0,0 +1,103 @@ +"""XPI detail model.""" + +from dataclasses import dataclass, field +from typing import Optional + +from ..utils.logging import get_logger +from ..utils.version import compare_versions + +logger = get_logger("models.xpi") + + +@dataclass +class XpiDetail: + """XPI addon details extracted from manifest files.""" + + id: Optional[str] = None + name: Optional[str] = None + version: Optional[str] = None + description: Optional[str] = None + update_url: Optional[str] = None + min_version: str = "*" + max_version: str = "*" + + def append_info(self, details: dict) -> None: + """Append information from parsed manifest details. + + Args: + details: Dictionary containing parsed manifest data. + """ + # Check ID consistency + if addon_id := details.get("id"): + if self.id and self.id != addon_id: + logger.warning(f"XPI ID mismatch: {self.id} != {addon_id}") + return + self.id = addon_id + + # Update basic info + if name := details.get("name"): + self.name = name + if version := details.get("version"): + self.version = version + if description := details.get("description"): + self.description = description + + # Update URL (support both keys) + if update_url := details.get("updateURL") or details.get("update_url"): + self.update_url = update_url + + # Update version range (take the widest range) + min_ver = details.get("min_version") + max_ver = details.get("max_version") + if min_ver and max_ver: + self._update_version_range(min_ver, max_ver) + + def _update_version_range(self, min_ver: str, max_ver: str) -> None: + """Update version range to include the new range. + + Takes the widest range between existing and new values. + + Args: + min_ver: New minimum version. + max_ver: New maximum version. + """ + # For min_version, take the smaller one + min_normalized = min_ver.replace("*", "0") + existing_min_normalized = self.min_version.replace("*", "999") + if compare_versions(min_normalized, existing_min_normalized) <= 0: + self.min_version = min_ver + + # For max_version, take the larger one + max_normalized = max_ver.replace("*", "999") + existing_max_normalized = self.max_version.replace("*", "0") + if compare_versions(max_normalized, existing_max_normalized) >= 0: + self.max_version = max_ver + + def check_compatible(self, version: str) -> bool: + """Check if addon is compatible with a Zotero version. + + Args: + version: Zotero version string (e.g., "7.*" or "6.*"). + + Returns: + True if compatible, False otherwise. + """ + # Normalize versions for comparison + min_normalized = self.min_version.replace("*", "0") + max_normalized = self.max_version.replace("*", "999") + version_normalized = version.replace("*", "999") + version_base = version.replace("*", "0") + + # Check if version is within range + return ( + compare_versions(min_normalized, version_normalized) <= 0 + and compare_versions(max_normalized, version_base) >= 0 + ) + + def is_valid(self) -> bool: + """Check if XPI detail has minimum required information. + + Returns: + True if has at least ID, False otherwise. + """ + return bool(self.id) diff --git a/src/zotero_scraper/parsers/__init__.py b/src/zotero_scraper/parsers/__init__.py new file mode 100644 index 0000000..dd7ad33 --- /dev/null +++ b/src/zotero_scraper/parsers/__init__.py @@ -0,0 +1,10 @@ +"""XPI parsers module.""" + +from .manifest import JsonManifestParser, RdfManifestParser +from .xpi_parser import XPIParser + +__all__ = [ + "JsonManifestParser", + "RdfManifestParser", + "XPIParser", +] diff --git a/src/zotero_scraper/parsers/manifest.py b/src/zotero_scraper/parsers/manifest.py new file mode 100644 index 0000000..8d5c0b0 --- /dev/null +++ b/src/zotero_scraper/parsers/manifest.py @@ -0,0 +1,349 @@ +"""Manifest file parsers for XPI addons.""" + +import os +import re +import zipfile +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Any, Optional, Union +from xml.dom import minidom + +import commentjson as json + +from ..config.constants import ZoteroApp +from ..utils.logging import get_logger +from ..utils.version import compare_versions + +logger = get_logger("parsers.manifest") + + +class ManifestParser(ABC): + """Abstract base class for manifest parsers.""" + + @abstractmethod + def parse(self, addon_path: Union[str, Path]) -> Optional[dict[str, Any]]: + """Parse manifest from addon path. + + Args: + addon_path: Path to XPI file or directory. + + Returns: + Dictionary with manifest data or None if not found. + """ + pass + + @abstractmethod + def extract_details( + self, addon_path: Union[str, Path], manifest: Any + ) -> Optional[dict[str, Any]]: + """Extract addon details from manifest. + + Args: + addon_path: Path to XPI file or directory. + manifest: Parsed manifest data. + + Returns: + Dictionary with addon details. + """ + pass + + +class JsonManifestParser(ManifestParser): + """Parser for manifest.json (WebExtensions format).""" + + def parse(self, addon_path: Union[str, Path]) -> Optional[dict[str, Any]]: + """Parse manifest.json from XPI file.""" + addon_path = Path(addon_path) + + try: + if zipfile.is_zipfile(addon_path): + with zipfile.ZipFile(addon_path, "r") as zf: + if "manifest.json" in zf.namelist(): + content = zf.read("manifest.json").decode("utf-8") + return json.loads(content) + elif addon_path.is_dir(): + manifest_path = addon_path / "manifest.json" + if manifest_path.exists(): + with open(manifest_path, "r", encoding="utf-8") as f: + return json.loads(f.read()) + except Exception as e: + logger.warning(f"Failed to parse manifest.json from {addon_path}: {e}") + + return None + + def extract_details( + self, addon_path: Union[str, Path], manifest: dict[str, Any] + ) -> Optional[dict[str, Any]]: + """Extract addon details from manifest.json.""" + details: dict[str, Any] = { + "name": manifest.get("name"), + "version": manifest.get("version"), + "description": manifest.get("description"), + "id": None, + "min_version": None, + "max_version": None, + "update_url": None, + } + + # Try to get addon info from applications/browser_specific_settings + for location in ("applications", "browser_specific_settings"): + if details["id"]: + break + for app in ("zotero", "gecko"): + try: + app_settings = manifest[location][app] + details["id"] = app_settings.get("id") + details["min_version"] = app_settings.get("strict_min_version") + details["max_version"] = app_settings.get("strict_max_version") + details["update_url"] = app_settings.get("update_url") + break + except (KeyError, TypeError): + continue + + # Handle __MSG_{}__ placeholders + self._resolve_message_placeholders(addon_path, manifest, details) + + return details + + def _resolve_message_placeholders( + self, + addon_path: Union[str, Path], + manifest: dict[str, Any], + details: dict[str, Any], + ) -> None: + """Resolve __MSG_xxx__ placeholders from locale files.""" + locale_data: Optional[dict[str, Any]] = None + + for key, value in details.items(): + if not isinstance(value, str): + continue + + match = re.search(r"__MSG_(.*?)__", value) + if not match: + continue + + placeholder = match.group(1) + + # Load locale data on first need + if locale_data is None: + locale_data = self._load_locale(addon_path, manifest) + if locale_data is None: + break + + # Resolve placeholder + msg = locale_data.get(placeholder, {}).get("message") + if msg: + details[key] = msg + + def _load_locale( + self, addon_path: Union[str, Path], manifest: dict[str, Any] + ) -> Optional[dict[str, Any]]: + """Load locale messages file.""" + default_locale = manifest.get("default_locale") + if not default_locale: + return None + + locale_filename = f"_locales/{default_locale}/messages.json" + addon_path = Path(addon_path) + + try: + if zipfile.is_zipfile(addon_path): + with zipfile.ZipFile(addon_path, "r") as zf: + if locale_filename in zf.namelist(): + content = zf.read(locale_filename).decode("utf-8") + return json.loads(content) + elif addon_path.is_dir(): + locale_path = addon_path / locale_filename + if locale_path.exists(): + with open(locale_path, "r", encoding="utf-8") as f: + return json.loads(f.read()) + except Exception as e: + logger.warning(f"Failed to load locale file: {e}") + + return None + + +class RdfManifestParser(ManifestParser): + """Parser for install.rdf (legacy XUL format).""" + + def parse(self, addon_path: Union[str, Path]) -> Optional[bytes]: + """Parse install.rdf from XPI file.""" + addon_path = Path(addon_path) + + try: + if zipfile.is_zipfile(addon_path): + with zipfile.ZipFile(addon_path, "r") as zf: + if "install.rdf" in zf.namelist(): + return zf.read("install.rdf") + elif addon_path.is_dir(): + rdf_path = addon_path / "install.rdf" + if rdf_path.exists(): + with open(rdf_path, "rb") as f: + return f.read() + except Exception as e: + logger.warning(f"Failed to parse install.rdf from {addon_path}: {e}") + + return None + + def extract_details( + self, addon_path: Union[str, Path], manifest: bytes + ) -> Optional[dict[str, Any]]: + """Extract addon details from install.rdf.""" + details: dict[str, Any] = { + "id": None, + "name": None, + "version": None, + "description": None, + "min_version": None, + "max_version": None, + "update_url": None, + } + + try: + doc = minidom.parseString(manifest) + + # Get namespace prefixes + em = self._get_namespace_id(doc, "http://www.mozilla.org/2004/em-rdf#") + rdf = self._get_namespace_id(doc, "http://www.w3.org/1999/02/22-rdf-syntax-ns#") + + # Find the main Description element + description = self._find_main_description(doc, rdf, em) + if description is None: + return None + + # Extract basic info + self._extract_info(description, details, em) + + # Validate ID + addon_id = details.get("id") + if not addon_id or (addon_id.startswith("__") and addon_id.endswith("__")): + return None + + # Extract version info from targetApplication + self._extract_target_application_info(description, details, em) + + # Handle updateURL + if update_url := details.get("updateURL"): + details["update_url"] = update_url + + return details + + except Exception as e: + logger.warning(f"Failed to extract details from install.rdf: {e}") + return None + + def _get_namespace_id(self, doc: minidom.Document, url: str) -> str: + """Get namespace prefix for a given URL.""" + attributes = doc.documentElement.attributes + for i in range(attributes.length): + attr = attributes.item(i) + if attr.value == url and ":" in attr.name: + return attr.name.split(":")[1] + ":" + return "" + + def _find_main_description( + self, doc: minidom.Document, rdf: str, em: str + ) -> Optional[minidom.Element]: + """Find the main Description element.""" + descriptions = doc.getElementsByTagName(rdf + "Description") + if not descriptions: + return None + + # Try to find description with targetApplication + for desc in descriptions: + if desc.getElementsByTagName(em + "targetApplication"): + return desc + + return descriptions.item(0) + + def _extract_info( + self, + node: minidom.Element, + result: dict[str, Any], + em: str, + ) -> None: + """Extract info from node attributes and children.""" + try: + # From attributes + for entry, value in node.attributes.items(): + entry = entry.replace(em, "") + if entry in result: + result[entry] = value + + # From child nodes + for child in node.childNodes: + if child.nodeType != child.ELEMENT_NODE: + continue + entry = child.nodeName.replace(em, "") + if entry in result: + result[entry] = self._get_text(child) + except Exception: + pass + + def _get_text(self, element: minidom.Element) -> str: + """Get text content of an element.""" + rc = [] + for node in element.childNodes: + if node.nodeType == node.TEXT_NODE: + rc.append(node.data) + return "".join(rc).strip() + + def _extract_target_application_info( + self, + description: minidom.Element, + details: dict[str, Any], + em: str, + ) -> None: + """Extract version info from targetApplication elements.""" + for target_app in description.getElementsByTagName(em + "targetApplication"): + version_info = {"id": None, "minVersion": None, "maxVersion": None} + self._extract_info(target_app, version_info, em) + self._update_version_from_target(details, version_info) + + # Check child nodes as well + for child in target_app.childNodes: + if child.nodeType != child.ELEMENT_NODE: + continue + version_info = {"id": None, "minVersion": None, "maxVersion": None} + self._extract_info(child, version_info, em) + self._update_version_from_target(details, version_info) + + def _update_version_from_target( + self, + details: dict[str, Any], + version_info: dict[str, Any], + ) -> None: + """Update version range from targetApplication info.""" + # Only process Zotero application + if version_info.get("id") != ZoteroApp.APP_ID: + return + + min_ver = version_info.get("minVersion") + max_ver = version_info.get("maxVersion") + + if not min_ver or not max_ver: + return + + # Update min version (take smaller) + existing_min = details.get("min_version") + if existing_min: + min_normalized = min_ver.replace("*", "0") + existing_normalized = existing_min.replace("*", "999") + if compare_versions(min_normalized, existing_normalized) <= 0: + details["min_version"] = min_ver + else: + details["min_version"] = min_ver + + # Update max version (take larger, but cap at 6.* for RDF) + # RDF format doesn't support Zotero 7 + if compare_versions(max_ver.replace("*", "999"), "6.*") > 0: + max_ver = "6.*" + + existing_max = details.get("max_version") + if existing_max: + max_normalized = max_ver.replace("*", "999") + existing_normalized = existing_max.replace("*", "0") + if compare_versions(max_normalized, existing_normalized) >= 0: + details["max_version"] = max_ver + else: + details["max_version"] = max_ver diff --git a/src/zotero_scraper/parsers/xpi_parser.py b/src/zotero_scraper/parsers/xpi_parser.py new file mode 100644 index 0000000..345137a --- /dev/null +++ b/src/zotero_scraper/parsers/xpi_parser.py @@ -0,0 +1,80 @@ +"""XPI file parser.""" + +import os +from pathlib import Path +from typing import Optional, Union + +from ..exceptions import XPIParseError +from ..models.xpi import XpiDetail +from ..utils.logging import get_logger +from .manifest import JsonManifestParser, RdfManifestParser + +logger = get_logger("parsers.xpi") + + +class XPIParser: + """Parser for XPI addon files.""" + + def __init__(self): + """Initialize XPI parser with manifest parsers.""" + self.json_parser = JsonManifestParser() + self.rdf_parser = RdfManifestParser() + + def parse( + self, + addon_path: Union[str, Path], + priority_sources: Optional[list[str]] = None, + ) -> Optional[XpiDetail]: + """Parse XPI file and extract addon details. + + Args: + addon_path: Path to XPI file or directory. + priority_sources: List of sources to try, in order. + Options: "json", "rdf". + Default: ["json", "rdf"]. + + Returns: + XpiDetail with extracted information, or None if parsing failed. + + Raises: + XPIParseError: If addon path doesn't exist. + """ + addon_path = Path(addon_path) + + if not addon_path.exists(): + raise XPIParseError(f"Addon path does not exist: {addon_path}") + + if priority_sources is None: + priority_sources = ["json", "rdf"] + + xpi_detail = XpiDetail() + + for source in priority_sources: + if source == "json": + self._parse_json(addon_path, xpi_detail) + elif source == "rdf": + self._parse_rdf(addon_path, xpi_detail) + + return xpi_detail if xpi_detail.is_valid() else None + + def _parse_json(self, addon_path: Path, xpi_detail: XpiDetail) -> None: + """Parse manifest.json and update XpiDetail.""" + try: + manifest = self.json_parser.parse(addon_path) + if manifest: + details = self.json_parser.extract_details(addon_path, manifest) + if details: + xpi_detail.append_info(details) + except Exception as e: + logger.debug(f"JSON parsing failed for {addon_path}: {e}") + + def _parse_rdf(self, addon_path: Path, xpi_detail: XpiDetail) -> None: + """Parse install.rdf and update XpiDetail.""" + try: + manifest = self.rdf_parser.parse(addon_path) + if manifest: + details = self.rdf_parser.extract_details(addon_path, manifest) + if details: + xpi_detail.append_info(details) + except Exception as e: + logger.debug(f"RDF parsing failed for {addon_path}: {e}") diff --git a/src/zotero_scraper/services/__init__.py b/src/zotero_scraper/services/__init__.py new file mode 100644 index 0000000..e7fbe10 --- /dev/null +++ b/src/zotero_scraper/services/__init__.py @@ -0,0 +1,11 @@ +"""Business services module.""" + +from .fallback import FallbackService +from .publisher import PublisherService +from .scraper import AddonScraper + +__all__ = [ + "AddonScraper", + "FallbackService", + "PublisherService", +] diff --git a/src/zotero_scraper/services/fallback.py b/src/zotero_scraper/services/fallback.py new file mode 100644 index 0000000..48e9cec --- /dev/null +++ b/src/zotero_scraper/services/fallback.py @@ -0,0 +1,185 @@ +"""Fallback service for merging addon data.""" + +from typing import Any, Optional + +import requests + +from ..utils.logging import get_logger + +logger = get_logger("services.fallback") + + +# Constants for fallback association mapping +_UNIQUE_ID_KEY = "p_unique_associated_id##80502300-9DE6-4510-8768-EC42B0EF14E6" +_EXCLUDE_IDS_KEY = "p_exclude_associated_ids##073F5543-C70A-46AA-8529-E05168852D8F" + +# Default association map for addon data structure +DEFAULT_ASSOCIATION_MAP: dict[str, Any] = { + _UNIQUE_ID_KEY: "repo", + _EXCLUDE_IDS_KEY: ["id"], + "repo": { + "releases": { + _UNIQUE_ID_KEY: "tagName", + _EXCLUDE_IDS_KEY: ["currentVersion"], + } + }, +} + + +class FallbackService: + """Service for falling back to previous addon data.""" + + def __init__(self, association_map: Optional[dict[str, Any]] = None): + """Initialize fallback service. + + Args: + association_map: Custom association map for merging. + Defaults to DEFAULT_ASSOCIATION_MAP. + """ + self.association_map = association_map or DEFAULT_ASSOCIATION_MAP + + def apply_fallback( + self, + current_data: list[dict[str, Any]], + previous_url: str, + ) -> list[dict[str, Any]]: + """Apply fallback from previous data URL. + + Args: + current_data: Current addon data list. + previous_url: URL to fetch previous data from. + + Returns: + Merged data list. + """ + try: + response = requests.get(previous_url, timeout=30) + response.raise_for_status() + previous_data = response.json() + + if isinstance(previous_data, list): + return self._merge_lists( + current_data, previous_data, self.association_map + ) + return current_data + + except Exception as e: + logger.warning(f"Failed to fetch fallback data from {previous_url}: {e}") + return current_data + + def _merge_if_needed( + self, + current: Any, + previous: Any, + association_map: Optional[dict[str, Any]] = None, + ) -> Any: + """Recursively merge previous data into current. + + Args: + current: Current data (dict or list). + previous: Previous data to merge from. + association_map: Association mapping for this level. + + Returns: + Merged data. + """ + if association_map is None: + association_map = self.association_map + + if not previous or not current: + return current + + if not isinstance(previous, type(current)): + return current + + if isinstance(current, dict): + return self._merge_dicts(current, previous, association_map) + elif isinstance(current, list): + return self._merge_lists(current, previous, association_map) + + return current + + def _merge_dicts( + self, + current: dict[str, Any], + previous: dict[str, Any], + association_map: dict[str, Any], + ) -> dict[str, Any]: + """Merge two dictionaries. + + Args: + current: Current dictionary. + previous: Previous dictionary. + association_map: Association mapping. + + Returns: + Merged dictionary. + """ + exclude_keys = association_map.get(_EXCLUDE_IDS_KEY, []) + + for key, value in previous.items(): + # Skip excluded keys + if key in exclude_keys: + continue + + if key in current: + # If current value exists and is a container, recurse + if isinstance(value, (list, dict)): + if current[key]: + nested_map = association_map.get(key, {}) + self._merge_if_needed(current[key], value, nested_map) + else: + # Current is empty, use previous + logger.debug(f"Fallback: {key}") + current[key] = value + else: + # Key doesn't exist in current, add from previous + logger.debug(f"Fallback: {key}") + current[key] = value + + return current + + def _merge_lists( + self, + current: list[Any], + previous: list[Any], + association_map: dict[str, Any], + ) -> list[Any]: + """Merge two lists using unique ID matching. + + Args: + current: Current list. + previous: Previous list. + association_map: Association mapping. + + Returns: + Merged list. + """ + unique_key = association_map.get(_UNIQUE_ID_KEY) + if not unique_key: + return current + + for prev_item in previous: + if not isinstance(prev_item, dict): + continue + + prev_id = prev_item.get(unique_key) + if not prev_id: + continue + + # Find matching item in current + current_item = next( + ( + item + for item in current + if isinstance(item, dict) and item.get(unique_key) == prev_id + ), + None, + ) + + if current_item: + # Merge matching items + nested_map = association_map.get(unique_key, {}) + self._merge_if_needed(current_item, prev_item, nested_map) + + return current diff --git a/src/zotero_scraper/services/publisher.py b/src/zotero_scraper/services/publisher.py new file mode 100644 index 0000000..5b4a426 --- /dev/null +++ b/src/zotero_scraper/services/publisher.py @@ -0,0 +1,117 @@ +"""Publisher service for releasing addon information.""" + +from pathlib import Path +from typing import Optional + +from ..clients.github import GitHubClient +from ..config.settings import GitHubConfig +from ..utils.logging import get_logger + +logger = get_logger("services.publisher") + + +class PublisherService: + """Service for publishing addon information to GitHub releases.""" + + def __init__(self, github_client: GitHubClient, config: GitHubConfig): + """Initialize publisher service. + + Args: + github_client: GitHub API client. + config: GitHub configuration. + """ + self.github = github_client + self.config = config + + def publish(self, output_file: Path) -> bool: + """Publish addon info file to GitHub release. + + Args: + output_file: Path to addon_infos.json file. + + Returns: + True if successful. + """ + if not self.config.owner or not self.config.repo_name: + logger.error("GitHub repository not configured") + return False + + if not output_file.exists(): + logger.error(f"Output file not found: {output_file}") + return False + + owner = self.config.owner + repo = self.config.repo_name + + # Clean up old releases and tags + logger.info("Cleaning up old releases and tags...") + self.github.delete_old_releases(owner, repo, keep_count=2) + self.github.delete_old_tags(owner, repo, keep_count=2) + + # Create new release + logger.info("Creating new release...") + release_id = self.github.create_release(owner, repo) + if not release_id: + logger.error("Failed to create release") + return False + + # Upload file + logger.info(f"Uploading {output_file.name}...") + success = self.github.upload_release_asset( + owner=owner, + repo=repo, + release_id=release_id, + filename="addon_infos.json", + filepath=str(output_file), + ) + + if success: + logger.info("Published successfully") + else: + logger.error("Failed to upload release asset") + + return success + + def cleanup_caches(self, keep_count: int = 1) -> None: + """Clean up old GitHub Actions caches. + + Args: + keep_count: Number of caches to keep. + """ + if not self.config.owner or not self.config.repo_name: + return + + logger.info("Cleaning up old caches...") + self.github.delete_old_caches( + self.config.owner, + self.config.repo_name, + keep_count=keep_count, + ) + + def report_issue( + self, + title: str, + body: str, + check_id: Optional[str] = None, + ) -> bool: + """Report an issue to GitHub. + + Args: + title: Issue title. + body: Issue body. + check_id: Optional ID for duplicate checking. + + Returns: + True if successful or skipped as duplicate. + """ + if not self.config.owner or not self.config.repo_name: + logger.warning("GitHub repository not configured for issue reporting") + return False + + return self.github.create_issue( + owner=self.config.owner, + repo=self.config.repo_name, + title=title, + body=body, + check_duplicate_id=check_id, + ) diff --git a/src/zotero_scraper/services/scraper.py b/src/zotero_scraper/services/scraper.py new file mode 100644 index 0000000..235390c --- /dev/null +++ b/src/zotero_scraper/services/scraper.py @@ -0,0 +1,357 @@ +"""Core scraper service for fetching addon information.""" + +import json +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path +from typing import Any, Optional + +from ..clients.downloader import XPIDownloader +from ..clients.github import GitHubClient, ReleaseAsset +from ..config.constants import ContentTypes, XPIProxy +from ..config.settings import ScraperConfig +from ..models.addon import AddonInfo, AddonRelease, Author, XpiDownloadUrls +from ..parsers.xpi_parser import XPIParser +from ..utils.logging import get_logger +from ..utils.version import compare_versions +from .fallback import FallbackService +from .publisher import PublisherService + +logger = get_logger("services.scraper") + + +class AddonScraper: + """Core service for scraping Zotero addon information.""" + + def __init__(self, config: ScraperConfig): + """Initialize scraper service. + + Args: + config: Scraper configuration. + """ + self.config = config + self.github = GitHubClient(config.github) + self.downloader = XPIDownloader(config.cache) + self.xpi_parser = XPIParser() + self.fallback = FallbackService() + self.publisher = PublisherService(self.github, config.github) + + def scrape_all(self) -> list[dict[str, Any]]: + """Scrape all addon information. + + Returns: + List of addon info dictionaries. + """ + # Load input configs + plugins = self._load_input_configs() + logger.info(f"Loaded {len(plugins)} addon configs") + + # Scrape in parallel + addon_infos: list[AddonInfo] = [] + with ThreadPoolExecutor(max_workers=self.config.max_workers) as executor: + futures = { + executor.submit(self._scrape_addon, plugin): plugin + for plugin in plugins + } + + for future in as_completed(futures): + plugin = futures[future] + try: + if result := future.result(): + addon_infos.append(result) + except Exception as e: + logger.error(f"Failed to scrape {plugin.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_input_configs(self) -> list[AddonInfo]: + """Load addon configs from input directory.""" + plugins: list[AddonInfo] = [] + + if not self.config.input_dir.exists(): + logger.error(f"Input directory not found: {self.config.input_dir}") + return plugins + + 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) + plugins.append(AddonInfo.from_dict(data)) + except Exception as e: + logger.error(f"Failed to load {config_file}: {e}") + + return plugins + + def _scrape_addon(self, plugin: AddonInfo) -> Optional[AddonInfo]: + """Scrape information for a single addon.""" + if not plugin.owner or not plugin.releases: + logger.warning(f"Invalid plugin config: {plugin.repo}") + return None + + plugin.name = plugin.repository + + # Fetch author info + self._fetch_author_info(plugin) + + # Fetch repository info + self._fetch_repo_info(plugin) + + # Process releases + valid_releases: list[AddonRelease] = [] + invalid_releases: list[AddonRelease] = [] + + for release in plugin.releases: + processed = self._process_release(plugin, release) + if processed: + if processed.get("valid"): + valid_releases.append(release) + else: + invalid_releases.append(release) + + # Report invalid releases + for invalid_release in invalid_releases: + logger.warning( + f"{plugin.repo} invalid for Zotero {invalid_release.zotero_check_version}" + ) + self.publisher.report_issue( + title=f"Invalid {plugin.repo} xpi with zotero version {invalid_release.zotero_check_version}", + body=( + f"xpi: https://github.com/{plugin.repo} @{invalid_release.tagName}\n" + f"min zotero Version: {invalid_release.minZoteroVersion}\n" + f"max Zotero version: {invalid_release.maxZoteroVersion}\n" + f"expect Zotero version: {invalid_release.zotero_check_version}\n" + ), + check_id=f"Target zotero version not match: {plugin.repo}+{invalid_release.tagName}@{invalid_release.targetZoteroVersion}", + ) + + plugin.releases = valid_releases + return plugin if valid_releases else None + + def _fetch_author_info(self, plugin: AddonInfo) -> None: + """Fetch author info from GitHub.""" + if user_info := self.github.get_user(plugin.owner): + plugin.author = Author( + name=user_info.name, + url=user_info.html_url, + avatar=user_info.avatar_url, + ) + + def _fetch_repo_info(self, plugin: AddonInfo) -> None: + """Fetch repository info from GitHub.""" + if repo_info := self.github.get_repo(plugin.owner, plugin.repository): + if not plugin.description and repo_info.description: + plugin.description = repo_info.description + plugin.stars = repo_info.stargazers_count + + def _process_release( + self, plugin: AddonInfo, release: AddonRelease + ) -> Optional[dict[str, Any]]: + """Process a single release. + + Returns: + Dict with 'valid' key indicating compatibility, or None if failed. + """ + if not release.tagName: + return None + + # Get release info from GitHub + release_info = self.github.get_release( + plugin.owner, plugin.repository, release.tagName + ) + if not release_info: + return None + + release.tagName = release_info.tag_name + + # Find XPI asset + xpi_asset = self._find_xpi_asset(release_info.assets) + if not xpi_asset: + logger.warning(f"No XPI found for {plugin.repo}@{release.tagName}") + return None + + # Set download URLs + release.xpiDownloadUrl = XpiDownloadUrls( + github=xpi_asset.browser_download_url, + ghProxy=XPIProxy.ghproxy_url(xpi_asset.browser_download_url), + kgithub=XPIProxy.kkgithub_url(xpi_asset.browser_download_url), + ) + release.releaseDate = xpi_asset.updated_at + + # Download and parse XPI + xpi_filename = ( + f"{plugin.owner}#{plugin.repository}+{release.tagName}@{xpi_asset.id}.xpi" + ) + + # Determine priority sources based on Zotero version + priority_sources = ["rdf", "json"] + if release.targetZoteroVersion == "6": + priority_sources = ["json", "rdf"] + + xpi_path = self.downloader.download( + xpi_asset.browser_download_url, xpi_filename + ) + + if not xpi_path: + self.publisher.report_issue( + title=f"Parse {plugin.repo} addon details failed", + body=f"xpi: https://github.com/{plugin.repo} @{release.tagName} on {release.targetZoteroVersion}\n", + check_id=f"Parse details failed: {plugin.repo}+{release.tagName}@{release.targetZoteroVersion}", + ) + return None + + details = self.xpi_parser.parse(xpi_path, priority_sources) + + if details: + # Update release with XPI details + if details.id: + release.id = details.id + if details.name: + release.name = details.name + if details.version: + release.xpiVersion = details.version + if details.description: + release.description = details.description + release.minZoteroVersion = details.min_version + release.maxZoteroVersion = details.max_version + + # Check for updates via update_url + self._check_for_updates(plugin, release, details, priority_sources) + + # Validate compatibility + if not details.check_compatible(release.zotero_check_version): + return {"valid": False} + + # Report if version info is missing + if ( + not release.minZoteroVersion + or release.minZoteroVersion == "*" + ) and ( + not release.maxZoteroVersion + or release.maxZoteroVersion == "*" + ): + self.publisher.report_issue( + title=f"Parse {plugin.repo} of zotero version failed", + body=f"xpi: https://github.com/{plugin.repo} @{release.tagName} on {release.targetZoteroVersion}\n", + check_id=f"Parse min/max version failed: {plugin.repo}+{release.tagName}@{release.targetZoteroVersion}", + ) + + return {"valid": True} + else: + self.publisher.report_issue( + title=f"Parse {plugin.repo} addon details failed", + body=f"xpi: https://github.com/{plugin.repo} @{release.tagName} on {release.targetZoteroVersion}\n", + check_id=f"Parse details failed: {plugin.repo}+{release.tagName}@{release.targetZoteroVersion}", + ) + return None + + def _find_xpi_asset( + self, assets: list[ReleaseAsset] + ) -> Optional[ReleaseAsset]: + """Find XPI asset from release assets.""" + # Sort by updated_at descending + sorted_assets = sorted( + assets, key=lambda a: a.updated_at, reverse=True + ) + + # First try XPI content type + for asset in sorted_assets: + if asset.content_type == ContentTypes.XPI: + return asset + + # Fallback to ZIP content type + for asset in sorted_assets: + if asset.content_type == ContentTypes.ZIP: + return asset + + return None + + def _check_for_updates( + self, + plugin: AddonInfo, + release: AddonRelease, + details: Any, + priority_sources: list[str], + ) -> None: + """Check for updates via update_url.""" + if not details.update_url or not details.id or not details.version: + return + + try: + import requests + + response = requests.get(details.update_url, timeout=30) + if response.status_code != 200: + return + + 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 + + update_filename = ( + f"{plugin.owner}#{plugin.repository}+update{details.version}.xpi" + ) + update_path = self.downloader.download(xpi_url, update_filename) + + if update_path: + update_details = self.xpi_parser.parse( + update_path, priority_sources + ) + if ( + update_details + and update_details.id + and update_details.check_compatible( + release.zotero_check_version + ) + ): + # Update release with newer version + if update_details.id: + release.id = update_details.id + if update_details.name: + release.name = update_details.name + if update_details.version: + release.xpiVersion = update_details.version + if update_details.description: + release.description = update_details.description + release.minZoteroVersion = update_details.min_version + release.maxZoteroVersion = update_details.max_version + release.xpiDownloadUrl = XpiDownloadUrls(github=xpi_url) + break + + except Exception as e: + logger.debug(f"Failed to check updates for {plugin.repo}: {e}") + + 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}") diff --git a/src/zotero_scraper/utils/__init__.py b/src/zotero_scraper/utils/__init__.py new file mode 100644 index 0000000..d3cab77 --- /dev/null +++ b/src/zotero_scraper/utils/__init__.py @@ -0,0 +1,10 @@ +"""Utility functions module.""" + +from .logging import get_logger, setup_logging +from .version import compare_versions + +__all__ = [ + "compare_versions", + "get_logger", + "setup_logging", +] diff --git a/src/zotero_scraper/utils/logging.py b/src/zotero_scraper/utils/logging.py new file mode 100644 index 0000000..8ddc3a5 --- /dev/null +++ b/src/zotero_scraper/utils/logging.py @@ -0,0 +1,60 @@ +"""Logging configuration for the Zotero addons scraper.""" + +import logging +import sys +from pathlib import Path +from typing import Optional + + +def setup_logging( + level: int = logging.INFO, + log_file: Optional[Path] = None, + format_string: Optional[str] = None, +) -> logging.Logger: + """Configure the logging system. + + Args: + level: Logging level (default: INFO). + log_file: Optional path to log file. + format_string: Optional custom format string. + + Returns: + Configured root logger for the package. + """ + if format_string is None: + format_string = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + + # Create root logger for the package + logger = logging.getLogger("zotero_scraper") + logger.setLevel(level) + + # Remove existing handlers to avoid duplicates + logger.handlers.clear() + + # Console handler + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(level) + console_handler.setFormatter(logging.Formatter(format_string)) + logger.addHandler(console_handler) + + # File handler (optional) + if log_file: + log_file.parent.mkdir(parents=True, exist_ok=True) + file_handler = logging.FileHandler(log_file, encoding="utf-8") + file_handler.setLevel(level) + file_handler.setFormatter(logging.Formatter(format_string)) + logger.addHandler(file_handler) + + return logger + + +def get_logger(name: str) -> logging.Logger: + """Get a child logger for a specific module. + + Args: + name: Module name (will be prefixed with 'zotero_scraper.'). + + Returns: + Logger instance for the module. + """ + return logging.getLogger(f"zotero_scraper.{name}") diff --git a/src/zotero_scraper/utils/version.py b/src/zotero_scraper/utils/version.py new file mode 100644 index 0000000..b083a5a --- /dev/null +++ b/src/zotero_scraper/utils/version.py @@ -0,0 +1,75 @@ +"""Version comparison utilities.""" + +from typing import Union + + +def compare_versions(version1: Union[str, int], version2: Union[str, int]) -> int: + """Compare two version strings. + + Supports version formats like: + - "1.0.0", "1.2.3" + - "1.0.0-beta", "1.0.0-alpha" + - "7.*", "6.*" + + Args: + version1: First version string. + version2: Second version string. + + Returns: + -1 if version1 < version2 + 0 if version1 == version2 + 1 if version1 > version2 + """ + v1 = str(version1) + v2 = str(version2) + + # Normalize: replace '-' with '.' for comparison + parts1 = v1.replace("-", ".").split(".") + parts2 = v2.replace("-", ".").split(".") + + # Pad shorter list with zeros + max_len = max(len(parts1), len(parts2)) + parts1.extend(["0"] * (max_len - len(parts1))) + parts2.extend(["0"] * (max_len - len(parts2))) + + for p1, p2 in zip(parts1, parts2): + # Handle wildcard + if p1 == "*" or p2 == "*": + continue + + # Try numeric comparison + try: + n1, n2 = int(p1), int(p2) + if n1 < n2: + return -1 + elif n1 > n2: + return 1 + except ValueError: + # Fall back to string comparison for non-numeric parts + if p1 < p2: + return -1 + elif p1 > p2: + return 1 + + return 0 + + +def version_in_range( + version: str, + min_version: str, + max_version: str, +) -> bool: + """Check if a version is within a given range. + + Args: + version: Version to check. + min_version: Minimum version (inclusive). + max_version: Maximum version (inclusive). + + Returns: + True if version is within [min_version, max_version]. + """ + return ( + compare_versions(version, min_version) >= 0 + and compare_versions(version, max_version) <= 0 + ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..46816dd --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests package.""" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a5e24e1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,126 @@ +"""Pytest fixtures for Zotero addons scraper tests.""" + +import json +import tempfile +from pathlib import Path +from typing import Any + +import pytest + + +@pytest.fixture +def fixtures_dir() -> Path: + """Return path to fixtures directory.""" + return Path(__file__).parent / "fixtures" + + +@pytest.fixture +def temp_dir(): + """Create a temporary directory for tests.""" + with tempfile.TemporaryDirectory() as tmpdir: + yield Path(tmpdir) + + +@pytest.fixture +def sample_manifest_json() -> dict[str, Any]: + """Return sample manifest.json data.""" + return { + "name": "Test Addon", + "version": "1.0.0", + "description": "A test addon for Zotero", + "browser_specific_settings": { + "zotero": { + "id": "test@addon.com", + "strict_min_version": "6.0", + "strict_max_version": "7.*", + } + }, + } + + +@pytest.fixture +def sample_addon_config() -> dict[str, Any]: + """Return sample addon configuration.""" + return { + "repo": "owner/repo", + "releases": [ + {"targetZoteroVersion": "7", "tagName": "latest"}, + {"targetZoteroVersion": "6", "tagName": "v1.0"}, + ], + } + + +@pytest.fixture +def sample_addon_info_output() -> dict[str, Any]: + """Return sample output addon info.""" + return { + "repo": "owner/repo", + "releases": [ + { + "targetZoteroVersion": "7", + "tagName": "v2.0.0", + "xpiDownloadUrl": { + "github": "https://github.com/owner/repo/releases/download/v2.0.0/addon.xpi", + "ghProxy": "https://gh-proxy.com/?q=https%3A//github.com/owner/repo/releases/download/v2.0.0/addon.xpi", + "kgithub": "https://kkgithub.com/owner/repo/releases/download/v2.0.0/addon.xpi", + }, + "releaseDate": "2024-01-01T00:00:00Z", + "id": "test@addon.com", + "xpiVersion": "2.0.0", + "name": "Test Addon", + "description": "A test addon", + "minZoteroVersion": "7.0", + "maxZoteroVersion": "7.*", + } + ], + "name": "repo", + "description": "Test repository", + "stars": 100, + "star": 100, + "author": { + "name": "Test Author", + "url": "https://github.com/owner", + "avatar": "https://avatars.githubusercontent.com/owner", + }, + } + + +@pytest.fixture +def github_user_response() -> dict[str, Any]: + """Return sample GitHub user API response.""" + return { + "login": "owner", + "name": "Test Owner", + "html_url": "https://github.com/owner", + "avatar_url": "https://avatars.githubusercontent.com/owner", + } + + +@pytest.fixture +def github_repo_response() -> dict[str, Any]: + """Return sample GitHub repo API response.""" + return { + "name": "repo", + "full_name": "owner/repo", + "description": "Test repository description", + "stargazers_count": 100, + } + + +@pytest.fixture +def github_release_response() -> dict[str, Any]: + """Return sample GitHub release API response.""" + return { + "tag_name": "v2.0.0", + "prerelease": False, + "published_at": "2024-01-01T00:00:00Z", + "assets": [ + { + "id": 12345, + "name": "addon.xpi", + "browser_download_url": "https://github.com/owner/repo/releases/download/v2.0.0/addon.xpi", + "content_type": "application/x-xpinstall", + "updated_at": "2024-01-01T00:00:00Z", + } + ], + } diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..c66cd71 --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1 @@ +"""Integration tests package.""" diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..ea3f8b9 --- /dev/null +++ b/tests/unit/__init__.py @@ -0,0 +1 @@ +"""Unit tests package.""" diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py new file mode 100644 index 0000000..8dc3efa --- /dev/null +++ b/tests/unit/test_models.py @@ -0,0 +1,189 @@ +"""Unit tests for data models.""" + +import pytest + +from zotero_scraper.models.addon import ( + AddonInfo, + AddonRelease, + Author, + XpiDownloadUrls, +) + + +class TestAuthor: + """Tests for Author model.""" + + def test_to_dict_full(self): + """Test to_dict with all fields.""" + author = Author( + name="Test Author", + url="https://github.com/test", + avatar="https://avatars.githubusercontent.com/test", + ) + result = author.to_dict() + assert result == { + "name": "Test Author", + "url": "https://github.com/test", + "avatar": "https://avatars.githubusercontent.com/test", + } + + def test_to_dict_excludes_none(self): + """Test to_dict excludes None values.""" + author = Author(name="Test", url=None, avatar=None) + result = author.to_dict() + assert result == {"name": "Test"} + + def test_from_dict(self): + """Test from_dict factory method.""" + data = {"name": "Test", "url": "https://example.com"} + author = Author.from_dict(data) + assert author.name == "Test" + assert author.url == "https://example.com" + assert author.avatar is None + + def test_from_dict_none(self): + """Test from_dict with None input.""" + author = Author.from_dict(None) + assert author.name is None + assert author.url is None + assert author.avatar is None + + +class TestXpiDownloadUrls: + """Tests for XpiDownloadUrls model.""" + + def test_to_dict_full(self): + """Test to_dict with all fields.""" + urls = XpiDownloadUrls( + github="https://github.com/example.xpi", + ghProxy="https://gh-proxy.com/example.xpi", + kgithub="https://kkgithub.com/example.xpi", + ) + result = urls.to_dict() + assert result == { + "github": "https://github.com/example.xpi", + "ghProxy": "https://gh-proxy.com/example.xpi", + "kgithub": "https://kkgithub.com/example.xpi", + } + + def test_to_dict_minimal(self): + """Test to_dict with only required fields.""" + urls = XpiDownloadUrls(github="https://github.com/example.xpi") + result = urls.to_dict() + assert result == {"github": "https://github.com/example.xpi"} + + def test_from_dict_valid(self): + """Test from_dict with valid data.""" + data = {"github": "https://github.com/example.xpi"} + urls = XpiDownloadUrls.from_dict(data) + assert urls is not None + assert urls.github == "https://github.com/example.xpi" + + def test_from_dict_missing_github(self): + """Test from_dict returns None if github is missing.""" + data = {"ghProxy": "https://gh-proxy.com/example.xpi"} + urls = XpiDownloadUrls.from_dict(data) + assert urls is None + + +class TestAddonRelease: + """Tests for AddonRelease model.""" + + def test_zotero_check_version_7(self): + """Test zotero_check_version for Zotero 7.""" + release = AddonRelease(targetZoteroVersion="7", tagName="latest") + assert release.zotero_check_version == "7.*" + + def test_zotero_check_version_6(self): + """Test zotero_check_version for Zotero 6.""" + release = AddonRelease(targetZoteroVersion="6", tagName="v1.0") + assert release.zotero_check_version == "6.*" + + def test_zotero_check_version_invalid(self): + """Test zotero_check_version raises for invalid version.""" + release = AddonRelease(targetZoteroVersion="5", tagName="v1.0") + with pytest.raises(ValueError, match="Invalid targetZoteroVersion"): + _ = release.zotero_check_version + + def test_to_dict_minimal(self): + """Test to_dict with minimal fields.""" + release = AddonRelease(targetZoteroVersion="7", tagName="latest") + result = release.to_dict() + assert result == {"targetZoteroVersion": "7", "tagName": "latest"} + + def test_to_dict_full(self): + """Test to_dict with all fields.""" + release = AddonRelease( + targetZoteroVersion="7", + tagName="v1.0.0", + xpiDownloadUrl=XpiDownloadUrls(github="https://example.com"), + releaseDate="2024-01-01T00:00:00Z", + id="test@addon.com", + xpiVersion="1.0.0", + name="Test Addon", + description="Test description", + minZoteroVersion="7.0", + maxZoteroVersion="7.*", + ) + result = release.to_dict() + assert result["targetZoteroVersion"] == "7" + assert result["tagName"] == "v1.0.0" + assert result["xpiDownloadUrl"] == {"github": "https://example.com"} + assert result["id"] == "test@addon.com" + + def test_from_dict(self, sample_addon_config): + """Test from_dict factory method.""" + release_data = sample_addon_config["releases"][0] + release = AddonRelease.from_dict(release_data) + assert release.targetZoteroVersion == "7" + assert release.tagName == "latest" + + +class TestAddonInfo: + """Tests for AddonInfo model.""" + + def test_owner_valid(self): + """Test owner property with valid repo.""" + addon = AddonInfo(repo="owner/repo", releases=[]) + assert addon.owner == "owner" + + def test_owner_invalid(self): + """Test owner property with invalid repo.""" + addon = AddonInfo(repo="invalid", releases=[]) + assert addon.owner is None + + def test_repository_valid(self): + """Test repository property with valid repo.""" + addon = AddonInfo(repo="owner/repo", releases=[]) + assert addon.repository == "repo" + + def test_repository_invalid(self): + """Test repository property with invalid repo.""" + addon = AddonInfo(repo="invalid", releases=[]) + assert addon.repository is None + + def test_to_dict_backward_compatible(self): + """Test to_dict maintains backward compatibility.""" + addon = AddonInfo( + repo="owner/repo", + releases=[AddonRelease(targetZoteroVersion="7", tagName="v1.0")], + name="Test", + stars=100, + author=Author(name="Author"), + ) + result = addon.to_dict() + + # Check backward compatibility: both "stars" and "star" should exist + assert result["stars"] == 100 + assert result["star"] == 100 + assert result["repo"] == "owner/repo" + assert result["name"] == "Test" + assert result["author"] == {"name": "Author"} + + def test_from_dict(self, sample_addon_config): + """Test from_dict factory method.""" + addon = AddonInfo.from_dict(sample_addon_config) + assert addon.repo == "owner/repo" + assert len(addon.releases) == 2 + assert addon.releases[0].targetZoteroVersion == "7" + assert addon.releases[1].tagName == "v1.0" diff --git a/tests/unit/test_parsers.py b/tests/unit/test_parsers.py new file mode 100644 index 0000000..aa892fe --- /dev/null +++ b/tests/unit/test_parsers.py @@ -0,0 +1,149 @@ +"""Unit tests for parsers.""" + +import json +import tempfile +import zipfile +from pathlib import Path + +import pytest + +from zotero_scraper.models.xpi import XpiDetail +from zotero_scraper.parsers.manifest import JsonManifestParser, RdfManifestParser +from zotero_scraper.parsers.xpi_parser import XPIParser + + +class TestJsonManifestParser: + """Tests for JsonManifestParser.""" + + @pytest.fixture + def parser(self): + return JsonManifestParser() + + @pytest.fixture + def sample_xpi(self, sample_manifest_json, temp_dir): + """Create a sample XPI file with manifest.json.""" + xpi_path = temp_dir / "test.xpi" + with zipfile.ZipFile(xpi_path, "w") as zf: + zf.writestr("manifest.json", json.dumps(sample_manifest_json)) + return xpi_path + + def test_parse_from_xpi(self, parser, sample_xpi, sample_manifest_json): + """Test parsing manifest.json from XPI file.""" + manifest = parser.parse(sample_xpi) + assert manifest is not None + assert manifest["name"] == sample_manifest_json["name"] + assert manifest["version"] == sample_manifest_json["version"] + + def test_parse_from_directory(self, parser, sample_manifest_json, temp_dir): + """Test parsing manifest.json from directory.""" + manifest_path = temp_dir / "manifest.json" + with open(manifest_path, "w") as f: + json.dump(sample_manifest_json, f) + + manifest = parser.parse(temp_dir) + assert manifest is not None + assert manifest["name"] == sample_manifest_json["name"] + + def test_parse_nonexistent(self, parser, temp_dir): + """Test parsing from nonexistent path.""" + manifest = parser.parse(temp_dir / "nonexistent.xpi") + assert manifest is None + + def test_extract_details(self, parser, sample_xpi, sample_manifest_json): + """Test extracting details from manifest.""" + manifest = parser.parse(sample_xpi) + details = parser.extract_details(sample_xpi, manifest) + + assert details["name"] == "Test Addon" + assert details["version"] == "1.0.0" + assert details["id"] == "test@addon.com" + assert details["min_version"] == "6.0" + assert details["max_version"] == "7.*" + + +class TestXpiDetail: + """Tests for XpiDetail model.""" + + def test_append_info(self): + """Test appending info to XpiDetail.""" + detail = XpiDetail() + detail.append_info({ + "id": "test@addon.com", + "name": "Test Addon", + "version": "1.0.0", + "min_version": "6.0", + "max_version": "7.*", + }) + + assert detail.id == "test@addon.com" + assert detail.name == "Test Addon" + assert detail.version == "1.0.0" + assert detail.min_version == "6.0" + assert detail.max_version == "7.*" + + def test_check_compatible_zotero7(self): + """Test compatibility check for Zotero 7.""" + detail = XpiDetail() + detail.min_version = "7.0" + detail.max_version = "7.*" + + assert detail.check_compatible("7.*") is True + assert detail.check_compatible("6.*") is False + + def test_check_compatible_both_versions(self): + """Test compatibility check for both Zotero versions.""" + detail = XpiDetail() + detail.min_version = "6.0" + detail.max_version = "7.*" + + assert detail.check_compatible("7.*") is True + assert detail.check_compatible("6.*") is True + + def test_is_valid(self): + """Test validity check.""" + detail = XpiDetail() + assert detail.is_valid() is False + + detail.id = "test@addon.com" + assert detail.is_valid() is True + + +class TestXPIParser: + """Tests for XPIParser.""" + + @pytest.fixture + def parser(self): + return XPIParser() + + def test_parse_json_manifest(self, parser, sample_manifest_json, temp_dir): + """Test parsing XPI with JSON manifest.""" + xpi_path = temp_dir / "test.xpi" + with zipfile.ZipFile(xpi_path, "w") as zf: + zf.writestr("manifest.json", json.dumps(sample_manifest_json)) + + details = parser.parse(xpi_path) + assert details is not None + assert details.id == "test@addon.com" + assert details.name == "Test Addon" + + def test_parse_nonexistent_raises(self, parser, temp_dir): + """Test parsing nonexistent file raises error.""" + from zotero_scraper.exceptions import XPIParseError + + with pytest.raises(XPIParseError): + parser.parse(temp_dir / "nonexistent.xpi") + + def test_parse_priority_sources(self, parser, sample_manifest_json, temp_dir): + """Test priority sources parameter.""" + xpi_path = temp_dir / "test.xpi" + with zipfile.ZipFile(xpi_path, "w") as zf: + zf.writestr("manifest.json", json.dumps(sample_manifest_json)) + + # Should work with json priority + details = parser.parse(xpi_path, priority_sources=["json"]) + assert details is not None + assert details.id == "test@addon.com" + + # Should return None with only rdf priority (no install.rdf) + details = parser.parse(xpi_path, priority_sources=["rdf"]) + assert details is None diff --git a/tests/unit/test_version.py b/tests/unit/test_version.py new file mode 100644 index 0000000..f7f0ec4 --- /dev/null +++ b/tests/unit/test_version.py @@ -0,0 +1,71 @@ +"""Unit tests for version utilities.""" + +import pytest + +from zotero_scraper.utils.version import compare_versions, version_in_range + + +class TestCompareVersions: + """Tests for compare_versions function.""" + + @pytest.mark.parametrize( + "v1,v2,expected", + [ + # Equal versions + ("1.0.0", "1.0.0", 0), + ("2.0", "2.0", 0), + ("1", "1", 0), + # v1 > v2 + ("2.0.0", "1.0.0", 1), + ("1.1.0", "1.0.0", 1), + ("1.0.1", "1.0.0", 1), + ("10.0.0", "9.0.0", 1), + ("1.0.0", "0.9.9", 1), + # v1 < v2 + ("1.0.0", "2.0.0", -1), + ("1.0.0", "1.1.0", -1), + ("1.0.0", "1.0.1", -1), + ("9.0.0", "10.0.0", -1), + # Different length versions + ("1.0", "1.0.0", 0), + ("1.0.0", "1.0", 0), + ("1.0.1", "1.0", 1), + # Pre-release versions (using - separator) + ("1.0.0-beta", "1.0.0-alpha", 1), + ("1.0.0-alpha", "1.0.0-beta", -1), + # Wildcard versions + ("7.*", "6.*", 1), + ("6.*", "7.*", -1), + ("7.*", "7.*", 0), + ], + ) + def test_compare_versions(self, v1: str, v2: str, expected: int): + """Test version comparison.""" + assert compare_versions(v1, v2) == expected + + def test_compare_versions_with_int(self): + """Test version comparison with integer input.""" + assert compare_versions(1, "1.0.0") == 0 + assert compare_versions("1.0.0", 1) == 0 + + +class TestVersionInRange: + """Tests for version_in_range function.""" + + @pytest.mark.parametrize( + "version,min_ver,max_ver,expected", + [ + ("7.0", "6.0", "8.0", True), + ("6.0", "6.0", "7.0", True), + ("7.0", "6.0", "7.0", True), + ("5.0", "6.0", "7.0", False), + ("8.0", "6.0", "7.0", False), + ("7.*", "6.*", "7.*", True), + ("6.5", "6.*", "7.*", True), + ], + ) + def test_version_in_range( + self, version: str, min_ver: str, max_ver: str, expected: bool + ): + """Test version range checking.""" + assert version_in_range(version, min_ver, max_ver) == expected