Update app.py

app.py

@@ -12,6 +12,7 @@ import logging
 import tempfile
 import csv
 import requests
+import json
 from typing import Optional, Dict, Tuple, Any, Union
 from pathlib import Path
 from dataclasses import dataclass
@@ -93,8 +94,10 @@ SUPPORTED_FILE_TYPES = [
 # Audio file extensions
 AUDIO_EXTENSIONS = [".wav", ".mp3", ".m4a"]
 
-# Evaluation CSV URL
-EVALUATION_CSV_URL = "https://github.com/mgolomanta/Models_Evaluation/blob/main/evaluation.csv"
+# GitHub repository details
+GITHUB_REPO = "mgolomanta/Models_Evaluation"
+EVALUATION_FILE = "evaluation.csv"
+GITHUB_TOKEN = os.getenv("git_tk")  # Set this in your environment variables
 
 # ================================
 # Logging Configuration
@@ -437,12 +440,42 @@ class AudioProcessor:
 # ================================
 
 class EvaluationService:
-    """Handles evaluation submissions and storage."""
+    """Handles evaluation submissions and GitHub storage."""
     
-    EVALUATION_CSV_PATH = "evaluation.csv"
+    @staticmethod
+    def get_github_file_sha() -> Optional[str]:
+        """Get the SHA of the existing evaluation file on GitHub."""
+        try:
+            url = f"https://api.github.com/repos/{GITHUB_REPO}/contents/{EVALUATION_FILE}"
+            headers = {"Authorization": f"token {GITHUB_TOKEN}"} if GITHUB_TOKEN else {}
+            response = requests.get(url, headers=headers)
+            
+            if response.status_code == 200:
+                return response.json().get("sha")
+            return None
+        except Exception as e:
+            logger.error(f"Error getting file SHA: {e}")
+            return None
+    
+    @staticmethod
+    def read_existing_csv_content() -> str:
+        """Read existing CSV content from GitHub."""
+        try:
+            url = f"https://api.github.com/repos/{GITHUB_REPO}/contents/{EVALUATION_FILE}"
+            headers = {"Authorization": f"token {GITHUB_TOKEN}"} if GITHUB_TOKEN else {}
+            response = requests.get(url, headers=headers)
+            
+            if response.status_code == 200:
+                content = response.json().get("content", "")
+                import base64
+                return base64.b64decode(content).decode('utf-8')
+            return ""
+        except Exception as e:
+            logger.error(f"Error reading existing CSV: {e}")
+            return ""
     
     @staticmethod
-    def save_evaluation(
+    def save_evaluation_to_github(
         source_lang: str,
         target_lang: str,
         user_input: str,
@@ -451,7 +484,7 @@ class EvaluationService:
         correct_answer: Optional[str] = None
     ) -> str:
         """
-        Save evaluation to CSV file.
+        Save evaluation to GitHub CSV file.
         
         Args:
             source_lang: Source language name
@@ -465,37 +498,56 @@ class EvaluationService:
             Status message
         """
         try:
-            # Check if CSV file exists, create it with headers if not
-            file_exists = os.path.exists(EvaluationService.EVALUATION_CSV_PATH)
-            
-            with open(EvaluationService.EVALUATION_CSV_PATH, mode='a', newline='', encoding='utf-8') as file:
-                writer = csv.writer(file)
-                
-                # Write headers if file is new
-                if not file_exists:
-                    writer.writerow([
-                        "source_language_name",
-                        "target_language_name", 
-                        "user_input",
-                        "model_output",
-                        "notation_value",
-                        "correct_answer"
-                    ])
-                
-                # Write evaluation data
-                writer.writerow([
-                    source_lang,
-                    target_lang,
-                    user_input,
-                    model_output,
-                    notation if notation else "",
-                    correct_answer if correct_answer else ""
-                ])
+            # Prepare the new evaluation data
+            new_row = f'"{source_lang}","{target_lang}","{user_input.replace(\'"\', \'""\')}","{model_output.replace(\'"\', \'""\')}","{notation if notation else \'\'}","{correct_answer if correct_answer else \'\'}"\n'
             
-            return "✅ Evaluation submitted successfully!"
+            # Get existing content
+            existing_content = EvaluationService.read_existing_csv_content()
             
+            # Check if file exists and has headers
+            if existing_content.strip():
+                # File exists, append new row
+                csv_content = existing_content + new_row
+            else:
+                # File doesn't exist, create with headers
+                headers = "source_language_name,target_language_name,user_input,model_output,notation_value,correct_answer\n"
+                csv_content = headers + new_row
+            
+            # Encode content for GitHub API
+            import base64
+            content_encoded = base64.b64encode(csv_content.encode('utf-8')).decode('utf-8')
+            
+            # Prepare GitHub API request
+            url = f"https://api.github.com/repos/{GITHUB_REPO}/contents/{EVALUATION_FILE}"
+            headers = {
+                "Authorization": f"token {GITHUB_TOKEN}",
+                "Accept": "application/vnd.github.v3+json"
+            }
+            
+            # Check if file exists to get SHA
+            file_sha = EvaluationService.get_github_file_sha()
+            
+            # Prepare payload
+            payload = {
+                "message": "Add new evaluation",
+                "content": content_encoded
+            }
+            
+            # Add SHA if file exists (for update)
+            if file_sha:
+                payload["sha"] = file_sha
+            
+            # Send request to GitHub API
+            response = requests.put(url, headers=headers, json=payload)
+            
+            if response.status_code in [200, 201]:
+                return "✅ Evaluation submitted successfully to GitHub!"
+            else:
+                logger.error(f"GitHub API error: {response.status_code} - {response.text}")
+                return f"❌ Error saving evaluation to GitHub: {response.status_code}"
+                
         except Exception as e:
-            logger.error(f"Failed to save evaluation: {e}")
+            logger.error(f"Failed to save evaluation to GitHub: {e}")
             return f"❌ Error saving evaluation: {str(e)}"
 
 # ================================
@@ -559,8 +611,14 @@ class TranslationApp:
         notation: Optional[str],
         correct_answer: Optional[str]
     ) -> str:
-        """Submit evaluation data."""
-        return self.evaluation_service.save_evaluation(
+        """Submit evaluation data to GitHub."""
+        if not GITHUB_TOKEN:
+            return "❌ GitHub token not configured. Please set GITHUB_TOKEN environment variable."
+        
+        if not user_input.strip() or not model_output.strip():
+            return "⚠️ Please translate text before submitting evaluation."
+            
+        return self.evaluation_service.save_evaluation_to_github(
             source_lang, target_lang, user_input, model_output, notation, correct_answer
         )
     
@@ -635,6 +693,10 @@ class TranslationApp:
                 interactive=False
             )
             
+            # Store the last translation data for evaluation
+            last_input_state = gr.State("")
+            last_output_state = gr.State("")
+            
             # Evaluation section
             gr.Markdown("### 📝 Model Evaluation")
             with gr.Group():
@@ -673,7 +735,7 @@ class TranslationApp:
                 text_input: str,
                 audio_file: Optional[str],
                 file_obj: Optional[gr.FileData]
-            ) -> Tuple[str, str]:
+            ) -> Tuple[str, str, str, str]:
                 """Handle initial input processing."""
                 try:
                     processed_text = self.process_input(
@@ -683,28 +745,29 @@ class TranslationApp:
                         audio_file,
                         file_obj
                     )
