Mgolo commited on
Commit
1aee8e8
·
verified ·
1 Parent(s): 8c3c59b

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +536 -225
app.py CHANGED
@@ -1,9 +1,22 @@
 
 
 
 
 
 
 
 
 
 
 
1
  import os
2
  import re
3
- import tempfile
4
  import logging
5
- from typing import Optional, Dict, Tuple, Any
 
6
  from pathlib import Path
 
 
7
 
8
  import gradio as gr
9
  import torch
@@ -11,256 +24,554 @@ import whisper
11
  import fitz # PyMuPDF
12
  import docx
13
  from bs4 import BeautifulSoup
14
- import markdown2
15
  import chardet
16
  from transformers import pipeline, MarianTokenizer, AutoModelForSeq2SeqLM
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
17
 
 
 
 
 
 
18
 
19
- # -------------------------------
20
- # Configuration & Logging Setup
21
- # -------------------------------
22
-
23
- logging.basicConfig(level=logging.INFO)
24
- logger = logging.getLogger(__name__)
25
 
26
- DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
27
- HF_TOKEN = os.getenv("hffff")
28
-
29
- # Language Pair Models
30
- MODELS: Dict[Tuple[str, str], Dict[str, str]] = {
31
- ("English", "Wolof"): {"model_name": "LocaleNLP/localenlp-eng-wol-0.03", "tag": ">>wol<<"},
32
- ("Wolof", "English"): {"model_name": "LocaleNLP/localenlp-wol-eng-0.03", "tag": ">>eng<<"},
33
- ("English", "Hausa"): {"model_name": "LocaleNLP/localenlp-eng-hau-0.01", "tag": ">>hau<<"},
34
- ("Hausa", "English"): {"model_name": "LocaleNLP/localenlp-hau-eng-0.01", "tag": ">>eng<<"},
35
- ("English", "Darija"): {"model_name": "LocaleNLP/english_darija", "tag": ">>dar<<"}
36
- }
37
 
38
- SUPPORTED_LANGUAGES = ["English", "Wolof", "Hausa", "Darija"]
39
- INPUT_MODES = ["Text", "Audio", "File"]
40
- SUPPORTED_FILE_TYPES = [".pdf", ".docx", ".html", ".htm", ".md", ".srt", ".txt"]
 
 
41
 
42
- # -------------------------------
43
- # Model Manager
44
- # -------------------------------
45
 
46
  class ModelManager:
47
- """Manages loading and caching of translation and transcription models."""
48
 
49
  def __init__(self):
50
- self.translation_pipeline = None
51
- self.whisper_model = None
52
-
53
- def load_translation_model(self, src_lang: str, tgt_lang: str) -> Tuple[Any, str]:
54
- key = (src_lang, tgt_lang)
55
- if key not in MODELS:
56
- raise ValueError(f"Unsupported language pair: {src_lang} -> {tgt_lang}")
57
 
58
- config = MODELS[key]
59
- model_name = config["model_name"]
60
- lang_tag = config["tag"]
61
-
62
- if self.translation_pipeline is None or self.translation_pipeline.model.config._name_or_path != model_name:
63
- logger.info(f"Loading translation model: {model_name}")
64
- model = AutoModelForSeq2SeqLM.from_pretrained(model_name, token=HF_TOKEN).to(DEVICE)
65
- tokenizer = MarianTokenizer.from_pretrained(model_name, token=HF_TOKEN)
66
- self.translation_pipeline = pipeline(
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
67
  "translation",
68
  model=model,
69
  tokenizer=tokenizer,
70
- device=0 if DEVICE.type == "cuda" else -1
71
  )
72
- return self.translation_pipeline, lang_tag
73
-
74
- def load_whisper_model(self) -> Any:
75
- if self.whisper_model is None:
 
 
 
 
 
 
 
 
 
76
  logger.info("Loading Whisper base model...")
