uxoxo/eb2ab
0
1# NOTE!!NOTE!!!NOTE!!NOTE!!!NOTE!!NOTE!!!NOTE!!NOTE!!!2# THE WORD "CHAPTER" IN THE CODE DOES NOT MEAN3# IT'S THE REAL CHAPTER OF THE EBOOK SINCE NO STANDARDS4# ARE DEFINING A CHAPTER ON .EPUB FORMAT. THE WORD "BLOCK"5# IS USED TO PRINT IT OUT TO THE TERMINAL, AND "CHAPTER" TO THE CODE6# WHICH IS LESS GENERIC FOR THE DEVELOPERS7 8import argparse, asyncio, csv, fnmatch, hashlib, io, json, math, os, platform, random, shutil, socket, subprocess, sys, tempfile, threading, time, traceback9import unicodedata, urllib.request, uuid, zipfile, ebooklib, gradio as gr, psutil, pymupdf4llm, regex as re, requests, stanza, torch, uvicorn10 11# Import UI helpers for design system integration12from lib.ui_helpers import (13 create_primary_button, create_secondary_button, create_outline_button,14 create_destructive_button, create_section_card, create_status_tag,15 create_alert_box, create_progress_indicator, create_status_display,16 create_cost_estimate_box, create_loading_spinner, get_design_tokens,17 create_hero_section, create_info_card, create_parameter_group,18 create_action_row, create_two_column_layout19)20 21from soynlp.tokenizer import LTokenizer22from pythainlp.tokenize import word_tokenize23from sudachipy import dictionary, tokenizer24from PIL import Image25from tqdm import tqdm26from bs4 import BeautifulSoup, NavigableString, Tag27from collections import Counter28from collections.abc import Mapping29from collections.abc import MutableMapping30from datetime import datetime31from ebooklib import epub32from glob import glob33from iso639 import languages34from markdown import markdown35from multiprocessing import Pool, cpu_count36from multiprocessing import Manager, Event37from multiprocessing.managers import DictProxy, ListProxy38from num2words import num2words39from pathlib import Path40from pydub import AudioSegment41from pydub.utils import mediainfo42from queue import Queue, Empty43from types import MappingProxyType44from urllib.parse import urlparse45from starlette.requests import ClientDisconnect46 47from lib import *48from lib.classes.voice_extractor import VoiceExtractor49from lib.classes.tts_manager import TTSManager50#from lib.classes.redirect_console import RedirectConsole51#from lib.classes.argos_translator import ArgosTranslator52 53context = None54is_gui_process = False55active_sessions = set()56 57#import logging58#logging.basicConfig(59# level=logging.INFO, # DEBUG for more verbosity60# format="%(asctime)s [%(levelname)s] %(message)s"61#)62 63class DependencyError(Exception):64 def __init__(self, message=None):65 super().__init__(message)66 print(message)67 # Automatically handle the exception when it's raised68 self.handle_exception()69 70 def handle_exception(self):71 # Print the full traceback of the exception72 traceback.print_exc() 73 # Print the exception message74 error = f'Caught DependencyError: {self}'75 print(error) 76 # Exit the script if it's not a web process77 if not is_gui_process:78 sys.exit(1)79 80class SessionTracker:81 def __init__(self):82 self.lock = threading.Lock()83 84 def start_session(self, id):85 with self.lock:86 session = context.get_session(id)87 if session['status'] is None:88 session['status'] = 'ready'89 return True90 return False91 92 def end_session(self, id, socket_hash):93 active_sessions.discard(socket_hash)94 with self.lock:95 session = context.get_session(id)96 session['cancellation_requested'] = True97 session['tab_id'] = None98 session['status'] = None99 session[socket_hash] = None100 101class SessionContext:102 def __init__(self):103 self.manager = Manager()104 self.sessions = self.manager.dict()105 self.cancellation_events = {}106 107 def get_session(self, id):108 if id not in self.sessions:109 self.sessions[id] = recursive_proxy({110 "script_mode": NATIVE,111 "id": id,112 "tab_id": None,113 "process_id": None,114 "status": None,115 "event": None,116 "progress": 0,117 "cancellation_requested": False,118 "device": default_device,119 "system": None,120 "client": None,121 "language": default_language_code,122 "language_iso1": None,123 "audiobook": None,124 "audiobooks_dir": None,125 "process_dir": None,126 "ebook": None,127 "ebook_list": None,128 "ebook_mode": "single",129 "chapters_dir": None,130 "chapters_dir_sentences": None,131 "epub_path": None,132 "filename_noext": None,133 "tts_engine": default_tts_engine,134 "fine_tuned": default_fine_tuned,135 "voice": None,136 "voice_dir": None,137 "custom_model": None,138 "custom_model_dir": None,139 "temperature": default_engine_settings[TTS_ENGINES['XTTSv2']]['temperature'],140 "length_penalty": default_engine_settings[TTS_ENGINES['XTTSv2']]['length_penalty'],141 "num_beams": default_engine_settings[TTS_ENGINES['XTTSv2']]['num_beams'],142 "repetition_penalty": default_engine_settings[TTS_ENGINES['XTTSv2']]['repetition_penalty'],143 "top_k": default_engine_settings[TTS_ENGINES['XTTSv2']]['top_k'],144 "top_p": default_engine_settings[TTS_ENGINES['XTTSv2']]['top_p'],145 "speed": default_engine_settings[TTS_ENGINES['XTTSv2']]['speed'],146 "enable_text_splitting": default_engine_settings[TTS_ENGINES['XTTSv2']]['enable_text_splitting'],147 "text_temp": default_engine_settings[TTS_ENGINES['BARK']]['text_temp'],148 "waveform_temp": default_engine_settings[TTS_ENGINES['BARK']]['waveform_temp'],149 "final_name": None,150 "output_format": default_output_format,151 "output_split": default_output_split,152 "output_split_hours": default_output_split_hours,153 "metadata": {154 "title": None, 155 "creator": None,156 "contributor": None,157 "language": None,158 "identifier": None,159 "publisher": None,160 "date": None,161 "description": None,162 "subject": None,163 "rights": None,164 "format": None,165 "type": None,166 "coverage": None,167 "relation": None,168 "Source": None,169 "Modified": None,170 },171 "toc": None,172 "chapters": None,173 "cover": None,174 "duration": 0,175 "playback_time": 0176 }, manager=self.manager)177 return self.sessions[id]178 179 def find_id_by_hash(self, socket_hash):180 for id, session in self.sessions.items():181 if socket_hash in session:182 return session.get('id')183 return None184 185ctx_tracker = SessionTracker()186 187def recursive_proxy(data, manager=None):188 if manager is None:189 manager = Manager()190 if isinstance(data, dict):191 proxy_dict = manager.dict()192 for key, value in data.items():193 proxy_dict[key] = recursive_proxy(value, manager)194 return proxy_dict195 elif isinstance(data, list):196 proxy_list = manager.list()197 for item in data:198 proxy_list.append(recursive_proxy(item, manager))199 return proxy_list200 elif isinstance(data, (str, int, float, bool, type(None))):201 return data202 else:203 error = f"Unsupported data type: {type(data)}"204 print(error)205 return206 207def prepare_dirs(src, session):208 try:209 resume = False210 os.makedirs(os.path.join(models_dir,'tts'), exist_ok=True)211 os.makedirs(session['session_dir'], exist_ok=True)212 os.makedirs(session['process_dir'], exist_ok=True)213 os.makedirs(session['custom_model_dir'], exist_ok=True)214 os.makedirs(session['voice_dir'], exist_ok=True)215 os.makedirs(session['audiobooks_dir'], exist_ok=True)216 session['ebook'] = os.path.join(session['process_dir'], os.path.basename(src))217 if os.path.exists(session['ebook']):218 if compare_files_by_hash(session['ebook'], src):219 resume = True220 if not resume:221 shutil.rmtree(session['chapters_dir'], ignore_errors=True)222 os.makedirs(session['chapters_dir'], exist_ok=True)223 os.makedirs(session['chapters_dir_sentences'], exist_ok=True)224 shutil.copy(src, session['ebook']) 225 return True226 except Exception as e:227 DependencyError(e)228 return False229 230def check_programs(prog_name, command, options):231 try:232 subprocess.run(233 [command, options],234 stdout=subprocess.PIPE, 235 stderr=subprocess.PIPE,236 check=True,237 text=True,238 encoding='utf-8'239 )240 return True, None241 except FileNotFoundError:242 e = f'''********** Error: {prog_name} is not installed! if your OS calibre package version 243 is not compatible you still can run ebook2audiobook.sh (linux/mac) or ebook2audiobook.cmd (windows) **********'''244 DependencyError(e)245 return False, None246 except subprocess.CalledProcessError:247 e = f'Error: There was an issue running {prog_name}.'248 DependencyError(e)249 return False, None250 251def analyze_uploaded_file(zip_path, required_files):252 try:253 if not os.path.exists(zip_path):254 error = f"The file does not exist: {os.path.basename(zip_path)}"255 print(error)256 return False257 files_in_zip = {}258 empty_files = set()259 with zipfile.ZipFile(zip_path, 'r') as zf:260 for file_info in zf.infolist():261 file_name = file_info.filename262 if file_info.is_dir():263 continue264 base_name = os.path.basename(file_name)265 files_in_zip[base_name.lower()] = file_info.file_size266 if file_info.file_size == 0:267 empty_files.add(base_name.lower())268 required_files = [file.lower() for file in required_files]269 missing_files = [f for f in required_files if f not in files_in_zip]270 required_empty_files = [f for f in required_files if f in empty_files]271 if missing_files:272 print(f"Missing required files: {missing_files}")273 if required_empty_files:274 print(f"Required files with 0 KB: {required_empty_files}")275 return not missing_files and not required_empty_files276 except zipfile.BadZipFile:277 error = "The file is not a valid ZIP archive."278 raise ValueError(error)279 except Exception as e:280 error = f"An error occurred: {e}"281 raise RuntimeError(error)282 283def extract_custom_model(file_src, session, required_files=None):284 try:285 model_path = None286 if required_files is None:287 required_files = models[session['tts_engine']][default_fine_tuned]['files']288 model_name = re.sub('.zip', '', os.path.basename(file_src), flags=re.IGNORECASE)289 model_name = get_sanitized(model_name)290 with zipfile.ZipFile(file_src, 'r') as zip_ref:291 files = zip_ref.namelist()292 files_length = len(files)293 tts_dir = session['tts_engine']294 model_path = os.path.join(session['custom_model_dir'], tts_dir, model_name)295 if os.path.exists(model_path):296 print(f'{model_path} already exists, bypassing files extraction')297 return model_path298 os.makedirs(model_path, exist_ok=True)299 required_files_lc = set(x.lower() for x in required_files)300 with tqdm(total=files_length, unit='files') as t:301 for f in files:302 base_f = os.path.basename(f).lower()303 if base_f in required_files_lc:304 out_path = os.path.join(model_path, base_f)305 with zip_ref.open(f) as src, open(out_path, 'wb') as dst:306 shutil.copyfileobj(src, dst)307 t.update(1)308 if is_gui_process:309 os.remove(file_src)310 if model_path is not None:311 msg = f'Extracted files to {model_path}'312 print(msg)313 return model_path314 else:315 error = f'An error occured when unzip {file_src}'316 return None317 except asyncio.exceptions.CancelledError as e:318 DependencyError(e)319 if is_gui_process:320 os.remove(file_src)321 return None 322 except Exception as e:323 DependencyError(e)324 if is_gui_process:325 os.remove(file_src)326 return None327 328def hash_proxy_dict(proxy_dict):329 return hashlib.md5(str(proxy_dict).encode('utf-8')).hexdigest()330 331def calculate_hash(filepath, hash_algorithm='sha256'):332 hash_func = hashlib.new(hash_algorithm)333 with open(filepath, 'rb') as f:334 while chunk := f.read(8192): # Read in chunks to handle large files335 hash_func.update(chunk)336 return hash_func.hexdigest()337 338def compare_files_by_hash(file1, file2, hash_algorithm='sha256'):339 return calculate_hash(file1, hash_algorithm) == calculate_hash(file2, hash_algorithm)340 341def compare_dict_keys(d1, d2):342 if not isinstance(d1, Mapping) or not isinstance(d2, Mapping):343 return d1 == d2344 d1_keys = set(d1.keys())345 d2_keys = set(d2.keys())346 missing_in_d2 = d1_keys - d2_keys347 missing_in_d1 = d2_keys - d1_keys348 if missing_in_d2 or missing_in_d1:349 return {350 "missing_in_d2": missing_in_d2,351 "missing_in_d1": missing_in_d1,352 }353 for key in d1_keys.intersection(d2_keys):354 nested_result = compare_keys(d1[key], d2[key])355 if nested_result:356 return {key: nested_result}357 return None358 359def proxy2dict(proxy_obj):360 def recursive_copy(source, visited):361 # Handle circular references by tracking visited objects362 if id(source) in visited:363 return None # Stop processing circular references364 visited.add(id(source)) # Mark as visited365 if isinstance(source, dict):366 result = {}367 for key, value in source.items():368 result[key] = recursive_copy(value, visited)369 return result370 elif isinstance(source, list):371 return [recursive_copy(item, visited) for item in source]372 elif isinstance(source, set):373 return list(source)374 elif isinstance(source, (int, float, str, bool, type(None))):375 return source376 elif isinstance(source, DictProxy):377 # Explicitly handle DictProxy objects378 return recursive_copy(dict(source), visited) # Convert DictProxy to dict379 else:380 return str(source) # Convert non-serializable types to strings381 return recursive_copy(proxy_obj, set())382 383def convert2epub(id):384 session = context.get_session(id)385 if session['cancellation_requested']:386 print('Cancel requested')387 return False388 try:389 title = False390 author = False391 util_app = shutil.which('ebook-convert')392 if not util_app:393 error = "The 'ebook-convert' utility is not installed or not found."394 print(error)395 return False396 file_input = session['ebook']397 if os.path.getsize(file_input) == 0:398 error = f"Input file is empty: {file_input}"399 print(error)400 return False401 file_ext = os.path.splitext(file_input)[1].lower()402 if file_ext not in ebook_formats:403 error = f'Unsupported file format: {file_ext}'404 print(error)405 return False406 if file_ext == '.pdf':407 import fitz408 msg = 'File input is a PDF. flatten it in MarkDown...'409 print(msg)410 doc = fitz.open(session['ebook'])411 pdf_metadata = doc.metadata412 filename_no_ext = os.path.splitext(os.path.basename(session['ebook']))[0]413 title = pdf_metadata.get('title') or filename_no_ext414 author = pdf_metadata.get('author') or False415 markdown_text = pymupdf4llm.to_markdown(session['ebook'])416 # Remove single asterisks for italics (but not bold **)417 markdown_text = re.sub(r'(?<!\*)\*(?!\*)(.*?)\*(?!\*)', r'\1', markdown_text)418 # Remove single underscores for italics (but not bold __)419 markdown_text = re.sub(r'(?<!_)_(?!_)(.*?)_(?!_)', r'\1', markdown_text)420 file_input = os.path.join(session['process_dir'], f'{filename_no_ext}.md')421 with open(file_input, "w", encoding="utf-8") as html_file:422 html_file.write(markdown_text)423 msg = f"Running command: {util_app} {file_input} {session['epub_path']}"424 print(msg)425 cmd = [426 util_app, file_input, session['epub_path'],427 '--input-encoding=utf-8',428 '--output-profile=generic_eink',429 '--epub-version=3',430 '--flow-size=0',431 '--chapter-mark=pagebreak',432 '--page-breaks-before', "//*[name()='h1' or name()='h2' or name()='h3' or name()='h4' or name()='h5']",433 '--disable-font-rescaling',434 '--pretty-print',435 '--smarten-punctuation',436 '--verbose'437 ]438 if title:439 cmd += ['--title', title]440 if author:441 cmd += ['--authors', author]442 result = subprocess.run(443 cmd,444 stdout=subprocess.PIPE,445 stderr=subprocess.PIPE,446 text=True,447 encoding='utf-8'448 )449 print(result.stdout)450 return True451 except subprocess.CalledProcessError as e:452 print(f"Subprocess error: {e.stderr}")453 DependencyError(e)454 return False455 except FileNotFoundError as e:456 print(f"Utility not found: {e}")457 DependencyError(e)458 return False459 460def get_ebook_title(epubBook, all_docs):461 # 1. Try metadata (official EPUB title)462 meta_title = epubBook.get_metadata("DC", "title")463 if meta_title and meta_title[0][0].strip():464 return meta_title[0][0].strip()465 # 2. Try <title> in the head of the first XHTML document466 if all_docs:467 html = all_docs[0].get_content().decode("utf-8")468 soup = BeautifulSoup(html, "html.parser")469 title_tag = soup.select_one("head > title")470 if title_tag and title_tag.text.strip():471 return title_tag.text.strip()472 # 3. Try <img alt="..."> if no visible <title>473 img = soup.find("img", alt=True)474 if img:475 alt = img['alt'].strip()476 if alt and "cover" not in alt.lower():477 return alt478 return None479 480def get_cover(epubBook, session):481 try:482 if session['cancellation_requested']:483 msg = 'Cancel requested'484 print(msg)485 return False486 cover_image = None487 cover_path = os.path.join(session['process_dir'], session['filename_noext'] + '.jpg')488 for item in epubBook.get_items_of_type(ebooklib.ITEM_COVER):489 cover_image = item.get_content()490 break491 if not cover_image:492 for item in epubBook.get_items_of_type(ebooklib.ITEM_IMAGE):493 if 'cover' in item.file_name.lower() or 'cover' in item.get_id().lower():494 cover_image = item.get_content()495 break496 if cover_image:497 # Open the image from bytes498 image = Image.open(io.BytesIO(cover_image))499 # Convert to RGB if needed (JPEG doesn't support alpha)500 if image.mode in ('RGBA', 'P'):501 image = image.convert('RGB')502 image.save(cover_path, format='JPEG')503 return cover_path504 return True505 except Exception as e:506 DependencyError(e)507 return False508 509def get_chapters(epubBook, session):510 try:511 msg = r'''512*******************************************************************************513NOTE:514The warning "Character xx not found in the vocabulary."515MEANS THE MODEL CANNOT INTERPRET THE CHARACTER AND WILL MAYBE GENERATE516(AS WELL AS WRONG PUNCTUATION POSITION) AN HALLUCINATION TO IMPROVE THIS MODEL,517IT NEEDS TO ADD THIS CHARACTER INTO A NEW TRAINING MODEL.518YOU CAN IMPROVE IT OR ASK TO A TRAINING MODEL EXPERT.519*******************************************************************************520 '''521 print(msg)522 if session['cancellation_requested']:523 print('Cancel requested')524 return False525 # Step 1: Extract TOC (Table of Contents)526 try:527 toc = epubBook.toc # Extract TOC528 toc_list = [529 nt for item in toc if hasattr(item, 'title')530 if (nt := normalize_text(531 str(item.title),532 session['language'],533 session['language_iso1'],534 session['tts_engine']535 )) is not None536 ]537 except Exception as toc_error:538 error = f"Error extracting TOC: {toc_error}"539 print(error)540 # Get spine item IDs541 spine_ids = [item[0] for item in epubBook.spine]542 # Filter only spine documents (i.e., reading order)543 all_docs = [544 item for item in epubBook.get_items_of_type(ebooklib.ITEM_DOCUMENT)545 if item.id in spine_ids546 ]547 if not all_docs:548 return [], []549 title = get_ebook_title(epubBook, all_docs)550 chapters = []551 stanza_nlp = False552 if session['language'] in year_to_decades_languages:553 stanza.download(session['language_iso1'])554 stanza_nlp = stanza.Pipeline(session['language_iso1'], processors='tokenize,ner')555 is_num2words_compat = get_num2words_compat(session['language_iso1'])556 msg = 'Analyzing numbers, maths signs, dates and time to convert in words...'557 print(msg)558 for doc in all_docs:559 sentences_list = filter_chapter(doc, session['language'], session['language_iso1'], session['tts_engine'], stanza_nlp, is_num2words_compat)560 if sentences_list is None:561 break562 elif len(sentences_list) > 0:563 chapters.append(sentences_list)564 if len(chapters) == 0:565 error = 'No chapters found!'566 return None, None567 return toc, chapters568 except Exception as e:569 error = f'Error extracting main content pages: {e}'570 DependencyError(error)571 return None, None572 573def filter_chapter(doc, lang, lang_iso1, tts_engine, stanza_nlp, is_num2words_compat):574 575 def tuple_row(node, last_text_char=None):576 try:577 for child in node.children:578 if isinstance(child, NavigableString):579 text = child.strip()580 if text:581 yield ("text", text)582 last_text_char = text[-1] if text else last_text_char583 584 elif isinstance(child, Tag):585 name = child.name.lower()586 if name in heading_tags:587 title = child.get_text(strip=True)588 if title:589 yield ("heading", title)590 last_text_char = title[-1] if title else last_text_char591 592 elif name == "table":593 yield ("table", child)594 595 else:596 return_data = False597 if name in proc_tags:598 for inner in tuple_row(child, last_text_char):599 return_data = True600 yield inner601 # Track last char if this is text or heading602 if inner[0] in ("text", "heading") and inner[1]:603 last_text_char = inner[1][-1]604 605 if return_data:606 if name in break_tags:607 # Only yield break if last char is NOT alnum or space608 if not (last_text_char and (last_text_char.isalnum() or last_text_char.isspace())):609 yield ("break", TTS_SML['break'])610 elif name in heading_tags or name in pause_tags:611 yield ("pause", TTS_SML['pause'])612 613 else:614 yield from tuple_row(child, last_text_char)615 616 except Exception as e:617 error = f'filter_chapter() tuple_row() error: {e}'618 DependencyError(error)619 return None620 621 try:622 heading_tags = [f'h{i}' for i in range(1, 5)]623 break_tags = ['br', 'p']624 pause_tags = ['div', 'span']625 proc_tags = heading_tags + break_tags + pause_tags626 raw_html = doc.get_body_content().decode("utf-8")627 soup = BeautifulSoup(raw_html, 'html.parser')628 body = soup.body629 if not body or not body.get_text(strip=True):630 return []631 # Skip known non-chapter types632 epub_type = body.get("epub:type", "").lower()633 if not epub_type:634 section_tag = soup.find("section")635 if section_tag:636 epub_type = section_tag.get("epub:type", "").lower()637 excluded = {638 "frontmatter", "backmatter", "toc", "titlepage", "colophon",639 "acknowledgments", "dedication", "glossary", "index",640 "appendix", "bibliography", "copyright-page", "landmark"641 }642 if any(part in epub_type for part in excluded):643 return []644 # remove scripts/styles645 for tag in soup(["script", "style"]):646 tag.decompose()647 tuples_list = list(tuple_row(body))648 if not tuples_list:649 error = 'No tuples_list from body created!'650 print(error)651 return None652 text_list = []653 handled_tables = set()654 prev_typ = None655 for typ, payload in tuples_list:656 if typ == "heading":657 text_list.append(payload.strip())658 elif typ == "break":659 if prev_typ != 'break':660 text_list.append(TTS_SML['break'])661 elif typ == 'pause':662 if prev_typ != 'pause':663 text_list.append(TTS_SML['pause'])664 elif typ == "table":665 table = payload666 if table in handled_tables:667 prev_typ = typ668 continue669 handled_tables.add(table)670 rows = table.find_all("tr")671 if not rows:672 prev_typ = typ673 continue674 headers = [c.get_text(strip=True) for c in rows[0].find_all(["td", "th"])]675 for row in rows[1:]:676 cells = [c.get_text(strip=True).replace('\xa0', ' ') for c in row.find_all("td")]677 if not cells:678 continue679 if len(cells) == len(headers) and headers:680 line = " — ".join(f"{h}: {c}" for h, c in zip(headers, cells))681 else:682 line = " — ".join(cells)683 if line:684 text_list.append(line.strip())685 else:686 text = payload.strip()687 if text:688 text_list.append(text)689 prev_typ = typ690 max_chars = language_mapping[lang]['max_chars'] - 4691 clean_list = []692 i = 0693 while i < len(text_list):694 current = text_list[i]695 if current == "‡break‡":696 if clean_list:697 prev = clean_list[-1]698 if prev in ("‡break‡", "‡pause‡"):699 i += 1700 continue701 if prev and (prev[-1].isalnum() or prev[-1] == ' '):702 if i + 1 < len(text_list):703 next_sentence = text_list[i + 1]704 merged_length = len(prev.rstrip()) + 1 + len(next_sentence.lstrip())705 if merged_length <= max_chars:706 # Merge with space handling707 if not prev.endswith(" ") and not next_sentence.startswith(" "):708 clean_list[-1] = prev + " " + next_sentence709 else:710 clean_list[-1] = prev + next_sentence711 i += 2712 continue713 else:714 clean_list.append(current)715 i += 1716 continue717 clean_list.append(current)718 i += 1719 text = ' '.join(clean_list)720 if not re.search(r"[^\W_]", text):721 error = 'No valid text found!'722 print(error)723 return None724 if stanza_nlp:725 # Check if there are positive integers so possible date to convert726 re_ordinal = re.compile(727 r'(?<!\w)(0?[1-9]|[12][0-9]|3[01])(?:\s|\u00A0)*(?:st|nd|rd|th)(?!\w)',728 re.IGNORECASE729 )730 re_num = re.compile(r'(?<!\w)[-+]?\d+(?:\.\d+)?(?!\w)')731 text = unicodedata.normalize('NFKC', text).replace('\u00A0', ' ')732 if re_num.search(text) and re_ordinal.search(text):733 date_spans = get_date_entities(text, stanza_nlp)734 if date_spans:735 result = []736 last_pos = 0737 for start, end, date_text in date_spans:738 result.append(text[last_pos:start])739 # 1) convert 4-digit years (your original behavior)740 processed = re.sub(741 r"\b\d{4}\b",742 lambda m: year2words(m.group(), lang, lang_iso1, is_num2words_compat),743 date_text744 )745 # 2) convert ordinal days like "16th"/"16 th" -> "sixteenth"746 if is_num2words_compat:747 processed = re_ordinal.sub(748 lambda m: num2words(int(m.group(1)), to="ordinal", lang=(lang_iso1 or "en")),749 processed750 )751 else:752 processed = re_ordinal.sub(753 lambda m: math2words(m.group(), lang, lang_iso1, tts_engine, is_num2words_compat),754 processed755 )756 # 3) convert other numbers (skip 4-digit years)757 def _num_repl(m):758 s = m.group(0)759 # leave years alone (already handled above)760 if re.fullmatch(r"\d{4}", s):761 return s762 n = float(s) if "." in s else int(s)763 if is_num2words_compat:764 return num2words(n, lang=(lang_iso1 or "en"))765 else:766 return math2words(m, lang, lang_iso1, tts_engine, is_num2words_compat)767 768 processed = re_num.sub(_num_repl, processed)769 result.append(processed)770 last_pos = end771 result.append(text[last_pos:])772 text = ''.join(result)773 else:774 if is_num2words_compat:775 text = re_ordinal.sub(776 lambda m: num2words(int(m.group(1)), to="ordinal", lang=(lang_iso1 or "en")),777 text778 )779 else:780 text = re_ordinal.sub(781 lambda m: math2words(int(m.group(1)), lang, lang_iso1, tts_engine, is_num2words_compat),782 text783 )784 text = re.sub(785 r"\b\d{4}\b",786 lambda m: year2words(m.group(), lang, lang_iso1, is_num2words_compat),787 text788 )789 text = roman2number(text)790 text = clock2words(text, lang, lang_iso1, tts_engine, is_num2words_compat)791 text = math2words(text, lang, lang_iso1, tts_engine, is_num2words_compat)792 # build a translation table mapping each bad char to a space793 specialchars_remove_table = str.maketrans({ch: ' ' for ch in specialchars_remove})794 text = text.translate(specialchars_remove_table)795 text = normalize_text(text, lang, lang_iso1, tts_engine)796 # Ensure space before and after punctuation_list797 #pattern_space = re.escape(''.join(punctuation_list))798 #punctuation_pattern_space = r'(?<!\s)([{}])'.format(pattern_space)799 #text = re.sub(punctuation_pattern_space, r' \1', text)800 sentences = get_sentences(text, lang, tts_engine)801 if len(sentences) == 0:802 error = 'No sentences found!'803 print(error)804 return None805 return get_sentences(text, lang, tts_engine)806 except Exception as e:807 error = f'filter_chapter() error: {e}'808 DependencyError(error)809 return None810 811def get_sentences(text, lang, tts_engine):812 813 def split_inclusive(text, pattern):814 result = []815 last_end = 0816 for match in pattern.finditer(text):817 result.append(text[last_end:match.end()].strip())818 last_end = match.end()819 if last_end < len(text):820 tail = text[last_end:].strip()821 if tail:822 result.append(tail)823 return result824 825 def segment_ideogramms(text):826 sml_pattern = "|".join(re.escape(token) for token in sml_tokens)827 segments = re.split(f"({sml_pattern})", text)828 result = []829 try:830 for segment in segments:831 if not segment:832 continue833 # If the segment is a SML token, keep as its own834 if re.fullmatch(sml_pattern, segment):835 result.append(segment)836 else:837 if lang == 'zho':838 import jieba839 result.extend([t for t in jieba.cut(segment) if t.strip()])840 elif lang == 'jpn':841 sudachi = dictionary.Dictionary().create()842 mode = tokenizer.Tokenizer.SplitMode.C843 result.extend([m.surface() for m in sudachi.tokenize(segment, mode) if m.surface().strip()])844 elif lang == 'kor':845 ltokenizer = LTokenizer()846 result.extend([t for t in ltokenizer.tokenize(segment) if t.strip()])847 elif lang in ['tha', 'lao', 'mya', 'khm']:848 result.extend([t for t in word_tokenize(segment, engine='newmm') if t.strip()])849 else:850 result.append(segment.strip())851 return result852 except Exception as e:853 DependencyError(e)854 return [text] 855 856 def join_ideogramms(idg_list):857 try:858 buffer = ''859 for token in idg_list:860 # 1) On sml token: flush & emit buffer, then emit the token861 if token.strip() in sml_tokens:862 if buffer:863 yield buffer864 buffer = ''865 yield token866 continue867 # 2) If adding this token would overflow, flush current buffer first868 if buffer and len(buffer) + len(token) > max_chars:869 yield buffer870 buffer = ''871 # 3) Append the token (word, punctuation, whatever) unless it's a sml token (already checked)872 buffer += token873 # 4) Flush any trailing text874 if buffer:875 yield buffer876 except Exception as e:877 DependencyError(e)878 if buffer:879 yield buffer880 881 try:882 max_chars = language_mapping[lang]['max_chars'] - 4883 min_tokens = 5884 # List or tuple of tokens that must never be appended to buffer885 sml_tokens = tuple(TTS_SML.values())886 sml_list = re.split(rf"({'|'.join(map(re.escape, sml_tokens))})", text)887 sml_list = [s for s in sml_list if s.strip() or s in sml_tokens]888 pattern_split = '|'.join(map(re.escape, punctuation_split_hard_set))889 pattern = re.compile(rf"(.*?(?:{pattern_split}){''.join(punctuation_list_set)})(?=\s|$)", re.DOTALL)890 hard_list = []891 for s in sml_list:892 if s in [TTS_SML['break'], TTS_SML['pause']] or len(s) <= max_chars:893 hard_list.append(s)894 else:895 parts = split_inclusive(s, pattern)896 if parts:897 for text_part in parts:898 text_part = text_part.strip()899 if text_part:900 hard_list.append(text_part)901 else:902 s = s.strip()903 if s:904 hard_list.append(s)905 # Check if some hard_list entries exceed max_chars, so split on soft punctuation906 pattern_split = '|'.join(map(re.escape, punctuation_split_soft_set))907 pattern = re.compile(rf"(.*?(?:{pattern_split}))(?=\s|$)", re.DOTALL)908 soft_list = []909 for s in hard_list:910 if s in [TTS_SML['break'], TTS_SML['pause']] or len(s) <= max_chars:911 soft_list.append(s)912 elif len(s) > max_chars:913 parts = [p for p in split_inclusive(s, pattern) if p]914 if parts:915 buffer = ''916 for idx, part in enumerate(parts):917 # Predict length if we glue this part918 predicted_length = len(buffer) + (1 if buffer else 0) + len(part)919 # Peek ahead to see if gluing will exceed max_chars920 if predicted_length <= max_chars:921 buffer = (buffer + ' ' + part).strip() if buffer else part922 else:923 # If we overshoot, check if buffer ends with punctuation924 if buffer and not any(buffer.rstrip().endswith(p) for p in punctuation_split_soft_set):925 # Try to backtrack to last punctuation inside buffer926 last_punct_idx = max((buffer.rfind(p) for p in punctuation_split_soft_set if p in buffer), default=-1)927 if last_punct_idx != -1:928 soft_list.append(buffer[:last_punct_idx+1].strip())929 leftover = buffer[last_punct_idx+1:].strip()930 buffer = leftover + ' ' + part if leftover else part931 else:932 # No punctuation, just split as-is933 soft_list.append(buffer.strip())934 buffer = part935 else:936 soft_list.append(buffer.strip())937 buffer = part938 if buffer:939 cleaned = re.sub(r'[^\p{L}\p{N} ]+', '', buffer)940 if any(ch.isalnum() for ch in cleaned):941 soft_list.append(buffer.strip())942 else:943 cleaned = re.sub(r'[^\p{L}\p{N} ]+', '', s)944 if any(ch.isalnum() for ch in cleaned):945 soft_list.append(s.strip())946 else:947 cleaned = re.sub(r'[^\p{L}\p{N} ]+', '', s)948 if any(ch.isalnum() for ch in cleaned):949 soft_list.append(s.strip())950 951 if lang in ['zho', 'jpn', 'kor', 'tha', 'lao', 'mya', 'khm']:952 result = []953 for s in soft_list:954 if s in [TTS_SML['break'], TTS_SML['pause']]:955 result.append(s)956 else:957 tokens = segment_ideogramms(s)958 if isinstance(tokens, list):959 result.extend([t for t in tokens if t.strip()])960 else:961 tokens = tokens.strip()962 if tokens:963 result.append(tokens)964 return list(join_ideogramms(result))965 else:966 sentences = []967 for s in soft_list:968 if s in [TTS_SML['break'], TTS_SML['pause']] or len(s) <= max_chars:969 sentences.append(s)970 else:971 words = s.split(' ')972 text_part = words[0]973 for w in words[1:]:974 if len(text_part) + 1 + len(w) <= max_chars:975 text_part += ' ' + w976 else:977 text_part = text_part.strip()978 if text_part:979 sentences.append(text_part)980 text_part = w981 if text_part:982 cleaned = re.sub(r'[^\p{L}\p{N} ]+', '', text_part).strip()983 if not any(ch.isalnum() for ch in cleaned):984 continue985 sentences.append(text_part)986 return sentences987 except Exception as e:988 error = f'get_sentences() error: {e}'989 print(error)990 return None991 992def get_ram():993 vm = psutil.virtual_memory()994 return vm.total // (1024 ** 3)995 996def get_vram():997 os_name = platform.system()998 # NVIDIA (Cross-Platform: Windows, Linux, macOS)999 try:1000 from pynvml import nvmlInit, nvmlDeviceGetHandleByIndex, nvmlDeviceGetMemoryInfo1001 nvmlInit()1002 handle = nvmlDeviceGetHandleByIndex(0) # First GPU1003 info = nvmlDeviceGetMemoryInfo(handle)1004 vram = info.total1005 return int(vram // (1024 ** 3)) # Convert to GB1006 except ImportError:1007 pass1008 except Exception as e:1009 pass1010 # AMD (Windows)1011 if os_name == "Windows":1012 try:1013 cmd = 'wmic path Win32_VideoController get AdapterRAM'1014 output = subprocess.run(cmd, capture_output=True, text=True, shell=True)1015 lines = output.stdout.splitlines()1016 vram_values = [int(line.strip()) for line in lines if line.strip().isdigit()]1017 if vram_values:1018 return int(vram_values[0] // (1024 ** 3))1019 except Exception as e:1020 pass1021 # AMD (Linux)1022 if os_name == "Linux":1023 try:1024 cmd = "lspci -v | grep -i 'VGA' -A 12 | grep -i 'preallocated' | awk '{print $2}'"1025 output = subprocess.run(cmd, capture_output=True, text=True, shell=True)1026 if output.stdout.strip().isdigit():1027 return int(output.stdout.strip()) // 10241028 except Exception as e:1029 pass1030 # Intel (Linux Only)1031 intel_vram_paths = [1032 "/sys/kernel/debug/dri/0/i915_vram_total", # Intel dedicated GPUs1033 "/sys/class/drm/card0/device/resource0" # Some integrated GPUs1034 ]1035 for path in intel_vram_paths:1036 if os.path.exists(path):1037 try:1038 with open(path, "r") as f:1039 vram = int(f.read().strip()) // (1024 ** 3)1040 return vram1041 except Exception as e:1042 pass1043 # macOS (OpenGL Alternative)1044 if os_name == "Darwin":1045 try:1046 from OpenGL.GL import glGetIntegerv1047 from OpenGL.GLX import GLX_RENDERER_VIDEO_MEMORY_MB_MESA1048 vram = int(glGetIntegerv(GLX_RENDERER_VIDEO_MEMORY_MB_MESA) // 1024)1049 return vram1050 except ImportError:1051 pass1052 except Exception as e:1053 pass1054 msg = 'Could not detect GPU VRAM Capacity!'1055 return 01056 1057def get_sanitized(str, replacement="_"):1058 str = str.replace('&', 'And')1059 forbidden_chars = r'[<>:"/\\|?*\x00-\x1F ()]'1060 sanitized = re.sub(r'\s+', replacement, str)1061 sanitized = re.sub(forbidden_chars, replacement, sanitized)1062 sanitized = sanitized.strip("_")1063 return sanitized1064 1065def get_date_entities(text, stanza_nlp):1066 try:1067 doc = stanza_nlp(text)1068 date_spans = []1069 for ent in doc.ents:1070 if ent.type == 'DATE':1071 date_spans.append((ent.start_char, ent.end_char, ent.text))1072 return date_spans1073 except Exception as e:1074 error = f'get_date_entities() error: {e}'1075 print(error)1076 return False1077 1078def get_num2words_compat(lang_iso1):1079 try:1080 test = num2words(1, lang=lang_iso1.replace('zh', 'zh_CN'))1081 return True1082 except NotImplementedError:1083 return False1084 except Exception as e:1085 return False1086 1087def set_formatted_number(text: str, lang, lang_iso1: str, is_num2words_compat: bool, max_single_value: int = 999_999_999_999_999_999):1088 # match up to 18 digits, optional “,…” groups (allowing spaces or NBSP after comma), optional decimal of up to 12 digits1089 # handle optional range with dash/en dash/em dash between numbers, and allow trailing punctuation1090 number_re = re.compile(1091 r'(?<!\w)'1092 r'(\d{1,18}(?:,\s*\d{1,18})*(?:\.\d{1,12})?)' # first number1093 r'(?:\s*([-–—])\s*' # dash type1094 r'(\d{1,18}(?:,\s*\d{1,18})*(?:\.\d{1,12})?))?' # optional second number1095 r'([^\w\s]*)', # optional trailing punctuation1096 re.UNICODE1097 )1098 1099 def normalize_commas(num_str: str) -> str:1100 """Normalize number string to standard comma format: 1,234,567"""1101 tok = num_str.replace('\u00A0', '').replace(' ', '')1102 if '.' in tok:1103 integer_part, decimal_part = tok.split('.', 1)1104 integer_part = integer_part.replace(',', '')1105 integer_part = "{:,}".format(int(integer_part))1106 return f"{integer_part}.{decimal_part}"1107 else:1108 integer_part = tok.replace(',', '')1109 return "{:,}".format(int(integer_part))1110 1111 def clean_single_num(num_str):1112 tok = unicodedata.normalize('NFKC', num_str)1113 if tok.lower() in ('inf', 'infinity', 'nan'):1114 return tok1115 clean = tok.replace(',', '').replace('\u00A0', '').replace(' ', '')1116 try:1117 num = float(clean) if '.' in clean else int(clean)1118 except (ValueError, OverflowError):1119 return tok1120 if not math.isfinite(num) or abs(num) > max_single_value:1121 return tok1122 1123 # Normalize commas before final output1124 tok = normalize_commas(tok)1125 1126 if is_num2words_compat:1127 new_lang_iso1 = lang_iso1.replace('zh', 'zh_CN')1128 return num2words(num, lang=new_lang_iso1)1129 else:1130 phoneme_map = language_math_phonemes.get(1131 lang,1132 language_math_phonemes.get(default_language_code, language_math_phonemes['eng'])1133 )1134 return ' '.join(phoneme_map.get(ch, ch) for ch in str(num))1135 1136 def clean_match(match):1137 first_num = clean_single_num(match.group(1))1138 dash_char = match.group(2) or ''1139 second_num = clean_single_num(match.group(3)) if match.group(3) else ''1140 trailing = match.group(4) or ''1141 if second_num:1142 return f"{first_num}{dash_char}{second_num}{trailing}"1143 else:1144 return f"{first_num}{trailing}"1145 1146 return number_re.sub(clean_match, text)1147 1148def year2words(year_str, lang, lang_iso1, is_num2words_compat):1149 try:1150 year = int(year_str)1151 first_two = int(year_str[:2])1152 last_two = int(year_str[2:])1153 lang_iso1 = lang_iso1 if lang in language_math_phonemes.keys() else default_language_code1154 lang_iso1 = lang_iso1.replace('zh', 'zh_CN')1155 if not year_str.isdigit() or len(year_str) != 4 or last_two < 10:1156 if is_num2words_compat:1157 return num2words(year, lang=lang_iso1)1158 else:1159 return ' '.join(language_math_phonemes[lang].get(ch, ch) for ch in year_str)1160 if is_num2words_compat:1161 return f"{num2words(first_two, lang=lang_iso1)} {num2words(last_two, lang=lang_iso1)}" 1162 else:1163 return ' '.join(language_math_phonemes[lang].get(ch, ch) for ch in first_two) + ' ' + ' '.join(language_math_phonemes[lang].get(ch, ch) for ch in last_two)1164 except Exception as e:1165 error = f'year2words() error: {e}'1166 print(error)1167 raise1168 return False1169 1170def clock2words(text, lang, lang_iso1, tts_engine, is_num2words_compat):1171 time_rx = re.compile(r'(\d{1,2})[:.](\d{1,2})(?:[:.](\d{1,2}))?')1172 lang_lc = (lang or "").lower()1173 lc = language_clock.get(lang_lc) if 'language_clock' in globals() else None1174 _n2w_cache = {}1175 1176 def n2w(n: int) -> str:1177 key = (n, lang_lc, is_num2words_compat)1178 if key in _n2w_cache:1179 return _n2w_cache[key]1180 if is_num2words_compat:1181 word = num2words(n, lang=lang_lc)1182 else:1183 word = math2words(n, lang, lang_iso1, tts_engine, is_num2words_compat)1184 _n2w_cache[key] = word1185 return word1186 1187 def repl_num(m: re.Match) -> str:1188 # Parse hh[:mm[:ss]]1189 try:1190 h = int(m.group(1))1191 mnt = int(m.group(2))1192 sec = m.group(3)1193 sec = int(sec) if sec is not None else None1194 except Exception:1195 return m.group(0)1196 # basic validation; if out of range, keep original1197 if not (0 <= h <= 23 and 0 <= mnt <= 59 and (sec is None or 0 <= sec <= 59)):1198 return m.group(0)1199 # If no language clock rules, just say numbers plainly1200 if not lc: