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

refactor: refactor proj

Browse Source
This commit is contained in:
ytshen 2025-12-23 16:22:54 +08:00
parent bd1de7ca46
commit dc2d1bd336
43 changed files with 3539 additions and 987 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
.claude/settings.local.json Normal file
View File

@ -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:*)"
]
}
}

118
addon_info.py
View File

@ -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

17
convert_addon_from_zchinese.py
View File

@ -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=' ')

40
fallback_infos.py
View File

@ -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

27
file_cache.py
View File

@ -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}')

186
github_operations.py
View File

@ -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}')

336
main.py
View File

@ -1,320 +1,30 @@
import concurrent.futures #!/usr/bin/env python3
import urllib.parse """
import argparse Backward compatible entry point for Zotero addons scraper.
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 *
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): Usage:
try: python main.py -i addons -o addon_infos.json \\
download_filepath = os.path.join(download_dir, unique_name) --github_repository owner/repo \\
folder = os.path.dirname(download_filepath) --github_token TOKEN \\
if folder and not os.path.exists(folder): --cache_directory caches \\
os.makedirs(folder) --runtime_xpi_directory xpis \\
--previous_info_urls URL1 URL2 \\
--create_release True
"""
if cache_dir := kwargs.get('cache_dir'): import sys
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
if not force_download and os.path.isfile(download_filepath): # Add src to path for development
return download_filepath from pathlib import Path
response = requests.get(xpi_url, stream=True) src_path = Path(__file__).parent / "src"
if response.status_code == 200: if src_path.exists():
print(f'download {unique_name} from {xpi_url}') sys.path.insert(0, str(src_path))
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}')
from zotero_scraper.__main__ import main
# 输出格式参考 [zotero-chinese/zotero-plugins](https://github.com/zotero-chinese/zotero-plugins) if __name__ == "__main__":
def parse(plugin: AddonInfo, **kwargs): sys.exit(main())
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)

283
moz_addons.py
View File

@ -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

48
pyproject.toml Normal file
View File

@ -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"]

13
requirements-dev.txt Normal file
View File

@ -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

5
requirements.txt
View File

@ -1,3 +1,2 @@
requests requests>=2.28.0
argparse commentjson>=0.9.0
commentjson

3
src/zotero_scraper/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""Zotero Addons Scraper - Scrape Zotero addon information from GitHub."""
__version__ = "2.0.0"

148
src/zotero_scraper/__main__.py Normal file
View File

@ -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())

5
src/zotero_scraper/cache/__init__.py vendored Normal file
View File

@ -0,0 +1,5 @@
"""Cache module."""
from .file_cache import FileCache
__all__ = ["FileCache"]

129
src/zotero_scraper/cache/file_cache.py vendored Normal file
View File

@ -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}")

11
src/zotero_scraper/clients/__init__.py Normal file
View File

@ -0,0 +1,11 @@
"""HTTP clients module."""
from .base import BaseHTTPClient
from .downloader import XPIDownloader
from .github import GitHubClient
__all__ = [
"BaseHTTPClient",
"GitHubClient",
"XPIDownloader",
]

142
src/zotero_scraper/clients/base.py Normal file
View File

@ -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()

149
src/zotero_scraper/clients/downloader.py Normal file
View File

@ -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}")

476
src/zotero_scraper/clients/github.py Normal file
View File

@ -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

13
src/zotero_scraper/config/__init__.py Normal file
View File

@ -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",
]

113
src/zotero_scraper/config/constants.py Normal file
View File

@ -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"

77
src/zotero_scraper/config/settings.py Normal file
View File

@ -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,
)

49
src/zotero_scraper/exceptions.py Normal file
View File

@ -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

12
src/zotero_scraper/models/__init__.py Normal file
View File

@ -0,0 +1,12 @@
"""Data models module."""
from .addon import AddonInfo, AddonRelease, Author, XpiDownloadUrls
from .xpi import XpiDetail
__all__ = [
"AddonInfo",
"AddonRelease",
"Author",
"XpiDownloadUrls",
"XpiDetail",
]

218
src/zotero_scraper/models/addon.py Normal file
View File

@ -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,
)

103
src/zotero_scraper/models/xpi.py Normal file
View File

@ -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)

10
src/zotero_scraper/parsers/__init__.py Normal file
View File

@ -0,0 +1,10 @@
"""XPI parsers module."""
from .manifest import JsonManifestParser, RdfManifestParser
from .xpi_parser import XPIParser
__all__ = [
"JsonManifestParser",
"RdfManifestParser",
"XPIParser",
]

349
src/zotero_scraper/parsers/manifest.py Normal file
View File

@ -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

80
src/zotero_scraper/parsers/xpi_parser.py Normal file
View File

@ -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}")

11
src/zotero_scraper/services/__init__.py Normal file
View File

@ -0,0 +1,11 @@
"""Business services module."""
from .fallback import FallbackService
from .publisher import PublisherService
from .scraper import AddonScraper
__all__ = [
"AddonScraper",
"FallbackService",
"PublisherService",
]

185
src/zotero_scraper/services/fallback.py Normal file
View File

@ -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

117
src/zotero_scraper/services/publisher.py Normal file
View File

@ -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,
)

357
src/zotero_scraper/services/scraper.py Normal file
View File

@ -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}")

10
src/zotero_scraper/utils/__init__.py Normal file
View File

@ -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",
]

60
src/zotero_scraper/utils/logging.py Normal file
View File

@ -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}")

75
src/zotero_scraper/utils/version.py Normal file
View File

@ -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
)

1
tests/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests package."""

126
tests/conftest.py Normal file
View File

@ -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",
}
],
}

1
tests/integration/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Integration tests package."""

1
tests/unit/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Unit tests package."""

189
tests/unit/test_models.py Normal file
View File

@ -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"

149
tests/unit/test_parsers.py Normal file
View File

@ -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

71
tests/unit/test_version.py Normal file
View File

@ -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