77
- self.whisper_model = whisper.load_model("base")
78
- return self.whisper_model
79
-
80
-
81
- # -------------------------------
82
- # File Processing Utilities
83
- # -------------------------------
84
 
85
- def extract_text_from_file(file_path: str) -> str:
86
- """Extracts text from various file types."""
87
- ext = Path(file_path).suffix.lower()
88
- content = Path(file_path).read_bytes()
89
 
90
- if ext == ".pdf":
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
91
  with fitz.open(stream=content, filetype="pdf") as doc:
92
  return "\n".join(page.get_text() for page in doc)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
93
 
94
- elif ext == ".docx":
95
- doc = docx.Document(file_path)
96
- return "\n".join(p.text for p in doc.paragraphs)
97
-
98
- elif ext in (".html", ".htm"):
99
- return BeautifulSoup(content.decode("utf-8", errors="ignore"), "html.parser").get_text()
100
-
101
- elif ext == ".md":
102
- html = markdown2.markdown(content.decode("utf-8", errors="ignore"))
103
- return BeautifulSoup(html, "html.parser").get_text()
104
-
105
- elif ext == ".srt":
106
- decoded = content.decode("utf-8", errors="ignore")
107
- return re.sub(r"\d+\n\d{2}:\d{2}:\d{2},\d{3} --> .*?\n", "", decoded)
108
-
109
- elif ext in (".txt", ".text"):
110
- encoding = chardet.detect(content)["encoding"]
111
- return content.decode(encoding or "utf-8", errors="ignore")
112
-
113
- else:
114
- raise ValueError(f"Unsupported file type: {ext}")
115
-
116
-
117
- # -------------------------------
118
- # Translation Logic
119
- # -------------------------------
120
-
121
- def translate_text(text: str, src_lang: str, tgt_lang: str, model_manager: ModelManager) -> str:
122
- """Translates input text using the specified language pair."""
123
- pipe, tag = model_manager.load_translation_model(src_lang, tgt_lang)
124
- paragraphs = text.splitlines()
125
- translated_output = []
126
 
127
- with torch.no_grad():
128
- for para in paragraphs:
129
- if not para.strip():
130
- translated_output.append("")
131
- continue
132
- sentences = [s.strip() for s in para.split(". ") if s.strip()]
133
- formatted = [f"{tag} {sentence}" for sentence in sentences]
134
- results = pipe(
135
- formatted,
136
- max_length=5000,
137
- num_beams=5,
138
- early_stopping=True,
139
- no_repeat_ngram_size=3,
140
- repetition_penalty=1.5,
141
- length_penalty=1.2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
142
  )
143
- translated_sentences = [r["translation_text"].capitalize() for r in results]
144
- translated_output.append(". ".join(translated_sentences))
145
-
146
- return "\n".join(translated_output)
147
-
148
-
149
- # -------------------------------
150
- # Audio Transcription
151
- # -------------------------------
152
-
153
- def transcribe_audio(file_path: str, model_manager: ModelManager) -> str:
154
- """Transcribes audio file using Whisper."""
155
- model = model_manager.load_whisper_model()
156
- result = model.transcribe(file_path)
157
- return result["text"]
158
-
159
-
160
- # -------------------------------
161
- # Main Processing Function
162
- # -------------------------------
163
-
164
- def process_input(
165
- mode: str,
166
- src_lang: str,
167
- text_input: str,
168
- audio_path: Optional[str],
169
- file_obj: Optional[gr.FileData]
170
- ) -> str:
171
- """Processes input based on selected mode."""
172
- if mode == "Text":
173
- return text_input
174
- elif mode == "Audio":
175
- if src_lang != "English":
176
- raise ValueError("Audio input must be in English.")
177
- if not audio_path:
178
- raise ValueError("No audio file uploaded.")
179
- return transcribe_audio(audio_path, model_manager)
180
- elif mode == "File":
181
- if not file_obj:
182
- raise ValueError("No file uploaded.")
183
- return extract_text_from_file(file_obj.name)
184
- return ""
185
-
186
-
187
- # -------------------------------
188
- # Gradio UI Logic
189
- # -------------------------------
190
-
191
- model_manager = ModelManager()
192
-
193
-
194
- def update_visibility(mode: str) -> Dict[str, Any]:
195
- """Update visibility of input components based on selected mode."""
196
- return {
197
- input_text: gr.update(visible=(mode == "Text")),
198
- audio_input: gr.update(visible=(mode == "Audio")),
199
- file_input: gr.update(visible=(mode == "File")),
200
- extracted_text: gr.update(value="", visible=True),
201
- output_text: gr.update(value="")
202
- }
203
-
204
-
205
- def handle_process(
206
- mode: str,
207
- src_lang: str,
208
- text_input: str,
209
- audio_path: Optional[str],
210
- file_obj: Optional[gr.FileData]
211
- ) -> Tuple[str, str]:
212
- """Handles the initial processing of input."""
213
- try:
214
- extracted = process_input(mode, src_lang, text_input, audio_path, file_obj)
215
- return extracted, ""
216
- except Exception as e:
217
- logger.error(f"Processing error: {e}")
218
- return "", f"Error: {str(e)}"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
219
 
 
 
 
220
 