-                    return processed_text, ""
+                    return processed_text, "", processed_text, ""
                 except Exception as e:
                     logger.error(f"Processing error: {e}")
-                    return "", f"❌ Error: {str(e)}"
+                    return "", f"❌ Error: {str(e)}", "", ""
             
             def handle_translate(
                 extracted_text: str,
                 source_lang: str,
                 target_lang: str
-            ) -> str:
+            ) -> Tuple[str, str, str]:
                 """Handle translation of processed text."""
                 if not extracted_text.strip():
-                    return "📝 No text to translate."
+                    return "📝 No text to translate.", extracted_text, ""
                 try:
-                    return self.translation_service.translate(
+                    result = self.translation_service.translate(
                         extracted_text,
                         Language(source_lang),
                         Language(target_lang)
                     )
+                    return result, extracted_text, result
                 except Exception as e:
                     logger.error(f"Translation error: {e}")
-                    return f"❌ Translation error: {str(e)}"
+                    return f"❌ Translation error: {str(e)}", extracted_text, ""
             
             def handle_evaluation(
                 source_lang: str,
@@ -715,9 +778,6 @@ class TranslationApp:
                 correct_answer: Optional[str]
             ) -> str:
                 """Handle evaluation submission."""
-                if not user_input.strip() or not model_output.strip():
-                    return "⚠️ Please translate text before submitting evaluation."
-                
                 return self.submit_evaluation(
                     source_lang,
                     target_lang,
@@ -727,14 +787,6 @@ class TranslationApp:
                     correct_answer
                 )
             
-            def clear_evaluation_fields() -> Dict:
-                """Clear evaluation fields after submission."""
-                return {
-                    notation: gr.update(value=None),
-                    correct_translation: gr.update(value=""),
-                    evaluation_status: gr.update(value="Evaluation cleared. Ready for next submission.")
-                }
-            
             # Connect events
             input_mode.change(
                 fn=update_visibility,
@@ -742,14 +794,14 @@ class TranslationApp:
                 outputs=[input_text, audio_input, file_input, extracted_text, output_text]
             )
             
-            translate_btn.click(
+            process_result = translate_btn.click(
                 fn=handle_process,
                 inputs=[input_mode, input_lang, input_text, audio_input, file_input],
-                outputs=[extracted_text, output_text]
+                outputs=[extracted_text, output_text, last_input_state, last_output_state]
             ).then(
                 fn=handle_translate,
                 inputs=[extracted_text, input_lang, output_lang],
-                outputs=output_text
+                outputs=[output_text, last_input_state, last_output_state]
             )
             
             submit_evaluation_btn.click(
@@ -757,16 +809,12 @@ class TranslationApp:
                 inputs=[
                     input_lang, 
                     output_lang, 
-                    extracted_text, 
-                    output_text, 
+                    last_input_state, 
+                    last_output_state, 
                     notation, 
                     correct_translation
                 ],
                 outputs=evaluation_status
-            ).then(
-                fn=clear_evaluation_fields,
-                inputs=[],
-                outputs=[notation, correct_translation, evaluation_status]
             )
         
         return interface
@@ -777,6 +825,12 @@ class TranslationApp:
 
 def main():
     """Main application entry point."""
+    # Check if GitHub token is set
+    if not os.getenv("git_tk"):
+        logger.warning("GITHUB_TOKEN environment variable not set. Evaluation submissions will fail.")
+        print("⚠️  WARNING: GITHUB_TOKEN environment variable not set!")
+        print("   Please set it to enable evaluation submissions to GitHub.")
+    
     try:
         app = TranslationApp()
         interface = app.create_interface()