html added and working
This commit is contained in:
parent
3caa9f52cf
commit
bd6e42d36a
6 changed files with 196 additions and 72 deletions
|
|
@ -6,6 +6,7 @@ import genanki
|
|||
import edge_tts
|
||||
import shutil
|
||||
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."""
|
||||
|
|
@ -29,14 +30,7 @@ def compile_anki_package(records, output_path, deck_name):
|
|||
|
||||
# Global Configuration Pace Resolver Mapping
|
||||
settings = database.load_all_settings() or {}
|
||||
config_speed = settings.get("tts_playback_speed", "1.0")
|
||||
|
||||
# Transform numeric string floats (e.g., 1.2) into Edge-TTS percentage strings (e.g., +20%)
|
||||
try:
|
||||
pct = int((float(config_speed) - 1.0) * 100)
|
||||
rate_string = f"{'+' if pct >= 0 else ''}{pct}%"
|
||||
except Exception:
|
||||
rate_string = "+0%"
|
||||
rate_string = get_configured_tts_rate(settings)
|
||||
|
||||
anki_model = genanki.Model(
|
||||
model_id,
|
||||
|
|
@ -83,10 +77,14 @@ def compile_anki_package(records, output_path, deck_name):
|
|||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
for idx, record in enumerate(records):
|
||||
en_text = record["en_text"]
|
||||
es_text = record["es_text"]
|
||||
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"
|
||||
|
|
@ -105,14 +103,14 @@ def compile_anki_package(records, output_path, deck_name):
|
|||
es_audio_path = os.path.join(tmpdir, es_audio_filename)
|
||||
|
||||
# English Audio Synthesis
|
||||
if loop.run_until_complete(generate_edge_audio(en_text, english_voice, en_audio_path, rate_string)):
|
||||
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 loop.run_until_complete(generate_edge_audio(es_text, spanish_voice, es_audio_path, rate_string)):
|
||||
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:
|
||||
|
|
@ -121,7 +119,7 @@ def compile_anki_package(records, output_path, deck_name):
|
|||
# Clean sequential matching fields array matching schema mapping above
|
||||
note = genanki.Note(
|
||||
model=anki_model,
|
||||
fields=[en_text, es_text, anki_notes_html, en_audio_field, es_audio_field],
|
||||
fields=[en_raw, es_raw, anki_notes_html, en_audio_field, es_audio_field],
|
||||
tags=note_tags
|
||||
)
|
||||
deck.add_note(note)
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -2,14 +2,19 @@
|
|||
import random
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import asyncio
|
||||
from PyQt6.QtWidgets import (
|
||||
QWidget, QVBoxLayout, QHBoxLayout, QLabel,
|
||||
QLineEdit, QPushButton, QTableWidget, QTableWidgetItem,
|
||||
QHeaderView, QFormLayout, QMessageBox, QFrame, QStackedWidget
|
||||
)
|
||||
from PyQt6.QtCore import Qt, pyqtSlot
|
||||
import edge_tts
|
||||
import database
|
||||
import anki_exporter
|
||||
from tts_utils import parse_text_for_edgetts, get_configured_tts_rate
|
||||
|
||||
class ReviewTab(QWidget):
|
||||
def __init__(self, parent=None):
|
||||
|
|
@ -262,21 +267,74 @@ class ReviewTab(QWidget):
|
|||
self.is_flipped = False
|
||||
self.display_current_card()
|
||||
|
||||
def _async_edge_speech_worker(self, text, voice, rate_modifier):
|
||||
"""Background thread worker to render neural speech with terminal debug logging."""
|
||||
async def stream_audio():
|
||||
temp_file = os.path.join(tempfile.gettempdir(), "review_card_audio.mp3")
|
||||
|
||||
# Clean up old file if present
|
||||
if os.path.exists(temp_file):
|
||||
try:
|
||||
os.remove(temp_file)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
print(f"[TTS Debug] Generating TTS -> Voice: {voice} | Rate: {rate_modifier} | Text: '{text}'")
|
||||
communicate = edge_tts.Communicate(text=text, voice=voice, rate=rate_modifier)
|
||||
await communicate.save(temp_file)
|
||||
|
||||
if os.path.exists(temp_file) and os.path.getsize(temp_file) > 0:
|
||||
print(f"[TTS Debug] Audio ready ({os.path.getsize(temp_file)} bytes). Playing via afplay...")
|
||||
result = subprocess.run(["afplay", temp_file], capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
print(f"[TTS Debug] afplay failed: {result.stderr}")
|
||||
else:
|
||||
print("[TTS Debug] Playback finished successfully.")
|
||||
else:
|
||||
print("[TTS Debug] Error: Audio file was not created or is 0 bytes.")
|
||||
except Exception as e:
|
||||
print(f"[TTS Debug] Exception during speech synthesis: {e}")
|
||||
|
||||
# Explicitly set up and run a clean event loop for this thread
|
||||
try:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
loop.run_until_complete(stream_audio())
|
||||
loop.close()
|
||||
except Exception as e:
|
||||
print(f"[TTS Debug] Event loop error: {e}")
|
||||
|
||||
@pyqtSlot()
|
||||
def play_card_audio(self):
|
||||
"""Auditions native voice files based on active visual canvas sides."""
|
||||
"""Auditions neural edge-tts voice based on active flashcard side."""
|
||||
if not (0 <= self.current_index < len(self.filtered_review_pool)):
|
||||
return
|
||||
|
||||
record = self.filtered_review_pool[self.current_index]
|
||||
settings = database.load_all_settings() or {}
|
||||
rate_string = get_configured_tts_rate(settings)
|
||||
|
||||
if not self.is_flipped:
|
||||
# Play English text utilizing standard native subsystem default output
|
||||
if record["en_text"]:
|
||||
subprocess.Popen(["say", record["en_text"]])
|
||||
raw_text = record.get("en_text", "")
|
||||
spoken_text = parse_text_for_edgetts(raw_text)
|
||||
voice = "en-US-EmmaNeural"
|
||||
else:
|
||||
# Play target translation explicitly targeting the Castilian voice profile Mónica
|
||||
if record["es_text"]:
|
||||
subprocess.Popen(["say", "-v", "Monica", record["es_text"]])
|
||||
raw_text = record.get("es_text", "")
|
||||
spoken_text = parse_text_for_edgetts(raw_text)
|
||||
is_male = (record.get("gender") == "Male")
|
||||
voice = "es-ES-AlvaroNeural" if is_male else "es-ES-ElviraNeural"
|
||||
|
||||
# Respect <meta sound-off> flags or empty entries
|
||||
if not spoken_text.strip():
|
||||
print("[TTS Debug] Skipped: Parsed text is empty or muted via sound-off tag.")
|
||||
return
|
||||
|
||||
threading.Thread(
|
||||
target=self._async_edge_speech_worker,
|
||||
args=(spoken_text, voice, rate_string),
|
||||
daemon=True
|
||||
).start()
|
||||
|
||||
@pyqtSlot()
|
||||
def generate_deck_action(self):
|
||||
|
|
@ -288,14 +346,10 @@ class ReviewTab(QWidget):
|
|||
|
||||
try:
|
||||
# Load active settings dictionary directly from your database configurations
|
||||
settings = database.load_all_settings()
|
||||
# print("--- CURRENT DATABASE SETTINGS ---")
|
||||
# for key, value in settings.items():
|
||||
# print(f"{key}: {value}")
|
||||
# print("---------------------------------")
|
||||
settings = database.load_all_settings() or {}
|
||||
|
||||
# Extract configurations targeting exact database schema names found in settings
|
||||
target_dir = settings.get("anki_export_directory")
|
||||
#root_deck_name = settings.get("default_deck_name")
|
||||
root_deck_name = settings.get("anki_root_deck_name")
|
||||
sub_deck_hierarchy = settings.get("anki_sub_deck_name")
|
||||
|
||||
|
|
@ -316,8 +370,6 @@ class ReviewTab(QWidget):
|
|||
if sub_deck_hierarchy and str(sub_deck_hierarchy).strip():
|
||||
deck_tree_parts.append(str(sub_deck_hierarchy).strip())
|
||||
|
||||
## deck_tree_parts.append("Filtered Review Session")
|
||||
|
||||
# Join parts using Anki double-colon syntax (::)
|
||||
full_deck_namespace = "::".join(deck_tree_parts)
|
||||
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ from PyQt6.QtWidgets import (
|
|||
from PyQt6.QtCore import pyqtSignal, pyqtSlot, Qt
|
||||
import edge_tts
|
||||
import database
|
||||
from tts_utils import parse_text_for_edgetts, get_configured_tts_rate
|
||||
|
||||
|
||||
class TextEditorDialog(QDialog):
|
||||
"""A pop-up modal containing a large text field workspace for copy-pasting extra text blocks or drafting HTML content."""
|
||||
|
|
@ -360,42 +362,35 @@ class SandboxTab(QWidget):
|
|||
|
||||
@pyqtSlot()
|
||||
def audition_english(self):
|
||||
txt = self.txt_english.text().strip()
|
||||
if not txt:
|
||||
raw_txt = self.txt_english.text().strip()
|
||||
spoken_text = parse_text_for_edgetts(raw_txt)
|
||||
|
||||
if not spoken_text.strip():
|
||||
return
|
||||
|
||||
settings = database.load_all_settings() or {}
|
||||
config_speed = settings.get("tts_playback_speed", "1.0")
|
||||
try:
|
||||
pct = int((float(config_speed) - 1.0) * 100)
|
||||
rate_string = f"{'+' if pct >= 0 else ''}{pct}%"
|
||||
except Exception:
|
||||
rate_string = "+0%"
|
||||
rate_string = get_configured_tts_rate(settings)
|
||||
|
||||
threading.Thread(
|
||||
target=self._async_edge_speech_worker,
|
||||
args=(txt, "en-US-EmmaNeural", rate_string),
|
||||
args=(spoken_text, "en-US-EmmaNeural", rate_string),
|
||||
daemon=True
|
||||
).start()
|
||||
|
||||
@pyqtSlot()
|
||||
def audition_spanish(self):
|
||||
txt = self.txt_spanish.text().strip()
|
||||
if not txt:
|
||||
raw_txt = self.txt_spanish.text().strip()
|
||||
spoken_text = parse_text_for_edgetts(raw_txt)
|
||||
|
||||
if not spoken_text.strip():
|
||||
return
|
||||
|
||||
voice = "es-ES-AlvaroNeural" if self.rb_male.isChecked() else "es-ES-ElviraNeural"
|
||||
|
||||
settings = database.load_all_settings() or {}
|
||||
config_speed = settings.get("tts_playback_speed", "1.0")
|
||||
try:
|
||||
pct = int((float(config_speed) - 1.0) * 100)
|
||||
rate_string = f"{'+' if pct >= 0 else ''}{pct}%"
|
||||
except Exception:
|
||||
rate_string = "+0%"
|
||||
rate_string = get_configured_tts_rate(settings)
|
||||
|
||||
threading.Thread(
|
||||
target=self._async_edge_speech_worker,
|
||||
args=(txt, voice, rate_string),
|
||||
args=(spoken_text, voice, rate_string),
|
||||
daemon=True
|
||||
).start()
|
||||
77
test_tts.py
77
test_tts.py
|
|
@ -1,28 +1,55 @@
|
|||
import asyncio
|
||||
import os
|
||||
import edge_tts
|
||||
# tts_utils.py
|
||||
import re
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
# 1. Define the phrase, output path, and target Castellano voice
|
||||
SPANISH_TEXT = "¡Buenos días! ¿Cómo estás? Bienvenido a tu curso de español."
|
||||
OUTPUT_FILE = "media/buenos_dias_castellano-Female.mp3"
|
||||
#VOICE = "es-ES-AlvaroNeural" # Swap to "es-ES-ElviraNeural" if you prefer a female tone
|
||||
VOICE = "es-ES-ElviraNeural" # Swap to "es-ES-ElviraNeural" if you prefer a female tone
|
||||
|
||||
async def generate_castilian_audio():
|
||||
# Ensure our target media folder exists locally
|
||||
os.makedirs(os.path.dirname(OUTPUT_FILE), exist_ok=True)
|
||||
|
||||
print(f"🔄 Synthesizing text using Castilian voice: {VOICE}...")
|
||||
|
||||
# 2. Configure the Communicate engine
|
||||
communicate = edge_tts.Communicate(SPANISH_TEXT, VOICE)
|
||||
|
||||
# 3. Stream and write the data packets to disk
|
||||
await communicate.save(OUTPUT_FILE)
|
||||
|
||||
print(f"✨ Success! MP3 file exported safely to: {OUTPUT_FILE}")
|
||||
print(f"📂 File size: {os.path.getsize(OUTPUT_FILE)} bytes")
|
||||
def get_configured_tts_rate(settings: dict) -> str:
|
||||
"""Converts a numerical speed multiplier (e.g., 0.75, 1.0, 1.25) from app settings
|
||||
into Edge TTS percentage format string (e.g., '-25%', '+0%', '+25%').
|
||||
"""
|
||||
# Check the actual database key 'tts_playback_speed' with fallbacks
|
||||
raw_val = (
|
||||
settings.get("tts_playback_speed")
|
||||
or settings.get("tts_speed_multiplier")
|
||||
or 1.0
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run the async loop loop natively
|
||||
asyncio.run(generate_castilian_audio())
|
||||
try:
|
||||
raw_rate = float(raw_val)
|
||||
except (TypeError, ValueError):
|
||||
raw_rate = 1.0
|
||||
|
||||
# Calculate percentage shift relative to baseline 1.0
|
||||
pct = int(round((raw_rate - 1.0) * 100))
|
||||
if pct >= 0:
|
||||
return f"+{pct}%"
|
||||
return f"{pct}%"
|
||||
|
||||
|
||||
def parse_text_for_edgetts(html_content: str) -> str:
|
||||
"""Strips content between <meta sound-off> and <meta sound-on> tags,
|
||||
handling optional whitespace inside the tag brackets (e.g., <meta sound-off >).
|
||||
Converts remaining HTML tags to clean spoken text.
|
||||
"""
|
||||
if not html_content:
|
||||
return ""
|
||||
|
||||
# Flexible regex to slice out everything from <meta sound-off ...> through <meta sound-on ...>
|
||||
pattern = re.compile(
|
||||
r"<meta\s+sound-off\s*\/?>.*?<meta\s+sound-on\s*\/?>",
|
||||
re.DOTALL | re.IGNORECASE,
|
||||
)
|
||||
cleaned_html = re.sub(pattern, "", html_content)
|
||||
|
||||
# Handle unclosed <meta sound-off> (mute rest of string from that point)
|
||||
if re.search(r"<meta\s+sound-off\s*\/?>", cleaned_html, re.IGNORECASE):
|
||||
cleaned_html = re.split(
|
||||
r"<meta\s+sound-off\s*\/?>", cleaned_html, flags=re.IGNORECASE
|
||||
)[0]
|
||||
|
||||
# Convert remaining HTML into plain text for speech
|
||||
soup = BeautifulSoup(cleaned_html, "html.parser")
|
||||
text = soup.get_text(separator=" ")
|
||||
|
||||
# Normalize extra whitespace
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
52
tts_utils.py
Normal file
52
tts_utils.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
# tts_utils.py
|
||||
import re
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
|
||||
def get_configured_tts_rate(settings: dict) -> str:
|
||||
"""Extracts speed multiplier from settings dict and converts to Edge TTS percentage string (e.g., '-25%')."""
|
||||
# Query the exact key name shown in SQLite database: 'tts_playback_speed'
|
||||
raw_val = settings.get("tts_playback_speed", 1.0) if settings else 1.0
|
||||
|
||||
try:
|
||||
raw_rate = float(raw_val)
|
||||
except (TypeError, ValueError):
|
||||
raw_rate = 1.0
|
||||
|
||||
# Calculate percentage shift relative to baseline 1.0 (e.g., 0.75 -> -25%)
|
||||
pct = int(round((raw_rate - 1.0) * 100))
|
||||
rate_str = f"+{pct}%" if pct >= 0 else f"{pct}%"
|
||||
|
||||
print(
|
||||
f"[TTS Debug] DB Key 'tts_playback_speed': {raw_val} -> Formatted Rate: {rate_str}"
|
||||
)
|
||||
return rate_str
|
||||
|
||||
|
||||
def parse_text_for_edgetts(html_content: str) -> str:
|
||||
"""Strips content between <meta sound-off> and <meta sound-on> tags,
|
||||
handling optional whitespace inside the tag brackets (e.g., <meta sound-off >).
|
||||
Converts remaining HTML tags to clean spoken text.
|
||||
"""
|
||||
if not html_content:
|
||||
return ""
|
||||
|
||||
# Flexible regex to slice out everything from <meta sound-off ...> through <meta sound-on ...>
|
||||
pattern = re.compile(
|
||||
r"<meta\s+sound-off\s*\/?>.*?<meta\s+sound-on\s*\/?>",
|
||||
re.DOTALL | re.IGNORECASE,
|
||||
)
|
||||
cleaned_html = re.sub(pattern, "", html_content)
|
||||
|
||||
# Handle unclosed <meta sound-off> (mute rest of string from that point)
|
||||
if re.search(r"<meta\s+sound-off\s*\/?>", cleaned_html, re.IGNORECASE):
|
||||
cleaned_html = re.split(
|
||||
r"<meta\s+sound-off\s*\/?>", cleaned_html, flags=re.IGNORECASE
|
||||
)[0]
|
||||
|
||||
# Convert remaining HTML into plain text for speech
|
||||
soup = BeautifulSoup(cleaned_html, "html.parser")
|
||||
text = soup.get_text(separator=" ")
|
||||
|
||||
# Normalize extra whitespace
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
Loading…
Reference in a new issue