221
- def handle_translate(extracted_text: str, src_lang: str, tgt_lang: str) -> str:
222
- """Handles translation of extracted text."""
223
- if not extracted_text.strip():
224
- return "No input text to translate."
225
  try:
226
- return translate_text(extracted_text, src_lang, tgt_lang, model_manager)
 
 
 
 
 
 
227
  except Exception as e:
228
- logger.error(f"Translation error: {e}")
229
- return f"Translation error: {str(e)}"
230
-
231
-
232
- # -------------------------------
233
- # Gradio Interface
234
- # -------------------------------
235
-
236
- with gr.Blocks(title="LocaleNLP Translator") as demo:
237
- gr.Markdown("## 🌍 LocaleNLP Multi-language Translator")
238
- gr.Markdown("Supports translation between English, Wolof, Hausa, and Darija. Audio input must be in English.")
239
-
240
- with gr.Row():
241
- input_mode = gr.Radio(choices=INPUT_MODES, label="Input Type", value="Text")
242
- input_lang = gr.Dropdown(choices=SUPPORTED_LANGUAGES[:-1], label="Input Language", value="English")
243
- output_lang = gr.Dropdown(choices=SUPPORTED_LANGUAGES, label="Output Language", value="Wolof")
244
-
245
- input_text = gr.Textbox(label="Enter Text", lines=10, visible=True)
246
- audio_input = gr.Audio(label="Upload Audio (.wav, .mp3, .m4a)", type="filepath", visible=False)
247
- file_input = gr.File(file_types=SUPPORTED_FILE_TYPES, label="Upload Document", visible=False)
248
-
249
- extracted_text = gr.Textbox(label="Extracted / Transcribed Text", lines=10, interactive=False)
250
- translate_button = gr.Button("Translate")
251
- output_text = gr.Textbox(label="Translated Text", lines=10, interactive=False)
252
-
253
- input_mode.change(fn=update_visibility, inputs=input_mode, outputs=[input_text, audio_input, file_input, extracted_text, output_text])
254
-
255
- translate_button.click(
256
- fn=handle_process,
257
- inputs=[input_mode, input_lang, input_text, audio_input, file_input],
258
- outputs=[extracted_text, output_text]
259
- ).then(
260
- fn=handle_translate,
261
- inputs=[extracted_text, input_lang, output_lang],
262
- outputs=output_text
263
- )
264
 
265
  if __name__ == "__main__":
266
- demo.launch()
 
1
+ """
2
+ LocaleNLP Translation Service
3
+ ============================
4
+
5
+ A multi-language translation application supporting English, Wolof, Hausa, and Darija.
6
+ Features text, audio, and document translation with a modern web interface.
7
+
8
+ Author: LocaleNLP Team
9
+ Version: 1.0.0
10
+ """
11
+
12
  import os
