add bulk procesing

.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

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 = [

src/anki_hsk_creator/__about__.py

@@ -3,4 +3,4 @@
 # SPDX-FileCopyrightText: 2026-present Wolfang Torres <wolfang.torres@gmail.com>
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
-__version__ = "0.1.3"
+__version__ = "0.2.0"

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()
+        option = cli_choose_work_type()
-        if DICT_TYPE in input_file.input_file.suffixes:
+        if option == "e":
-            dict_selected = cli_select_dictionay_tsv()
+            break
-            if dict_selected:
+        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)
+                print(f"processing file {input_file} with language {language_id}")
-            else:
+                process_a_phrases_file(input_file, language_id)
-                language_id = cli_select_language(
+            elif DICTATION_TYPE in input_file.input_file.suffixes:
-                    input_file.available_dictionary_languages
+                language_id = cli_select_language()
-                )
+                print(f"processing file {input_file} with language {language_id}")
-                process_a_dictionary_file(input_file, language_id)
+                process_a_dictation_file(input_file, language_id)
-        elif PHRASES_TYPE in input_file.input_file.suffixes:
+        elif option == "f":
+            in_folder = cli_select_folder()
             language_id = cli_select_language()
-            print(
+            print(f"Selected: {in_folder} with language {language_id}")
-                f"processing file {input_file.input_file} with language {language_id}"
+            folder_proccess(in_folder, language_id=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)
 
 
 if __name__ == "__main__":

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)

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)

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",
     }

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:
+                # if pinyin is not None:
-                    entries = list(filter(lambda x: x.pinyin == pinyin, entries))
+                #     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(
+                        if pos_meanings and hint in meaning or not pos_meanings:
-                            {
+                            tsv_writer.writerow(
-                                "simplified": entry.simplified,
+                                {
-                                "traditional": entry.traditional,
+                                    "n": n,
-                                "pinyin": entry.pinyin,
+                                    "simplified": entry.simplified,
-                                "meaning": meaning,
+                                    "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]:

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"Installing in_package {in_package.from_code}"
-                                f"{in_package.to_code}"
+                                f"->{in_package.to_code}"
                             )
                             print(
-                                f"Check out_package {out_package.from_code}"
+                                f"Installing out_package {out_package.from_code}"
-                                f"{out_package.to_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"""