from result import Result, Ok, Err import requests import logging from typing import List, Dict, Tuple, Any logger = logging.getLogger("flashcard-creator") class AnkiError(Exception): """Base class for Anki-related errors""" pass class ConnectionError(AnkiError): """Error connecting to AnkiConnect""" pass class DeckError(AnkiError): """Error related to deck operations""" pass class CardError(AnkiError): """Error related to card operations""" pass class AnkiConnect: def __init__(self, url="http://127.0.0.1:8765"): self.url = url def get_decks(self) -> Result[List[str], AnkiError]: """ Get the list of decks from Anki using AnkiConnect. Returns Ok(list_of_deck_names) on success, Err(AnkiError) on failure. """ try: response = requests.post( self.url, json={"action": "deckNames", "version": 6} ) data = response.json() return Ok(data.get("result", [])) except requests.exceptions.RequestException as e: logger.error(f"Connection error getting Anki decks: {e}") return Err(ConnectionError(f"Failed to connect to Anki: {e}")) except Exception as e: logger.error(f"Error getting Anki decks: {e}") return Err(DeckError(f"Failed to get decks: {e}")) def create_cards( self, deck: str, cards: List[Dict[str, str]], source_language: str, target_language: str, ) -> Result[Tuple[int, List[Dict[str, Any]]], AnkiError]: """Create flashcards in Anki using AnkiConnect, adding each note one by one. Returns: Ok((successful_count, failed_cards_list)): Where failed_cards_list contains dictionaries like {'card': original_card_dict, 'error': error_message}. Err(ConnectionError): If connection to AnkiConnect fails. Err(CardError): If a non-connection error occurs during the process. """ try: logger.info(f"Preparing to create {len(cards)} Anki cards in deck: {deck}") failed_cards = [] successful_count = 0 # Add each note individually for card in cards: note = { "deckName": deck, "modelName": "Basic", # Using the basic model "fields": {"Front": card["front"], "Back": card["back"]}, "options": {"allowDuplicate": False}, "tags": [f"{source_language}-{target_language}"], } # Send request for a single note response = requests.post( self.url, json={"action": "addNote", "version": 6, "params": {"note": note}}, ) data = response.json() # Check if the note was added successfully if data.get("error") is not None or data.get("result") is None: error_msg = data.get("error", "Unknown error") failed_cards.append({"card": card, "error": error_msg}) logger.warning( f"Failed to add card: {card['front']} - Error: {error_msg}" ) else: successful_count += 1 logger.debug(f"Successfully added card: {card['front']}") # Report results if failed_cards: logger.warning( f"Some cards failed to be added to Anki. Added {successful_count}/{len(cards)}" ) elif successful_count > 0: logger.info(f"Successfully added all {successful_count} cards to Anki") else: logger.info( "No cards were added." ) # Should ideally not happen if input list wasn't empty # Always return Ok with success count and failed list (which might be empty) # The caller can decide if the presence of failed_cards constitutes an overall "error" state for the UI return Ok((successful_count, failed_cards)) except requests.exceptions.RequestException as e: logger.error(f"Connection error creating Anki cards: {e}") return Err(ConnectionError(f"Failed to connect to Anki: {e}")) except Exception as e: logger.error(f"Error creating Anki cards: {e}") return Err(CardError(f"Failed to create cards: {e}"))