13
  import re
 
14
  import logging
15
+ import tempfile
16
+ from typing import Optional, Dict, Tuple, Any, Union
17
  from pathlib import Path
18
+ from dataclasses import dataclass
19
+ from enum import Enum
20
 
21
  import gradio as gr
22
  import torch
 
24
  import fitz # PyMuPDF
25
  import docx
26
  from bs4 import BeautifulSoup
27
+ from markdown import markdown
28
  import chardet
29
  from transformers import pipeline, MarianTokenizer, AutoModelForSeq2SeqLM
30
+ from huggingface_hub import login
31
+
32
+ # ================================
33
+ # Configuration & Constants
34
+ # ================================
35
+
36
+ class Language(str, Enum):
37
+ """Supported languages for translation."""
38
+ ENGLISH = "English"
39
+ WOLOF = "Wolof"
40
+ HAUSA = "Hausa"
41
+ DARIJA = "Darija"
42
+
43
+ class InputMode(str, Enum):
44
+ """Supported input modes."""
45
+ TEXT = "Text"
46
+ AUDIO = "Audio"
47
+ FILE = "File"
48
+
49
+ @dataclass
50
+ class ModelConfig:
51
+ """Configuration for translation models."""
52
+ model_name: str
53
+ language_tag: str
54
+
55
+ # Language pair configurations
56
+ TRANSLATION_MODELS: Dict[Tuple[Language, Language], ModelConfig] = {
57
+ (Language.ENGLISH, Language.WOLOF): ModelConfig(
58
+ "LocaleNLP/localenlp-eng-wol-0.03", ">>wol<<"
59
+ ),
60
+ (Language.WOLOF, Language.ENGLISH): ModelConfig(
61
+ "LocaleNLP/localenlp-wol-eng-0.03", ">>eng<<"
62
+ ),
63
+ (Language.ENGLISH, Language.HAUSA): ModelConfig(
64
+ "LocaleNLP/localenlp-eng-hau-0.01", ">>hau<<"
65
+ ),
66
+ (Language.HAUSA, Language.ENGLISH): ModelConfig(
67
+ "LocaleNLP/localenlp-hau-eng-0.01", ">>eng<<"
68
+ ),
69
+ (Language.ENGLISH, Language.DARIJA): ModelConfig(
70
+ "LocaleNLP/english_darija", ">>dar<<"
71
+ )
72
+ }
73
 
74
+ # File type support
75
+ SUPPORTED_FILE_TYPES = [
76
+ ".pdf", ".docx", ".html", ".htm", ".md",
77
+ ".srt", ".txt", ".text"
78
+ ]
79
 
80
+ # Audio file extensions
81
+ AUDIO_EXTENSIONS = [".wav", ".mp3", ".m4a"]
 
 
 
 
82
 
83
+ # ================================
84
+ # Logging Configuration
85
+ # ================================
 
 
 
 
 
 
 
 
86
 
87
+ logging.basicConfig(
88
+ level=logging.INFO,
89
+ format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
90
+ )
91
+ logger = logging.getLogger(__name__)
92
 
93
+ # ================================
94
+ # Model Management
95
+ # ================================
96
 
97
  class ModelManager:
98
+ """Centralized model management for translation and transcription."""
99
 
100
  def __init__(self):
101
+ self._translation_pipeline = None
102
+ self._whisper_model = None
103
+ self._current_model_name = None
 
 
 
 
104
 
105
+ def get_translation_pipeline(
106
+ self,
107
+ source_lang: Language,
108
+ target_lang: Language
109
+ ) -> Tuple[Any, str]:
110
+ """
111
+ Load and return translation pipeline for given language pair.
112
+
113
+ Args:
114
+ source_lang: Source language
115
+ target_lang: Target language
116
+
117
+ Returns:
118
+ Tuple of (pipeline, language_tag)
119
+
120
+ Raises:
121
+ ValueError: If language pair is not supported
122
+ """
123
+ key = (source_lang, target_lang)
124
+ if key not in TRANSLATION_MODELS:
125
+ raise ValueError(f"Unsupported translation pair: {source_lang} -> {target_lang}")
126
+
127
+ config = TRANSLATION_MODELS[key]
128
+
129
+ # Load model if not loaded or different model needed
130
+ if (self._translation_pipeline is None or
131
+ self._current_model_name != config.model_name):
132
+
133
+ logger.info(f"Loading translation model: {config.model_name}")
134
+
135
+ # Authenticate with Hugging Face if token provided
136
+ if hf_token := os.getenv("hffff"):
137
+ login(token=hf_token)
138
+
139
+ model = AutoModelForSeq2SeqLM.from_pretrained(
140
+ config.model_name,
141
+ token=hf_token
142
+ ).to(self._get_device())
143
+
144
+ tokenizer = MarianTokenizer.from_pretrained(
145
+ config.model_name,
146
+ token=hf_token
147
+ )
148
+
149
+ self._translation_pipeline = pipeline(
150
  "translation",
151
  model=model,
152
  tokenizer=tokenizer,
153
+ device=0 if self._get_device().type == "cuda" else -1
154
  )
155
+
156
+ self._current_model_name = config.model_name
157
+
158
+ return self._translation_pipeline, config.language_tag
159
+
160
+ def get_whisper_model(self) -> Any:
161
+ """
162
+ Load and return Whisper transcription model.
163
+
164
+ Returns:
165
+ Whisper model instance
166
+ """
167
+ if self._whisper_model is None:
168
  logger.info("Loading Whisper base model...")
169
+ self._whisper_model = whisper.load_model("base")
170
+ return self._whisper_model
171
+
172
+ def _get_device(self) -> torch.device:
173
+ """Get appropriate device for model execution."""
174
+ return torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
175
 
176
+ # ================================
177
+ # Content Processing
178
+ # ================================
 
179
 
180
+ class ContentProcessor:
181
+ """Handles extraction and processing of content from various sources."""
182
+
183
+ @staticmethod
184
+ def extract_text_from_file(file_path: Union[str, Path]) -> str:
185
+ """
186
+ Extract text content from various file formats.
187
+
188
+ Args:
189
+ file_path: Path to the file
190
+
191
+ Returns:
192
+ Extracted text content
193
+
194
+ Raises:
195
+ ValueError: If file type is unsupported
196
+ Exception: If file processing fails
197
+ """
198
+ file_path = Path(file_path)
199
+ extension = file_path.suffix.lower()
200
+
201
+ try:
202
+ content = file_path.read_bytes()
203
+
204
+ if extension == ".pdf":
205
+ return ContentProcessor._extract_pdf_text(content)
206
+ elif extension == ".docx":
207
+ return ContentProcessor._extract_docx_text(file_path)
208
+ elif extension in (".html", ".htm"):
209
+ return ContentProcessor._extract_html_text(content)
210
+ elif extension == ".md":
211
+ return ContentProcessor._extract_markdown_text(content)
212
+ elif extension == ".srt":
213
+ return ContentProcessor._extract_srt_text(content)
214
+ elif extension in (".txt", ".text"):
215
+ return ContentProcessor._extract_plain_text(content)
216
+ else:
217
+ raise ValueError(f"Unsupported file type: {extension}")
218
+
219
+ except Exception as e:
220
+ logger.error(f"Failed to extract text from {file_path}: {e}")
221
+ raise
222
+
223
+ @staticmethod
224
+ def _extract_pdf_text(content: bytes) -> str:
225
+ """Extract text from PDF file."""
226
  with fitz.open(stream=content, filetype="pdf") as doc:
227
  return "\n".join(page.get_text() for page in doc)
