Files
Amateurfunk-Anki/amateurfunk_anki.py
T

1356 lines
44 KiB
Python

#!/usr/bin/env python3
"""Build Anki deck packages from an extracted BNetzA question catalog.
The full design and contract live in DESIGN.md. As an overview the
script does:
1. Read the per-edition directory written by `amateurfunk_fetch.py`
(the JSON catalog, the `svgs/` folder, and the per-edition
`manifest.json`).
2. Split the catalog into five categories. Betriebliche and
Vorschriften get one each. Technische is additionally fanned
out per license class into three decks (N / E / A) via a
strict equality split on the question's `class` field.
3. For each category, render every question as an Anki note: front
shows the (shuffled) A/B/C/D choices, back names the displayed
position of the correct answer.
4. Hand-roll a v11 Anki collection — SQLite database plus JSON
config blobs — and package it into a `.apkg` ZIP with fixed
timestamps. Same input → byte-identical output.
The script is intentionally a single file with stdlib only. Readability
beats cleverness here — most of the bytes below are docstrings and
comments. Performance is a non-goal (we build a handful of small
decks once per upstream edition).
"""
import argparse
import dataclasses
import datetime as dt
import hashlib
import html
import json
import os
import random
import re
import shutil
import sqlite3
import sys
import unicodedata
import zipfile
from pathlib import Path
# ============================================================================
# Constants
# ============================================================================
# Default location of the fetcher's output (one directory up from a
# per-edition slug, containing `manifest-latest.json`).
DEFAULT_DATA_DIR = Path("data")
# Default destination for the generated `.apkg` files.
DEFAULT_OUT_DIR = Path("anki")
# Exit codes. The builder is much simpler than the fetcher — there is
# no "operator must resolve local state" case here, so two codes are
# enough.
EXIT_OK = 0
EXIT_ERROR = 1
# Anki stores all of a note's field values in a single string column
# (`flds`), separated by the Unit Separator control character. Never
# changes; spelled out as a constant so the intent is obvious.
FIELD_SEP = "\x1f"
# The German top-level section titles all share this prefix
# (e.g. "Prüfungsfragen im Prüfungsteil: Technische Kenntnisse").
# We strip it for human-facing display so cards and tags aren't
# dominated by boilerplate.
TOP_LEVEL_PREFIX = "Prüfungsfragen im Prüfungsteil: "
# Map from the catalog's class digits to the user-facing license
# letters documented in DESIGN.md §3 ("Important consumer-side
# conventions"). Tags use the letters because that's what learners
# search for.
CLASS_TAGS = {"1": "N", "2": "E", "3": "A"}
# Ordered class digits used when fanning Technische Kenntnisse out
# into one deck per license class. Order controls the on-disk file
# ordering and the Anki deck-tree ordering.
CLASS_ORDER = [("1", "N"), ("2", "E"), ("3", "A")]
# Short title of the top-level Prüfungsteil that gets fanned out
# per license class. The other two (Betriebliche, Vorschriften) are
# shared across all candidates and stay as a single deck each.
TECHNISCHE_SHORT_TITLE = "Technische Kenntnisse"
# Fallback build epoch if neither the manifest nor `--epoch` supplies
# one. Picked as 0 so missing-metadata builds are still deterministic
# and obviously wrong (timestamps would all show 1970).
DEFAULT_BUILD_EPOCH = 0
# The catalog uses inline `<u>...</u>` for emphasis (typically marking
# negation in question stems — "_NICHT_"). We preserve those tags and
# escape everything else. Anything not in this list goes through
# `html.escape` so plain `<` and `>` remain safe.
SAFE_INLINE_TAG_RE = re.compile(r"(</?u>)", flags=re.IGNORECASE)
# ============================================================================
# Exception types and data classes
# ============================================================================
class AnkiBuildError(Exception):
"""Raised when the local catalog cannot be converted into decks."""
@dataclasses.dataclass
class QuestionItem:
"""A catalog question together with its path through the section tree.
`question` is the raw question dict (with `number`, `class`,
`question`, `answer_a` .. `answer_d`, optional `picture_*`).
`path` is a tuple of section titles from root to leaf — used to
build the card breadcrumb and the path tags.
"""
question: dict
path: tuple
@dataclasses.dataclass
class Category:
"""One top-level exam part, emitted as one Anki `.apkg`.
`title` is the original German title (with the Prüfungsteil
prefix); `short_title` has the prefix stripped (used in the deck
name and the slug); `questions` is the flat list of every question
that lives anywhere under this category.
"""
title: str
short_title: str
slug: str
questions: list
# ============================================================================
# Loading the fetched catalog
# ============================================================================
def load_latest_catalog(data_dir):
"""Return `(edition_dir, manifest, catalog)` from a fetch output dir.
Reads `data_dir/manifest-latest.json` for the slug pointer, then
`data_dir/<slug>/manifest.json` for the per-edition manifest, then
the catalog JSON file the manifest names. Any missing file or
malformed JSON is surfaced as an `AnkiBuildError` so `main()` can
report it cleanly.
"""
latest_path = data_dir / "manifest-latest.json"
try:
latest = json.loads(latest_path.read_text("utf-8"))
slug = latest["slug"]
edition_dir = data_dir / slug
manifest = json.loads(
(edition_dir / "manifest.json").read_text("utf-8")
)
catalog = json.loads(
(edition_dir / manifest["json_filename"]).read_text("utf-8")
)
except (OSError, KeyError, json.JSONDecodeError) as e:
raise AnkiBuildError(
f"could not read fetched catalog under {data_dir}: {e}"
) from e
return edition_dir, manifest, catalog
# ============================================================================
# Category collection
# ============================================================================
def collect_categories(catalog):
"""Split the catalog into one `Category` per output `.apkg`.
Per DESIGN.md §3 axis 1 the catalog has three top-level sections
(the three Prüfungsteile). Technische Kenntnisse is additionally
fanned out per license class (N/E/A) — see DESIGN.md §7 "Output
layout". Betriebliche and Vorschriften are class-1-only in the
data but apply to every candidate, so they stay as one category
each.
For every category we flatten the subtree into a list of
`QuestionItem`s carrying the section path, used downstream for
the breadcrumb and the `pfad-*` tags.
"""
sections = catalog.get("sections")
if not isinstance(sections, list):
raise AnkiBuildError("catalog has no top-level sections list")
categories = []
for section in sections:
title = section.get("title", "Unbenannte Kategorie")
questions = []
_collect_questions(section, (title,), questions)
short_title = _short_category_title(title)
if short_title == TECHNISCHE_SHORT_TITLE:
categories.extend(_split_by_class(title, short_title, questions))
else:
categories.append(
Category(
title=title,
short_title=short_title,
slug=slugify(short_title),
questions=questions,
)
)
return categories
def _split_by_class(title, short_title, questions):
"""Fan one Prüfungsteil out into one `Category` per license class.
Strict equality split on the question's `class` field: a class-1
question lands in the N deck only, class-2 in E only, class-3 in
A only. This matches the catalog's `class` field as written, not
the cumulative study pool (a class-A candidate who wants every
Technische question imports all three decks).
The short title uses Anki's `::` deck-hierarchy separator so the
three decks render as children of a shared parent in Anki's deck
browser (e.g. `Amateurfunk::Technische Kenntnisse::N`).
"""
base_slug = slugify(short_title)
for digit, letter in CLASS_ORDER:
subset = [
item for item in questions
if str(item.question.get("class")) == digit
]
yield Category(
title=f"{title} — Klasse {letter}",
short_title=f"{short_title}::{letter}",
slug=f"{base_slug}-{letter.lower()}",
questions=subset,
)
def _collect_questions(section, path, out):
"""Recurse into a section node and append every question to `out`.
`path` is the tuple of section titles from the top of the catalog
down to the current node — extended by one element on each
recursive call.
"""
for question in section.get("questions") or []:
out.append(QuestionItem(question=question, path=path))
for child in section.get("sections") or []:
title = child.get("title", "Unbenannt")
_collect_questions(child, path + (title,), out)
def _short_category_title(title):
"""Strip the boilerplate `Prüfungsfragen im Prüfungsteil:` prefix."""
if title.startswith(TOP_LEVEL_PREFIX):
return title[len(TOP_LEVEL_PREFIX):]
return title
# ============================================================================
# Slugs and stable IDs
# ============================================================================
def slugify(value):
"""Reduce a German title to a readable lowercase ASCII filename slug.
Umlauts and ß map to digraphs (`ä → ae`, `ß → ss`) before NFKD
normalization strips any remaining diacritics; everything that
isn't `[A-Za-z0-9]` becomes a single hyphen, and the result is
lowercased. Returns `"unbenannt"` for inputs that collapse to the
empty string.
"""
replacements = {
"ä": "ae",
"ö": "oe",
"ü": "ue",
"Ä": "Ae",
"Ö": "Oe",
"Ü": "Ue",
"ß": "ss",
}
for src, dst in replacements.items():
value = value.replace(src, dst)
value = unicodedata.normalize("NFKD", value)
value = value.encode("ascii", "ignore").decode("ascii")
value = re.sub(r"[^A-Za-z0-9]+", "-", value).strip("-").lower()
return value or "unbenannt"
def stable_id(namespace, text, digits=13):
"""Return a deterministic positive integer ID for Anki row IDs.
Anki uses 13-digit integers (ms-since-epoch-shaped) for deck,
model, note, and card IDs. We hash a namespaced text key with
SHA-1, take the first 60 bits, and squash them into the
`[10**(digits-1), 10**digits)` range. Different namespaces (e.g.
`"deck"` vs `"note"`) keep IDs from colliding across kinds.
"""
digest = hashlib.sha1(f"{namespace}:{text}".encode("utf-8")).hexdigest()
modulus = 10 ** digits
floor = 10 ** (digits - 1)
return floor + (int(digest[:15], 16) % (modulus - floor))
def stable_guid(text):
"""Return a deterministic 20-character note GUID.
Anki uses the GUID to deduplicate notes when an `.apkg` is
re-imported. Deriving it from a stable per-question key (the
category slug plus the question number) means importing the same
deck twice updates rather than duplicates.
"""
return hashlib.sha1(text.encode("utf-8")).hexdigest()[:20]
def checksum_sort_field(sort_field):
"""Compute Anki's `notes.csum`: first 32 bits of SHA1(sort field).
Anki uses this for fast duplicate-detection in the note browser.
"""
digest = hashlib.sha1(sort_field.encode("utf-8")).hexdigest()
return int(digest[:8], 16)
# ============================================================================
# Answer shuffling
# ============================================================================
def randomized_answers(question, seed):
"""Return `(choices, correct_label, correct_choice)` for one question.
Upstream always stores the correct answer in `answer_a` and the
three distractors in `answer_b/c/d` (see DESIGN.md §3 "Important
consumer-side conventions"). We shuffle the four choices using a
seed derived from the CLI seed plus the question number, so the
same input always produces the same display order. After shuffle,
each choice gets a displayed label A/B/C/D, and we return which
label happened to land on the correct answer.
"""
choices = [
{
"source": "answer_a",
"text": question["answer_a"],
"picture": question.get("picture_a"),
"correct": True,
},
{
"source": "answer_b",
"text": question["answer_b"],
"picture": question.get("picture_b"),
"correct": False,
},
{
"source": "answer_c",
"text": question["answer_c"],
"picture": question.get("picture_c"),
"correct": False,
},
{
"source": "answer_d",
"text": question["answer_d"],
"picture": question.get("picture_d"),
"correct": False,
},
]
rnd_seed = hashlib.sha256(
f"{seed}:{question.get('number', '')}".encode("utf-8")
).hexdigest()
rnd = random.Random(int(rnd_seed[:16], 16))
rnd.shuffle(choices)
for index, choice in enumerate(choices):
choice["label"] = "ABCD"[index]
correct = next(choice for choice in choices if choice["correct"])
return choices, correct["label"], correct
# ============================================================================
# HTML rendering
# ============================================================================
def render_question(item, media, seed):
"""Render one question as `(front_html, back_html, correct_label)`.
The front shows the question stem, an optional figure, and the
four shuffled answer choices as an ordered list with `type="A"`.
The back names the displayed position of the correct answer and
repeats its text/figure.
`media` is the `MediaRegistry` for this category; it records
which figure files are actually used so the packager can include
only those.
"""
question = item.question
choices, correct_label, correct = randomized_answers(question, seed)
number = html.escape(str(question.get("number", "")))
path = html.escape(display_path(item.path))
front_parts = [
'<div class="af-card">',
f'<div class="af-meta"><span class="af-number">{number}</span>'
f'<span class="af-path">{path}</span></div>',
f'<div class="af-question">{text_html(question.get("question", ""))}</div>',
]
q_image = media.image_html(question.get("picture_question"))
if q_image:
front_parts.append(
f'<div class="af-image af-question-image">{q_image}</div>'
)
front_parts.append('<ol class="af-answers" type="A">')
for choice in choices:
choice_image = media.image_html(choice.get("picture"))
image_html_part = (
f'<div class="af-image">{choice_image}</div>' if choice_image else ""
)
front_parts.append(
"<li>"
f'<div class="af-answer-text">{text_html(choice["text"])}</div>'
f"{image_html_part}"
"</li>"
)
front_parts.append("</ol></div>")
correct_image = media.image_html(correct.get("picture"))
back_parts = [
'<div class="af-back">',
f'<div class="af-correct-label">Richtige Antwort: {correct_label}</div>',
f'<div class="af-correct-answer">{text_html(correct["text"])}</div>',
]
if correct_image:
back_parts.append(
f'<div class="af-image af-correct-image">{correct_image}</div>'
)
back_parts.append("</div>")
return "".join(front_parts), "".join(back_parts), correct_label
def display_path(path):
"""Return the user-facing section path with the top prefix stripped.
Used for both the visible card breadcrumb and the stored `Path`
note field so the two never drift.
"""
if not path:
return ""
return " / ".join([_short_category_title(path[0]), *path[1:]])
# ============================================================================
# LaTeX and HTML escaping
# ============================================================================
def text_html(value):
"""Escape text for HTML, preserve line breaks, and enable MathJax.
The catalog uses bare `$...$` for inline LaTeX (DESIGN.md §3
quotes the upstream README on this). Anki 2.1+ ships built-in
MathJax that recognizes `\\(...\\)` but not `$...$`, so we
tokenize the input into alternating math / non-math runs, escape
each run for HTML, and wrap math runs in `\\(...\\)` afterwards.
Non-math newlines become `<br>` and the catalog's safe inline
markup (`<u>...</u>`) survives escaping.
"""
parts = []
for is_math, chunk in tokenize_dollar_math(str(value)):
if is_math:
escaped = html.escape(chunk)
parts.append(r"\(" + escaped + r"\)")
else:
escaped = escape_text_preserving_safe_tags(chunk)
parts.append("<br>".join(escaped.splitlines()))
return "".join(parts)
def escape_text_preserving_safe_tags(text):
"""HTML-escape `text` while preserving exact `<u>` / `</u>` tags.
The live catalog uses `<u>...</u>` to emphasize negation in
question stems (e.g. "Welche der folgenden Aussagen ist <u>nicht</u>
korrekt"). Preserving those tags keeps that emphasis visible on
the rendered card. Anything else — including ordinary mathematical
comparisons like `2 < 3` — still goes through `html.escape`.
"""
escaped_parts = []
for part in SAFE_INLINE_TAG_RE.split(text):
if SAFE_INLINE_TAG_RE.fullmatch(part):
escaped_parts.append(part.lower())
else:
escaped_parts.append(html.escape(part))
return "".join(escaped_parts)
def tokenize_dollar_math(text):
"""Yield `(is_math, chunk)` pairs, splitting on unescaped `$...$`.
A backslash-escaped `\\$` is treated as a literal dollar and does
not open or close a math span. Unmatched dollars (an opening `$`
with no closing partner) are treated as ordinary text.
"""
index = 0
while index < len(text):
start = find_unescaped_dollar(text, index)
if start is None:
yield False, text[index:]
return
end = find_unescaped_dollar(text, start + 1)
if end is None:
yield False, text[index:]
return
if start > index:
yield False, text[index:start]
yield True, text[start + 1:end]
index = end + 1
def find_unescaped_dollar(text, start):
"""Return the index of the next `$` not preceded by an odd run of `\\`."""
while True:
pos = text.find("$", start)
if pos == -1:
return None
backslashes = 0
cursor = pos - 1
while cursor >= 0 and text[cursor] == "\\":
backslashes += 1
cursor -= 1
if backslashes % 2 == 0:
return pos
start = pos + 1
# ============================================================================
# Media registry
# ============================================================================
class MediaRegistry:
"""Resolve catalog image stems and collect media files for one `.apkg`.
Indexes every file in `media_dir` by filename. Each call to
`resolve(reference)` accepts the catalog's bare stem
(e.g. `"AB109_q"`) or a filename with extension; on a hit, the
`Path` is recorded in `used_paths` so the packager can include
only the files we actually referenced.
"""
def __init__(self, media_dir):
self.media_dir = media_dir
# filename → Path for every file under `media_dir`
self.by_filename = {}
# files referenced by at least one resolved reference
self.used_paths = set()
# references that didn't resolve (recorded for diagnostics)
self.missing = []
self._index_media_dir()
def _index_media_dir(self):
"""Populate `by_filename` from a single pass over `media_dir`."""
if not self.media_dir.exists():
return
for path in self.media_dir.iterdir():
if path.is_file():
self.by_filename[path.name] = path
def resolve(self, reference):
"""Return the resolved `Path`, or `None` if the reference is
missing.
Picture references in the catalog are stems without
extensions (e.g. `"AB109_q"`); we try the stem with `.svg`
first, then `.png`. A reference that already carries an
extension is tried as-is. References containing path
separators are rejected outright — the catalog only ever
names a bare basename.
"""
if not reference:
return None
ref = str(reference)
if "/" in ref or "\\" in ref:
self.missing.append(ref)
return None
if Path(ref).suffix:
candidates = [ref]
else:
candidates = [f"{ref}.svg", f"{ref}.png"]
for candidate in candidates:
path = self.by_filename.get(candidate)
if path is not None:
self.used_paths.add(path)
return path
self.missing.append(ref)
return None
def image_html(self, reference):
"""Return an `<img>` tag for `reference`, or `""` if unresolved."""
path = self.resolve(reference)
if path is None:
return ""
filename = html.escape(path.name, quote=True)
return f'<img src="{filename}" class="af-media">'
# ============================================================================
# Tagging
# ============================================================================
def tags_for_item(item):
"""Return the Anki tags field for one note.
Tags are space-separated and the field carries a leading and
trailing space — that's the convention Anki itself uses. We emit
a single `klasse-N/E/A` tag plus one `pfad-<section>` tag per
section level below the top-level Prüfungsteil. The question
number itself is intentionally NOT a tag (it's already in the
`Number` field; per-question tags would clutter Anki's tag tree
with 1750 singletons).
"""
question = item.question
class_tag = CLASS_TAGS.get(str(question.get("class")), "unknown")
tags = [f"klasse-{class_tag}"]
tags.extend(f"pfad-{slugify(part)}" for part in item.path[1:])
return " " + " ".join(tags) + " "
# ============================================================================
# Anki package writer (apkg ZIP + SVG post-processing)
# ============================================================================
def build_apkg_for_category(category, edition_dir, out_path, seed, build_epoch):
"""Write one category as an Anki `.apkg` file.
Renders every question to a note, hand-rolls the v11 SQLite
collection, and writes the final ZIP with deterministic
timestamps. Returns a small result dict describing what was
written (for the CLI summary).
"""
tmp_dir = out_path.parent / f".{out_path.name}.tmp"
db_path = tmp_dir / "collection.anki2"
# Unlike the fetcher, this builder has no operator-recoverable
# state in its package temp file. A stale `.apkg.tmp` is just an
# incomplete output artifact, so we overwrite it on the next
# (deterministic) build.
if tmp_dir.exists():
shutil.rmtree(tmp_dir)
tmp_dir.mkdir(parents=True)
media = MediaRegistry(edition_dir / "svgs")
deck_name = f"Amateurfunk::{category.short_title}"
deck_id = stable_id("deck", deck_name)
model_id = stable_id("model", "Amateurfunk Multiple Choice")
try:
notes = []
for ordinal, item in enumerate(category.questions):
front, back, correct_label = render_question(item, media, seed)
question = item.question
number = str(question.get("number", f"q{ordinal}"))
note_id = stable_id("note", f"{category.slug}:{number}")
card_id = stable_id("card", f"{category.slug}:{number}")
fields = [
number,
category.short_title,
display_path(item.path),
front,
back,
correct_label,
]
notes.append({
"note_id": note_id,
"card_id": card_id,
"guid": stable_guid(f"{category.slug}:{number}"),
"fields": FIELD_SEP.join(fields),
"sort": number,
"tags": tags_for_item(item),
})
# Only the files actually referenced make it into the package.
media_paths = {path.name: path for path in media.used_paths}
create_collection_db(
db_path,
deck_id=deck_id,
deck_name=deck_name,
model_id=model_id,
notes=notes,
now=build_epoch,
)
write_apkg(
out_path, db_path, media_paths, build_epoch=build_epoch,
)
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
return {
"path": out_path,
"deck": deck_name,
"questions": len(category.questions),
"media": len(media_paths),
"missing_media": sorted(set(media.missing)),
}
def write_apkg(out_path, db_path, media_paths, build_epoch):
"""Write the Anki package ZIP atomically.
The ZIP contains:
* `collection.anki2` — the SQLite database
* one numbered entry per media file (Anki addresses media by
sequential integer keys, not by filename)
* `media` — a JSON object mapping those keys back to filenames
Every member is written with a fixed timestamp so the output is
byte-identical across runs of the same input.
"""
out_path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = out_path.with_suffix(out_path.suffix + ".tmp")
media_map = {}
try:
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zip_writestr_fixed(
zf, "collection.anki2", db_path.read_bytes(),
build_epoch=build_epoch,
)
for index, filename in enumerate(sorted(media_paths)):
archive_name = str(index)
media_map[archive_name] = filename
zip_writestr_fixed(
zf, archive_name,
media_bytes_for_package(media_paths[filename]),
build_epoch=build_epoch,
)
zip_writestr_fixed(
zf, "media",
json.dumps(media_map, ensure_ascii=False).encode("utf-8"),
build_epoch=build_epoch,
)
os.replace(tmp_path, out_path)
except Exception:
tmp_path.unlink(missing_ok=True)
raise
def zip_writestr_fixed(zf, name, payload, build_epoch):
"""Write one ZIP member with a deterministic timestamp + attrs."""
info = zipfile.ZipInfo(name, zip_datetime(build_epoch))
info.compress_type = zipfile.ZIP_DEFLATED
info.external_attr = 0o644 << 16
zf.writestr(info, payload)
def zip_datetime(epoch):
"""Return a ZIP-compatible UTC date-time tuple for `epoch`.
ZIP timestamps can't represent dates before 1980, so we clamp the
lower bound to 1980-01-01 UTC. This keeps the
`DEFAULT_BUILD_EPOCH = 0` fallback from blowing up.
"""
safe_epoch = max(int(epoch), 315532800)
value = dt.datetime.fromtimestamp(safe_epoch, dt.timezone.utc)
return (
value.year,
value.month,
value.day,
value.hour,
value.minute,
value.second,
)
def media_bytes_for_package(path):
"""Return media bytes as they should be stored in the package.
BNetzA SVGs are transparent. In Anki dark mode that makes the
black line drawings nearly invisible against the dark card
background. When packaging an SVG we inject an explicit white
background rectangle as the first painted element. Non-SVG files
(PNGs, etc.) are passed through untouched. The extracted source
files on disk are likewise never modified.
"""
raw = path.read_bytes()
if path.suffix.lower() != ".svg":
return raw
text = raw.decode("utf-8-sig", errors="replace")
return svg_with_white_background(text).encode("utf-8")
def svg_with_white_background(svg_text):
"""Inject a white background `<rect>` after the opening `<svg>`.
Idempotent: subsequent calls notice the `data-af-white-background`
marker and leave the SVG alone. If we can't find an opening
`<svg>` tag we return the input unchanged (better to ship the
catalog's SVG as-is than to refuse the whole package).
"""
if 'data-af-white-background="1"' in svg_text:
return svg_text
match = re.search(r"<svg\b[^>]*>", svg_text, flags=re.IGNORECASE)
if not match:
return svg_text
background = (
'<rect data-af-white-background="1" '
'width="100%" height="100%" fill="#fff"/>'
)
return svg_text[:match.end()] + background + svg_text[match.end():]
# ============================================================================
# SQLite collection schema
# ============================================================================
def create_collection_db(db_path, deck_id, deck_name, model_id, notes, now):
"""Create the `collection.anki2` SQLite database for one package.
Uses the v11 schema, which modern Anki still understands (it
upgrades the collection on first open). We pre-write a single
`col` row with the JSON config blobs, then insert one `notes`
row and one `cards` row per question.
"""
conn = sqlite3.connect(db_path)
try:
create_schema(conn)
insert_collection_metadata(
conn,
deck_id=deck_id,
deck_name=deck_name,
model_id=model_id,
now=now,
)
for due, note in enumerate(notes, start=1):
conn.execute(
"""
INSERT INTO notes
(id, guid, mid, mod, usn, tags, flds, sfld, csum, flags, data)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
note["note_id"],
note["guid"],
model_id,
now,
-1,
note["tags"],
note["fields"],
note["sort"],
checksum_sort_field(note["sort"]),
0,
"",
),
)
conn.execute(
"""
INSERT INTO cards
(id, nid, did, ord, mod, usn, type, queue, due, ivl, factor,
reps, lapses, left, odue, odid, flags, data)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
note["card_id"],
note["note_id"],
deck_id,
0,
now,
-1,
0,
0,
due,
0,
0,
0,
0,
0,
0,
0,
0,
"",
),
)
conn.commit()
finally:
conn.close()
def create_schema(conn):
"""Create the v11 Anki collection schema (tables + indices)."""
conn.executescript(
"""
CREATE TABLE col (
id integer primary key,
crt integer not null,
mod integer not null,
scm integer not null,
ver integer not null,
dty integer not null,
usn integer not null,
ls integer not null,
conf text not null,
models text not null,
decks text not null,
dconf text not null,
tags text not null
);
CREATE TABLE notes (
id integer primary key,
guid text not null,
mid integer not null,
mod integer not null,
usn integer not null,
tags text not null,
flds text not null,
sfld integer not null,
csum integer not null,
flags integer not null,
data text not null
);
CREATE TABLE cards (
id integer primary key,
nid integer not null,
did integer not null,
ord integer not null,
mod integer not null,
usn integer not null,
type integer not null,
queue integer not null,
due integer not null,
ivl integer not null,
factor integer not null,
reps integer not null,
lapses integer not null,
left integer not null,
odue integer not null,
odid integer not null,
flags integer not null,
data text not null
);
CREATE TABLE revlog (
id integer primary key,
cid integer not null,
usn integer not null,
ease integer not null,
ivl integer not null,
lastIvl integer not null,
factor integer not null,
time integer not null,
type integer not null
);
CREATE TABLE graves (
usn integer not null,
oid integer not null,
type integer not null
);
CREATE INDEX ix_notes_usn on notes (usn);
CREATE INDEX ix_cards_usn on cards (usn);
CREATE INDEX ix_cards_nid on cards (nid);
CREATE INDEX ix_cards_sched on cards (did, queue, due);
CREATE INDEX ix_revlog_usn on revlog (usn);
CREATE INDEX ix_revlog_cid on revlog (cid);
"""
)
def insert_collection_metadata(conn, deck_id, deck_name, model_id, now):
"""Write the single `col` row that carries the JSON config blobs.
`crt` is a seconds-epoch creation time; `mod` and `scm` are in
milliseconds (Anki's mixed convention, not ours to fix).
"""
conn.execute(
"""
INSERT INTO col
(id, crt, mod, scm, ver, dty, usn, ls, conf, models, decks, dconf, tags)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
1,
now,
now * 1000,
now * 1000,
11,
0,
0,
0,
json.dumps(collection_conf(deck_id), separators=(",", ":")),
json.dumps(
{str(model_id): model_json(model_id, now)},
separators=(",", ":"),
),
json.dumps(
{str(deck_id): deck_json(deck_id, deck_name, now)},
separators=(",", ":"),
),
json.dumps(default_deck_conf(now), separators=(",", ":")),
"{}",
),
)
# ============================================================================
# Anki JSON config (col.conf / models / decks / dconf entries)
# ============================================================================
def collection_conf(deck_id):
"""Return the JSON written into `col.conf`.
These knobs control review UI defaults (current deck, sort
column, etc.). They're not deck content — Anki rewrites most of
them on first open.
"""
return {
"nextPos": 1,
"estTimes": True,
"activeDecks": [deck_id],
"sortType": "noteFld",
"timeLim": 0,
"sortBackwards": False,
"addToCur": True,
"curDeck": deck_id,
"newBury": True,
"newSpread": 0,
"dueCounts": True,
"curModel": None,
"collapseTime": 1200,
}
def deck_json(deck_id, deck_name, now):
"""Return one deck entry for `col.decks`.
The `Amateurfunk::Foo` naming gives the user a single top-level
Amateurfunk node in Anki's deck browser with three child decks.
"""
return {
"id": deck_id,
"name": deck_name,
"mod": now,
"usn": -1,
"lrnToday": [0, 0],
"revToday": [0, 0],
"newToday": [0, 0],
"timeToday": [0, 0],
"collapsed": False,
"browserCollapsed": False,
"dyn": 0,
"conf": 1,
"desc": "",
"extendNew": 0,
"extendRev": 0,
}
def default_deck_conf(now):
"""Return a baseline `dconf` entry referenced by `deck_json`."""
return {
"1": {
"id": 1,
"name": "Default",
"mod": now,
"usn": -1,
"maxTaken": 60,
"autoplay": True,
"timer": 0,
"replayq": True,
"new": {
"delays": [1, 10],
"ints": [1, 4, 0],
"initialFactor": 2500,
"separate": True,
"order": 1,
"perDay": 20,
"bury": True,
},
"rev": {
"perDay": 200,
"ease4": 1.3,
"fuzz": 0.05,
"minSpace": 1,
"ivlFct": 1,
"maxIvl": 36500,
"bury": True,
},
"lapse": {
"delays": [10],
"mult": 0,
"minInt": 1,
"leechFails": 8,
"leechAction": 0,
},
}
}
def model_json(model_id, now):
"""Return the note-type entry for `col.models`.
A single template (`Card 1`) renders the `Front` field, then the
horizontal rule, then the `Back` field — the standard Anki
"front / divider / back" layout. The other fields (`Number`,
`Category`, `Path`, `CorrectLabel`) are stored for searchability
and exports.
"""
fields = [
field_json("Number", 0),
field_json("Category", 1),
field_json("Path", 2),
field_json("Front", 3),
field_json("Back", 4),
field_json("CorrectLabel", 5),
]
return {
"id": model_id,
"name": "Amateurfunk Multiple Choice",
"type": 0,
"mod": now,
"usn": -1,
"sortf": 0,
"did": None,
"tmpls": [
{
"name": "Card 1",
"ord": 0,
"qfmt": "{{Front}}",
"afmt": "{{FrontSide}}\n<hr id=answer>\n{{Back}}",
"did": None,
"bqfmt": "",
"bafmt": "",
}
],
"flds": fields,
"css": CARD_CSS,
"latexPre": (
"\\documentclass[12pt]{article}\n"
"\\special{papersize=3in,5in}\n"
"\\usepackage[utf8]{inputenc}\n"
"\\usepackage{amssymb,amsmath}\n"
"\\pagestyle{empty}\n"
"\\begin{document}"
),
"latexPost": "\\end{document}",
"req": [[0, "any", [3]]],
"vers": [],
}
def field_json(name, ord_):
"""Return one field-definition entry for a model's `flds` list."""
return {
"name": name,
"ord": ord_,
"sticky": False,
"rtl": False,
"font": "Arial",
"size": 20,
"description": "",
"plainText": False,
"collapsed": False,
"excludeFromSearch": False,
"tag": None,
"preventDeletion": False,
}
# ============================================================================
# Card styling
# ============================================================================
# Inlined into the model's `css` field. Kept readable here rather than
# minified; Anki doesn't care about whitespace.
CARD_CSS = """
.card {
font-family: Arial, sans-serif;
font-size: 18px;
line-height: 1.45;
color: #111;
background: #fff;
text-align: left;
}
.af-meta {
color: #555;
font-size: 13px;
margin-bottom: 0.75rem;
}
.af-number {
font-weight: bold;
margin-right: 0.75rem;
}
.af-question {
font-size: 20px;
margin-bottom: 1rem;
}
.af-answers {
padding-left: 1.7rem;
}
.af-answers li {
margin: 0.7rem 0;
}
.af-media {
max-width: 100%;
height: auto;
margin: 0.4rem 0;
}
.af-correct-label {
font-weight: bold;
color: #0b6b3a;
margin-bottom: 0.5rem;
}
"""
# ============================================================================
# Build epoch and orchestration
# ============================================================================
def build_epoch_from_manifest(manifest, override_epoch=None):
"""Decide which integer epoch to stamp into the package.
Priority is `--epoch` override → manifest `fetched_at` →
`DEFAULT_BUILD_EPOCH`. A non-empty but unparseable `fetched_at`
is treated as a hard error (the manifest is supposed to carry an
ISO-8601 UTC string).
"""
if override_epoch is not None:
return int(override_epoch)
fetched_at = manifest.get("fetched_at")
if not fetched_at:
return DEFAULT_BUILD_EPOCH
try:
normalized = str(fetched_at).replace("Z", "+00:00")
return int(dt.datetime.fromisoformat(normalized).timestamp())
except ValueError as e:
raise AnkiBuildError(
f"invalid manifest fetched_at value: {fetched_at!r}"
) from e
def build_all(data_dir, out_dir, seed, override_epoch=None):
"""Build every category's `.apkg` and return their result dicts.
Loads the latest fetched catalog, picks a build epoch, then walks
every category writing one `.apkg` each. Raises `AnkiBuildError`
on configuration / catalog problems.
"""
edition_dir, manifest, catalog = load_latest_catalog(data_dir)
build_epoch = build_epoch_from_manifest(manifest, override_epoch)
categories = collect_categories(catalog)
if not categories:
raise AnkiBuildError("catalog has no categories")
results = []
for category in categories:
out_path = out_dir / f"amateurfunk-{category.slug}.apkg"
results.append(
build_apkg_for_category(
category,
edition_dir,
out_path,
seed=seed,
build_epoch=build_epoch,
)
)
return results
# ============================================================================
# Main entry point
# ============================================================================
def _parse_args(argv):
"""Build the argparse object and parse `argv`."""
parser = argparse.ArgumentParser(
description=(
"Build Anki .apkg decks from an extracted BNetzA "
"question catalog."
),
)
parser.add_argument(
"--data",
type=Path,
default=DEFAULT_DATA_DIR,
help=(
"fetch output directory containing manifest-latest.json "
"(default: ./data)"
),
)
parser.add_argument(
"--out",
type=Path,
default=DEFAULT_OUT_DIR,
help="output directory for .apkg files (default: ./anki)",
)
parser.add_argument(
"--seed",
default="amateurfunk-anki-v1",
help="deterministic seed for answer shuffling",
)
parser.add_argument(
"--epoch",
type=int,
default=None,
help=(
"override the package timestamp epoch; by default this is "
"derived from manifest.json's fetched_at"
),
)
return parser.parse_args(argv)
def main(argv=None):
"""Top-level entry point. Returns an exit code; never raises."""
args = _parse_args(argv)
try:
results = build_all(
args.data,
args.out,
seed=args.seed,
override_epoch=args.epoch,
)
except AnkiBuildError as e:
print(f"error: {e}", file=sys.stderr)
return EXIT_ERROR
except Exception as e: # noqa: BLE001 (main() promises not to raise)
print(
f"error: failed to build Anki packages: {e}", file=sys.stderr,
)
return EXIT_ERROR
for result in results:
print(
f"wrote {result['path']} "
f"({result['questions']} cards, {result['media']} media files)"
)
if result["missing_media"]:
print(
f"warning: {len(result['missing_media'])} missing media "
f"reference(s) in {result['deck']}",
file=sys.stderr,
)
return EXIT_OK
if __name__ == "__main__":
sys.exit(main())