From 7cb0894abd4ffa31919bf7ee663fd9a1f6e468c9 Mon Sep 17 00:00:00 2001 From: Wolfang Torres Date: Sat, 27 Jun 2026 12:52:16 +0800 Subject: [PATCH] add bulk procesing --- .editorconfig | 4 + pyproject.toml | 4 +- src/anki_hsk_creator/__about__.py | 2 +- src/anki_hsk_creator/__main__.py | 89 +++++++++++++++----- src/anki_hsk_creator/anki_generation.py | 37 +++++++- src/anki_hsk_creator/api.py | 41 ++++++++- src/anki_hsk_creator/constants.py | 4 +- src/anki_hsk_creator/proccessor.py | 104 ++++++++++++++++++----- src/anki_hsk_creator/utility.py | 107 ++++++++++++++++++++++-- 9 files changed, 336 insertions(+), 56 deletions(-) diff --git a/.editorconfig b/.editorconfig index 53c0a85..d5fc940 100644 --- a/.editorconfig +++ b/.editorconfig @@ -24,3 +24,7 @@ indent_style = tab [*.{js,css,scss,html,xml}] indent_style = space indent_size = 2 + +[*.{dictionary,dictation,phrases}.txt] +indent_style = tab +indent_size = 8 diff --git a/pyproject.toml b/pyproject.toml index 18a7a33..c59eabf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ dependencies = [ "torchcodec", "python-dotenv", "qwen-tts", - + # "flash-attn" ] [project.optional-dependencies] @@ -75,7 +75,7 @@ extra-dependencies = [ [tool.hatch.envs.default.scripts] format = "black --target-version=py314 src tests && isort src tests" -lint = "flake8 src" +lint = "flake8 src" [tool.hatch.envs.types] extra-dependencies = [ diff --git a/src/anki_hsk_creator/__about__.py b/src/anki_hsk_creator/__about__.py index a88217b..fa228b1 100644 --- a/src/anki_hsk_creator/__about__.py +++ b/src/anki_hsk_creator/__about__.py @@ -3,4 +3,4 @@ # SPDX-FileCopyrightText: 2026-present Wolfang Torres # # SPDX-License-Identifier: GPL-3.0-or-later -__version__ = "0.1.3" +__version__ = "0.2.0" diff --git a/src/anki_hsk_creator/__main__.py b/src/anki_hsk_creator/__main__.py index 4047ae4..cfd49c1 100644 --- a/src/anki_hsk_creator/__main__.py +++ b/src/anki_hsk_creator/__main__.py @@ -5,6 +5,7 @@ from pathlib import Path # Local from .api import ( + folder_proccess, is_file, list_input_files, pre_process_a_dictionary_file, @@ -12,9 +13,20 @@ from .api import ( process_a_dictionary_file, process_a_phrases_file, select_file, + select_folder, ) from .constants import DICTATION_TYPE, DICT_TYPE, LANGUAGES, PHRASES_TYPE -from .utility import ProcessFile +from .utility import ProcessFile, ProcessFolder + + +def cli_choose_work_type() -> str: + """entry point for interactive interface""" + option = None + while not option in ["s", "f", "e"]: + option = input( + "Please select the work option:\n" "s: single file, f:folder, e:exit\n" + ) + return option def cli_select_files() -> ProcessFile: @@ -38,6 +50,30 @@ def cli_select_files() -> ProcessFile: return input_file +def cli_select_folder() -> ProcessFile: + """Loops until it finds a valid input_folder""" + print("Select data folder:") + in_folder = None + level = Path() + while not in_folder: + files = list_input_files(level) + for n, file in enumerate(files): + print(f"{n+1} - {file}") + s = None + while not s or not s.isnumeric() or not 1 <= int(s) <= len(files): + s = input(f"Please select the file [1-{len(files)}]: ") + selected = files[int(s) - 1] + print(f"Selected {selected}") + s = None + while s not in ("yes", "y", "no", "n"): + s = input("if this the folder? (yes/no, go inside)") + if s in ("yes", "y"): + in_folder = selected + else: + level = selected + return select_folder(in_folder) + + def cli_select_dictionay_tsv() -> bool: """If a dictionary file is selected, ask if the user wants to proccess it""" s = None @@ -74,29 +110,38 @@ def cli_select_language(languages: list = None) -> str: def main(): """CLI interface for the module""" while True: - input_file = cli_select_files() - if DICT_TYPE in input_file.input_file.suffixes: - dict_selected = cli_select_dictionay_tsv() - if dict_selected: + option = cli_choose_work_type() + if option == "e": + break + elif option == "s": + input_file = cli_select_files() + if DICT_TYPE in input_file.input_file.suffixes: + dict_selected = cli_select_dictionay_tsv() + if dict_selected: + language_id = cli_select_language() + print( + f"pre-processing file {input_file} with language {language_id}" + ) + pre_process_a_dictionary_file(input_file, language_id) + else: + print(f"Processing file {input_file} with language {language_id}") + language_id = cli_select_language( + input_file.available_dictionary_languages + ) + process_a_dictionary_file(input_file, language_id) + elif PHRASES_TYPE in input_file.input_file.suffixes: language_id = cli_select_language() - pre_process_a_dictionary_file(input_file, language_id) - else: - language_id = cli_select_language( - input_file.available_dictionary_languages - ) - process_a_dictionary_file(input_file, language_id) - elif PHRASES_TYPE in input_file.input_file.suffixes: + print(f"processing file {input_file} with language {language_id}") + process_a_phrases_file(input_file, language_id) + elif DICTATION_TYPE in input_file.input_file.suffixes: + language_id = cli_select_language() + print(f"processing file {input_file} with language {language_id}") + process_a_dictation_file(input_file, language_id) + elif option == "f": + in_folder = cli_select_folder() language_id = cli_select_language() - print( - f"processing file {input_file.input_file} with language {language_id}" - ) - process_a_phrases_file(input_file, language_id) - elif DICTATION_TYPE in input_file.input_file.suffixes: - language_id = cli_select_language() - print( - f"processing file {input_file.input_file} with language {language_id}" - ) - process_a_dictation_file(input_file, language_id) + print(f"Selected: {in_folder} with language {language_id}") + folder_proccess(in_folder, language_id=language_id) if __name__ == "__main__": diff --git a/src/anki_hsk_creator/anki_generation.py b/src/anki_hsk_creator/anki_generation.py index d735415..3791e6b 100644 --- a/src/anki_hsk_creator/anki_generation.py +++ b/src/anki_hsk_creator/anki_generation.py @@ -12,7 +12,7 @@ from genanki import Deck, Model, Note, Package from pinyin_tone_converter.pinyin_tone_converter import PinyinToneConverter # Local -from .utility import DictionaryResult, ProcessFile, TranslationResult +from .utility import DictionaryResult, ProcessFile, ProcessFolder, TranslationResult # Constants @@ -153,7 +153,6 @@ def output_anki_dictation( f"Deck for {final_file.name}, " "created in https://www.wolfang.info.ve/hskankicreator/", ) - package = Package(deck) audios = [] for result in results: note = Note( @@ -166,6 +165,7 @@ def output_anki_dictation( ) deck.add_note(note) audios.append(result.audio_path) + package = Package(deck) package.media_files = audios package.write_to_file(final_file) return final_file @@ -236,3 +236,36 @@ def output_anki_phrase( package.media_files = audios package.write_to_file(final_file) return final_file + + +def output_anki_package( + process_folder: ProcessFolder, results: dict[ProcessFile, list[TranslationResult]] +): + final_file = process_file.output_name.with_suffix(".apkg") + decks = [] + audios = [] + for process_file, results in results.items(): + deck_name = "::".join( + ProcessFolder.input_folder.parts[:-1] + (ProcessFolder.output_name.stem,) + ) + deck = Deck( + random.randrange(1 << 30, 1 << 31), + deck_name, + f"Deck for {final_file.name}, " + "created in https://www.wolfang.info.ve/hskankicreator/", + ) + for result in results: + note = Note( + model=DICTATION_MODEL, + fields=[ + result.translated, + result.line, + f"[sound:{result.audio_path}]", + ], + ) + deck.add_note(note) + audios.append(result.audio_path) + decks.append(deck) + package = Package(decks) + package.media_files = audios + package.write_to_file(final_file) diff --git a/src/anki_hsk_creator/api.py b/src/anki_hsk_creator/api.py index 16a1af2..8345fdc 100644 --- a/src/anki_hsk_creator/api.py +++ b/src/anki_hsk_creator/api.py @@ -24,11 +24,12 @@ from .constants import ( ) from .proccessor import ( dictation_process, + dictionary_bulk_process, dictionary_pre_process, dictionary_process, translator_process, ) -from .utility import CCCEDICT, TRANS, TTS, ProcessFile +from .utility import CCCEDICT, TRANS, TTS, ProcessFile, ProcessFolder # interface @@ -80,6 +81,14 @@ def select_file(file_path: Path) -> ProcessFile: raise ValueError(f"{file_path} is not a file") +def select_folder(file_path: Path) -> ProcessFile: + """Given a relative path from `list_input_files`, return a ProcessFile""" + if (INPUT / file_path).is_dir(): + return ProcessFile(file_path) + else: + raise ValueError(f"{file_path} is not a folder") + + def create_folder(file_path: Path) -> ProcessFile: """Creates a folder in a file_path""" input_folder = INPUT / file_path @@ -237,3 +246,33 @@ def process_a_phrases_file(process_file: ProcessFile, language_id: str) -> Path: text_lines = [line.strip() for line in file.readlines() if line.strip()] results = translator_process(text_lines, process_file) return output_anki_phrase(process_file, results) + + +def folder_proccess(process_folder: ProcessFolder, language_id: str): + process_folder.language_id = language_id + TTS.create_tts() + TRANS.create_translator(LANGUAGES.CN, language_id) + CCCEDICT.create_cedict(language_id) + results = {} + for file in process_folder.input_files: + if file.file_type is DICT_TYPE: + with file.absolute_input_file.open( + "r", encoding="utf8", newline="\n" + ) as file: + words_list = [word.strip() for word in file.readlines() if word] + dictionary_bulk_process(words_list, file) + results[file] = dictionary_process(file) + elif file.file_type is DICTATION_TYPE: + with file.absolute_input_file.open( + "r", encoding="utf8", newline="\n" + ) as file: + text_lines = [ + line.strip() for line in file.read().split("。") if line.strip() + ] + results[file] = dictation_process(text_lines, file) + elif file.file_type is PHRASES_TYPE: + with file.absolute_input_file.open( + "r", encoding="utf8", newline="\n" + ) as file: + text_lines = [line.strip() for line in file.readlines() if line.strip()] + results[file] = translator_process(text_lines, file) diff --git a/src/anki_hsk_creator/constants.py b/src/anki_hsk_creator/constants.py index d9689a6..b3cbf0e 100644 --- a/src/anki_hsk_creator/constants.py +++ b/src/anki_hsk_creator/constants.py @@ -34,7 +34,8 @@ class LANGUAGES: RU = "ru" TR = "tr" TH = "th" - AvailableLanguages = (EN, ES, FR, RU, TR, TH) + JP = "jp" + AvailableLanguages = (EN, ES, FR, RU, TR, TH, JP) LanguageNames = { EN: "English", ES: "Spanish", @@ -42,4 +43,5 @@ class LANGUAGES: RU: "Russian", TR: "Turkish", TH: "Thai", + JP: "Japanese", } diff --git a/src/anki_hsk_creator/proccessor.py b/src/anki_hsk_creator/proccessor.py index b90f71b..4ffc727 100644 --- a/src/anki_hsk_creator/proccessor.py +++ b/src/anki_hsk_creator/proccessor.py @@ -2,6 +2,7 @@ # Standard Library import csv +import uuid # Pip import argostranslate.translate @@ -13,7 +14,7 @@ from .utility import CCCEDICT, TTS, DictionaryResult, ProcessFile, TranslationRe # Constants -FIELDNAMES = ["simplified", "traditional", "pinyin", "meaning"] +FIELDNAMES = ["n" "simplified", "traditional", "pinyin", "meaning"] DIALECT = "excel-tab" # Results Classes @@ -26,9 +27,6 @@ def dictation_process( results = [] for n, line in enumerate(text_lines): audio_line = line.strip() - # line = " ".join(line.split()) - # audio_line = " ".join(line) - # audio_line = audio_line.replace("，", "，。。。]") audio_path = process_file.resources / f"N{n:03n}.wav" if not audio_path.exists(): wavs, sr = TTS.generate(f"{audio_line}。") @@ -49,9 +47,6 @@ def translator_process( results = [] for n, line in enumerate(text_lines): audio_line = line.strip() - # line = " ".join(line.split()) - # audio_line = " ".join(line) - # audio_line = audio_line.replace("，", "，。。。]") audio_path = process_file.resources / f"N{n:03n}.wav" if not audio_path.exists(): wavs, sr = TTS.generate(f"{audio_line}。") @@ -65,6 +60,61 @@ def translator_process( return results +def dictionary_bulk_process(words_list: list[str], process_file: ProcessFile): + dictionary = CCCEDICT.create_cedict(process_file.language_id) + with process_file.dictionary_resource_file.open( + "w", encoding="utf8", newline="" + ) as resource_file: + tsv_writer = csv.DictWriter( + resource_file, dialect=DIALECT, fieldnames=FIELDNAMES + ) + tsv_writer.writeheader() + number = 1 + for words in words_list: + word = words.split()[0] + hint = " ".join(words.split()[1:]) if len(words.split()) > 1 else None + if entries := dictionary.get(word): + if len(entries) > 1: + print(f"\nWARNING: {word} has multiple meanings:") + pos_meanings = ( + [ + entry + for entry in entries + for meaning in entry.meaning + if hint in meaning + ] + if hint + else [] + ) + pos_entries = pos_meanings or entries + meanings = "\n".join( + f"{i} - {entry.meaning}" for i, entry in enumerate(pos_entries) + ) + tsv_writer.writerow( + { + "n": n, + "simplified": entry.simplified, + "traditional": entry.traditional, + "pinyin": entry.pinyin, + "meaning": meanings, + } + ) + else: + print("============================================") + print(f"===================>ERROR: {word} not found") + print("============================================") + tsv_writer.writerow( + { + "n": number, + "simplified": word, + "traditional": None, + "pinyin": None, + "meaning": None, + } + ) + number += 1 + + def dictionary_pre_process(words_list: list[str], process_file: ProcessFile): """Pre Process dictionary files into a intermediary resources file""" dictionary = CCCEDICT.create_cedict(process_file.language_id) @@ -75,36 +125,52 @@ def dictionary_pre_process(words_list: list[str], process_file: ProcessFile): resource_file, dialect=DIALECT, fieldnames=FIELDNAMES ) tsv_writer.writeheader() + n = 1 for words in words_list: word = words.split()[0] - pinyin = " ".join(words.split()[1:]) if len(words.split()) > 1 else None + hint = " ".join(words.split()[1:]) if len(words.split()) > 1 else None if entries := dictionary.get(word): - if pinyin is not None: - entries = list(filter(lambda x: x.pinyin == pinyin, entries)) + # if pinyin is not None: + # filtered_entries = tuple(filter(lambda x: x.pinyin == pinyin, entries)) if len(entries) > 1: print(f"\nWARNING: {word} has multiple meanings:") - for entry in entries: + pos_meanings = ( + [ + entry + for entry in entries + for meaning in entry.meaning + if hint in meaning + ] + if hint + else [] + ) + pos_entries = pos_meanings or entries + for entry in pos_entries: for meaning in entry.meanings: - tsv_writer.writerow( - { - "simplified": entry.simplified, - "traditional": entry.traditional, - "pinyin": entry.pinyin, - "meaning": meaning, - } - ) + if pos_meanings and hint in meaning or not pos_meanings: + tsv_writer.writerow( + { + "n": n, + "simplified": entry.simplified, + "traditional": entry.traditional, + "pinyin": entry.pinyin, + "meaning": meaning, + } + ) else: print("============================================") print(f"===================>ERROR: {word} not found") print("============================================") tsv_writer.writerow( { + "n": n, "simplified": word, "traditional": None, "pinyin": None, "meaning": None, } ) + n += 1 def dictionary_process(process_file: ProcessFile) -> list[DictionaryResult]: diff --git a/src/anki_hsk_creator/utility.py b/src/anki_hsk_creator/utility.py index 7469aeb..83ca6e0 100644 --- a/src/anki_hsk_creator/utility.py +++ b/src/anki_hsk_creator/utility.py @@ -5,6 +5,7 @@ Static clasess and functions for general use """ # Standard Library +import random from pathlib import Path # Pip @@ -15,7 +16,17 @@ from cedict_utils.cedict import CedictEntry, CedictParser from qwen_tts import Qwen3TTSModel # Local -from .constants import CCCEDICT_PATH, GWEN_TTS, INPUT, LANGUAGES, OUTPUT, RESOURCES +from .constants import ( + CCCEDICT_PATH, + DICTATION_TYPE, + DICT_TYPE, + GWEN_TTS, + INPUT, + LANGUAGES, + OUTPUT, + PHRASES_TYPE, + RESOURCES, +) # Static Clases @@ -46,19 +57,28 @@ class TRANS: packages_to_install = [] for in_package in packages: if in_package.from_code == from_code: + if in_package.to_code == to_code: + # Single package between 2 languages + print( + f"Installing package {in_package.from_code}" + f"->{in_package.to_code}" + ) + packages_to_install.append(in_package) + break for out_package in packages: if out_package.to_code == to_code: if in_package.to_code == out_package.from_code: print( - f"Check in_package {in_package.from_code}" - f"{in_package.to_code}" + f"Installing in_package {in_package.from_code}" + f"->{in_package.to_code}" ) print( - f"Check out_package {out_package.from_code}" - f"{out_package.to_code}" + f"Installing out_package {out_package.from_code}" + f"->{out_package.to_code}" ) packages_to_install.append(in_package) packages_to_install.append(out_package) + break for package in packages_to_install: print(f"instaling package {package}") argostranslate.package.install_from_path(package.download()) @@ -143,7 +163,9 @@ class TTS: "language": "Chinese", "speaker": "Uncle_Fu", "instruct": "语速缓慢而审慎，每个音节的语调都拿捏得恰到好处，宛如教授在指导新生。", + "instruct": "Speak in a slow pace and enuntiate every word, as a teacher to a learning student", } + VOICES = ["Vivian", "Serena", "Uncle_Fu", "Dylan", "Eric"] @staticmethod def create_tts(): @@ -161,18 +183,73 @@ class TTS: GWEN_TTS, device_map=TTS.DEVICE, dtype=torch.bfloat16, - attn_implementation="flash_attention_2", + # attn_implementation="flash_attention_2", ) @staticmethod def generate(text: str): """Generates a Waw using the defaulst values""" - return TTS.MODEL.generate_custom_voice(text=text, **TTS.DEFAULTS) + speaker = random.choice(TTS.VOICES) + return TTS.MODEL.generate_custom_voice( + text=text, **TTS.DEFAULTS, speaker=speaker + ) # Clases +class ProcessFolder: + """Class that represents a folder to processs + + diferent input files has direfent process_files depending on language + """ + + def __init__(self, input_folder: Path, language_id: str = None): + self.input_folder = input_folder + self._language_id = language_id + # process file type + self.out_folder = OUTPUT / input_folder + self.out_folder.mkdir(parents=True, exist_ok=True) + self.resources = RESOURCES / input_folder + self.resources.mkdir(parents=True, exist_ok=True) + + @property + def output_name(): + """Posible name for the output file, still missing the filetype""" + if self.language_id is None: + raise ValueError("Not a valid language selected") + return self.out_folder / f"{self.input_file.stem}.{self.language_id}.temp" + + @property + def absolute_input_file(self): + """Absolute input file""" + return INPUT / self.input_file + + @property + def input_files(self): + input_files = [] + for file in self.absolute_input_file.glob(f"*.txt"): + for file_type in (DICT_TYPE, PHRASES_TYPE, DICTATION_TYPE): + if file_type in file.suffixes: + input_files.append(ProcessFile(file, language_id=self.language_id)) + return input_files + + @property + def language_id(self): + """language for this trasnlation process""" + return self._language_id + + @language_id.setter + def language_id(self, value): + self._language_id = value + + def __str__(self): + return ( + f"Proccess {self.input_folder}" + f"(out: {self.out_folder}, res: {self.resources})" + ) + + class ProcessFile: """Class that represents a file to processs @@ -189,6 +266,17 @@ class ProcessFile: self.resources = resources.parent / resources.stem self.resources.mkdir(parents=True, exist_ok=True) + @property + def file_type(self): + if DICTATION_TYPE in self.input_file.suffixes: + return DICTATION_TYPE + elif DICT_TYPE in self.input_file.suffixes: + return DICT_TYPE + elif PHRASES_TYPE in self.input_file.suffixes: + return PHRASES_TYPE + else: + raise ValueError("File type not recognized") + @property def absolute_input_file(self): """Absolute input file""" @@ -196,7 +284,7 @@ class ProcessFile: @property def language_id(self): - """language for this trasnlation process""" + """language for this trasnlation proccess""" return self._language_id @language_id.setter @@ -226,6 +314,9 @@ class ProcessFile: """for a Dictionary file loads the avaliable proceced languages""" return [lan.suffixes[0][1:] for lan in self.resources.glob("dictionary.*.tsv")] + def __str__(self): + return f"Proccess:{self.input_file}" + class TranslationResult: """Result of a translated process"""