228
+
229
+ @staticmethod
230
+ def _extract_docx_text(file_path: Path) -> str:
231
+ """Extract text from DOCX file."""
232
+ doc = docx.Document(str(file_path))
233
+ return "\n".join(paragraph.text for paragraph in doc.paragraphs)
234
+
235
+ @staticmethod
236
+ def _extract_html_text(content: bytes) -> str:
237
+ """Extract text from HTML file."""
238
+ encoding = chardet.detect(content)["encoding"] or "utf-8"
239
+ text = content.decode(encoding, errors="ignore")
240
+ soup = BeautifulSoup(text, "html.parser")
241
+ return soup.get_text()
242
+
243
+ @staticmethod
244
+ def _extract_markdown_text(content: bytes) -> str:
245
+ """Extract text from Markdown file."""
246
+ encoding = chardet.detect(content)["encoding"] or "utf-8"
247
+ text = content.decode(encoding, errors="ignore")
248
+ html = markdown(text)
249
+ soup = BeautifulSoup(html, "html.parser")
250
+ return soup.get_text()
251
+
252
+ @staticmethod
253
+ def _extract_srt_text(content: bytes) -> str:
254
+ """Extract text from SRT subtitle file."""
255
+ encoding = chardet.detect(content)["encoding"] or "utf-8"
256
+ text = content.decode(encoding, errors="ignore")
257
+ # Remove timestamp lines
258
+ return re.sub(r"\d+\n\d{2}:\d{2}:\d{2},\d{3} --> .*?\n", "", text)
259
+
260
+ @staticmethod
261
+ def _extract_plain_text(content: bytes) -> str:
262
+ """Extract text from plain text file."""
263
+ encoding = chardet.detect(content)["encoding"] or "utf-8"
264
+ return content.decode(encoding, errors="ignore")
265
+
266
+ # ================================
267
+ # Translation Service
268
+ # ================================
269
+
270
+ class TranslationService:
271
+ """Core translation service with advanced processing capabilities."""
272
+
273
+ def __init__(self, model_manager: ModelManager):
274
+ self.model_manager = model_manager
275
+
276
+ def translate(
277
+ self,
278
+ text: str,
279
+ source_lang: Language,
280
+ target_lang: Language
281
+ ) -> str:
282
+ """
283
+ Translate text from source to target language.
284
+
285
+ Args:
286
+ text: Input text to translate
287
+ source_lang: Source language
288
+ target_lang: Target language
289
+
290
+ Returns:
291
+ Translated text
292
+ """
293
+ if not text.strip():
294
+ return "No input text to translate."
295
+
296
+ pipeline_obj, lang_tag = self.model_manager.get_translation_pipeline(
297
+ source_lang, target_lang
298
+ )
299
+
300
+ # Process text in paragraphs
301
+ paragraphs = text.splitlines()
302
+ translated_paragraphs = []
303
+
304
+ with torch.no_grad():
305
+ for paragraph in paragraphs:
306
+ if not paragraph.strip():
307
+ translated_paragraphs.append("")
308
+ continue
309
+
310
+ # Split into sentences and translate
311
+ sentences = [
312
+ s.strip() for s in paragraph.split(". ")
313
+ if s.strip()
314
+ ]
315
+
316
+ # Add language tag to each sentence
317
+ formatted_sentences = [
318
+ f"{lang_tag} {sentence}"
319
+ for sentence in sentences
320
+ ]
321
+
322
+ # Perform translation
323
+ results = pipeline_obj(
324
+ formatted_sentences,
325
+ max_length=5000,
326
+ num_beams=5,
327
+ early_stopping=True,
328
+ no_repeat_ngram_size=3,
329
+ repetition_penalty=1.5,
330
+ length_penalty=1.2
331
+ )
332
+
333
+ # Process results
334
+ translated_sentences = [
335
+ result["translation_text"].capitalize()
336
+ for result in results
337
+ ]
338
+
339
+ translated_paragraphs.append(". ".join(translated_sentences))
340
+
341
+ return "\n".join(translated_paragraphs)
342
 
