Update app.py

app.py

@@ -1,4 +1,6 @@
+import gradio as gr
 from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
 from transformers import pipeline
 import langdetect
 import logging
@@ -7,39 +9,32 @@ from typing import Optional
 import re
 from functools import lru_cache
 import asyncio
+import threading
+import time
+
+# Create necessary directories
+os.makedirs("./cache", exist_ok=True)
+os.makedirs("./logs", exist_ok=True)
 
 # Set environment variables for Hugging Face cache
-os.environ["HF_HOME"] = "/app/cache"
-os.environ["TRANSFORMERS_CACHE"] = "/app/cache"
+os.environ["HF_HOME"] = "./cache"
+os.environ["TRANSFORMERS_CACHE"] = "./cache"
 
 # Environment configuration
-USE_8BIT = False
-try:
-    import bitsandbytes  # hanya untuk memastikan modul tersedia
-    USE_8BIT = os.getenv("USE_QUANTIZATION", "0") == "1"
-except ImportError:
-    USE_8BIT = False
-
-DEVICE = int(os.getenv("DEVICE", "-1"))  # -1 for CPU, 0+ for GPU
+DEVICE = -1  # Always use CPU for HF Spaces
 MAX_TEXT_LENGTH = int(os.getenv("MAX_TEXT_LENGTH", "5000"))
 
-app = FastAPI()
-
-# Configure logging with timestamp and level
+# Configure logging
 logging.basicConfig(
     format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-    level=logging.INFO,
-    handlers=[
-        logging.handlers.RotatingFileHandler("/app/logs/app.log", maxBytes=1000000, backupCount=1),
-        logging.StreamHandler()
-    ]
+    level=logging.INFO
 )
 logger = logging.getLogger(__name__)
 
 # Map of supported language models
 MODEL_MAP = {
     "th": "Helsinki-NLP/opus-mt-th-en",
-    "ja": "Helsinki-NLP/opus-mt-ja-en",
+    "ja": "Helsinki-NLP/opus-mt-ja-en", 
     "zh": "Helsinki-NLP/opus-mt-zh-en",
     "vi": "Helsinki-NLP/opus-mt-vi-en",
 }
