#!/usr/bin/env python3.13 """Hold Loudwalk's Anki export to Anki itself. Loudwalk tells an Anki user "try it, and if it is not for you, go back with everything you did here". The only honest proof of that sentence is Anki's own reading of the file, so this script runs the whole trip with the Anki installed on this Mac: 1. Anki builds a small but awkward collection (FSRS on, two options presets, sub-decks, a filtered deck, tags, a flag, suspended cards, pictures and sounds, cards in every state, real answers given through Anki's scheduler) and exports it the way a user would: File > Export, as a .colpkg - in the current format, in the legacy one, and as an .apkg. 2. Loudwalk imports it into a throwaway collection and things happen there: reviews over several days, a lapse, a correction, an undo, cards written, edited, moved, retired, a deck renamed, a picture attached. 3. Loudwalk exports. Anki imports the .colpkg into an empty profile folder, and every card is read back through Anki and compared with what Loudwalk's own store says it should be: deck, state, due date, interval, memory state, counters, fields. Cards nobody touched in Loudwalk must come back byte for byte, and so must every review row Anki gave us. Anki's Check Database must find nothing to repair, and Anki must be able to schedule the cards. 4. The .apkg is merged twice - into an empty collection, and into the collection the cards came from - and what Anki did with it is reported, because that decides what Loudwalk may promise about it. It works under a temporary directory with LOUDWALK_HOME pointed into it, never opens an Anki profile, and checks that Loudwalk reports "scratch": true before going on. No Anki code is linked into the app: Anki is AGPL, so it is only ever run, as a black box. PYTHONPATH=/Applications/Anki.app/Contents/Resources/app_packages \ python3.13 tools/anki_export_check.py [--keep] [--legacy] python3.13 tools/anki_export_check.py verify The second form asks the same questions of a real collection's export (`./ankiwalk collection export out.colpkg --expect expectations.json`), reading all three files and writing to none. Checked against Anki 26.08.1. Exit status 0 means every check passed. """ import hashlib import json import os import shutil import subprocess import sys import tempfile import time from datetime import datetime, timezone from anki.buildinfo import version as anki_version from anki.collection import Collection from anki.generic_pb2 import Empty from anki.import_export_pb2 import (ExportAnkiPackageOptions, ExportLimit, ImportAnkiPackageOptions, ImportAnkiPackageRequest) from anki.scheduler_pb2 import CardAnswer try: from anki.cards import FSRSMemoryState except ImportError: # older layouts from anki.cards_pb2 import FsrsMemoryState as FSRSMemoryState REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # LOUDWALK_CLI aims the whole trip at another binary - a released, notarized app's own # executable, say - instead of the repo's ./ankiwalk wrapper. CLI = os.environ.get("LOUDWALK_CLI") or os.path.join(REPO, "ankiwalk") RATINGS = [CardAnswer.AGAIN, CardAnswer.HARD, CardAnswer.GOOD, CardAnswer.EASY] CUSTOM_PARAMS = [0.3, 1.1, 2.9, 12.0, 6.2, 0.9, 2.8, 0.01, 1.7, 0.2, 0.85, 1.5, 0.07, 0.3, 1.7, 0.55, 1.9, 0.6, 0.1, 0.07, 0.2] PNG = bytes.fromhex( "89504e470d0a1a0a0000000d4948445200000001000000010802000000907753de0000000c4944415408d7" "63f8cfc000000301010018dd8db00000000049454e44ae426082") failures = [] passed = 0 def check(label, ok, detail=""): global passed if ok: passed += 1 else: failures.append(label) print(f" {'ok ' if ok else 'FAIL'} {label}{(' - ' + str(detail)) if detail else ''}") return ok # --------------------------------------------------------------------------- the fixture def answer(col, cid, rating): card = col.get_card(cid) card.start_timer() states = col._backend.get_scheduling_states(cid) col.sched.answer_card(col.sched.build_answer(card=card, states=states, rating=RATINGS[rating - 1])) def put(col, cid, memory=None, history=True, **fields): """Puts a card in a state, with one believable review behind it: a review card with no review on record is not something a real collection contains.""" card = col.get_card(cid) for key, value in fields.items(): setattr(card, key, value) if memory is not None: card.memory_state = FSRSMemoryState(stability=memory[0], difficulty=memory[1]) col.update_card(card, skip_undo_entry=True) if history and fields.get("type", 0) != 0: last = fields.get("last_review_time") or int(time.time()) - 86400 rid = last * 1000 while col.db.scalar("select count() from revlog where id = ?", rid): rid += 1 col.db.execute( "insert into revlog (id,cid,usn,ease,ivl,lastIvl,factor,time,type) values (?,?,?,?,?,?,?,?,?)", rid, cid, -1, 3, fields.get("ivl", 1), max(1, fields.get("ivl", 1) // 2), 500, 6000, 1) def build_fixture(tmp, days_back=200): """The collection a user would hand over. Returns paths and the cards by role.""" folder = os.path.join(tmp, "anki-original") os.makedirs(folder) path = os.path.join(folder, "collection.anki2") col = Collection(path) col.db.execute("update col set crt = crt - ?", days_back * 86400) col.close() col = Collection(path) assert col.sched.today == days_back, col.sched.today col.set_config("fsrs", True) art = col.decks.id("Histoire de l'art") renaissance = col.decks.id("Histoire de l'art::Renaissance") russian = col.decks.id("Langues::Russe") old = col.decks.id("Vieux deck") conf = col.decks.config_dict_for_deck_id(art) conf["new"]["delays"], conf["lapse"]["delays"] = [1.0, 10.0], [10.0] conf["desiredRetention"] = 0.9 col.decks.update_config(conf) langues = col.decks.add_config_returning_id("Langues") lconf = col.decks.get_config(langues) lconf["new"]["delays"], lconf["lapse"]["delays"] = [2.0, 15.0], [20.0] lconf["fsrsParams6"], lconf["desiredRetention"] = CUSTOM_PARAMS, 0.85 lconf["new"]["perDay"], lconf["rev"]["perDay"] = 7, 111 col.decks.update_config(lconf) deck = col.decks.get(russian) deck["conf"] = langues col.decks.save(deck) media = col.media.dir() with open(os.path.join(media, "tableau.png"), "wb") as f: f.write(PNG) with open(os.path.join(media, "thème été.mp3"), "wb") as f: f.write(b"ID3" + bytes(range(256)) * 8) basic = col.models.by_name("Basic") both = col.models.by_name("Basic (and reversed card)") cloze = col.models.by_name("Cloze") roles = {} def note(model, did, role, tags=(), **fields): n = col.new_note(model) for key, value in fields.items(): n[key] = value n.tags = list(tags) col.add_note(n, did) ids = [c.id for c in n.cards()] roles[role] = ids[0] if len(ids) == 1 else ids time.sleep(0.002) return ids today, now, cutoff = col.sched.today, int(time.time()), col.sched.day_cutoff for i, role in enumerate(["R1", "R2", "R3", "R4", "R5", "R8", "R9", "R10", "R11"]): tags = ["peinture", "siècle::XVe"] if role == "R9" else [] (cid,) = note(basic, renaissance if i % 2 else art, role, tags, Front=f"Qui a peint l'œuvre {i} ?", Back=f"Peintre n°{i}") ivl = [20, 35, 12, 90, 8, 15, 40, 25, 18][i] late = [0, 5, 0, -30, 0, 2, 0, 0, 0][i] put(col, cid, type=2, queue=-1 if role == "R5" else 2, ivl=ivl, due=today - late, reps=3 + i, lapses=i % 3, factor=2300 + 10 * i, memory=(ivl * 1.1, 4.0 + i * 0.5), last_review_time=cutoff - 86400 * (ivl + late) - 3600 - i) card = col.get_card(roles["R9"]) card.flags = 3 col.update_card(card, skip_undo_entry=True) for role in ["R6", "R7"]: (cid,) = note(basic, old, role, Front=f"Question du vieux deck {role}", Back=f"Réponse {role}") put(col, cid, type=2, queue=2, ivl=9, due=today, reps=4, memory=(9.5, 6.0), last_review_time=cutoff - 86400 * 9 - 500) for role in ["L1", "L2"]: # real answers: genuine learning cards, genuine review rows (cid,) = note(basic, russian, role, Front=f"слово {role}", Back=f"mot {role}") answer(col, cid, 3) for role in ["N1", "N2", "N3", "N4", "N5"]: note(basic, art if role in ("N1", "N3") else russian, role, Front=f"Nouvelle carte {role}", Back=f"Réponse {role}") note(cloze, art, "C", Text="Le {{c1::Discobole}} est une œuvre de {{c2::Myron}}.", **{"Back Extra": "Vers 450 av. J.-C."}) note(both, russian, "RV", Front="дом", Back="maison") note(basic, art, "M1", Front='Quel tableau ? ', Back="La Joconde [sound:thème été.mp3]") (cid,) = note(basic, art, "G", Front="Carte révisée dans Anki", Back="Oui") answer(col, cid, 3) answer(col, cid, 3) # graduated by Anki itself # A filtered deck holding R6: answered in Loudwalk, it has to come home. roles["filtered"] = False try: filtered = col.sched.get_or_create_filtered_deck(deck_id=0) filtered.name = "Session filtrée" filtered.config.search_terms[0].search = f"cid:{roles['R6']}" filtered.config.reschedule = True col.sched.add_or_update_filtered_deck(filtered) roles["filtered"] = col.db.scalar("select odid from cards where id = ?", roles["R6"]) != 0 except Exception as error: # noqa: BLE001 - a fixture nicety, not a requirement print(f" (no filtered deck in the fixture: {error})") # Written yesterday, not this second: Anki's merge compares modification times, and a # note edited in Loudwalk a second after it was made in Anki is not a real situation. col.db.execute("update notes set mod = mod - 86400") out = { "folder": folder, "path": path, "roles": roles, "colpkg": os.path.join(tmp, "from-anki.colpkg"), "legacy": os.path.join(tmp, "from-anki-legacy.colpkg"), "apkg": os.path.join(tmp, "from-anki.apkg"), } col.export_anki_package( out_path=out["apkg"], limit=ExportLimit(whole_collection=Empty()), options=ExportAnkiPackageOptions(with_scheduling=True, with_deck_configs=True, with_media=True, legacy=False)) col.export_collection_package(out["colpkg"], include_media=True, legacy=False) col = Collection(path) col.export_collection_package(out["legacy"], include_media=True, legacy=True) col = Collection(path) out["rows"] = snapshot_rows(col) out["today"] = col.sched.today col.close() return out def snapshot_rows(col): return { "cards": {r[0]: r for r in col.db.all("select * from cards")}, "notes": {r[0]: r for r in col.db.all("select * from notes")}, "revlog": {r[0]: r for r in col.db.all("select * from revlog")}, } # --------------------------------------------------------------------------- Loudwalk class Loudwalk: def __init__(self, home): self.home = home self.env = dict(os.environ, LOUDWALK_HOME=home, LOUDWALK_SYNC_DIR=os.path.join(home, "cloud")) self.env.pop("OPENAI_API_KEY", None) def run(self, *args, ok=True): done = subprocess.run([CLI, *args], env=self.env, capture_output=True, text=True, cwd=self.home) try: reply = json.loads(done.stdout) except ValueError: raise SystemExit(f"ankiwalk {' '.join(args)} did not answer in JSON:\n{done.stdout}\n{done.stderr}") if reply.get("scratch") is not True: raise SystemExit(f"REFUSING TO GO ON: ankiwalk {args[0]} did not report a scratch collection: {reply}") if ok and (done.returncode != 0 or "error" in reply): raise SystemExit(f"ankiwalk {' '.join(args)} failed: {reply}") return reply def review(self, cid, grade, at, mode="visual", overrides=None, total_ms=None): entry = { "ts": at, "iso": datetime.fromtimestamp(at, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "card_id": cid, "mode": mode, "grade_source": "ai" if mode == "voice" else "self", "final_grade": grade, } if overrides is not None: entry["overrides_ts"], entry["grade_source"] = overrides, "override" if total_ms is not None: entry["total_ms"] = total_ms # A scratch collection's own log, in the app's own line format (one JSON object # per line, appended) - the same thing tools/site-shots does. with open(os.path.join(self.home, "review_log.jsonl"), "a") as log: log.write(json.dumps(entry, sort_keys=True, separators=(",", ":")) + "\n") return at def drive_loudwalk(tmp, fixture, package): """Everything a person might do in Loudwalk between importing and leaving.""" home = os.path.join(tmp, "loudwalk-home") os.makedirs(home) lw = Loudwalk(home) imported = lw.run("collection", "import", package) check("Loudwalk imports the package Anki exported", imported["cards"] == len(fixture["rows"]["cards"]), f"{imported['cards']} cards, {imported['reviews']} reviews, {imported['media']} media") roles, now = fixture["roles"], time.time() day = 86400 lw.review(roles["R1"], 3, now - 60) lw.review(roles["R2"], 1, now - 3600) # a lapse... lw.review(roles["R2"], 3, now - 2800) # ...and out of relearning lw.review(roles["R3"], 2, now - 5 * day) # before Anki's rules took over in the app lw.review(roles["R3"], 3, now - 100) lw.review(roles["R5"], 3, now - 90) # suspended in Anki, alive here lw.review(roles["R6"], 3, now - 80) # sitting in a filtered deck first = lw.review(roles["R10"], 3, now - 7200) lw.review(roles["R10"], 1, now - 7100, overrides=first) # corrected: it was a miss undone = lw.review(roles["R11"], 3, now - 600) lw.review(roles["R11"], 0, now - 590, overrides=undone) # undone: never happened lw.review(roles["L2"], 3, now - 70) lw.review(roles["N2"], 3, now - 50) lw.review(roles["N3"], 3, now - 3 * day) lw.review(roles["N3"], 3, now - 3 * day + 700) lw.review(roles["N3"], 3, now - 40) lw.review(roles["N4"], 1, now - 30) lw.review(roles["N5"], 4, now - 20) lw.review(roles["C"][1], 3, now - 15) lw.review(roles["RV"][1], 3, now - 10, mode="voice", total_ms=4200) lw.review(999, 3, now - 5) # a card this collection has never had picture = os.path.join(tmp, "attachment.png") with open(picture, "wb") as f: f.write(PNG) lw.run("cards", "deck-new", "Nouveau::Sous-deck") w1 = lw.run("cards", "add", "--deck", "Nouveau::Sous-deck", "--front", "Quelle est cette façade ?", "--back", "Saint-Front de Périgueux", "--image", f"Back={picture}") w2 = lw.run("cards", "add", "--deck", "Histoire de l'art", "--type", "Cloze", "--text", "{{c1::Guernica}} a été peint par {{c2::Picasso}} en 1937.") w3 = lw.run("cards", "add", "--deck", "Histoire de l'art", "--front", "Carte écrite puis retirée", "--back", "Retirée") ids = lambda reply: [int(c["card_id"]) for c in reply.get("cards", [reply.get("card", {})]) if c] # noqa: E731 written = {"W1": ids(w1), "W2": ids(w2), "W3": ids(w3)} lw.run("cards", "delete", str(written["W3"][0])) lw.run("cards", "delete", str(roles["R8"])) lw.run("cards", "edit", str(roles["R9"]), "--back", "Piero della Francesca & atelier") lw.run("cards", "deck-rename", "Vieux deck", "Deck renommé") lw.review(written["W1"][0], 3, time.time() - 1) exports = {} for extension in ("colpkg", "apkg"): out = os.path.join(tmp, f"from-loudwalk.{extension}") reply = lw.run("collection", "export", out, "--expect", os.path.join(tmp, "expect.json")) exports[extension] = out exports[extension + "_report"] = reply["report"] with open(os.path.join(tmp, "expect.json")) as f: exports["expect"] = json.load(f) exports["written"] = written exports["home"] = home check("the export names what it left out", exports["colpkg_report"]["orphan_reviews"] == 1, f"{exports['colpkg_report']['orphan_reviews']} answer for a card the collection never had") return exports # --------------------------------------------------------------------------- Anki reads it back def import_colpkg(tmp, name, package): folder = os.path.join(tmp, name) os.makedirs(folder) path = os.path.join(folder, "collection.anki2") col = Collection(path) backend = col._backend col.close() backend.import_collection_package( col_path=path, backup_path=package, media_folder=os.path.join(folder, "collection.media"), media_db=os.path.join(folder, "collection.media.db2")) return Collection(path), folder def memory_of(card): state = card.memory_state return (state.stability, state.difficulty) if state is not None and state.stability > 0 else (None, None) def verify_cards(col, fixture, exports, relative_to_today=None, rows_speak_for_untouched=False): """Every card, read through Anki, against what Loudwalk's store says about it.""" expect, roles, rows = exports["expect"], fixture["roles"], fixture["rows"] crt = expect["crt"] by_id = {c["cid"]: c for c in expect["cards"]} found = set(col.db.list("select id from cards")) check("every card Loudwalk holds is in Anki, and no other", found == set(by_id), f"{len(found)} cards; missing {sorted(set(by_id) - found)[:3]}, extra {sorted(found - set(by_id))[:3]}") check("cards retired in Loudwalk are gone", not (set(expect["removed"]) & found), f"{len(expect['removed'])} retired") kinds = {0: "new", 1: "learning", 2: "review", 3: "relearning"} today_there = col.sched.today today_here = int((time.time() - crt) // 86400) if not relative_to_today: check("Anki and Loudwalk agree on what day it is", today_there == today_here, f"day {today_there}") wrong = {"deck": [], "kind": [], "due": [], "memory": [], "counters": [], "fields": [], "suspended": []} for cid in sorted(found & set(by_id)): want, card = by_id[cid], col.get_card(cid) for key in ("due_day", "due_at", "stability", "difficulty"): # Swift leaves out what is nil want.setdefault(key, None) note = card.note() if col.decks.name(card.odid or card.did) != want["deck"]: wrong["deck"].append((cid, col.decks.name(card.odid or card.did), want["deck"])) original = rows["cards"].get(cid) untouched = want["answers_here"] == 0 and not want["written_here"] # A review card Anki has no review on record for reads as new in Loudwalk until it # is answered; its row is compared byte for byte below instead. if not (untouched and original is not None): if kinds[card.type] != want["kind"]: wrong["kind"].append((cid, kinds[card.type], want["kind"])) due = card.odue if card.odid and card.odue else card.due if rows_speak_for_untouched and untouched and original is not None: # Its row is compared with Anki's original, byte for byte, by the caller. Where # the import itself read that row differently (a card reset in Anki that still # has history, say), that is the importer's reading, and nothing was exported. if (want["due_day"] is not None and due != want["due_day"]) or \ (want["due_at"] is not None and due != want["due_at"]): wrong.setdefault("read differently at import", []).append(cid) elif want["due_day"] is not None: shift = (today_there - today_here) if relative_to_today else 0 if due != want["due_day"] + shift: wrong["due"].append((cid, due, want["due_day"] + shift)) elif want["due_at"] is not None and due != want["due_at"]: wrong["due"].append((cid, due, want["due_at"])) stability, difficulty = memory_of(card) if rows_speak_for_untouched and untouched and original is not None: pass elif want["stability"] is not None and want["kind"] != "new": if stability is None or abs(stability - want["stability"]) > 1e-4 + 1e-4 * want["stability"] \ or abs(difficulty - want["difficulty"]) > 1e-3: wrong["memory"].append((cid, (stability, difficulty), (want["stability"], want["difficulty"]))) if original is not None and not relative_to_today: if card.reps != original[11] + want["answers_here"] or card.lapses != original[12] + want["lapses_here"]: wrong["counters"].append((cid, (card.reps, card.lapses), (original[11], original[12]))) names = [f["name"] for f in note.note_type()["flds"]] if any(note[name] != want["fields"].get(name, "") for name in names): wrong["fields"].append((cid, dict(note.items()), want["fields"])) if want["answers_here"] > 0 and card.queue < 0: wrong["suspended"].append(cid) check("every card is in the deck Loudwalk had it in", not wrong["deck"], wrong["deck"][:2]) check("every card is in the state Loudwalk had it in", not wrong["kind"], wrong["kind"][:2]) check("every card is due when Loudwalk said", not wrong["due"], wrong["due"][:3]) check("every card has the memory state Loudwalk computed", not wrong["memory"], wrong["memory"][:2]) if not relative_to_today: check("reviews and lapses are Anki's count plus what happened here", not wrong["counters"], wrong["counters"][:2]) check("every note says what it said in Loudwalk", not wrong["fields"], wrong["fields"][:1]) check("a card answered here is back in rotation", not wrong["suspended"], wrong["suspended"]) if wrong.get("read differently at import"): print(f" info {len(wrong['read differently at import'])} card(s) nobody answered here were read differently " f"by the import than Anki stores them; they go back as Anki had them: " f"{wrong['read differently at import'][:3]}") return by_id def verify_colpkg(tmp, fixture, exports, label): print(f"\nAnki {anki_version} imports Loudwalk's .colpkg ({label})") col, folder = import_colpkg(tmp, f"anki-after-{label}", exports["colpkg"]) expect, roles, rows = exports["expect"], fixture["roles"], fixture["rows"] by_id = verify_cards(col, fixture, exports) # What nobody touched comes back exactly as Anki wrote it. moved = {roles["R7"]} same = [cid for cid, want in by_id.items() if cid in rows["cards"] and want["answers_here"] == 0 and cid not in moved] now_rows = {r[0]: r for r in col.db.all("select * from cards")} differing = [cid for cid in same if list(now_rows[cid]) != list(rows["cards"][cid])] check("cards nobody answered here come back byte for byte", not differing and len(same) >= 8, f"{len(same)} cards identical" if not differing else (differing[:3], now_rows[differing[0]], rows["cards"][differing[0]])) check("an undone answer leaves no trace", roles["R11"] in same) notes = {r[0]: r for r in col.db.all("select * from notes")} edited = col.db.scalar("select nid from cards where id = ?", roles["R9"]) kept = [nid for nid in rows["notes"] if nid in notes and nid != edited] check("notes nobody edited come back byte for byte", all(list(notes[nid]) == list(rows["notes"][nid]) for nid in kept), f"{len(kept)} notes") original = rows["notes"][edited] check("an edited note keeps its id, tags and flags", notes[edited][1] == original[1] and notes[edited][5] == original[5] and "peinture" in notes[edited][5], notes[edited][5].strip()) check("...and its card keeps its flag", col.get_card(roles["R9"]).user_flag() == 3) # Sort field and checksum of what Loudwalk wrote, against Anki's own arithmetic. touched = [edited] + [col.db.scalar("select nid from cards where id = ?", c) for c in exports["written"]["W1"] + exports["written"]["W2"]] bad = [] for nid in set(touched): n = col.get_note(nid) stored = col.db.first("select sfld, csum from notes where id = ?", nid) col.update_note(n) # Anki recomputes both n.fields[0] += " " col.update_note(n) n.fields[0] = n.fields[0][:-1] col.update_note(n) again = col.db.first("select sfld, csum from notes where id = ?", nid) if list(stored) != list(again): bad.append((nid, stored, again)) check("sort field and checksum are what Anki computes itself", not bad, bad[:2]) # The history. revlog = {r[0]: r for r in col.db.all("select * from revlog")} lost = [rid for rid, row in rows["revlog"].items() if list(revlog.get(rid, [])) != list(row)] check("every review Anki gave us goes back unchanged", not lost, f"{len(rows['revlog'])} rows") ours = sorted(set(revlog) - set(rows["revlog"])) wanted = sorted((r["at_ms"], r["cid"], r["ease"]) for r in expect["reviews"]) got = sorted((rid, revlog[rid][1], revlog[rid][3]) for rid in ours) check("every answer given in Loudwalk is a review in Anki", len(got) == len(wanted) and all(abs(g[0] - w[0]) <= 3 and g[1:] == w[1:] for g, w in zip(got, wanted)), f"{len(got)} rows for {len(wanted)} answers") by_card = {} for rid in ours: by_card.setdefault(revlog[rid][1], []).append(revlog[rid]) corrected = by_card.get(roles["R10"], []) check("a corrected grade is one review, with the grade that stands", len(corrected) == 1 and corrected[0][3] == 1 and corrected[0][8] == 1, corrected) lapse = by_card.get(roles["R2"], []) check("a lapse is logged as Anki logs one", len(lapse) == 2 and lapse[0][3] == 1 and lapse[0][4] < 0 and lapse[0][5] == 35 and lapse[0][8] == 1 and lapse[1][8] == 2 and lapse[1][4] > 0, lapse) check("a lapse counts as a lapse", col.get_card(roles["R2"]).lapses == rows["cards"][roles["R2"]][12] + 1) voice = by_card.get(roles["RV"][1], []) check("the time an answer took is kept", len(voice) == 1 and voice[0][7] == 4200, voice) first = by_card.get(roles["N5"], []) check("a first answer is logged as learning", len(first) == 1 and first[0][8] == 0 and first[0][5] == 0, first) # Decks, media, settings. names = {d.name for d in col.decks.all_names_and_ids()} check("decks made here exist in Anki, parents included", {"Nouveau", "Nouveau::Sous-deck", "Deck renommé"} <= names, sorted(names)) check("a deck renamed here does not come back under its old name", "Vieux deck" not in names) check("cards in a renamed deck moved with it", col.decks.name(col.get_card(roles["R7"]).did) == "Deck renommé") if roles["filtered"]: r6 = col.get_card(roles["R6"]) check("a card answered here leaves its filtered deck for home", r6.odid == 0 and col.decks.name(r6.did) == "Deck renommé", (r6.did, r6.odid)) russian = col.decks.by_name("Langues::Russe") conf = col.decks.config_dict_for_deck_id(russian["id"]) check("options presets survive, FSRS parameters included", conf["name"] == "Langues" and len(conf["fsrsParams6"]) == 21 and abs(conf["desiredRetention"] - 0.85) < 1e-6 and conf["new"]["perDay"] == 7, conf["name"]) check("FSRS is still on", col.get_config("fsrs") is True) media = os.path.join(folder, "collection.media") original_media = os.path.join(fixture["folder"], "collection.media") digest = lambda p: hashlib.sha1(open(p, "rb").read()).hexdigest() # noqa: E731 same_media = all(os.path.exists(os.path.join(media, name)) and digest(os.path.join(media, name)) == digest(os.path.join(original_media, name)) for name in os.listdir(original_media)) check("pictures and sounds come back intact", same_media, sorted(os.listdir(media))) w1 = col.get_card(exports["written"]["W1"][0]).note() attached = [name for name in os.listdir(media) if name.startswith("aw-")] check("a picture attached here travels with its card", len(attached) == 1 and attached[0] in w1["Back"], attached) new_cards = [col.get_card(c) for c in exports["written"]["W2"]] taken = {r[8] for cid, r in rows["cards"].items() if r[6] == 0} check("new cards written here queue after Anki's own, in order", all(c.type == 0 and c.queue == 0 for c in new_cards) and [c.due for c in new_cards] == sorted(c.due for c in new_cards) and min(c.due for c in new_cards) > max(taken), [c.due for c in new_cards]) check("the next-position counter moved past them", col.get_config("nextPos") > max(c.due for c in new_cards)) # Anki's own judgement. problems, ok = col.fix_integrity() check("Anki's Check Database finds nothing to repair", ok and "rebuilt and optimized" in problems.lower() and "\n" not in problems.strip(), problems.strip().replace("\n", " | ")) schedulable = 0 for cid in by_id: col._backend.get_scheduling_states(cid) schedulable += 1 check("Anki can schedule every card", schedulable == len(by_id), f"{schedulable} cards") col.decks.select(col.decks.id("Histoire de l'art")) queued = col.sched.get_queued_cards(fetch_limit=500) check("Anki builds a study queue from it", len(queued.cards) > 0, f"{queued.new_count} new, {queued.learning_count} learning, {queued.review_count} to review") # And FSRS itself. Asked to recompute a card's memory from the exported history alone, # does Anki arrive where Loudwalk did? Held to it for cards whose whole history is # real answers (new in the fixture, or answered by Anki there); the fixture's other # review cards were *given* a memory state that no history produced, so for those the # gap is the fixture's, and it is only reported. genuine = {roles[r] for r in ("N2", "N3", "N4", "N5", "L2")} | {roles["C"][1], roles["RV"][1]} \ | set(exports["written"]["W1"]) drift, other = [], [] for cid, want in by_id.items(): if want["answers_here"] == 0 or want["stability"] is None: continue computed = col._backend.compute_memory_state(cid) if not computed.HasField("state"): continue gap = abs(computed.state.stability - want["stability"]) / want["stability"] (drift if cid in genuine else other).append( (gap, cid, round(computed.state.stability, 4), round(want["stability"], 4), round(computed.state.difficulty, 3), round(want["difficulty"], 3))) drift.sort(reverse=True) other.sort(reverse=True) check("Anki, recomputing memory from the exported history alone, lands where Loudwalk did", len(drift) >= 6 and all(d[0] < 0.002 and abs(d[4] - d[5]) < 0.01 for d in drift), f"{len(drift)} cards, largest gap {drift[0][0]:.3%}" if drift else "no card to compare") if other: print(f" info ({len(other)} fixture cards carry a memory state no history produced; recomputed, " f"they differ by up to {other[0][0]:.0%} - in the original collection too)") col.close() def verify_apkg(tmp, fixture, exports): print(f"\nAnki {anki_version} merges Loudwalk's .apkg") request = ImportAnkiPackageRequest( package_path=exports["apkg"], options=ImportAnkiPackageOptions(merge_notetypes=True, with_scheduling=True, with_deck_configs=True, update_notes=1, update_notetypes=1)) # 1 = always # Into a collection that has none of these cards: everything should arrive. folder = os.path.join(tmp, "anki-empty-then-apkg") os.makedirs(folder) col = Collection(os.path.join(folder, "collection.anki2")) col.set_config("fsrs", True) log = col.import_anki_package(request) print(f" info into an empty collection: {len(log.log.new)} notes added, {len(log.log.updated)} updated, " f"{len(log.log.duplicate)} skipped") verify_cards(col, fixture, exports, relative_to_today=True) revlog = col.db.scalar("select count() from revlog") carried = {c["cid"] for c in exports["expect"]["cards"]} history = sum(1 for r in fixture["rows"]["revlog"].values() if r[1] in carried) \ + sum(1 for r in exports["expect"]["reviews"] if r["cid"] in carried) check("...with the whole history of those cards, Anki's rows and Loudwalk's", revlog == history, f"{revlog} reviews") check("...and Anki finds nothing to repair", col.fix_integrity()[1]) col.close() # Into the collection the cards came from, where every one of them still exists. folder = os.path.join(tmp, "anki-original-then-apkg") shutil.copytree(fixture["folder"], folder) col = Collection(os.path.join(folder, "collection.anki2")) before = snapshot_rows(col) log = col.import_anki_package(request) after = snapshot_rows(col) roles = fixture["roles"] rescheduled = [cid for cid in before["cards"] if cid in after["cards"] and list(after["cards"][cid][6:14]) != list(before["cards"][cid][6:14])] gained = len(after["revlog"]) - len(before["revlog"]) edited = col.get_card(roles["R9"]).note()["Back"] added = [cid for cid in after["cards"] if cid not in before["cards"]] print(f" info into the collection they came from: {len(log.log.new)} notes added, " f"{len(log.log.updated)} updated, {len(log.log.duplicate)} skipped") print(f" info cards that already existed and were rescheduled: {len(rescheduled)} of " f"{sum(1 for c in exports['expect']['cards'] if c['answers_here'] and c['cid'] in before['cards'])} answered in Loudwalk") print(f" info review rows gained: {gained} of {len(exports['expect']['reviews'])} given in Loudwalk") print(f" info cards added: {len(added)}; the edited note now reads: {edited!r}") print(f" info retired card still there: {roles['R8'] in after['cards']}") check("merging never damages the collection it is merged into", col.fix_integrity()[1]) check("an edit made here does reach a note Anki still has", len(log.log.updated) == 1 and "Francesca" in edited, edited) check("Anki's merge leaves cards it already has alone (so Loudwalk must not promise otherwise)", len(rescheduled) == 0 and gained <= 2, f"{len(rescheduled)} rescheduled, {gained} rows gained") col.close() # The route for someone who kept using Anki for other decks: delete the decks that went # to Loudwalk, then merge. Everything has to arrive, beside what was already there. folder = os.path.join(tmp, "anki-original-emptied-then-apkg") shutil.copytree(fixture["folder"], folder) col = Collection(os.path.join(folder, "collection.anki2")) col.remove_notes(col.find_notes("")) bystander = col.new_note(col.models.by_name("Basic")) bystander["Front"], bystander["Back"] = "Une carte restée dans Anki", "Intacte" col.add_note(bystander, col.decks.id("Resté dans Anki")) kept = bystander.cards()[0].id log = col.import_anki_package(request) print(f" info after deleting those decks in Anki first: {len(log.log.new)} notes added, " f"{len(log.log.duplicate)} skipped") expect_cards = exports["expect"]["cards"] exports["expect"]["cards"] = expect_cards + [{ "cid": kept, "nid": bystander.id, "deck": "Resté dans Anki", "ord": 0, "kind": "new", "answers_here": 0, "lapses_here": 0, "written_here": True, "fields": {"Front": "Une carte restée dans Anki", "Back": "Intacte"}}] verify_cards(col, fixture, exports, relative_to_today=True) exports["expect"]["cards"] = expect_cards revlog = col.db.scalar("select count() from revlog where cid in (select id from cards)") check("...with the whole history of those cards", revlog == history, f"{revlog} reviews") check("...and Anki finds nothing to repair", col.fix_integrity()[1]) col.close() return {"rescheduled": len(rescheduled), "gained": gained, "added": len(added)} def verify_born_here(tmp): """A collection started empty in Loudwalk has no Anki file behind it: the export writes one from nothing, and Anki has to take it just the same.""" print(f"\nA collection born in Loudwalk, exported, imported by Anki {anki_version}") home = os.path.join(tmp, "loudwalk-born-here") os.makedirs(home) lw = Loudwalk(home) lw.run("collection", "new") # Born ten days ago rather than this minute, so that "which day is it?" is a real # question. (A scratch collection's own bundle, edited before anything reads it.) bundle = os.path.join(home, "data", "bundle.json") with open(bundle) as f: born = json.load(f) born["crt"] -= 10 * 86400 with open(bundle, "w") as f: json.dump(born, f) lw.run("cards", "deck-new", "Histoire::Antiquité") ids = lambda reply: [int(c["card_id"]) for c in reply.get("cards", [reply.get("card", {})]) if c] # noqa: E731 basic = ids(lw.run("cards", "add", "--deck", "Histoire::Antiquité", "--front", "Qui a sculpté le Discobole ?", "--back", "Myron")) cloze = ids(lw.run("cards", "add", "--deck", "Histoire", "--type", "Cloze", "--text", "{{c1::Guernica}} a été peint par {{c2::Picasso}}.", )) fresh = ids(lw.run("cards", "add", "--deck", "Histoire", "--front", "Jamais révisée", "--back", "Non")) now, day = time.time(), 86400 lw.review(basic[0], 3, now - 4 * day) lw.review(basic[0], 3, now - 4 * day + 700) lw.review(basic[0], 1, now - 120) lw.review(cloze[0], 4, now - 60) lw.review(cloze[1], 3, now - 30, mode="voice", total_ms=3100) out, expect_path = os.path.join(tmp, "born-here.colpkg"), os.path.join(tmp, "born-here-expect.json") report = lw.run("collection", "export", out, "--expect", expect_path)["report"] check("it is written from nothing, and says so", report["source"] == "rebuilt" and report["cards"] == 4 and report["loudwalk_reviews"] == 5, report["source"]) with open(expect_path) as f: expect = json.load(f) col, _ = import_colpkg(tmp, "anki-after-born-here", out) verify_cards(col, {"roles": {}, "rows": {"cards": {}}}, {"expect": expect}) check("every answer is a review in Anki", col.db.scalar("select count() from revlog") == 5) check("FSRS is on, with the parameters Loudwalk scheduled by", col.get_config("fsrs") is True and len(col.decks.config_dict_for_deck_id(col.decks.id("Histoire"))["fsrsParams6"]) == 21) names = {m["name"] for m in col.models.all()} check("its notetypes arrive as Anki notetypes", {"Basic", "Cloze"} <= names and col.models.by_name("Cloze")["type"] == 1, sorted(names)) rendered = col.get_card(cloze[0]).question() check("...and Anki renders a cloze from them", "[...]" in rendered and "Picasso" in rendered, rendered[-60:]) gaps = [] for cid in basic + cloze: computed = col._backend.compute_memory_state(cid) want = next(c for c in expect["cards"] if c["cid"] == cid) gaps.append(abs(computed.state.stability - want["stability"]) / want["stability"]) check("Anki, recomputing memory from that history, lands where Loudwalk did", max(gaps) < 0.002, f"largest gap {max(gaps):.3%}") problems, ok = col.fix_integrity() check("Anki's Check Database finds nothing to repair", ok, problems.strip().replace("\n", " | ")) for cid in basic + cloze + fresh: col._backend.get_scheduling_states(cid) col.decks.select(col.decks.id("Histoire")) queued = col.sched.get_queued_cards(fetch_limit=50) check("Anki builds a study queue from it", len(queued.cards) > 0, f"{queued.new_count} new, {queued.learning_count} learning, {queued.review_count} to review") col.close() def repairs(report): """Check Database's report as {what it repaired: how many}, numbers taken out of the wording.""" import re out = {} for line in report.splitlines(): line = line.strip().replace("\u2068", "").replace("\u2069", "") if not line or "rebuilt and optimized" in line.lower(): continue numbers = re.findall(r"\d+", line) out[re.sub(r"\d+", "N", line)] = int(numbers[0]) if numbers else 1 return out def verify_real(package, expect_path, original): """The same questions, asked of a real collection's export: `verify `. Reads all three; the original is opened immutable.""" import sqlite3 print(f"Anki {anki_version} imports {os.path.basename(package)}") source = sqlite3.connect(f"file:{original}?mode=ro&immutable=1", uri=True) rows = { "cards": {r[0]: list(r) for r in source.execute("select * from cards")}, "notes": {r[0]: list(r) for r in source.execute("select * from notes")}, "revlog": {r[0]: list(r) for r in source.execute("select * from revlog")}, } source.close() with open(expect_path) as f: expect = json.load(f) tmp = tempfile.mkdtemp(prefix="anki-export-verify-") try: # What Anki's Check Database already repairs in the collection as it was, before # Loudwalk ever saw it (on a copy): an export is not to blame for those. os.makedirs(os.path.join(tmp, "before")) shutil.copy(original, os.path.join(tmp, "before", "collection.anki2")) os.chmod(os.path.join(tmp, "before", "collection.anki2"), 0o600) before = Collection(os.path.join(tmp, "before", "collection.anki2")) already = repairs(before.fix_integrity()[0]) before.close() col, _ = import_colpkg(tmp, "anki", package) by_id = verify_cards(col, {"roles": {}, "rows": rows}, {"expect": expect}, rows_speak_for_untouched=True) now_cards = {r[0]: list(r) for r in col.db.all("select * from cards")} quiet = [cid for cid, want in by_id.items() if cid in rows["cards"] and want["answers_here"] == 0] identical = [cid for cid in quiet if now_cards.get(cid) == rows["cards"][cid]] moved_only = [cid for cid in quiet if cid not in set(identical) and now_cards[cid][6:] == rows["cards"][cid][6:]] # type, queue, due ... data check("cards nobody answered here keep Anki's schedule, byte for byte", len(identical) + len(moved_only) == len(quiet), f"{len(identical)} rows identical, {len(moved_only)} only moved to another deck, of {len(quiet)}") now_notes = {r[0]: list(r) for r in col.db.all("select * from notes")} same_notes = sum(1 for nid, row in rows["notes"].items() if now_notes.get(nid) == row) print(f" info {same_notes} of {len(rows['notes'])} notes identical to Anki's; " f"{len(now_notes) - len(set(now_notes) & set(rows['notes']))} written in Loudwalk") kept_tags = all(now_notes[nid][1] == row[1] and now_notes[nid][5] == row[5] for nid, row in rows["notes"].items() if nid in now_notes) check("every note keeps its id and its tags", kept_tags) revlog = {r[0]: list(r) for r in col.db.all("select * from revlog")} lost = [rid for rid, row in rows["revlog"].items() if revlog.get(rid) != row] check("every review Anki gave us goes back unchanged", not lost, f"{len(rows['revlog'])} rows") ours = sorted(set(revlog) - set(rows["revlog"])) wanted = sorted((r["at_ms"], r["cid"], r["ease"]) for r in expect["reviews"]) got = sorted((rid, revlog[rid][1], revlog[rid][3]) for rid in ours) check("every answer given in Loudwalk is a review in Anki", len(got) == len(wanted) and all(abs(g[0] - w[0]) <= 50 and g[1:] == w[1:] for g, w in zip(got, wanted)), f"{len(got)} rows for {len(wanted)} answers") found = repairs(col.fix_integrity()[0]) added = {what: n for what, n in found.items() if n > already.get(what, 0)} check("Anki's Check Database finds nothing the original collection did not already have", not added, f"repairs now: {found or 'none'}; in the original: {already or 'none'}") for cid in by_id: col._backend.get_scheduling_states(cid) check("Anki can schedule every card", True, f"{len(by_id)} cards") queued = col.sched.get_queued_cards(fetch_limit=100) check("Anki builds a study queue from it", len(queued.cards) > 0, f"{queued.new_count} new, {queued.learning_count} learning, {queued.review_count} to review") gaps = [] for cid, want in by_id.items(): if want["answers_here"] == 0 or want.get("stability") is None: continue computed = col._backend.compute_memory_state(cid) if computed.HasField("state"): gaps.append(abs(computed.state.stability - want["stability"]) / want["stability"]) if gaps: gaps.sort() close = sum(1 for g in gaps if g < 0.01) print(f" info Anki, recomputing memory from the exported history alone, is within 1% of Loudwalk on " f"{close} of {len(gaps)} answered cards (median gap {gaps[len(gaps) // 2]:.2%}, largest {gaps[-1]:.0%})") col.close() finally: shutil.rmtree(tmp, ignore_errors=True) print(f"\n{passed} checks passed, {len(failures)} failed" + (": " + "; ".join(failures) if failures else "")) return 1 if failures else 0 def main(): if len(sys.argv) >= 5 and sys.argv[1] == "verify": return verify_real(*sys.argv[2:5]) keep = "--keep" in sys.argv legacy = "--legacy" in sys.argv tmp = tempfile.mkdtemp(prefix="anki-export-check-") print(f"Anki {anki_version}; Loudwalk: {CLI}; working in {tmp}") try: print("\nAnki builds a collection and exports it") fixture = build_fixture(tmp) check("the fixture has cards in every state", len(fixture["rows"]["cards"]) >= 20 and len(fixture["rows"]["revlog"]) >= 12, f"{len(fixture['rows']['cards'])} cards, {len(fixture['rows']['revlog'])} reviews, " f"filtered deck: {fixture['roles']['filtered']}") print("\nLoudwalk imports it, and things happen there") exports = drive_loudwalk(tmp, fixture, fixture["legacy" if legacy else "colpkg"]) report = exports["colpkg_report"] print(f" info export: {report['cards']} cards, {report['anki_reviews']} reviews from Anki + " f"{report['loudwalk_reviews']} from Loudwalk, {report['rescheduled']} rescheduled, " f"{report['created_cards']} written, {report['edited_notes']} edited, {report['moved_cards']} moved, " f"{report['removed_cards']} retired, {report['media_files']} media, source: {report['source']}") verify_colpkg(tmp, fixture, exports, "legacy-source" if legacy else "current-format") verify_apkg(tmp, fixture, exports) verify_born_here(tmp) finally: if keep: print(f"\nkept: {tmp}") else: shutil.rmtree(tmp, ignore_errors=True) print(f"\n{passed} checks passed, {len(failures)} failed" + (": " + "; ".join(failures) if failures else "")) return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())