feat: add lyric music to romanji tool

app/api/v1/lyric_romanji_translator.py

@@ -0,0 +1,15 @@
+from fastapi import APIRouter
+from app.models.lyric_romanji_translator import LyricRomanjiTranslatorRequest, LyricRomanjiTranslatorResponse
+from app.services.lyric_romanji_translator import translate_lyric_romanji
+
+
+router = APIRouter()
+
+@router.post("/", response_model=LyricRomanjiTranslatorResponse)
+async def lyric_romanji_translator(request: LyricRomanjiTranslatorRequest):
+    lyric_romanji = await translate_lyric_romanji(request.folder_path)
+    return LyricRomanjiTranslatorResponse(
+        results=lyric_romanji["results"],
+        status=lyric_romanji["status"]
+    )
+

app/core/deepseek_client.py

@@ -1,38 +1,39 @@
-# app/services/openai_service.py
-import openai
 import os
-from openai import OpenAI
+import anyio
-from openai import OpenAIError
+from openai import OpenAI, OpenAIError
 from app.core.config import DEEPSEEK_API_BASE, DEEPSEEK_MODEL, DEEPSEEK_API_KEY
 
 # Ensure the API key is properly set
 if not DEEPSEEK_API_KEY:
     raise ValueError("DEEPSEEK_API_KEY is not set in environment variables")
 
-# Initialize the client with proper configuration
+# Initialize the client
 client = OpenAI(
     api_key=DEEPSEEK_API_KEY,
     base_url=DEEPSEEK_API_BASE
 )
 
-async def chat_with_openai(messages: list):
+async def chat_with_openai(messages: list[dict[str, str]]) -> str:
     if not messages:
         raise ValueError("Messages list cannot be empty")
 
     try:
-        response = client.chat.completions.create(
+        # Run sync client in a thread (non-blocking for FastAPI)
-            model=DEEPSEEK_MODEL,
+        response = await anyio.to_thread.run_sync(
-            messages=messages,
+            lambda: client.chat.completions.create(
-            max_tokens=1000,
+                model=DEEPSEEK_MODEL,
-            temperature=0.7,
+                messages=messages,
-            stream=False
+                max_tokens=1000,
+                temperature=0.7,
+                stream=False
+            )
         )
-        
+
         if not response.choices or not response.choices[0].message.content:
             return "No response content from the model"
-            
+
-        return response.choices[0].message.content
+        return response.choices[0].message.content.strip()
-        
+
     except OpenAIError as e:
         error_msg = f"DeepSeek API Error: {str(e)}"
         print(error_msg)

app/main.py

@@ -1,9 +1,11 @@
 from fastapi import FastAPI
 from app.api.v1 import translate
 from app.api.v1 import voice
+from app.api.v1 import lyric_romanji_translator
 
 app = FastAPI()
 
 # Include your routes
 app.include_router(translate.router, prefix="/api/v1/translate", tags=["translate"])
 app.include_router(voice.router, prefix="/api/v1/voice", tags=["voice"])
+app.include_router(lyric_romanji_translator.router, prefix="/api/v1/lyric_romanji_translator", tags=["lyric_romanji_translator"])

app/models/lyric_romanji_translator.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+from typing import List
+
+class LyricRomanjiTranslatorRequest(BaseModel):
+    folder_path: str
+
+class FileResult(BaseModel):
+    file: str
+    processed: bool
+    added_lines: int
+
+class LyricRomanjiTranslatorResponse(BaseModel):
+    results: List[FileResult]
+    status: str

app/services/lyric_romanji_translator.py

@@ -0,0 +1,59 @@
+import os
+import re
+from app.core.deepseek_client import chat_with_openai
+from app.models.lyric_romanji_translator import FileResult
+
+timestamp_pattern = re.compile(r"^\[\d{2}:\d{2}\.\d{2}\]")
+
+def needs_romaji(lines, idx):
+    if idx + 1 < len(lines) and not timestamp_pattern.match(lines[idx + 1]):
+        return False
+    return True
+
+async def get_romaji(text: str) -> str:
+    messages = [
+        {"role": "system", "content": "Convert Japanese text into romaji only. Output romaji without explanation."},
+        {"role": "user", "content": text}
+    ]
+    return await chat_with_openai(messages)
+
+async def process_lrc_file(filepath: str) -> FileResult:
+    added_lines = 0
+    with open(filepath, "r", encoding="utf-8") as f:
+        lines = f.readlines()
+
+    new_lines = []
+    for idx, line in enumerate(lines):
+        new_lines.append(line)
+
+        if timestamp_pattern.match(line) and needs_romaji(lines, idx):
+            japanese = line.strip().split("]", 1)[-1].strip()
+            if japanese:
+                romaji = await get_romaji(japanese)
+                new_lines.append(f"{romaji}\n")
+                added_lines += 1
+
+    if added_lines > 0:
+        with open(filepath, "w", encoding="utf-8") as f:
+            f.writelines(new_lines)
+
+    return FileResult(file=filepath, processed=added_lines > 0, added_lines=added_lines)
+
+async def translate_lyric_romanji(folder_path: str):
+    results = []
+
+    if not os.path.exists(folder_path):
+        return {"results": [], "status": f"error: folder not found {folder_path}"}
+
+    for root, _, files in os.walk(folder_path):
+        for file in files:
+            if file.endswith(".lrc"):
+                filepath = os.path.join(root, file)
+                print(f"Processing: {filepath}")
+
+                result = await process_lrc_file(filepath)
+
+                # ✅ result is already a FileResult object
+                results.append(result)
+
+    return {"results": results, "status": "completed"}