@@ -47,19 +42,30 @@ MODEL_MAP = {
 # List of terms to protect from translation
 PROTECTED_TERMS = ["2030 Aspirations", "Griffith"]
 
-# Cache for translators to avoid reloading models unnecessarily
+# Cache for translators
 translators = {}
 
+# Pydantic models
+class TranslationRequest(BaseModel):
+    text: str
+    source_lang_override: Optional[str] = None
+
+class TranslationResponse(BaseModel):
+    translated_text: str
+    source_language: Optional[str] = None
+
+# FastAPI app
+app = FastAPI(title="Translation Service API")
+
 def get_translator(lang: str):
     """Load or retrieve cached translator for the given language."""
     if lang not in translators:
-        logger.info(f"Loading model for {lang} from {MODEL_MAP[lang]}...")
+        logger.info(f"Loading model for {lang}...")
         try:
             translators[lang] = pipeline(
                 "translation",
                 model=MODEL_MAP[lang],
-                device=DEVICE,
-                model_kwargs={"load_in_8bit": USE_8BIT}
+                device=-1
             )
             logger.info(f"Model for {lang} loaded successfully.")
         except Exception as e:
@@ -69,30 +75,24 @@ def get_translator(lang: str):
 
 @lru_cache(maxsize=100)
 def detect_language(text: str) -> str:
-    """Cached language detection to reduce overhead for repeated inputs."""
+    """Cached language detection."""
     try:
         detected_lang = langdetect.detect(text)
-        logger.debug(f"langdetect raw result: '{detected_lang}' for text: '{text[:50]}...'")
         if detected_lang.startswith('zh'):
-            logger.debug(f"Normalizing '{detected_lang}' to 'zh' for Mandarin.")
             return 'zh'
-        final_lang = detected_lang if detected_lang in MODEL_MAP else "en"
-        logger.debug(f"Final determined language: '{final_lang}'. (Based on raw detected: '{detected_lang}')")
-        return final_lang
+        return detected_lang if detected_lang in MODEL_MAP else "en"
     except Exception as e:
-        logger.warning(f"Language detection FAILED for text: '{text[:50]}...'. Error: {str(e)}. Defaulting to English.")
+        logger.warning(f"Language detection failed: {str(e)}")
         return "en"
 
 def protect_terms(text: str, protected_terms: list) -> tuple[str, dict]:
-    """Replace protected terms with placeholders using regex for efficiency."""
+    """Replace protected terms with placeholders."""
     modified_text = text
     replacements = {}
     for i, term in enumerate(protected_terms):
         placeholder = f"__PROTECTED_{i}__"
         replacements[placeholder] = term
-        modified_text = re.sub(r'\b' + re.escape(term) + r'\b', placeholder, modified_text)
-    if replacements:
-        logger.debug(f"Protected terms replaced: {replacements}")
+        modified_text = re.sub(r'\b' + re.escape(term) + r'\b', placeholder, modified_text, flags=re.IGNORECASE)
     return modified_text, replacements
 
 def restore_terms(text: str, replacements: dict) -> str:
@@ -102,13 +102,24 @@ def restore_terms(text: str, replacements: dict) -> str:
         restored_text = restored_text.replace(placeholder, term)
     return restored_text
 
-@app.post("/translate")
+# FastAPI endpoints
+@app.get("/")
+async def root():
+    return {"message": "Translation Service API is running"}
+
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy", "supported_languages": list(MODEL_MAP.keys())}
+
+@app.post("/translate", response_model=TranslationResponse)
+async def translate_api(request: TranslationRequest):
+    """API endpoint for translation."""
+    return await translate(request.text, request.source_lang_override)
+
+# Core translation function
 async def translate(text: str, source_lang_override: Optional[str] = None):
-    """
-    Translate text to English, preserving protected terms like '2030 Aspirations'.
-    Automatically detects source language or uses override.
-    """
-    if not text:
+    """Core translation function used by both API and Gradio."""
+    if not text or not text.strip():
         raise HTTPException(status_code=400, detail="Text input is required.")
 
     if len(text) > MAX_TEXT_LENGTH:
@@ -121,42 +132,200 @@ async def translate(text: str, source_lang_override: Optional[str] = None):
         # Determine source language
         if source_lang_override and source_lang_override in MODEL_MAP:
             source_lang = source_lang_override
-            logger.debug(f"Source language overridden by user to: '{source_lang_override}'.")
         else:
-            source_lang = await asyncio.to_thread(detect_language, text)
-            logger.debug(f"Determined source language for translation: '{source_lang}'.")
+            source_lang = detect_language(text)
 
         # If source language is English, return original text
         if source_lang == "en":
-            logger.debug("Source language is English or unrecognized, returning original text.")
-            return {"translated_text": text}
+            return TranslationResponse(
+                translated_text=text,
+                source_language=source_lang
+            )
 
-        # Get translator (lazy-loaded)
+        # Get translator
         translator = get_translator(source_lang)
-        if not translator:
-            logger.error(f"No translator found for language: '{source_lang}'.")
-            raise HTTPException(
-                status_code=400,
-                detail=f"Translation not supported for language: {source_lang}."
-            )
 
         # Protect terms before translation
         modified_text, replacements = protect_terms(text, PROTECTED_TERMS)
-        logger.debug(f"Text after protecting terms: '{modified_text[:50]}...'")
 
-        # Perform translation in a thread to avoid blocking the event loop
-        logger.debug(f"Translating text from '{source_lang}' to English...")
-        result = await asyncio.to_thread(translator, modified_text, max_length=512, num_beams=4)
+        # Perform translation
+        result = translator(modified_text, max_length=512, num_beams=4)
         translated_text = result[0]["translation_text"]
-        logger.debug(f"Translation successful. Original: '{modified_text[:50]}...', Translated: '{translated_text[:50]}...'")
 
         # Restore protected terms
         final_text = restore_terms(translated_text, replacements)
-        logger.debug(f"Final translated text with restored terms: '{final_text[:50]}...'")
 
-        return {"translated_text": final_text}
+        return TranslationResponse(
+            translated_text=final_text,
+            source_language=source_lang
+        )
+        
+    except Exception as e:
+        logger.error(f"Translation error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Translation failed: {str(e)}")
+
+# Gradio interface functions
+def translate_gradio(text: str, source_lang: str = "auto"):
+    """Gradio wrapper for translation function."""
+    if not text.strip():
+        return "Please enter some text to translate.", "N/A"
+    
+    try:
+        source_lang_param = source_lang if source_lang != "auto" else None
+        
+        # Call the async function synchronously for Gradio
+        import asyncio
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        
+        result = loop.run_until_complete(translate(text, source_lang_param))
+        
+        return result.translated_text, result.source_language or "Unknown"
+        
     except HTTPException as e:
-        raise e
+        return f"Error: {e.detail}", "Error"
     except Exception as e:
-        logger.error(f"An unexpected error occurred during processing: {str(e)}", exc_info=True)
-        raise HTTPException(status_code=500, detail=f"Processing failed: {str(e)}")
+        return f"Error: {str(e)}", "Error"
+
+# Create Gradio interface
+def create_gradio_interface():
+    with gr.Blocks(
+        title="Multi-Language Translation Service",
+        theme=gr.themes.Soft(),
+        css="""
+        .gradio-container {
+            max-width: 1200px !important;
+        }
+        """
+    ) as interface:
+        
+        gr.Markdown("""
+        # 🌐 Multi-Language Translation Service
+        
+        Translate text from **Thai**, **Japanese**, **Chinese**, or **Vietnamese** to **English**
+        
+        ✨ Features: Automatic language detection • Protected terms preservation • Fast Helsinki-NLP models
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=1):
+                text_input = gr.Textbox(
+                    label="📝 Input Text",
+                    placeholder="Enter text to translate...",
+                    lines=6,
+                    max_lines=10
+                )
+                
+                with gr.Row():
+                    lang_dropdown = gr.Dropdown(
+                        choices=[
+                            ("🔍 Auto-detect", "auto"),
+                            ("🇹🇭 Thai", "th"),
+                            ("🇯🇵 Japanese", "ja"), 
+                            ("🇨🇳 Chinese", "zh"),
+                            ("🇻🇳 Vietnamese", "vi")
+                        ],
+                        value="auto",
+                        label="Source Language"
+                    )
+                    
+                    translate_btn = gr.Button(
+                        "🚀 Translate", 
+                        variant="primary",
+                        size="lg"
+                    )
+                
+            with gr.Column(scale=1):
+                output_text = gr.Textbox(
+                    label="🎯 Translation Result",
+                    lines=6,
+                    max_lines=10,
+                    interactive=False
+                )
+                
+                detected_lang = gr.Textbox(
+                    label="🔍 Detected Language",
+                    interactive=False,
+                    max_lines=1
+                )
+        
+        # Examples section
+        with gr.Row():
+            gr.Examples(
+                examples=[
+                    ["สวัสดีครับ ยินดีที่ได้รู้จัก การพัฒนา 2030 Aspirations เป็นเป้าหมายสำคัญ", "th"],
+                    ["こんにちは、はじめまして。Griffith大学での研究が進んでいます。", "ja"],
+                    ["你好，很高兴认识你。我们正在为2030 Aspirations制定计划。", "zh"],
+                    ["Xin chào, rất vui được gặp bạn. Griffith là trường đại học tuyệt vời.", "vi"],
+                ],
+                inputs=[text_input, lang_dropdown],
+                outputs=[output_text, detected_lang],
+                fn=translate_gradio,
+                cache_examples=False,
+                label="📋 Try these examples:"
+            )
+        
+        # Event handlers
+        translate_btn.click(
+            fn=translate_gradio,
+            inputs=[text_input, lang_dropdown],
+            outputs=[output_text, detected_lang]
+        )
+        
+        text_input.submit(
+            fn=translate_gradio,
+            inputs=[text_input, lang_dropdown], 
+            outputs=[output_text, detected_lang]
+        )
+        
+        # Information accordion
+        with gr.Accordion("ℹ️ About this service", open=False):
+            gr.Markdown("""
+            ### 🎯 Supported Languages:
+            - **Thai (th)** → English
+            - **Japanese (ja)** → English  
+            - **Chinese (zh)** → English
+            - **Vietnamese (vi)** → English
+            
+            ### 🛡️ Special Features:
+            - **Protected Terms**: Certain terms like "2030 Aspirations" and "Griffith" are preserved during translation
+            - **Auto Detection**: Automatically detects the source language if not specified
+            - **Fast Processing**: Uses optimized Helsinki-NLP translation models
+            
+            ### 🚀 How to use:
+            1. Paste or type your text in the input box
+            2. Choose source language or leave as 'Auto-detect'
+            3. Click 'Translate' or press Enter
+            4. Get your English translation instantly!
+            
+            ### 🔧 API Access:
+            This service also provides REST API endpoints:
+            - `GET /health` - Check service status
+            - `POST /translate` - Translate text (JSON payload required)
+            """)
+    
+    return interface
+
+# Start FastAPI in a separate thread
+def start_fastapi():
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")
+
+# Main execution
+if __name__ == "__main__":
+    # Start FastAPI server in background thread
+    fastapi_thread = threading.Thread(target=start_fastapi, daemon=True)
+    fastapi_thread.start()
+    
+    # Give FastAPI time to start
+    time.sleep(2)
+    
+    # Create and launch Gradio interface
+    demo = create_gradio_interface()
+    demo.queue(max_size=10)
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7861,
+        share=False,
+        show_error=True
+    )