1310 lines
42 KiB
Python
1310 lines
42 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 three categories — one per top-level
|
|
Prüfungsteil: Technische / Betriebliche / Vorschriften. (The
|
|
license-class axis is mapped into tags, not into separate
|
|
packages.)
|
|
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 three 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"}
|
|
|
|
# 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 top-level section.
|
|
|
|
Per DESIGN.md §3 axis 1, the catalog's three top-level sections
|
|
are exactly the three Prüfungsteile. We walk each subtree and
|
|
flatten every question it contains, recording the section path so
|
|
cards can show a breadcrumb and we can emit path-shaped 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)
|
|
categories.append(
|
|
Category(
|
|
title=title,
|
|
short_title=short_title,
|
|
slug=slugify(short_title),
|
|
questions=questions,
|
|
)
|
|
)
|
|
return categories
|
|
|
|
|
|
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
|
|
the three top-level categories 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())
|