Merge pull request #85 from iautolab/chore/fix-bugs-and-hygiene

chore: fix bugs and code hygiene issues

Author: MaleicAcid

openlrc/__init__.py

@@ -5,5 +5,5 @@
 from openlrc.openlrc import LRCer, TranscriptionConfig, TranslationConfig
 
 __all__ = ("LRCer", "TranscriptionConfig", "TranslationConfig", "ModelConfig", "list_chatbot_models", "ModelProvider")
-__version__ = "1.5.2"
+__version__ = "1.6.1"
 __author__ = "zh-plus"

openlrc/agents.py

@@ -100,7 +100,7 @@ def __init__(
         self,
         src_lang,
         target_lang,
-        info: TranslateInfo = TranslateInfo(),
+        info: TranslateInfo | None = None,
         chatbot_model: str | ModelConfig = "gpt-4.1-nano",
         fee_limit: float = 0.8,
         proxy: str | None = None,
@@ -119,6 +119,8 @@ def __init__(
             base_url_config (Optional[dict]): Configuration for the base URL of the API.
         """
         super().__init__()
+        if info is None:
+            info = TranslateInfo()
         self.chatbot_model = chatbot_model
         self.info = info
         self.chatbot = self._initialize_chatbot(chatbot_model, fee_limit, proxy, base_url_config)
@@ -149,9 +151,9 @@ def _parse_responses(self, resp) -> tuple[list[str], str, str]:
             translations = self._extract_translations(content)
 
             return [t.strip() for t in translations], summary.strip(), scene.strip()
-        except Exception as e:
+        except Exception:
             logger.error(f"Failed to extract contents from response: {content}")
-            raise e
+            raise
 
     def _extract_tag_content(self, content: str, tag: str) -> str:
         """
@@ -203,7 +205,7 @@ def translate_chunk(
         self,
         chunk_id: int,
         chunk: list[tuple[int, str]],
-        context: TranslationContext = TranslationContext(),
+        context: TranslationContext | None = None,
         use_glossary: bool = True,
     ) -> tuple[list[str], TranslationContext]:
         """
@@ -218,6 +220,8 @@ def translate_chunk(
         Returns:
             Tuple[List[str], TranslationContext]: The translated texts and updated context.
         """
+        if context is None:
+            context = TranslationContext()
         user_input = self.prompter.format_texts(chunk)
         guideline = context.guideline if use_glossary else context.non_glossary_guideline
         messages_list = [
@@ -251,7 +255,7 @@ def __init__(
         self,
         src_lang,
         target_lang,
-        info: TranslateInfo = TranslateInfo(),
+        info: TranslateInfo | None = None,
         chatbot_model: str | ModelConfig = "gpt-4.1-nano",
         retry_model: str | ModelConfig | None = None,
         fee_limit: float = 0.8,
@@ -272,6 +276,8 @@ def __init__(
             base_url_config (Optional[dict]): Configuration for the base URL of the API.
         """
         super().__init__()
+        if info is None:
+            info = TranslateInfo()
         self.src_lang = src_lang
         self.target_lang = target_lang
         self.info = info
@@ -426,7 +432,7 @@ def __init__(
         self,
         src_lang,
         target_lang,
-        info: TranslateInfo = TranslateInfo(),
+        info: TranslateInfo | None = None,
         chatbot_model: str | ModelConfig = "gpt-4.1-nano",
         fee_limit: float = 0.8,
         proxy: str | None = None,
@@ -445,6 +451,8 @@ def __init__(
             base_url_config (Optional[dict]): Configuration for the base URL of the API.
         """
         super().__init__()
+        if info is None:
+            info = TranslateInfo()
         self.src_lang = src_lang
         self.target_lang = target_lang
         self.info = info

openlrc/chatbot.py

@@ -6,7 +6,6 @@
 import os
 import random
 import re
-import time
 from collections.abc import Callable
 from copy import deepcopy
 
@@ -80,7 +79,7 @@ def __init__(self, model_name, temperature=1, top_p=1, retry=8, max_async=16, fe
             self.model_info = Models.get_model(model_name, beta)
             self.model_name = model_name
         except ValueError:
-            raise ValueError(f"Invalid model {model_name}.")
+            raise ValueError(f"Invalid model {model_name}.") from None
 
         self.temperature = temperature
         self.top_p = top_p
@@ -175,7 +174,7 @@ def message(
             )
         except ChatBotException as e:
             logger.error(f"Failed to message with GPT. Error: {e}")
-            raise e
+            raise
         finally:
             logger.info(f"Translation fee for this call: {self.api_fees[-1]:.4f} USD")
             logger.info(f"Total bot translation fee: {sum(self.api_fees):.4f} USD")
@@ -257,6 +256,9 @@ def _resolve_client_settings(api_key: str | None, base_url_config: dict | None)
 
         raise ValueError("No API key found. Set OPENAI_API_KEY or pass api_key explicitly.")
 
+    def __enter__(self):
+        return self
+
     def __exit__(self, exc_type, exc_val, exc_tb):
         asyncio.run(self.async_client.close())
 
@@ -318,7 +320,7 @@ async def _create_achat(
             ) as e:
                 sleep_time = self._get_sleep_time(e)
                 logger.warning(f"{type(e).__name__}: {e}. Wait {sleep_time}s before retry. Retry num: {i + 1}.")
-                time.sleep(sleep_time)
+                await asyncio.sleep(sleep_time)
 
         if not response:
             raise ChatBotException("Failed to create a chat.")
@@ -424,7 +426,7 @@ async def _create_achat(
             ) as e:
                 sleep_time = self._get_sleep_time(e)
                 logger.warning(f"{type(e).__name__}: {e}. Wait {sleep_time}s before retry. Retry num: {i + 1}.")
-                time.sleep(sleep_time)
+                await asyncio.sleep(sleep_time)
 
         if not response:
             raise ChatBotException("Failed to create a chat.")
@@ -552,7 +554,7 @@ async def _create_achat(
             self.update_fee(response)
             if not response.text:
                 logger.warning(f"Get None response. Wait 15s. Retry num: {i + 1}.")
-                time.sleep(15)
+                await asyncio.sleep(15)
                 continue
 
             response_text = remove_stop(response.text, stop_sequences)

openlrc/logger.py

@@ -17,8 +17,9 @@
 )
 handler.setFormatter(formatter)
 
-logger = logging.getLogger()
+logger = logging.getLogger("openlrc")
 logger.handlers.clear()  # Clear existing handlers
 logger.addHandler(handler)
+logger.propagate = False
 
 logger.setLevel("INFO")

openlrc/openlrc.py

@@ -679,7 +679,7 @@ def run(
             logger.warning("No audio/video file given. Skip LRCer.run()")
             return []
 
-        if isinstance(paths, str) or isinstance(paths, Path):
+        if isinstance(paths, (str, Path)):
             paths = [paths]
 
         paths = list(map(Path, paths))
@@ -743,7 +743,7 @@ def clear_temp_files(self, paths):
 
         This method removes temporary folders and generated wave files from video processing.
         """
-        temp_folders = set([path.parent for path in paths])
+        temp_folders = {path.parent for path in paths}
         for folder in temp_folders:
             assert folder.name == "preprocessed", f"Not a temporary folder: {folder}"
 

openlrc/preprocess.py

@@ -43,8 +43,10 @@ def __init__(
         self,
         audio_paths: str | Path | list[str] | list[Path],
         output_folder: str = "preprocessed",
-        options: dict = default_preprocess_options,
+        options: dict | None = None,
     ):
+        if options is None:
+            options = dict(default_preprocess_options)
         paths_list = audio_paths if isinstance(audio_paths, list) else [audio_paths]
         self.audio_paths: list[Path] = [Path(p) for p in paths_list]
         self.output_paths = [p.parent / output_folder for p in self.audio_paths]
@@ -61,7 +63,7 @@ def noise_suppression(self, audio_paths: list[Path], atten_lim_db: int = 15):
         if not audio_paths:
             return []
 
-        if "atten_lim_db" in self.options.keys():
+        if "atten_lim_db" in self.options:
             atten_lim_db = self.options["atten_lim_db"]
 
         model, df_state, _ = init_df()

openlrc/subtitle.py

@@ -60,6 +60,8 @@ def from_file(filename):
             return Subtitle.from_lrc(filename)
         elif suffix == ".srt":
             return Subtitle.from_srt(filename)
+        else:
+            raise ValueError(f"Unsupported subtitle format: {suffix!r}")
 
     def __len__(self):
         return len(self.segments)

openlrc/transcribe.py

@@ -148,7 +148,7 @@ def sentence_split(self, segments: list[Segment], lang: str):
         Returns:
             list: List of sentence-split segments.
         """
-        if lang not in LANGUAGE_CODES.keys():
+        if lang not in LANGUAGE_CODES:
             logger.warning(f"Language {lang} not supported. Skipping sentence split.")
             return segments
 

openlrc/translate.py

@@ -162,7 +162,7 @@ def translate(
         texts: str | list[str],
         src_lang: str,
         target_lang: str,
-        info: TranslateInfo = TranslateInfo(),
+        info: TranslateInfo | None = None,
         compare_path: Path = Path("translate_intermediate.json"),
     ) -> list[str]:
         """
@@ -185,6 +185,9 @@ def translate(
         Returns:
             List[str]: List of translated texts.
         """
+        if info is None:
+            info = TranslateInfo()
+
         if not isinstance(texts, list):
             texts = [texts]
 
@@ -414,7 +417,7 @@ def translate(self, texts: str | list[str], src_lang, target_lang, info=None):
         try:
             request = requests.post(self.constructed_url, params=params, headers=self.headers, json=body, timeout=20)
         except TimeoutError:
-            raise RuntimeError("Failed to connect to Microsoft Translator API.")
+            raise RuntimeError("Failed to connect to Microsoft Translator API.") from None
         response = request.json()
 
         return json.dumps(response, sort_keys=True, ensure_ascii=False, indent=4, separators=(",", ": "))

openlrc/utils.py

@@ -310,24 +310,24 @@ def get_similarity(text1, text2):
 
 
 def merge_subtitle(video_path, subtitle_path, output_path):
-    def convert2ffmpeg_path(path):
-        abs_path = str(Path(path).absolute()).replace("\\", "/")
-
-        abs_path = abs_path[0] + "\\\\" + abs_path[1:]
-
-        return abs_path
-
     # check ffmpeg
     try:
-        subprocess.check_output("ffmpeg -version", shell=True)
+        subprocess.check_output(["ffmpeg", "-version"])
     except FileNotFoundError:
-        raise RuntimeError("ffmpeg is not installed. Please install ffmpeg first.")
+        raise RuntimeError("ffmpeg is not installed. Please install ffmpeg first.") from None
 
-    subtitle_path = convert2ffmpeg_path(subtitle_path)
+    subtitle_abs = str(Path(subtitle_path).absolute())
     style = "FontSize=24,Bold=1"
     subprocess.call(
-        f'ffmpeg -y -i "{video_path}" -vf subtitles="{subtitle_path}":force_style=\'{style}\' "{output_path}"',
-        shell=True,
+        [
+            "ffmpeg",
+            "-y",
+            "-i",
+            str(video_path),
+            "-vf",
+            f"subtitles={subtitle_abs}:force_style='{style}'",
+            str(output_path),
+        ]
     )
 
     logger.info(f"Subtitled video saved to {output_path}")