reya
/
wani-anki
1
0
Fork 0
Code Releases Activity

Finish implementing WaniKani sync

Browse Source
main
Marissa 5 years ago
parent 545fa265ed
commit 9d9e0e3746
3 changed files with 214 additions and 67 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 1
      .gitignore
  2. 272
      addon/__init__.py
  3. 6
      addon/config.json

1
.gitignore vendored
Unescape View File

@ -1,2 +1,3 @@
venv/
.idea/
addon/meta.json

272
addon/__init__.py
Unescape View File

@ -1,12 +1,20 @@
import os.path
import re
import json
import unicodedata
import urllib.request
from anki import find
from anki.find import Finder
from anki.hooks import addHook
from aqt import mw
from anki.sound import allSounds
from anki.utils import stripHTMLMedia
from anki.utils import stripHTMLMedia, splitFields, ids2str
from aqt.qt import QAction
from collections import namedtuple
from aqt.utils import showInfo, showText
from urllib.error import HTTPError
from functools import lru_cache
from time import time
def reshow_card():
@ -132,42 +140,67 @@ def wrap_list_and_add(owner, name, items):
def new_func(*args, **kwargs):
return old(*args, **kwargs) + items
setattr(owner, name, new_func)
class Config(object):
__slots__ = ["wk_api_key", "kanji_query", "kanji_list_cache"]
__slots__ = ["should_searching_be_fast", "wk_api_key", "kanji_global_query", "kanji_individual_query",
"kanji_list_cache"]
@classmethod
def from_config(cls):
return cls.from_json(mw.addonManager.getConfig(__name__))
@classmethod
def from_json(cls, json_obj):
return Config(wk_api_key=json_obj.get("wk_api_key", ""),
kanji_query=json_obj.get("kanji_query", ""),
return Config(should_searching_be_fast=json_obj.get("should_searching_be_fast", False),
wk_api_key=json_obj.get("wk_api_key", ""),
kanji_global_query=json_obj.get("kanji_global_query", ""),
kanji_individual_query=json_obj.get("kanji_individual_query", ""),
kanji_list_cache=KanjiListCache.from_json(json_obj.get("kanji_list_cache", None)))
def __init__(self, wk_api_key="", kanji_query="", kanji_list_cache=None):
def __init__(self, should_searching_be_fast=True, wk_api_key="", kanji_global_query="", kanji_individual_query="",
kanji_list_cache=None):
self.should_searching_be_fast = should_searching_be_fast
self.wk_api_key = wk_api_key
self.kanji_query = kanji_query
self.kanji_global_query = kanji_global_query
self.kanji_individual_query = kanji_individual_query
self.kanji_list_cache = kanji_list_cache or KanjiListCache()
def save(self):
mw.addonManager.writeConfig(__name__, {
"should_searching_be_fast": self.should_searching_be_fast,
"wk_api_key": self.wk_api_key,
"kanji_query": self.kanji_query,
"kanji_global_query": self.kanji_global_query,
"kanji_individual_query": self.kanji_individual_query,
"kanji_list_cache": self.kanji_list_cache.to_json(),
})
def kanji_query(self):
return self.kanji_global_query.format(
kanji=" or ".join(
self.kanji_individual_query.format(kanji=kanji)
for kanji in self.kanji_list_cache.definitions.values()))
class KanjiListCache(object):
__slots__ = ["definitions", "last_list_etag", "last_list_update", "last_definition_etag", "last_definition_update"]
@classmethod
def from_json(cls, json_obj=None):
if not json_obj:
return
def __init__(self, definitions=None, last_list_etag="", last_list_update="", last_definition_etag="", last_definition_update=""):
return KanjiListCache(
definitions=json_obj.get("definitions", None),
last_list_etag=json_obj.get("last_list_etag", None),
last_list_update=json_obj.get("last_list_update", None),
last_definition_etag=json_obj.get("last_definition_etag", None),
last_definition_update=json_obj.get("last_definition_update", None)
)
def __init__(self, definitions=None, last_list_etag="", last_list_update="", last_definition_etag="",
last_definition_update=""):
self.definitions = dict(definitions or {})
self.last_list_etag = last_list_etag
self.last_list_update = last_list_update
@ -182,54 +215,77 @@ class KanjiListCache(object):
"last_list_etag": self.last_list_etag,
"last_list_update": self.last_list_update}
def update_unlocked_kanji(config):
url = (
r"https://api.wanikani.com/v2/assignments"
r"?started=true"
r"&subject_types=kanji")
if config.kanji_list_cache.last_list_update:
url += r"&updated_after=" + config.kanji_list_cache.last_list_update
list_etag = None
list_updated = None
ids = set()
while url:
list_request = urllib.request.Request(url)
list_request.add_header("Authorization", "Bearer {wk_api_key}".format(wk_api_key=config.wk_api_key))
list_request.add_header("Wanikani-Revision", "20170710")
if config.kanji_list_cache.last_list_etag:
list_request.add_header("If-None-Match", config.kanji_list_cache.last_list_etag)
list_etag, list_updated, updated_ids = get_updated_kanji_assignments(config)
previous_ids = set(config.kanji_list_cache.definitions.keys())
all_ids = updated_ids | previous_ids
defs_updated, defs_etag, defs_updates = get_kanji_definitions(config, all_ids, updated=UPDATED_KANJI)
missing_ids = updated_ids - set(updated_def_id for updated_def_id in defs_updates.keys())
if missing_ids:
_, _, defs_new = get_kanji_definitions(config, missing_ids, updated=NEW_KANJI)
else:
defs_new = ()
if list_updated:
config.kanji_list_cache.last_list_update = list_updated
if list_etag:
config.kanji_list_cache.last_list_etag = list_etag
if defs_updated:
config.kanji_list_cache.last_definition_update = defs_updated
if defs_etag:
config.kanji_list_cache.last_definition_etag = defs_etag
if defs_updates:
config.kanji_list_cache.definitions.update(defs_updates)
if defs_new:
config.kanji_list_cache.definitions.update(defs_new)
config.save()
def get_updated_kanji_assignments(config):
url = (
r"https://api.wanikani.com/v2/assignments"
r"?started=true"
r"&subject_types=kanji")
if config.kanji_list_cache.last_list_update:
url += r"&updated_after=" + config.kanji_list_cache.last_list_update
list_etag = None
list_updated = None
updated_ids = set()
while url:
list_request = urllib.request.Request(url)
list_request.add_header("Authorization", "Bearer {wk_api_key}".format(wk_api_key=config.wk_api_key))
list_request.add_header("Wanikani-Revision", "20170710")
if config.kanji_list_cache.last_list_etag:
list_request.add_header("If-None-Match", config.kanji_list_cache.last_list_etag)
try:
with urllib.request.urlopen(list_request) as result:
code = result.getcode()
headers = result.info()
page = json.loads(result.read())
if code != 200:
raise ValueError("Got %s from list request %s: %s" % (code, url, json.dumps(page)))
list_etag = headers["ETag"]
list_updated = page["data_updated_at"]
ids.add(item["id"] for item in page["data"])
if page["pages"]["next_page"]:
url = page["pages"]["next_page"]
else:
url = None
previous_ids = set(int(cached_id) for cached_id in config.kanji_list_cache.definitions.keys())
previous_kanji = set(config.kanji_list_cache.definitions.values())
all_ids = ids | previous_ids
defs_updated, defs_etag, defs_updates = get_updated_kanji_definitions(config, all_ids)
config.kanji_list_cache.definitions.update(defs_updates)
missing_ids = set(config.kanji_list_cache.definitions.keys()) - previous_ids
# TODO:
# Retrieve any items corresponding to ids which are not present in the cache without any cache specifiers
# Update the four cache specifiers
# Save the configuration
except HTTPError as ex:
if ex.code == 304:
return None, None, set()
raise
list_etag = headers["ETag"]
list_updated = page["data_updated_at"]
updated_ids.update(str(item["data"]["subject_id"]) for item in page["data"])
if page["pages"].get("next_url", None):
url = page["pages"]["next_url"]
else:
url = None
return list_etag, list_updated, updated_ids
NEW_KANJI = "New"
UPDATED_KANJI = "Updated"
def get_updated_kanji_definitions(config, all_ids):
def get_kanji_definitions(config, kanji_ids, updated=UPDATED_KANJI):
url = (
r"https://api.wanikani.com/v2/subjects"
r"?types=kanji"
r"&ids=%s" % (",".join(all_ids),)
r"https://api.wanikani.com/v2/subjects"
r"?types=kanji"
r"&ids=%s" % (",".join(kanji_ids),)
)
if config.kanji_list_cache.last_definition_update:
if updated != NEW_KANJI and config.kanji_list_cache.last_definition_update:
url += r"&updated_after=" + config.kanji_list_cache.last_definition_update
etag = None
updated = None
@ -238,31 +294,77 @@ def get_updated_kanji_definitions(config, all_ids):
list_request = urllib.request.Request(url)
list_request.add_header("Authorization", "Bearer {wk_api_key}".format(wk_api_key=config.wk_api_key))
list_request.add_header("Wanikani-Revision", "20170710")
if config.kanji_list_cache.last_definition_etag:
if updated != NEW_KANJI and config.kanji_list_cache.last_definition_etag:
list_request.add_header("If-None-Match", config.kanji_list_cache.last_definition_etag)
with urllib.request.urlopen(list_request) as result:
code = result.getcode()
headers = result.info()
page = json.loads(result.read())
if code != 200:
raise ValueError("Got %s from definition request %s: %s" % (code, url, json.dumps(page)))
try:
with urllib.request.urlopen(list_request) as result:
headers = result.info()
page = json.loads(result.read())
except HTTPError as ex:
if ex.code == 304:
return None, None, {}
raise
etag = headers["ETag"]
updated = page["data_updated_at"]
for item in page["data"]:
updated_definitions[item["id"]] = item["characters"]
if page["pages"]["next_page"]:
url = page["pages"]["next_page"]
item_id = str(item["id"])
characters = item["data"]["characters"]
updated_definitions[item_id] = characters
if page["pages"].get("next_url", None):
url = page["pages"]["next_url"]
else:
url = None
return updated, etag, updated_definitions
def get_new_kanji_definitions(config, new_ids):
pass
def find_unsuspendable_kanji_cards(config):
return set(mw.col.findCards(query))
def sync_wani_kani():
pass
unlocked_kanji_search_time = time()
config = Config.from_config()
update_unlocked_kanji(config=config)
unlocked_kanji_search_time = time() - unlocked_kanji_search_time
card_search_time = time()
query = config.kanji_query()
kanji_to_unsuspend = set(mw.col.findCards(query))
card_search_time = time() - card_search_time
if not kanji_to_unsuspend:
showText(
txt=(
"Retrieved {unlocked_kanji} unlocked kanji from WaniKani in {unlocked_kanji_search_time:f} seconds.\n\n"
"No cards to unsuspend in {card_search_time:f} seconds when searching {speed}.\n\nUsed query:\n{query}"
).format(
unlocked_kanji=len(config.kanji_list_cache.definitions),
unlocked_kanji_search_time=unlocked_kanji_search_time,
card_search_time=card_search_time,
speed=("quickly" if config.should_searching_be_fast else "slowly"),
query=query),
copyBtn=True)
return
mw.col.sched.unsuspendCards(kanji_to_unsuspend)
mw.reset()
card_text = []
for card_id in kanji_to_unsuspend:
card = mw.col.getCard(card_id)
note = card.note()
card_text.append("cid#{id} {kanji} (Lv. {level}, {onyomi} / {kunyomi}, {meaning})".format(
id=note.id,
kanji=note["Kanji"],
level=note["level"], onyomi=note["ONyomi"], kunyomi=note["KUNyomi"], meaning=note["Meaning"]))
showText(
txt=("Retrieved {unlocked_kanji} unlocked kanji from WaniKani in {unlocked_kanji_search_time:f} seconds.\n\n"
"Found {unsuspend_cards} card(s) to unsuspend in {card_search_time:f} seconds "
"when searching {speed}.\n\nUsed query:\n{query}\n\nUnsuspended cards:\n{card_list}").format(
unlocked_kanji=len(config.kanji_list_cache.definitions),
unlocked_kanji_search_time=unlocked_kanji_search_time,
unsuspend_cards=len(kanji_to_unsuspend),
card_search_time=card_search_time,
speed=("quickly" if config.should_searching_be_fast else "slowly"),
query=query,
card_list="\n".join(card_text)),
copyBtn=True)
new_shortcuts = (
@ -270,7 +372,6 @@ new_shortcuts = (
["Shift+m", swap_meaning_and_extra_info],
)
new_context_menu_items = (
None,
["Toggle Sound", "Shift+S", toggle_sound],
@ -284,3 +385,46 @@ wrap_list_and_add(mw.reviewer, "_contextMenu", new_context_menu_items)
action = QAction("Load WaniKani data", mw)
action.triggered.connect(sync_wani_kani)
mw.form.menuTools.addAction(action)
@lru_cache(maxsize=None)
def normalize_field_name(name):
return unicodedata.normalize("NFC", name.lower())
_findFieldButSlowly = find.Finder._findField
def _findFieldButFaster(field, val):
field = normalize_field_name(field)
val = val.replace("*", "%")
models_with_field = [
str(m['id']) for m in mw.col.models.all()
if any(normalize_field_name(f['name']) == field for f in m['flds'])]
return "n.mid in %s and field_by_model_id_and_name(n.mid, n.flds, '%s') like '%s' escape '\\'" % (
ids2str(models_with_field), field.replace("'", "''"), val.replace("'", "''"))
class FinderButFast(find.Finder):
def __init__(self, *args, **kwargs):
if Config.from_config().should_searching_be_fast:
self._findField = _findFieldButFaster
super(FinderButFast, self).__init__(*args, **kwargs)
def make_searching_fast():
models = mw.col.models
def get_field_by_model_id_and_name(id, flds, name):
m = models.get(id)
if not m:
return
for f in m['flds']:
if normalize_field_name(f['name']) == name:
return splitFields(flds)[f['ord']]
mw.col.db._db.create_function("field_by_model_id_and_name", 3, get_field_by_model_id_and_name)
addHook("profileLoaded", make_searching_fast)
find.Finder = FinderButFast

6
addon/config.json
Unescape View File

@ -1,5 +1,7 @@
{
"should_searching_be_fast": true,
"wk_api_key": "",
"kanji_query": "tag:kanji tag:writing is:suspended id:k_{kanji}",
"kanji_list_cache": ""
"kanji_global_query": "\"deck:\u66f8\u304f\u7df4\u7fd2\" \"note:Writing Kanji\" is:suspended ({kanji})",
"kanji_individual_query": "id:k_{kanji}",
"kanji_list_cache": null
}
Write Preview
Loading…
Cancel
Save