a9db6a8771
- core/vocab.py: extract B1-B2 level vocabulary from English content via Gemini Flash - core/anki.py: register vocab cards to AnkiConnect (English::Vocabulary deck) - core/enricher.py: add language detection field + summary_ko (Korean summary) - core/obsidian.py: render Korean + English summary in note - daemon/worker.py: call vocab extraction and Anki registration for English content
107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
"""AnkiConnect HTTP client for adding vocabulary cards."""
|
|
|
|
import json
|
|
import logging
|
|
from urllib import request, error
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ANKI_URL = "http://localhost:8765"
|
|
DECK_NAME = "English::Vocabulary"
|
|
MODEL_NAME = "Basic"
|
|
|
|
|
|
def _invoke(action: str, **params) -> object:
|
|
"""Call AnkiConnect API.
|
|
|
|
Args:
|
|
action: AnkiConnect action name.
|
|
**params: Action parameters.
|
|
|
|
Returns:
|
|
The result field from AnkiConnect response.
|
|
|
|
Raises:
|
|
RuntimeError: If AnkiConnect returns an error.
|
|
"""
|
|
payload = json.dumps({"action": action, "version": 6, "params": params}).encode()
|
|
req = request.Request(ANKI_URL, payload, {"Content-Type": "application/json"})
|
|
with request.urlopen(req, timeout=5) as resp:
|
|
data = json.loads(resp.read())
|
|
if data.get("error"):
|
|
raise RuntimeError(f"AnkiConnect error: {data['error']}")
|
|
return data["result"]
|
|
|
|
|
|
def _ensure_deck() -> None:
|
|
"""Create deck if it doesn't exist."""
|
|
_invoke("createDeck", deck=DECK_NAME)
|
|
|
|
|
|
def _build_front(word: str, pos: str, example: str) -> str:
|
|
return f"<b>{word}</b> <i>({pos})</i><br><br>{example}"
|
|
|
|
|
|
def _build_back(definition_en: str, definition_ko: str) -> str:
|
|
return f"{definition_en}<br><br><b>한국어:</b> {definition_ko}"
|
|
|
|
|
|
def add_vocab_cards(vocab_list: list[dict], source_title: str = "") -> list[int]:
|
|
"""Add vocabulary cards to Anki.
|
|
|
|
Skips duplicates silently (AnkiConnect returns null for existing notes).
|
|
|
|
Args:
|
|
vocab_list: List of vocab dicts from extract_vocab().
|
|
source_title: Content title, added as a tag on each card.
|
|
|
|
Returns:
|
|
List of created note IDs (excludes skipped duplicates).
|
|
"""
|
|
if not vocab_list:
|
|
return []
|
|
|
|
try:
|
|
_ensure_deck()
|
|
except error.URLError:
|
|
logger.warning("AnkiConnect unreachable — skipping vocab card creation")
|
|
return []
|
|
|
|
tag = source_title[:50].replace(" ", "_") if source_title else "knowledge-inbox"
|
|
|
|
notes = [
|
|
{
|
|
"deckName": DECK_NAME,
|
|
"modelName": MODEL_NAME,
|
|
"fields": {
|
|
"Front": _build_front(
|
|
item.get("word", ""),
|
|
item.get("pos", ""),
|
|
item.get("example", ""),
|
|
),
|
|
"Back": _build_back(
|
|
item.get("definition_en", ""),
|
|
item.get("definition_ko", ""),
|
|
),
|
|
},
|
|
"options": {"allowDuplicate": False},
|
|
"tags": ["knowledge-inbox", tag],
|
|
}
|
|
for item in vocab_list
|
|
if item.get("word")
|
|
]
|
|
|
|
try:
|
|
results = _invoke("addNotes", notes=notes)
|
|
except Exception as exc:
|
|
logger.error("Failed to add Anki cards: %s", exc)
|
|
return []
|
|
|
|
created = [note_id for note_id in results if note_id is not None]
|
|
skipped = len(results) - len(created)
|
|
logger.info(
|
|
"Anki: added %d card(s), skipped %d duplicate(s) for '%s'",
|
|
len(created), skipped, source_title[:40],
|
|
)
|
|
return created
|