343
+ # ================================
344
+ # Audio Processing
345
+ # ================================
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
346
 
347
+ class AudioProcessor:
348
+ """Handles audio file transcription using Whisper."""
349
+
350
+ def __init__(self, model_manager: ModelManager):
351
+ self.model_manager = model_manager
352
+
353
+ def transcribe(self, audio_file_path: str) -> str:
354
+ """
355
+ Transcribe audio file to text.
356
+
357
+ Args:
358
+ audio_file_path: Path to audio file
359
+
360
+ Returns:
361
+ Transcribed text
362
+ """
363
+ model = self.model_manager.get_whisper_model()
364
+ result = model.transcribe(audio_file_path)
365
+ return result["text"]
366
+
367
+ # ================================
368
+ # Main Application
369
+ # ================================
370
+
371
+ class TranslationApp:
372
+ """Main application orchestrating all components."""
373
+
374
+ def __init__(self):
375
+ self.model_manager = ModelManager()
376
+ self.content_processor = ContentProcessor()
377
+ self.translation_service = TranslationService(self.model_manager)
378
+ self.audio_processor = AudioProcessor(self.model_manager)
379
+
380
+ def process_input(
381
+ self,
382
+ mode: InputMode,
383
+ source_lang: Language,
384
+ text_input: str,
385
+ audio_file: Optional[str],
386
+ file_obj: Optional[gr.FileData]
387
+ ) -> str:
388
+ """
389
+ Process input based on selected mode.
390
+
391
+ Args:
392
+ mode: Input mode
393
+ source_lang: Source language
394
+ text_input: Text input
395
+ audio_file: Audio file path
396
+ file_obj: Uploaded file object
397
+
398
+ Returns:
399
+ Processed text content
400
+ """
401
+ if mode == InputMode.TEXT:
402
+ return text_input
403
+
404
+ elif mode == InputMode.AUDIO:
405
+ if source_lang != Language.ENGLISH:
406
+ raise ValueError("Audio input must be in English.")
407
+ if not audio_file:
408
+ raise ValueError("No audio file provided.")
409
+ return self.audio_processor.transcribe(audio_file)
410
+
411
+ elif mode == InputMode.FILE:
412
+ if not file_obj:
413
+ raise ValueError("No file uploaded.")
414
+ return self.content_processor.extract_text_from_file(file_obj.name)
415
+
416
+ return ""
417
+
418
+ def create_interface(self) -> gr.Blocks:
419
+ """Create and return the Gradio interface."""
420
+
421
+ with gr.Blocks(
422
+ title="LocaleNLP Translation Service",
423
+ theme=gr.themes.Soft()
424
+ ) as interface:
425
+ # Header
426
+ gr.Markdown("""
427
+ # 🌍 LocaleNLP Translation Service
428
+ Translate between English, Wolof, Hausa, and Darija with support for text, audio, and documents.
429
+ """)
430
+
431
+ # Input controls
432
+ with gr.Row():
433
+ input_mode = gr.Radio(
434
+ choices=[mode.value for mode in InputMode],
435
+ label="Input Type",
436
+ value=InputMode.TEXT.value
437
+ )
438
+
439
+ input_lang = gr.Dropdown(
440
+ choices=[lang.value for lang in Language if lang != Language.DARIJA],
441
+ label="Input Language",
442
+ value=Language.ENGLISH.value
443
+ )
444
+
445
+ output_lang = gr.Dropdown(
446
+ choices=[lang.value for lang in Language],
447
+ label="Output Language",
448
+ value=Language.WOLOF.value
449
+ )
450
+
451
+ # Input components
452
+ input_text = gr.Textbox(
453
+ label="Enter Text",
454
+ lines=8,
455
+ visible=True,
456
+ placeholder="Type or paste your text here..."
457
  )
