ffd392de21
* Accept iterables as inputs to backend methods * Shift add-on check to backend; use new endpoint The new endpoint will return info on a suitable branch if found, instead of returning all branches. This simplifies the frontend code, and means that you can now drop support for certain versions without it also remotely disabling the add-on for people who are running one of the excluded versions, like in https://forums.ankiweb.net/t/prevent-add-ons-from-being-disabled-remote-stealthily-surreptitiously/33427 * Bump version to 23.09 This changes Anki's version numbering system to year.month.patch, as previously mentioned on https://forums.ankiweb.net/t/use-a-different-versioning-system-semver-perhaps/20046/5 This is shaping up to be a big release, with the introduction of FSRS and image occlusion, and it seems like a good time to be finally updating the version scheme as well. AnkiWeb has been updated to understand the new format, and add-on authors will now specify version compatibility using the full version number, as can be seen here: https://ankiweb.net/shared/info/3918629684 * Shift update check to backend, and tidy up update.py * Use the shared client for sync connections too
1635 lines
53 KiB
Python
1635 lines
53 KiB
Python
# Copyright: Ankitects Pty Ltd and contributors
|
|
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import zipfile
|
|
from collections import defaultdict
|
|
from concurrent.futures import Future
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import IO, Any, Callable, Iterable, Sequence, Union
|
|
from urllib.parse import parse_qs, urlparse
|
|
from zipfile import ZipFile
|
|
|
|
import jsonschema
|
|
import markdown
|
|
from jsonschema.exceptions import ValidationError
|
|
from markdown.extensions import md_in_html
|
|
|
|
import anki
|
|
import anki.utils
|
|
import aqt
|
|
import aqt.forms
|
|
import aqt.main
|
|
from anki.collection import AddonInfo
|
|
from anki.httpclient import HttpClient
|
|
from anki.lang import without_unicode_isolation
|
|
from aqt import gui_hooks
|
|
from aqt.qt import *
|
|
from aqt.utils import (
|
|
askUser,
|
|
disable_help_button,
|
|
getFile,
|
|
is_win,
|
|
openFolder,
|
|
openLink,
|
|
restoreGeom,
|
|
restoreSplitter,
|
|
saveGeom,
|
|
saveSplitter,
|
|
send_to_trash,
|
|
show_info,
|
|
showInfo,
|
|
showText,
|
|
showWarning,
|
|
tooltip,
|
|
tr,
|
|
)
|
|
|
|
|
|
class AbortAddonImport(Exception):
|
|
"""If raised during add-on import, Anki will silently ignore this exception.
|
|
This allows you to terminate loading without an error being shown."""
|
|
|
|
|
|
@dataclass
|
|
class InstallOk:
|
|
name: str
|
|
conflicts: set[str]
|
|
compatible: bool
|
|
|
|
|
|
@dataclass
|
|
class InstallError:
|
|
errmsg: str
|
|
|
|
|
|
@dataclass
|
|
class DownloadOk:
|
|
data: bytes
|
|
filename: str
|
|
mod_time: int
|
|
min_point_version: int
|
|
max_point_version: int
|
|
branch_index: int
|
|
|
|
|
|
@dataclass
|
|
class DownloadError:
|
|
# set if result was not 200
|
|
status_code: int | None = None
|
|
# set if an exception occurred
|
|
exception: Exception | None = None
|
|
|
|
|
|
# first arg is add-on id
|
|
DownloadLogEntry = tuple[int, Union[DownloadError, InstallError, InstallOk]]
|
|
|
|
|
|
ANKIWEB_ID_RE = re.compile(r"^\d+$")
|
|
|
|
_current_version = anki.utils.int_version()
|
|
|
|
|
|
@dataclass
|
|
class AddonMeta:
|
|
dir_name: str
|
|
provided_name: str | None
|
|
enabled: bool
|
|
installed_at: int
|
|
conflicts: list[str]
|
|
min_version: int
|
|
max_version: int
|
|
branch_index: int
|
|
human_version: str | None
|
|
update_enabled: bool
|
|
homepage: str | None
|
|
|
|
def human_name(self) -> str:
|
|
return self.provided_name or self.dir_name
|
|
|
|
def ankiweb_id(self) -> int | None:
|
|
m = ANKIWEB_ID_RE.match(self.dir_name)
|
|
if m:
|
|
return int(m.group(0))
|
|
else:
|
|
return None
|
|
|
|
def compatible(self) -> bool:
|
|
min = self.min_version
|
|
if min is not None and _current_version < min:
|
|
return False
|
|
max = self.max_version
|
|
if max is not None and max < 0 and _current_version > abs(max):
|
|
return False
|
|
return True
|
|
|
|
def is_latest(self, server_update_time: int) -> bool:
|
|
return self.installed_at >= server_update_time
|
|
|
|
def page(self) -> str | None:
|
|
if self.ankiweb_id():
|
|
return f"{aqt.appShared}info/{self.dir_name}"
|
|
return self.homepage
|
|
|
|
@staticmethod
|
|
def from_json_meta(dir_name: str, json_meta: dict[str, Any]) -> AddonMeta:
|
|
return AddonMeta(
|
|
dir_name=dir_name,
|
|
provided_name=json_meta.get("name"),
|
|
enabled=not json_meta.get("disabled"),
|
|
installed_at=json_meta.get("mod", 0),
|
|
conflicts=json_meta.get("conflicts", []),
|
|
min_version=json_meta.get("min_point_version", 0) or 0,
|
|
max_version=json_meta.get("max_point_version", 0) or 0,
|
|
branch_index=json_meta.get("branch_index", 0) or 0,
|
|
human_version=json_meta.get("human_version"),
|
|
update_enabled=json_meta.get("update_enabled", True),
|
|
homepage=json_meta.get("homepage"),
|
|
)
|
|
|
|
|
|
def package_name_valid(name: str) -> bool:
|
|
# embedded /?
|
|
base = os.path.basename(name)
|
|
if base != name:
|
|
return False
|
|
# tries to escape to parent?
|
|
root = os.getcwd()
|
|
subfolder = os.path.abspath(os.path.join(root, name))
|
|
if root.startswith(subfolder):
|
|
return False
|
|
return True
|
|
|
|
|
|
# fixme: this class should not have any GUI code in it
|
|
class AddonManager:
|
|
exts: list[str] = [".ankiaddon", ".zip"]
|
|
_manifest_schema: dict = {
|
|
"type": "object",
|
|
"properties": {
|
|
# the name of the folder
|
|
"package": {"type": "string", "minLength": 1, "meta": False},
|
|
# the displayed name to the user
|
|
"name": {"type": "string", "meta": True},
|
|
# the time the add-on was last modified
|
|
"mod": {"type": "number", "meta": True},
|
|
# a list of other packages that conflict
|
|
"conflicts": {"type": "array", "items": {"type": "string"}, "meta": True},
|
|
# x for anki 2.1.x; int_version() for more recent releases
|
|
"min_point_version": {"type": "number", "meta": True},
|
|
# x for anki 2.1.x; int_version() for more recent releases
|
|
# if negative, abs(n) is the maximum version this add-on supports
|
|
# if positive, indicates version tested on, and is ignored
|
|
"max_point_version": {"type": "number", "meta": True},
|
|
# AnkiWeb sends this to indicate which branch the user downloaded.
|
|
"branch_index": {"type": "number", "meta": True},
|
|
# version string set by the add-on creator
|
|
"human_version": {"type": "string", "meta": True},
|
|
# add-on page on AnkiWeb or some other webpage
|
|
"homepage": {"type": "string", "meta": True},
|
|
},
|
|
"required": ["package", "name"],
|
|
}
|
|
|
|
def __init__(self, mw: aqt.main.AnkiQt) -> None:
|
|
self.mw = mw
|
|
self.dirty = False
|
|
f = self.mw.form
|
|
qconnect(f.actionAdd_ons.triggered, self.onAddonsDialog)
|
|
sys.path.insert(0, self.addonsFolder())
|
|
|
|
# in new code, you may want all_addon_meta() instead
|
|
def allAddons(self) -> list[str]:
|
|
l = []
|
|
for d in os.listdir(self.addonsFolder()):
|
|
path = self.addonsFolder(d)
|
|
if not os.path.exists(os.path.join(path, "__init__.py")):
|
|
continue
|
|
l.append(d)
|
|
l.sort()
|
|
if os.getenv("ANKIREVADDONS", ""):
|
|
l = list(reversed(l))
|
|
return l
|
|
|
|
def all_addon_meta(self) -> Iterable[AddonMeta]:
|
|
return map(self.addon_meta, self.allAddons())
|
|
|
|
def addonsFolder(self, module: str | None = None) -> str:
|
|
root = self.mw.pm.addonFolder()
|
|
if module is None:
|
|
return root
|
|
return os.path.join(root, module)
|
|
|
|
def loadAddons(self) -> None:
|
|
for addon in self.all_addon_meta():
|
|
if not addon.enabled:
|
|
continue
|
|
if not addon.compatible():
|
|
continue
|
|
self.dirty = True
|
|
try:
|
|
__import__(addon.dir_name)
|
|
except AbortAddonImport:
|
|
pass
|
|
except:
|
|
error = html.escape(
|
|
tr.addons_failed_to_load(
|
|
name=addon.human_name(),
|
|
traceback=traceback.format_exc(),
|
|
)
|
|
)
|
|
txt = f"<h1>{tr.qt_misc_error()}</h1><div style='white-space: pre-wrap'>{error}</div>"
|
|
showText(
|
|
txt,
|
|
type="html",
|
|
copyBtn=True,
|
|
)
|
|
|
|
def onAddonsDialog(self) -> None:
|
|
aqt.dialogs.open("AddonsDialog", self)
|
|
|
|
# Metadata
|
|
######################################################################
|
|
|
|
def addon_meta(self, dir_name: str) -> AddonMeta:
|
|
"""Get info about an installed add-on."""
|
|
json_obj = self.addonMeta(dir_name)
|
|
return AddonMeta.from_json_meta(dir_name, json_obj)
|
|
|
|
def write_addon_meta(self, addon: AddonMeta) -> None:
|
|
# preserve any unknown attributes
|
|
json_obj = self.addonMeta(addon.dir_name)
|
|
|
|
if addon.provided_name is not None:
|
|
json_obj["name"] = addon.provided_name
|
|
json_obj["disabled"] = not addon.enabled
|
|
json_obj["mod"] = addon.installed_at
|
|
json_obj["conflicts"] = addon.conflicts
|
|
json_obj["max_point_version"] = addon.max_version
|
|
json_obj["min_point_version"] = addon.min_version
|
|
json_obj["branch_index"] = addon.branch_index
|
|
if addon.human_version is not None:
|
|
json_obj["human_version"] = addon.human_version
|
|
json_obj["update_enabled"] = addon.update_enabled
|
|
|
|
self.writeAddonMeta(addon.dir_name, json_obj)
|
|
|
|
def _addonMetaPath(self, module: str) -> str:
|
|
return os.path.join(self.addonsFolder(module), "meta.json")
|
|
|
|
# in new code, use self.addon_meta() instead
|
|
def addonMeta(self, module: str) -> dict[str, Any]:
|
|
path = self._addonMetaPath(module)
|
|
try:
|
|
with open(path, encoding="utf8") as f:
|
|
return json.load(f)
|
|
except json.JSONDecodeError as e:
|
|
print(f"json error in add-on {module}:\n{e}")
|
|
return dict()
|
|
except:
|
|
# missing meta file, etc
|
|
return dict()
|
|
|
|
# in new code, use write_addon_meta() instead
|
|
def writeAddonMeta(self, module: str, meta: dict[str, Any]) -> None:
|
|
path = self._addonMetaPath(module)
|
|
with open(path, "w", encoding="utf8") as f:
|
|
json.dump(meta, f)
|
|
|
|
def toggleEnabled(self, module: str, enable: bool | None = None) -> None:
|
|
addon = self.addon_meta(module)
|
|
should_enable = enable if enable is not None else not addon.enabled
|
|
if should_enable is True:
|
|
conflicting = self._disableConflicting(module)
|
|
if conflicting:
|
|
addons = ", ".join(self.addonName(f) for f in conflicting)
|
|
showInfo(
|
|
tr.addons_the_following_addons_are_incompatible_with(
|
|
name=addon.human_name(),
|
|
found=addons,
|
|
),
|
|
textFormat="plain",
|
|
)
|
|
|
|
addon.enabled = should_enable
|
|
self.write_addon_meta(addon)
|
|
|
|
def ankiweb_addons(self) -> list[int]:
|
|
ids = []
|
|
for meta in self.all_addon_meta():
|
|
if meta.ankiweb_id() is not None:
|
|
ids.append(meta.ankiweb_id())
|
|
return ids
|
|
|
|
# Legacy helpers
|
|
######################################################################
|
|
|
|
def isEnabled(self, module: str) -> bool:
|
|
return self.addon_meta(module).enabled
|
|
|
|
def addonName(self, module: str) -> str:
|
|
return self.addon_meta(module).human_name()
|
|
|
|
def addonConflicts(self, module: str) -> list[str]:
|
|
return self.addon_meta(module).conflicts
|
|
|
|
def annotatedName(self, module: str) -> str:
|
|
meta = self.addon_meta(module)
|
|
name = meta.human_name()
|
|
if not meta.enabled:
|
|
name += f" {tr.addons_disabled()}"
|
|
return name
|
|
|
|
# Conflict resolution
|
|
######################################################################
|
|
|
|
def allAddonConflicts(self) -> dict[str, list[str]]:
|
|
all_conflicts: dict[str, list[str]] = defaultdict(list)
|
|
for addon in self.all_addon_meta():
|
|
if not addon.enabled:
|
|
continue
|
|
for other_dir in addon.conflicts:
|
|
all_conflicts[other_dir].append(addon.dir_name)
|
|
return all_conflicts
|
|
|
|
def _disableConflicting(self, module: str, conflicts: list[str] = None) -> set[str]:
|
|
if not self.isEnabled(module):
|
|
# disabled add-ons should not trigger conflict handling
|
|
return set()
|
|
|
|
conflicts = conflicts or self.addonConflicts(module)
|
|
|
|
installed = self.allAddons()
|
|
found = {d for d in conflicts if d in installed and self.isEnabled(d)}
|
|
found.update(self.allAddonConflicts().get(module, []))
|
|
|
|
for package in found:
|
|
self.toggleEnabled(package, enable=False)
|
|
|
|
return found
|
|
|
|
# Installing and deleting add-ons
|
|
######################################################################
|
|
|
|
def readManifestFile(self, zfile: ZipFile) -> dict[Any, Any]:
|
|
try:
|
|
with zfile.open("manifest.json") as f:
|
|
data = json.loads(f.read())
|
|
jsonschema.validate(data, self._manifest_schema)
|
|
# build new manifest from recognized keys
|
|
schema = self._manifest_schema["properties"]
|
|
manifest = {key: data[key] for key in data.keys() & schema.keys()}
|
|
except (KeyError, json.decoder.JSONDecodeError, ValidationError):
|
|
# raised for missing manifest, invalid json, missing/invalid keys
|
|
return {}
|
|
return manifest
|
|
|
|
def install(
|
|
self,
|
|
file: IO | str,
|
|
manifest: dict[str, Any] | None = None,
|
|
force_enable: bool = False,
|
|
) -> InstallOk | InstallError:
|
|
"""Install add-on from path or file-like object. Metadata is read
|
|
from the manifest file, with keys overridden by supplying a 'manifest'
|
|
dictionary"""
|
|
try:
|
|
zfile = ZipFile(file)
|
|
except zipfile.BadZipfile:
|
|
return InstallError(errmsg="zip")
|
|
|
|
with zfile:
|
|
file_manifest = self.readManifestFile(zfile)
|
|
if manifest:
|
|
file_manifest.update(manifest)
|
|
manifest = file_manifest
|
|
if not manifest:
|
|
return InstallError(errmsg="manifest")
|
|
package = manifest["package"]
|
|
if not package_name_valid(package):
|
|
return InstallError(errmsg="invalid package")
|
|
conflicts = manifest.get("conflicts", [])
|
|
found_conflicts = self._disableConflicting(package, conflicts)
|
|
meta = self.addonMeta(package)
|
|
gui_hooks.addon_manager_will_install_addon(self, package)
|
|
self._install(package, zfile)
|
|
gui_hooks.addon_manager_did_install_addon(self, package)
|
|
|
|
schema = self._manifest_schema["properties"]
|
|
manifest_meta = {
|
|
k: v for k, v in manifest.items() if k in schema and schema[k]["meta"]
|
|
}
|
|
meta.update(manifest_meta)
|
|
|
|
if force_enable:
|
|
meta["disabled"] = False
|
|
|
|
self.writeAddonMeta(package, meta)
|
|
|
|
meta2 = self.addon_meta(package)
|
|
|
|
return InstallOk(
|
|
name=meta["name"], conflicts=found_conflicts, compatible=meta2.compatible()
|
|
)
|
|
|
|
def _install(self, module: str, zfile: ZipFile) -> None:
|
|
# previously installed?
|
|
base = self.addonsFolder(module)
|
|
if os.path.exists(base):
|
|
self.backupUserFiles(module)
|
|
try:
|
|
self.deleteAddon(module)
|
|
except Exception:
|
|
self.restoreUserFiles(module)
|
|
raise
|
|
os.mkdir(base)
|
|
self.restoreUserFiles(module)
|
|
|
|
# extract
|
|
for n in zfile.namelist():
|
|
if n.endswith("/"):
|
|
# folder; ignore
|
|
continue
|
|
|
|
path = os.path.join(base, n)
|
|
# skip existing user files
|
|
if os.path.exists(path) and n.startswith("user_files/"):
|
|
continue
|
|
zfile.extract(n, base)
|
|
|
|
def deleteAddon(self, module: str) -> None:
|
|
send_to_trash(Path(self.addonsFolder(module)))
|
|
|
|
# Processing local add-on files
|
|
######################################################################
|
|
|
|
def processPackages(
|
|
self,
|
|
paths: list[str],
|
|
parent: QWidget | None = None,
|
|
force_enable: bool = False,
|
|
) -> tuple[list[str], list[str]]:
|
|
log = []
|
|
errs = []
|
|
|
|
self.mw.progress.start(parent=parent)
|
|
try:
|
|
for path in paths:
|
|
base = os.path.basename(path)
|
|
result = self.install(path, force_enable=force_enable)
|
|
|
|
if isinstance(result, InstallError):
|
|
errs.extend(
|
|
self._installationErrorReport(result, base, mode="local")
|
|
)
|
|
else:
|
|
log.extend(
|
|
self._installationSuccessReport(result, base, mode="local")
|
|
)
|
|
finally:
|
|
self.mw.progress.finish()
|
|
|
|
return log, errs
|
|
|
|
# Installation messaging
|
|
######################################################################
|
|
|
|
def _installationErrorReport(
|
|
self, result: InstallError, base: str, mode: str = "download"
|
|
) -> list[str]:
|
|
messages = {
|
|
"zip": tr.addons_corrupt_addon_file(),
|
|
"manifest": tr.addons_invalid_addon_manifest(),
|
|
}
|
|
|
|
msg = messages.get(result.errmsg, tr.addons_unknown_error(val=result.errmsg))
|
|
|
|
if mode == "download":
|
|
template = tr.addons_error_downloading_ids_errors(id=base, error=msg)
|
|
else:
|
|
template = tr.addons_error_installing_bases_errors(base=base, error=msg)
|
|
|
|
return [template]
|
|
|
|
def _installationSuccessReport(
|
|
self, result: InstallOk, base: str, mode: str = "download"
|
|
) -> list[str]:
|
|
name = result.name or base
|
|
if mode == "download":
|
|
template = tr.addons_downloaded_fnames(fname=name)
|
|
else:
|
|
template = tr.addons_installed_names(name=name)
|
|
|
|
strings = [template]
|
|
|
|
if result.conflicts:
|
|
strings.append(
|
|
tr.addons_the_following_conflicting_addons_were_disabled()
|
|
+ " "
|
|
+ ", ".join(self.addonName(f) for f in result.conflicts)
|
|
)
|
|
|
|
if not result.compatible:
|
|
strings.append(tr.addons_this_addon_is_not_compatible_with())
|
|
|
|
return strings
|
|
|
|
# Updating
|
|
######################################################################
|
|
|
|
def update_supported_versions(self, items: list[AddonInfo]) -> None:
|
|
"""Adjust the supported version range after an update check.
|
|
|
|
AnkiWeb will not have sent us any add-ons that don't support our
|
|
version, so this cannot disable add-ons that users are using. It
|
|
does allow the add-on author to mark an add-on as not supporting
|
|
a future release, causing the add-on to be disabled when the user
|
|
upgrades.
|
|
"""
|
|
|
|
for item in items:
|
|
addon = self.addon_meta(str(item.id))
|
|
updated = False
|
|
|
|
if addon.max_version != item.max_version:
|
|
addon.max_version = item.max_version
|
|
updated = True
|
|
if addon.min_version != item.min_version:
|
|
addon.min_version = item.min_version
|
|
updated = True
|
|
|
|
if updated:
|
|
self.write_addon_meta(addon)
|
|
|
|
def get_updated_addons(self, items: list[AddonInfo]) -> list[AddonInfo]:
|
|
"""Return ids of add-ons requiring an update."""
|
|
need_update = []
|
|
for item in items:
|
|
addon = self.addon_meta(str(item.id))
|
|
# update if server mtime is newer
|
|
if not addon.is_latest(item.modified):
|
|
need_update.append(item)
|
|
elif not addon.compatible():
|
|
# Addon is currently disabled, and a suitable branch was found on the
|
|
# server. Ignore our stored mtime (which may have been set incorrectly
|
|
# in the past) and require an update.
|
|
need_update.append(item)
|
|
|
|
return need_update
|
|
|
|
# Add-on Config
|
|
######################################################################
|
|
|
|
_configButtonActions: dict[str, Callable[[], bool | None]] = {}
|
|
_configUpdatedActions: dict[str, Callable[[Any], None]] = {}
|
|
_config_help_actions: dict[str, Callable[[], str]] = {}
|
|
|
|
def addonConfigDefaults(self, module: str) -> dict[str, Any] | None:
|
|
path = os.path.join(self.addonsFolder(module), "config.json")
|
|
try:
|
|
with open(path, encoding="utf8") as f:
|
|
return json.load(f)
|
|
except:
|
|
return None
|
|
|
|
def set_config_help_action(self, module: str, action: Callable[[], str]) -> None:
|
|
"Set a callback used to produce config help."
|
|
addon = self.addonFromModule(module)
|
|
self._config_help_actions[addon] = action
|
|
|
|
def addonConfigHelp(self, module: str) -> str:
|
|
if action := self._config_help_actions.get(module, None):
|
|
contents = action()
|
|
else:
|
|
path = os.path.join(self.addonsFolder(module), "config.md")
|
|
if os.path.exists(path):
|
|
with open(path, encoding="utf-8") as f:
|
|
contents = f.read()
|
|
else:
|
|
return ""
|
|
|
|
return markdown.markdown(contents, extensions=[md_in_html.makeExtension()])
|
|
|
|
def addonFromModule(self, module: str) -> str:
|
|
return module.split(".")[0]
|
|
|
|
def configAction(self, module: str) -> Callable[[], bool | None]:
|
|
return self._configButtonActions.get(module)
|
|
|
|
def configUpdatedAction(self, module: str) -> Callable[[Any], None]:
|
|
return self._configUpdatedActions.get(module)
|
|
|
|
# Schema
|
|
######################################################################
|
|
|
|
def _addon_schema_path(self, module: str) -> str:
|
|
return os.path.join(self.addonsFolder(module), "config.schema.json")
|
|
|
|
def _addon_schema(self, module: str) -> Any:
|
|
path = self._addon_schema_path(module)
|
|
try:
|
|
if not os.path.exists(path):
|
|
# True is a schema accepting everything
|
|
return True
|
|
with open(path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except json.decoder.JSONDecodeError as e:
|
|
print("The schema is not valid:")
|
|
print(e)
|
|
|
|
# Add-on Config API
|
|
######################################################################
|
|
|
|
def getConfig(self, module: str) -> dict[str, Any] | None:
|
|
addon = self.addonFromModule(module)
|
|
# get default config
|
|
config = self.addonConfigDefaults(addon)
|
|
if config is None:
|
|
return None
|
|
# merge in user's keys
|
|
meta = self.addonMeta(addon)
|
|
userConf = meta.get("config", {})
|
|
config.update(userConf)
|
|
return config
|
|
|
|
def setConfigAction(self, module: str, fn: Callable[[], bool | None]) -> None:
|
|
addon = self.addonFromModule(module)
|
|
self._configButtonActions[addon] = fn
|
|
|
|
def setConfigUpdatedAction(self, module: str, fn: Callable[[Any], None]) -> None:
|
|
addon = self.addonFromModule(module)
|
|
self._configUpdatedActions[addon] = fn
|
|
|
|
def writeConfig(self, module: str, conf: dict) -> None:
|
|
addon = self.addonFromModule(module)
|
|
meta = self.addonMeta(addon)
|
|
meta["config"] = conf
|
|
self.writeAddonMeta(addon, meta)
|
|
|
|
# user_files
|
|
######################################################################
|
|
|
|
def _userFilesPath(self, sid: str) -> str:
|
|
return os.path.join(self.addonsFolder(sid), "user_files")
|
|
|
|
def _userFilesBackupPath(self) -> str:
|
|
return os.path.join(self.addonsFolder(), "files_backup")
|
|
|
|
def backupUserFiles(self, sid: str) -> None:
|
|
p = self._userFilesPath(sid)
|
|
if os.path.exists(p):
|
|
os.rename(p, self._userFilesBackupPath())
|
|
|
|
def restoreUserFiles(self, sid: str) -> None:
|
|
p = self._userFilesPath(sid)
|
|
bp = self._userFilesBackupPath()
|
|
# did we back up userFiles?
|
|
if not os.path.exists(bp):
|
|
return
|
|
os.rename(bp, p)
|
|
|
|
# Web Exports
|
|
######################################################################
|
|
|
|
_webExports: dict[str, str] = {}
|
|
|
|
def setWebExports(self, module: str, pattern: str) -> None:
|
|
addon = self.addonFromModule(module)
|
|
self._webExports[addon] = pattern
|
|
|
|
def getWebExports(self, module: str) -> str:
|
|
return self._webExports.get(module)
|
|
|
|
|
|
# Add-ons Dialog
|
|
######################################################################
|
|
|
|
|
|
class AddonsDialog(QDialog):
|
|
def __init__(self, addonsManager: AddonManager) -> None:
|
|
self.mgr = addonsManager
|
|
self.mw = addonsManager.mw
|
|
self._require_restart = False
|
|
|
|
super().__init__(self.mw)
|
|
|
|
f = self.form = aqt.forms.addons.Ui_Dialog()
|
|
f.setupUi(self)
|
|
qconnect(f.getAddons.clicked, self.onGetAddons)
|
|
qconnect(f.installFromFile.clicked, self.onInstallFiles)
|
|
qconnect(f.checkForUpdates.clicked, self.check_for_updates)
|
|
qconnect(f.toggleEnabled.clicked, self.onToggleEnabled)
|
|
qconnect(f.viewPage.clicked, self.onViewPage)
|
|
qconnect(f.viewFiles.clicked, self.onViewFiles)
|
|
qconnect(f.delete_2.clicked, self.onDelete)
|
|
qconnect(f.config.clicked, self.onConfig)
|
|
qconnect(self.form.addonList.itemDoubleClicked, self.onConfig)
|
|
qconnect(self.form.addonList.currentRowChanged, self._onAddonItemSelected)
|
|
self.setWindowTitle(tr.addons_window_title())
|
|
disable_help_button(self)
|
|
self.setAcceptDrops(True)
|
|
self.redrawAddons()
|
|
restoreGeom(self, "addons")
|
|
gui_hooks.addons_dialog_will_show(self)
|
|
self.show()
|
|
|
|
def dragEnterEvent(self, event: QDragEnterEvent) -> None:
|
|
mime = event.mimeData()
|
|
if not mime.hasUrls():
|
|
return None
|
|
urls = mime.urls()
|
|
exts = self.mgr.exts
|
|
if all(any(url.toLocalFile().endswith(ext) for ext in exts) for url in urls):
|
|
event.acceptProposedAction()
|
|
|
|
def dropEvent(self, event: QDropEvent) -> None:
|
|
mime = event.mimeData()
|
|
paths = []
|
|
for url in mime.urls():
|
|
path = url.toLocalFile()
|
|
if os.path.exists(path):
|
|
paths.append(path)
|
|
self.onInstallFiles(paths)
|
|
|
|
def reject(self) -> None:
|
|
if self._require_restart:
|
|
tooltip(tr.addons_changes_will_take_effect_when_anki(), parent=self.mw)
|
|
saveGeom(self, "addons")
|
|
aqt.dialogs.markClosed("AddonsDialog")
|
|
|
|
return QDialog.reject(self)
|
|
|
|
silentlyClose = True
|
|
|
|
def name_for_addon_list(self, addon: AddonMeta) -> str:
|
|
name = addon.human_name()
|
|
|
|
if not addon.enabled:
|
|
return f"{name} {tr.addons_disabled2()}"
|
|
elif not addon.compatible():
|
|
return f"{name} {tr.addons_requires(val=self.compatible_string(addon))}"
|
|
|
|
return name
|
|
|
|
def compatible_string(self, addon: AddonMeta) -> str:
|
|
min = addon.min_version
|
|
if min is not None and min > _current_version:
|
|
return f"Anki >= 2.1.{min}"
|
|
else:
|
|
max = abs(addon.max_version)
|
|
return f"Anki <= 2.1.{max}"
|
|
|
|
def should_grey(self, addon: AddonMeta) -> bool:
|
|
return not addon.enabled or not addon.compatible()
|
|
|
|
def redrawAddons(
|
|
self,
|
|
) -> None:
|
|
addonList = self.form.addonList
|
|
mgr = self.mgr
|
|
|
|
self.addons = list(mgr.all_addon_meta())
|
|
self.addons.sort(key=lambda a: a.human_name().lower())
|
|
self.addons.sort(key=self.should_grey)
|
|
|
|
selected = set(self.selectedAddons())
|
|
addonList.clear()
|
|
for addon in self.addons:
|
|
name = self.name_for_addon_list(addon)
|
|
item = QListWidgetItem(name, addonList)
|
|
if self.should_grey(addon):
|
|
item.setForeground(Qt.GlobalColor.gray)
|
|
if addon.dir_name in selected:
|
|
item.setSelected(True)
|
|
|
|
addonList.reset()
|
|
|
|
def _onAddonItemSelected(self, row_int: int) -> None:
|
|
try:
|
|
addon = self.addons[row_int]
|
|
except IndexError:
|
|
return
|
|
self.form.viewPage.setEnabled(addon.page() is not None)
|
|
self.form.config.setEnabled(
|
|
bool(
|
|
self.mgr.getConfig(addon.dir_name)
|
|
or self.mgr.configAction(addon.dir_name)
|
|
)
|
|
)
|
|
gui_hooks.addons_dialog_did_change_selected_addon(self, addon)
|
|
return
|
|
|
|
def selectedAddons(self) -> list[str]:
|
|
idxs = [x.row() for x in self.form.addonList.selectedIndexes()]
|
|
return [self.addons[idx].dir_name for idx in idxs]
|
|
|
|
def onlyOneSelected(self) -> str | None:
|
|
dirs = self.selectedAddons()
|
|
if len(dirs) != 1:
|
|
show_info(tr.addons_please_select_a_single_addon_first())
|
|
return None
|
|
return dirs[0]
|
|
|
|
def selected_addon_meta(self) -> AddonMeta | None:
|
|
idxs = [x.row() for x in self.form.addonList.selectedIndexes()]
|
|
if len(idxs) != 1:
|
|
show_info(tr.addons_please_select_a_single_addon_first())
|
|
return None
|
|
return self.addons[idxs[0]]
|
|
|
|
def onToggleEnabled(self) -> None:
|
|
for module in self.selectedAddons():
|
|
self.mgr.toggleEnabled(module)
|
|
self._require_restart = True
|
|
self.redrawAddons()
|
|
|
|
def onViewPage(self) -> None:
|
|
addon = self.selected_addon_meta()
|
|
if not addon:
|
|
return
|
|
if page := addon.page():
|
|
openLink(page)
|
|
|
|
def onViewFiles(self) -> None:
|
|
# if nothing selected, open top level folder
|
|
selected = self.selectedAddons()
|
|
if not selected:
|
|
openFolder(self.mgr.addonsFolder())
|
|
return
|
|
|
|
# otherwise require a single selection
|
|
addon = self.onlyOneSelected()
|
|
if not addon:
|
|
return
|
|
path = self.mgr.addonsFolder(addon)
|
|
openFolder(path)
|
|
|
|
def onDelete(self) -> None:
|
|
selected = self.selectedAddons()
|
|
if not selected:
|
|
return
|
|
if not askUser(tr.addons_delete_the_numd_selected_addon(count=len(selected))):
|
|
return
|
|
gui_hooks.addons_dialog_will_delete_addons(self, selected)
|
|
try:
|
|
for module in selected:
|
|
# doing this before deleting, as `enabled` is always True afterwards
|
|
if self.mgr.addon_meta(module).enabled:
|
|
self._require_restart = True
|
|
self.mgr.deleteAddon(module)
|
|
except OSError as e:
|
|
showWarning(
|
|
tr.addons_unable_to_update_or_delete_addon(val=str(e)),
|
|
textFormat="plain",
|
|
)
|
|
self.form.addonList.clearSelection()
|
|
self.redrawAddons()
|
|
|
|
def onGetAddons(self) -> None:
|
|
obj = GetAddons(self)
|
|
if obj.ids:
|
|
download_addons(
|
|
self, self.mgr, obj.ids, self.after_downloading, force_enable=True
|
|
)
|
|
|
|
def after_downloading(self, log: list[DownloadLogEntry]) -> None:
|
|
self.redrawAddons()
|
|
if log:
|
|
show_log_to_user(self, log)
|
|
else:
|
|
tooltip(tr.addons_no_updates_available())
|
|
|
|
def onInstallFiles(self, paths: list[str] | None = None) -> bool | None:
|
|
if not paths:
|
|
filter = f"{tr.addons_packaged_anki_addon()} " + "({})".format(
|
|
" ".join(f"*{ext}" for ext in self.mgr.exts)
|
|
)
|
|
paths_ = getFile(
|
|
self, tr.addons_install_addons(), None, filter, key="addons", multi=True
|
|
)
|
|
paths = paths_ # type: ignore
|
|
if not paths:
|
|
return False
|
|
|
|
installAddonPackages(self.mgr, paths, parent=self, force_enable=True)
|
|
|
|
self.redrawAddons()
|
|
return None
|
|
|
|
def check_for_updates(self) -> None:
|
|
tooltip(tr.addons_checking())
|
|
check_and_prompt_for_updates(self, self.mgr, self.after_downloading)
|
|
|
|
def onConfig(self) -> None:
|
|
addon = self.onlyOneSelected()
|
|
if not addon:
|
|
return
|
|
|
|
# does add-on manage its own config?
|
|
act = self.mgr.configAction(addon)
|
|
if act:
|
|
ret = act()
|
|
if ret is not False:
|
|
return
|
|
|
|
conf = self.mgr.getConfig(addon)
|
|
if conf is None:
|
|
showInfo(tr.addons_addon_has_no_configuration())
|
|
return
|
|
|
|
ConfigEditor(self, addon, conf)
|
|
|
|
|
|
# Fetching Add-ons
|
|
######################################################################
|
|
|
|
|
|
class GetAddons(QDialog):
|
|
def __init__(self, dlg: AddonsDialog) -> None:
|
|
QDialog.__init__(self, dlg)
|
|
self.addonsDlg = dlg
|
|
self.mgr = dlg.mgr
|
|
self.mw = self.mgr.mw
|
|
self.ids: list[int] = []
|
|
self.form = aqt.forms.getaddons.Ui_Dialog()
|
|
self.form.setupUi(self)
|
|
b = self.form.buttonBox.addButton(
|
|
tr.addons_browse_addons(), QDialogButtonBox.ButtonRole.ActionRole
|
|
)
|
|
qconnect(b.clicked, self.onBrowse)
|
|
disable_help_button(self)
|
|
restoreGeom(self, "getaddons", adjustSize=True)
|
|
self.exec()
|
|
saveGeom(self, "getaddons")
|
|
|
|
def onBrowse(self) -> None:
|
|
openLink(f"{aqt.appShared}addons/2.1")
|
|
|
|
def accept(self) -> None:
|
|
# get codes
|
|
try:
|
|
ids = [int(n) for n in self.form.code.text().split()]
|
|
except ValueError:
|
|
showWarning(tr.addons_invalid_code())
|
|
return
|
|
|
|
self.ids = ids
|
|
QDialog.accept(self)
|
|
|
|
|
|
# Downloading
|
|
######################################################################
|
|
|
|
|
|
def download_addon(client: HttpClient, id: int) -> DownloadOk | DownloadError:
|
|
"Fetch a single add-on from AnkiWeb."
|
|
try:
|
|
resp = client.get(f"{aqt.appShared}download/{id}?v=2.1&p={_current_version}")
|
|
if resp.status_code != 200:
|
|
return DownloadError(status_code=resp.status_code)
|
|
|
|
data = client.stream_content(resp)
|
|
|
|
fname = re.match(
|
|
"attachment; filename=(.+)", resp.headers["content-disposition"]
|
|
).group(1)
|
|
|
|
meta = extract_meta_from_download_url(resp.url)
|
|
|
|
return DownloadOk(
|
|
data=data,
|
|
filename=fname,
|
|
mod_time=meta.mod_time,
|
|
min_point_version=meta.min_point_version,
|
|
max_point_version=meta.max_point_version,
|
|
branch_index=meta.branch_index,
|
|
)
|
|
except Exception as e:
|
|
return DownloadError(exception=e)
|
|
|
|
|
|
@dataclass
|
|
class ExtractedDownloadMeta:
|
|
mod_time: int
|
|
min_point_version: int
|
|
max_point_version: int
|
|
branch_index: int
|
|
|
|
|
|
def extract_meta_from_download_url(url: str) -> ExtractedDownloadMeta:
|
|
urlobj = urlparse(url)
|
|
query = parse_qs(urlobj.query)
|
|
|
|
meta = ExtractedDownloadMeta(
|
|
mod_time=int(query.get("t")[0]),
|
|
min_point_version=int(query.get("minpt")[0]),
|
|
max_point_version=int(query.get("maxpt")[0]),
|
|
branch_index=int(query.get("bidx")[0]),
|
|
)
|
|
|
|
return meta
|
|
|
|
|
|
def download_log_to_html(log: list[DownloadLogEntry]) -> str:
|
|
return "<br>".join(map(describe_log_entry, log))
|
|
|
|
|
|
def describe_log_entry(id_and_entry: DownloadLogEntry) -> str:
|
|
(id, entry) = id_and_entry
|
|
buf = f"{id}: "
|
|
|
|
if isinstance(entry, DownloadError):
|
|
if entry.status_code is not None:
|
|
if entry.status_code in (403, 404):
|
|
buf += tr.addons_invalid_code_or_addon_not_available()
|
|
else:
|
|
buf += tr.qt_misc_unexpected_response_code(val=entry.status_code)
|
|
else:
|
|
buf += (
|
|
tr.addons_please_check_your_internet_connection()
|
|
+ "\n\n"
|
|
+ str(entry.exception)
|
|
)
|
|
elif isinstance(entry, InstallError):
|
|
buf += entry.errmsg
|
|
else:
|
|
buf += tr.addons_installed_successfully()
|
|
|
|
return buf
|
|
|
|
|
|
def download_encountered_problem(log: list[DownloadLogEntry]) -> bool:
|
|
return any(not isinstance(e[1], InstallOk) for e in log)
|
|
|
|
|
|
def download_and_install_addon(
|
|
mgr: AddonManager, client: HttpClient, id: int, force_enable: bool = False
|
|
) -> DownloadLogEntry:
|
|
"Download and install a single add-on."
|
|
result = download_addon(client, id)
|
|
if isinstance(result, DownloadError):
|
|
return (id, result)
|
|
|
|
fname = result.filename.replace("_", " ")
|
|
name = os.path.splitext(fname)[0].strip()
|
|
if not name:
|
|
name = str(id)
|
|
|
|
manifest = dict(
|
|
package=str(id),
|
|
name=name,
|
|
mod=result.mod_time,
|
|
min_point_version=result.min_point_version,
|
|
max_point_version=result.max_point_version,
|
|
branch_index=result.branch_index,
|
|
)
|
|
|
|
result2 = mgr.install(
|
|
io.BytesIO(result.data), manifest=manifest, force_enable=force_enable
|
|
)
|
|
|
|
return (id, result2)
|
|
|
|
|
|
class DownloaderInstaller(QObject):
|
|
progressSignal = pyqtSignal(int, int)
|
|
|
|
def __init__(self, parent: QWidget, mgr: AddonManager, client: HttpClient) -> None:
|
|
QObject.__init__(self, parent)
|
|
self.mgr = mgr
|
|
self.client = client
|
|
qconnect(self.progressSignal, self._progress_callback)
|
|
|
|
def bg_thread_progress(up: int, down: int) -> None:
|
|
self.progressSignal.emit(up, down) # type: ignore
|
|
|
|
self.client.progress_hook = bg_thread_progress
|
|
|
|
def download(
|
|
self,
|
|
ids: list[int],
|
|
on_done: Callable[[list[DownloadLogEntry]], None],
|
|
force_enable: bool = False,
|
|
) -> None:
|
|
self.ids = ids
|
|
self.log: list[DownloadLogEntry] = []
|
|
|
|
self.dl_bytes = 0
|
|
self.last_tooltip = 0
|
|
|
|
self.on_done = on_done
|
|
|
|
parent = self.parent()
|
|
assert isinstance(parent, QWidget)
|
|
self.mgr.mw.progress.start(immediate=True, parent=parent)
|
|
self.mgr.mw.taskman.run_in_background(
|
|
lambda: self._download_all(force_enable), self._download_done
|
|
)
|
|
|
|
def _progress_callback(self, up: int, down: int) -> None:
|
|
self.dl_bytes += down
|
|
self.mgr.mw.progress.update(
|
|
label=tr.addons_downloading_adbd_kb02fkb(
|
|
part=len(self.log) + 1,
|
|
total=len(self.ids),
|
|
kilobytes=self.dl_bytes // 1024,
|
|
)
|
|
)
|
|
|
|
def _download_all(self, force_enable: bool = False) -> None:
|
|
for id in self.ids:
|
|
self.log.append(
|
|
download_and_install_addon(
|
|
self.mgr, self.client, id, force_enable=force_enable
|
|
)
|
|
)
|
|
|
|
def _download_done(self, future: Future) -> None:
|
|
self.mgr.mw.progress.finish()
|
|
future.result()
|
|
# qt gets confused if on_done() opens new windows while the progress
|
|
# modal is still cleaning up
|
|
self.mgr.mw.progress.single_shot(50, lambda: self.on_done(self.log))
|
|
|
|
|
|
def show_log_to_user(parent: QWidget, log: list[DownloadLogEntry]) -> None:
|
|
have_problem = download_encountered_problem(log)
|
|
|
|
if have_problem:
|
|
text = tr.addons_one_or_more_errors_occurred()
|
|
else:
|
|
text = tr.addons_download_complete_please_restart_anki_to()
|
|
text += f"<br><br>{download_log_to_html(log)}"
|
|
|
|
if have_problem:
|
|
showWarning(text, textFormat="rich", parent=parent)
|
|
else:
|
|
showInfo(text, parent=parent)
|
|
|
|
|
|
def download_addons(
|
|
parent: QWidget,
|
|
mgr: AddonManager,
|
|
ids: list[int],
|
|
on_done: Callable[[list[DownloadLogEntry]], None],
|
|
client: HttpClient | None = None,
|
|
force_enable: bool = False,
|
|
) -> None:
|
|
if client is None:
|
|
client = HttpClient()
|
|
downloader = DownloaderInstaller(parent, mgr, client)
|
|
downloader.download(ids, on_done=on_done, force_enable=force_enable)
|
|
|
|
|
|
# Update checking
|
|
######################################################################
|
|
|
|
|
|
class ChooseAddonsToUpdateList(QListWidget):
|
|
ADDON_ID_ROLE = 101
|
|
|
|
def __init__(
|
|
self,
|
|
parent: QWidget,
|
|
mgr: AddonManager,
|
|
updated_addons: list[AddonInfo],
|
|
) -> None:
|
|
QListWidget.__init__(self, parent)
|
|
self.mgr = mgr
|
|
self.updated_addons = sorted(updated_addons, key=lambda addon: addon.modified)
|
|
self.ignore_check_evt = False
|
|
self.setup()
|
|
self.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
|
|
qconnect(self.itemClicked, self.on_click)
|
|
qconnect(self.itemChanged, self.on_check)
|
|
qconnect(self.itemDoubleClicked, self.on_double_click)
|
|
qconnect(self.customContextMenuRequested, self.on_context_menu)
|
|
|
|
def setup(self) -> None:
|
|
header_item = QListWidgetItem(tr.addons_choose_update_update_all(), self)
|
|
header_item.setFlags(
|
|
Qt.ItemFlag(Qt.ItemFlag.ItemIsUserCheckable | Qt.ItemFlag.ItemIsEnabled)
|
|
)
|
|
self.header_item = header_item
|
|
for update_info in self.updated_addons:
|
|
addon_id = update_info.id
|
|
addon_meta = self.mgr.addon_meta(str(addon_id))
|
|
update_enabled = addon_meta.update_enabled
|
|
addon_name = addon_meta.human_name()
|
|
update_timestamp = update_info.modified
|
|
update_time = datetime.fromtimestamp(update_timestamp)
|
|
|
|
addon_label = f"{update_time:%Y-%m-%d} {addon_name}"
|
|
item = QListWidgetItem(addon_label, self)
|
|
# Not user checkable because it overlaps with itemClicked signal
|
|
item.setFlags(Qt.ItemFlag(Qt.ItemFlag.ItemIsEnabled))
|
|
if update_enabled:
|
|
item.setCheckState(Qt.CheckState.Checked)
|
|
else:
|
|
item.setCheckState(Qt.CheckState.Unchecked)
|
|
item.setData(self.ADDON_ID_ROLE, addon_id)
|
|
self.refresh_header_check_state()
|
|
|
|
def bool_to_check(self, check_bool: bool) -> Qt.CheckState:
|
|
if check_bool:
|
|
return Qt.CheckState.Checked
|
|
else:
|
|
return Qt.CheckState.Unchecked
|
|
|
|
def checked(self, item: QListWidgetItem) -> bool:
|
|
return item.checkState() == Qt.CheckState.Checked
|
|
|
|
def on_click(self, item: QListWidgetItem) -> None:
|
|
if item == self.header_item:
|
|
return
|
|
checked = self.checked(item)
|
|
self.check_item(item, self.bool_to_check(not checked))
|
|
self.refresh_header_check_state()
|
|
|
|
def on_check(self, item: QListWidgetItem) -> None:
|
|
if self.ignore_check_evt:
|
|
return
|
|
if item == self.header_item:
|
|
self.header_checked(item.checkState())
|
|
|
|
def on_double_click(self, item: QListWidgetItem) -> None:
|
|
if item == self.header_item:
|
|
checked = self.checked(item)
|
|
self.check_item(self.header_item, self.bool_to_check(not checked))
|
|
self.header_checked(self.bool_to_check(not checked))
|
|
|
|
def on_context_menu(self, point: QPoint) -> None:
|
|
item = self.itemAt(point)
|
|
addon_id = item.data(self.ADDON_ID_ROLE)
|
|
m = QMenu()
|
|
a = m.addAction(tr.addons_view_addon_page())
|
|
qconnect(a.triggered, lambda _: openLink(f"{aqt.appShared}info/{addon_id}"))
|
|
m.exec(QCursor.pos())
|
|
|
|
def check_item(self, item: QListWidgetItem, check: Qt.CheckState) -> None:
|
|
"call item.setCheckState without triggering on_check"
|
|
self.ignore_check_evt = True
|
|
item.setCheckState(check)
|
|
self.ignore_check_evt = False
|
|
|
|
def header_checked(self, check: Qt.CheckState) -> None:
|
|
for i in range(1, self.count()):
|
|
self.check_item(self.item(i), check)
|
|
|
|
def refresh_header_check_state(self) -> None:
|
|
for i in range(1, self.count()):
|
|
item = self.item(i)
|
|
if not self.checked(item):
|
|
self.check_item(self.header_item, Qt.CheckState.Unchecked)
|
|
return
|
|
self.check_item(self.header_item, Qt.CheckState.Checked)
|
|
|
|
def get_selected_addon_ids(self) -> list[int]:
|
|
addon_ids = []
|
|
for i in range(1, self.count()):
|
|
item = self.item(i)
|
|
if self.checked(item):
|
|
addon_id = item.data(self.ADDON_ID_ROLE)
|
|
addon_ids.append(addon_id)
|
|
return addon_ids
|
|
|
|
def save_check_state(self) -> None:
|
|
for i in range(1, self.count()):
|
|
item = self.item(i)
|
|
addon_id = item.data(self.ADDON_ID_ROLE)
|
|
addon_meta = self.mgr.addon_meta(str(addon_id))
|
|
addon_meta.update_enabled = self.checked(item)
|
|
self.mgr.write_addon_meta(addon_meta)
|
|
|
|
|
|
class ChooseAddonsToUpdateDialog(QDialog):
|
|
def __init__(
|
|
self, parent: QWidget, mgr: AddonManager, updated_addons: list[AddonInfo]
|
|
) -> None:
|
|
QDialog.__init__(self, parent)
|
|
self.setWindowTitle(tr.addons_choose_update_window_title())
|
|
self.setWindowModality(Qt.WindowModality.WindowModal)
|
|
self.mgr = mgr
|
|
self.updated_addons = updated_addons
|
|
self.setup()
|
|
restoreGeom(self, "addonsChooseUpdate")
|
|
|
|
def setup(self) -> None:
|
|
layout = QVBoxLayout()
|
|
label = QLabel(tr.addons_the_following_addons_have_updates_available())
|
|
layout.addWidget(label)
|
|
addons_list_widget = ChooseAddonsToUpdateList(
|
|
self, self.mgr, self.updated_addons
|
|
)
|
|
layout.addWidget(addons_list_widget)
|
|
self.addons_list_widget = addons_list_widget
|
|
|
|
button_box = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel) # type: ignore
|
|
qconnect(
|
|
button_box.button(QDialogButtonBox.StandardButton.Ok).clicked, self.accept
|
|
)
|
|
qconnect(
|
|
button_box.button(QDialogButtonBox.StandardButton.Cancel).clicked,
|
|
self.reject,
|
|
)
|
|
layout.addWidget(button_box)
|
|
self.setLayout(layout)
|
|
|
|
def ask(self) -> list[int]:
|
|
"Returns a list of selected addons' ids"
|
|
ret = self.exec()
|
|
saveGeom(self, "addonsChooseUpdate")
|
|
self.addons_list_widget.save_check_state()
|
|
if ret == QDialog.DialogCode.Accepted:
|
|
return self.addons_list_widget.get_selected_addon_ids()
|
|
else:
|
|
return []
|
|
|
|
|
|
def fetch_update_info(ids: list[int]) -> list[AddonInfo]:
|
|
"""Fetch update info from AnkiWeb in one or more batches."""
|
|
all_info: list[AddonInfo] = []
|
|
|
|
while ids:
|
|
# get another chunk
|
|
chunk = ids[:25]
|
|
del ids[:25]
|
|
|
|
batch_results = _fetch_update_info_batch(chunk)
|
|
all_info.extend(batch_results)
|
|
|
|
return all_info
|
|
|
|
|
|
def _fetch_update_info_batch(chunk: Iterable[int]) -> Sequence[AddonInfo]:
|
|
return aqt.mw.backend.get_addon_info(
|
|
client_version=_current_version, addon_ids=chunk
|
|
)
|
|
|
|
|
|
def check_and_prompt_for_updates(
|
|
parent: QWidget,
|
|
mgr: AddonManager,
|
|
on_done: Callable[[list[DownloadLogEntry]], None],
|
|
requested_by_user: bool = True,
|
|
) -> None:
|
|
def on_updates_received(items: list[AddonInfo]) -> None:
|
|
handle_update_info(parent, mgr, items, on_done, requested_by_user)
|
|
|
|
check_for_updates(mgr, on_updates_received)
|
|
|
|
|
|
def check_for_updates(
|
|
mgr: AddonManager, on_done: Callable[[list[AddonInfo]], None]
|
|
) -> None:
|
|
def check() -> list[AddonInfo]:
|
|
return fetch_update_info(mgr.ankiweb_addons())
|
|
|
|
def update_info_received(future: Future) -> None:
|
|
# if syncing/in profile screen, defer message delivery
|
|
if not mgr.mw.col:
|
|
mgr.mw.progress.single_shot(
|
|
1000,
|
|
lambda: update_info_received(future),
|
|
False,
|
|
)
|
|
return
|
|
|
|
if future.exception():
|
|
# swallow network errors
|
|
print(str(future.exception()))
|
|
result = []
|
|
else:
|
|
result = future.result()
|
|
|
|
on_done(result)
|
|
|
|
mgr.mw.taskman.run_in_background(check, update_info_received)
|
|
|
|
|
|
def handle_update_info(
|
|
parent: QWidget,
|
|
mgr: AddonManager,
|
|
items: list[AddonInfo],
|
|
on_done: Callable[[list[DownloadLogEntry]], None],
|
|
requested_by_user: bool = True,
|
|
) -> None:
|
|
mgr.update_supported_versions(items)
|
|
updated_addons = mgr.get_updated_addons(items)
|
|
|
|
if not updated_addons:
|
|
on_done([])
|
|
return
|
|
|
|
prompt_to_update(parent, mgr, updated_addons, on_done, requested_by_user)
|
|
|
|
|
|
def prompt_to_update(
|
|
parent: QWidget,
|
|
mgr: AddonManager,
|
|
updated_addons: list[AddonInfo],
|
|
on_done: Callable[[list[DownloadLogEntry]], None],
|
|
requested_by_user: bool = True,
|
|
) -> None:
|
|
client = HttpClient()
|
|
if not requested_by_user:
|
|
prompt_update = False
|
|
for addon in updated_addons:
|
|
if mgr.addon_meta(str(addon.id)).update_enabled:
|
|
prompt_update = True
|
|
if not prompt_update:
|
|
return
|
|
|
|
ids = ChooseAddonsToUpdateDialog(parent, mgr, updated_addons).ask()
|
|
if not ids:
|
|
return
|
|
download_addons(parent, mgr, ids, on_done, client)
|
|
|
|
|
|
# Editing config
|
|
######################################################################
|
|
|
|
|
|
class ConfigEditor(QDialog):
|
|
def __init__(self, dlg: AddonsDialog, addon: str, conf: dict) -> None:
|
|
super().__init__(dlg)
|
|
self.addon = addon
|
|
self.conf = conf
|
|
self.mgr = dlg.mgr
|
|
self.form = aqt.forms.addonconf.Ui_Dialog()
|
|
self.form.setupUi(self)
|
|
restore = self.form.buttonBox.button(
|
|
QDialogButtonBox.StandardButton.RestoreDefaults
|
|
)
|
|
qconnect(restore.clicked, self.onRestoreDefaults)
|
|
ok = self.form.buttonBox.button(QDialogButtonBox.StandardButton.Ok)
|
|
ok.setShortcut(QKeySequence("Ctrl+Return"))
|
|
self.setupFonts()
|
|
self.updateHelp()
|
|
self.updateText(self.conf)
|
|
restoreGeom(self, "addonconf")
|
|
self.form.splitter.setSizes([2 * self.width() // 3, self.width() // 3])
|
|
restoreSplitter(self.form.splitter, "addonconf")
|
|
self.setWindowTitle(
|
|
without_unicode_isolation(
|
|
tr.addons_config_window_title(
|
|
name=self.mgr.addon_meta(addon).human_name(),
|
|
)
|
|
)
|
|
)
|
|
disable_help_button(self)
|
|
self.show()
|
|
|
|
def onRestoreDefaults(self) -> None:
|
|
default_conf = self.mgr.addonConfigDefaults(self.addon)
|
|
self.updateText(default_conf)
|
|
tooltip(tr.addons_restored_defaults(), parent=self)
|
|
|
|
def setupFonts(self) -> None:
|
|
font_mono = QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)
|
|
font_mono.setPointSize(font_mono.pointSize() + 1)
|
|
self.form.editor.setFont(font_mono)
|
|
|
|
def updateHelp(self) -> None:
|
|
txt = self.mgr.addonConfigHelp(self.addon)
|
|
if txt:
|
|
self.form.help.stdHtml(txt, js=[], css=["css/addonconf.css"], context=self)
|
|
else:
|
|
self.form.help.setVisible(False)
|
|
|
|
def updateText(self, conf: dict[str, Any]) -> None:
|
|
text = json.dumps(
|
|
conf,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
indent=4,
|
|
separators=(",", ": "),
|
|
)
|
|
text = gui_hooks.addon_config_editor_will_display_json(text)
|
|
self.form.editor.setPlainText(text)
|
|
if is_mac:
|
|
self.form.editor.repaint()
|
|
|
|
def onClose(self) -> None:
|
|
saveGeom(self, "addonconf")
|
|
saveSplitter(self.form.splitter, "addonconf")
|
|
|
|
def reject(self) -> None:
|
|
self.onClose()
|
|
super().reject()
|
|
|
|
def accept(self) -> None:
|
|
txt = self.form.editor.toPlainText()
|
|
txt = gui_hooks.addon_config_editor_will_update_json(txt, self.addon)
|
|
try:
|
|
new_conf = json.loads(txt)
|
|
jsonschema.validate(new_conf, self.mgr._addon_schema(self.addon))
|
|
except ValidationError as e:
|
|
# The user did edit the configuration and entered a value
|
|
# which can not be interpreted.
|
|
schema = e.schema
|
|
erroneous_conf = new_conf
|
|
for link in e.path:
|
|
erroneous_conf = erroneous_conf[link]
|
|
path = "/".join(str(path) for path in e.path)
|
|
if "error_msg" in schema:
|
|
msg = schema["error_msg"].format(
|
|
problem=e.message,
|
|
path=path,
|
|
schema=str(schema),
|
|
erroneous_conf=erroneous_conf,
|
|
)
|
|
else:
|
|
msg = tr.addons_config_validation_error(
|
|
problem=e.message,
|
|
path=path,
|
|
schema=str(schema),
|
|
)
|
|
showInfo(msg)
|
|
return
|
|
except Exception as e:
|
|
showInfo(f"{tr.addons_invalid_configuration()} {repr(e)}")
|
|
return
|
|
|
|
if not isinstance(new_conf, dict):
|
|
showInfo(tr.addons_invalid_configuration_top_level_object_must())
|
|
return
|
|
|
|
if new_conf != self.conf:
|
|
self.mgr.writeConfig(self.addon, new_conf)
|
|
# does the add-on define an action to be fired?
|
|
act = self.mgr.configUpdatedAction(self.addon)
|
|
if act:
|
|
act(new_conf)
|
|
|
|
self.onClose()
|
|
super().accept()
|
|
|
|
|
|
# .ankiaddon installation wizard
|
|
######################################################################
|
|
|
|
|
|
def installAddonPackages(
|
|
addonsManager: AddonManager,
|
|
paths: list[str],
|
|
parent: QWidget | None = None,
|
|
warn: bool = False,
|
|
strictly_modal: bool = False,
|
|
advise_restart: bool = False,
|
|
force_enable: bool = False,
|
|
) -> bool:
|
|
if warn:
|
|
names = ",<br>".join(f"<b>{os.path.basename(p)}</b>" for p in paths)
|
|
q = tr.addons_important_as_addons_are_programs_downloaded() % dict(names=names)
|
|
if (
|
|
not showInfo(
|
|
q,
|
|
parent=parent,
|
|
title=tr.addons_install_anki_addon(),
|
|
type="warning",
|
|
customBtns=[
|
|
QMessageBox.StandardButton.No,
|
|
QMessageBox.StandardButton.Yes,
|
|
],
|
|
)
|
|
== QMessageBox.StandardButton.Yes
|
|
):
|
|
return False
|
|
|
|
log, errs = addonsManager.processPackages(
|
|
paths, parent=parent, force_enable=force_enable
|
|
)
|
|
|
|
if log:
|
|
log_html = "<br>".join(log)
|
|
if advise_restart:
|
|
log_html += f"<br><br>{tr.addons_please_restart_anki_to_complete_the()}"
|
|
if len(log) == 1 and not strictly_modal:
|
|
tooltip(log_html, parent=parent)
|
|
else:
|
|
showInfo(
|
|
log_html,
|
|
parent=parent,
|
|
textFormat="rich",
|
|
title=tr.addons_installation_complete(),
|
|
)
|
|
if errs:
|
|
msg = tr.addons_please_report_this_to_the_respective()
|
|
showWarning(
|
|
"<br><br>".join(errs + [msg]),
|
|
parent=parent,
|
|
textFormat="rich",
|
|
title=tr.addons_addon_installation_error(),
|
|
)
|
|
|
|
return not errs
|