# anki_exporter.py
import os
import tempfile
import asyncio
import time
import genanki
import edge_tts
import database
from tts_utils import parse_text_for_edgetts, get_configured_tts_rate
async def generate_edge_audio(text, voice, output_path, rate_modifier="+0%"):
"""Asynchronously streams data packages via the Microsoft Edge API pipeline."""
try:
communicate = edge_tts.Communicate(text=text, voice=voice, rate=rate_modifier)
await communicate.save(output_path)
return True
except Exception as e:
print(f"Edge-TTS synthesis anomaly: {e}")
return False
def compile_anki_package(records, output_path, deck_name):
"""
Compiles database records into a bidirectional card payload package.
Resolves voice models dynamically by gender selection parameters and applies
global speed coefficient rates from the active configurations.
"""
# Generate deterministic positive 32-bit integers from deck name and model name
# to avoid collisions across different decks while keeping imports stable
deck_id = abs(hash(deck_name)) % (2**31)
model_id = abs(hash("Spanish Bidirectional Multi-Note HTML Model")) % (2**31)
# Unique timestamp prefix for media files to prevent overwriting prior exports in Anki
run_prefix = int(time.time())
# Global Configuration Pace Resolver Mapping
settings = database.load_all_settings() or {}
rate_string = get_configured_tts_rate(settings)
anki_model = genanki.Model(
model_id,
'Spanish Bidirectional Multi-Note HTML Model',
fields=[
{'name': 'EnglishText'},
{'name': 'SpanishText'},
{'name': 'AnkiNotes'},
{'name': 'EnglishAudio'},
{'name': 'SpanishAudio'}
],
templates=[
{
'name': 'Card 1: English ➔ Spanish',
'qfmt': '
TRANSLATE TO SPANISH:
'
'{{EnglishText}}
'
'{{EnglishAudio}}
',
'afmt': '{{FrontSide}}
'
'{{SpanishText}}
'
'{{#AnkiNotes}}Anki Meta: {{{AnkiNotes}}}
{{/AnkiNotes}}
'
'{{SpanishAudio}}
',
},
{
'name': 'Card 2: Spanish ➔ English',
'qfmt': 'TRANSLATE TO ENGLISH:
'
'{{SpanishText}}
'
'{{SpanishAudio}}
',
'afmt': '{{FrontSide}}
'
'{{EnglishText}}
'
'{{#AnkiNotes}}Anki Meta: {{{AnkiNotes}}}
{{/AnkiNotes}}
'
'{{EnglishAudio}}
',
}
]
)
deck = genanki.Deck(deck_id, deck_name)
media_files_to_pack = []
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
with tempfile.TemporaryDirectory() as tmpdir:
for idx, record in enumerate(records):
en_raw = record["en_text"]
es_raw = record["es_text"]
gender_flag = record.get("gender", "Female")
# Sanitize text payloads for TTS engine
en_tts_text = parse_text_for_edgetts(en_raw)
es_tts_text = parse_text_for_edgetts(es_raw)
# Map native neural voice files matching gender settings
spanish_voice = "es-ES-AlvaroNeural" if gender_flag == "Male" else "es-ES-ElviraNeural"
english_voice = "en-US-EmmaNeural"
raw_tags = record.get("tags") or ""
note_tags = [t.strip().replace(" ", "_") for t in raw_tags.split(",") if t.strip()]
# Safely isolate the raw text/HTML data string down to Anki notes field
anki_notes_html = record['anki_notes'].strip() if record.get('anki_notes') else ""
# Unique Media Filenames combining execution timestamp and index
en_audio_filename = f"edge_en_{run_prefix}_{idx}.mp3"
es_audio_filename = f"edge_es_{run_prefix}_{idx}.mp3"
en_audio_path = os.path.join(tmpdir, en_audio_filename)
es_audio_path = os.path.join(tmpdir, es_audio_filename)
# English Audio Synthesis
if en_tts_text.strip() and loop.run_until_complete(generate_edge_audio(en_tts_text, english_voice, en_audio_path, rate_string)):
media_files_to_pack.append(en_audio_path)
en_audio_field = f"[sound:{en_audio_filename}]"
else:
en_audio_field = ""
# Spanish Audio Synthesis
if es_tts_text.strip() and loop.run_until_complete(generate_edge_audio(es_tts_text, spanish_voice, es_audio_path, rate_string)):
media_files_to_pack.append(es_audio_path)
es_audio_field = f"[sound:{es_audio_filename}]"
else:
es_audio_field = ""
# Clean sequential matching fields array matching schema mapping above
note = genanki.Note(
model=anki_model,
fields=[en_raw, es_raw, anki_notes_html, en_audio_field, es_audio_field],
tags=note_tags
)
deck.add_note(note)
# Build Package while media assets are guaranteed contextually active inside tmpdir
package = genanki.Package(deck)
package.media_files = media_files_to_pack
package.write_to_file(output_path)