458
+
459
+ audio_input = gr.Audio(
460
+ label="Upload Audio",
461
+ type="filepath",
462
+ visible=False
463
+ )
464
+
465
+ file_input = gr.File(
466
+ file_types=SUPPORTED_FILE_TYPES,
467
+ label="Upload Document",
468
+ visible=False
469
+ )
470
+
471
+ # Processing area
472
+ extracted_text = gr.Textbox(
473
+ label="Extracted / Transcribed Text",
474
+ lines=8,
475
+ interactive=False
476
+ )
477
+
478
+ translate_btn = gr.Button(
479
+ "🔄 Process & Translate",
480
+ variant="primary"
481
+ )
482
+
483
+ output_text = gr.Textbox(
484
+ label="Translated Text",
485
+ lines=8,
486
+ interactive=False
487
+ )
488
+
489
+ # Event handlers
490
+ def update_visibility(mode: str) -> Dict[str, Any]:
491
+ """Update component visibility based on input mode."""
492
+ return {
493
+ input_text: gr.update(visible=(mode == InputMode.TEXT.value)),
494
+ audio_input: gr.update(visible=(mode == InputMode.AUDIO.value)),
495
+ file_input: gr.update(visible=(mode == InputMode.FILE.value)),
496
+ extracted_text: gr.update(value="", visible=True),
497
+ output_text: gr.update(value="")
498
+ }
499
+
500
+ def handle_process(
501
+ mode: str,
502
+ source_lang: str,
503
+ text_input: str,
504
+ audio_file: Optional[str],
505
+ file_obj: Optional[gr.FileData]
506
+ ) -> Tuple[str, str]:
507
+ """Handle initial input processing."""
508
+ try:
509
+ processed_text = self.process_input(
510
+ InputMode(mode),
511
+ Language(source_lang),
512
+ text_input,
513
+ audio_file,
514
+ file_obj
515
+ )
516
+ return processed_text, ""
517
+ except Exception as e:
518
+ logger.error(f"Processing error: {e}")
519
+ return "", f"❌ Error: {str(e)}"
520
+
521
+ def handle_translate(
522
+ extracted_text: str,
523
+ source_lang: str,
524
+ target_lang: str
525
+ ) -> str:
526
+ """Handle translation of processed text."""
527
+ if not extracted_text.strip():
528
+ return "📝 No text to translate."
529
+ try:
530
+ return self.translation_service.translate(
531
+ extracted_text,
532
+ Language(source_lang),
533
+ Language(target_lang)
534
+ )
535
+ except Exception as e:
536
+ logger.error(f"Translation error: {e}")
537
+ return f"❌ Translation error: {str(e)}"
538
+
539
+ # Connect events
540
+ input_mode.change(
541
+ fn=update_visibility,
542
+ inputs=input_mode,
543
+ outputs=[input_text, audio_input, file_input, extracted_text, output_text]
544
+ )
545
+
546
+ translate_btn.click(
547
+ fn=handle_process,
548
+ inputs=[input_mode, input_lang, input_text, audio_input, file_input],
549
+ outputs=[extracted_text, output_text]
550
+ ).then(
551
+ fn=handle_translate,
552
+ inputs=[extracted_text, input_lang, output_lang],
553
+ outputs=output_text
554
+ )
555
+
556
+ return interface
557
 
558
+ # ================================
559
+ # Application Entry Point
560
+ # ================================
561
 
562
+ def main():
563
+ """Main application entry point."""
 
 
564
  try:
565
+ app = TranslationApp()
566
+ interface = app.create_interface()
567
+ interface.launch(
568
+ server_name="0.0.0.0",
569
+ server_port=int(os.getenv("PORT", 7860)),
570
+ share=False
571
+ )
572
  except Exception as e:
573
+ logger.critical(f"Failed to start application: {e}")
574
+ raise
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
575
 
576
  if __name__ == "__main__":
577
+ main()