Update app.py

app.py

@@ -7,7 +7,7 @@ import mimetypes
 import zipfile
 import tempfile
 from datetime import datetime
-from typing import List, Dict, Optional, Union
+from typing import List, Dict, Optional
 from pathlib import Path
 import requests
 import validators
@@ -32,7 +32,7 @@ class URLProcessor:
         self.session = requests.Session()
         self.timeout = 10  # seconds
         self.session.headers.update({
-            'User-Agent': UserAgent().random,
+            'User -Agent': UserAgent().random,
             'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
             'Accept-Language': 'en-US,en;q=0.5',
             'Accept-Encoding': 'gzip, deflate, br',
@@ -96,7 +96,7 @@ class URLProcessor:
             if not file_id:
                 logger.error(f"Invalid Google Drive URL: {url}")
                 return None
-            
+                
             direct_url = f"https://drive.google.com/uc?export=download&id={file_id.group(1)}"
             response = self.session.get(direct_url, timeout=self.timeout)
             response.raise_for_status()
@@ -109,7 +109,7 @@ class URLProcessor:
         except Exception as e:
             logger.error(f"Google Drive processing failed: {e}")
             return None
-        
+
     def _handle_google_calendar(self, url: str) -> Optional[Dict]:
         """Process Google Calendar ICS feeds"""
         try:
@@ -165,68 +165,64 @@ class FileProcessor:
         except Exception:
             return False
 
-    def process_files(self, files: Union[List[gr.File], List[str]]) -> List[Dict]:
-        """Process multiple uploaded files and return a single JSON extraction"""
-        if not files:
+    def process_file(self, file) -> List[Dict]:
+        """Process uploaded file with enhanced error handling"""
+        if not file:
             return []
 
-        combined_data = []
+        dataset = []
         try:
-            for file in files:
-                # Check if the file is a Gradio File object or a string path
-                file_name = file.name if isinstance(file, gr.File) else file
-                if os.path.isfile(file_name):
-                    file_size = os.path.getsize(file_name)
-                    if file_size > self.max_file_size:
-                        logger.warning(f"File size ({file_size} bytes) exceeds maximum allowed size")
-                        continue  # Skip this file
+            file_size = os.path.getsize(file.name)
+            if file_size > self.max_file_size:
+                logger.warning(f"File size ({file_size} bytes) exceeds maximum allowed size")
+                return [{"error": f"File size ({file_size} bytes) exceeds maximum allowed size of {self.max_file_size} bytes."}]
 
-                    if zipfile.is_zipfile(file_name):
-                        combined_data.extend(self._process_zip_file(file_name))
-                    else:
-                        combined_data.extend(self._process_single_file(file_name))
+            with tempfile.TemporaryDirectory() as temp_dir:
+                if zipfile.is_zipfile(file.name):
+                    dataset.extend(self._process_zip_file(file.name, temp_dir))
                 else:
-                    logger.warning(f"Skipping directory: {file_name}")
+                    dataset.extend(self._process_single_file(file))
 
         except Exception as e:
-            logger.error(f"Error processing files: {str(e)}")
+            logger.error(f"Error processing file: {str(e)}")
             return []
 
-    def _process_zip_file(self, zip_path: str) -> List[Dict]:
+        return dataset
+
+    def _process_zip_file(self, zip_path: str, temp_dir: str) -> List[Dict]:
         """Process ZIP file contents"""
         results = []
         with zipfile.ZipFile(zip_path, 'r') as zip_ref:
-            with tempfile.TemporaryDirectory() as temp_dir:
-                zip_ref.extractall(temp_dir)
-                for root, _, files in os.walk(temp_dir):
-                    for filename in files:
-                        filepath = os.path.join(root, filename)
-                        if self.is_text_file(filepath):
-                            try:
-                                with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
-                                    content = f.read()
-                                if content.strip():
-                                    results.append({
-                                        "source": "file",
-                                        "filename": filename,
-                                        "content": content,
-                                        "timestamp": datetime.now().isoformat()
-                                    })
-                            except Exception as e:
-                                logger.error(f"Error reading file {filename}: {str(e)}")
+            zip_ref.extractall(temp_dir)
+            for root, _, files in os.walk(temp_dir):
+                for filename in files:
+                    filepath = os.path.join(root, filename)
+                    if self.is_text_file(filepath):
+                        try:
+                            with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
+                                content = f.read()
+                            if content.strip():
+                                results.append({
+                                    "source": "file",
+                                    "filename": filename,
+                                    "content": content,
+                                    "timestamp": datetime.now().isoformat()
+                                })
+                        except Exception as e:
+                            logger.error(f"Error reading file {filename}: {str(e)}")
         return results
 
-    def _process_single_file(self, file_path: str) -> List[Dict]:
+    def _process_single_file(self, file) -> List[Dict]:
         try:
-            file_stat = os.stat(file_path)
-            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
+            file_stat = os.stat(file.name)
+            with open(file.name, 'r', encoding='utf-8', errors='ignore') as f:
                 content = f.read()
             
             return [{
                 'source': 'file',
-                'filename': os.path.basename(file_path),
+                'filename': os.path.basename(file.name),
                 'file_size': file_stat.st_size,
-                'mime_type': mimetypes.guess_type(file_path)[0],
+                'mime_type': mimetypes.guess_type(file.name)[0],
                 'created': datetime.fromtimestamp(file_stat.st_ctime).isoformat(),
                 'modified': datetime.fromtimestamp(file_stat.st_mtime).isoformat(),
                 'content': content,
@@ -266,18 +262,13 @@ def create_interface():
     """Create a comprehensive Gradio interface with advanced features"""
     
     css = """
-    body { background-color: #f0f4f8; font-family: 'Arial', sans-serif; }
-    .container { max-width: 1200px; margin: auto; padding: 20px; border-radius: 8px; box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1); }
-    h1 { color: #333; }
-    .tab { background-color: #ffffff; border-radius: 8px; padding: 20px; margin-bottom: 20px; }
-    .button { background-color: #007bff; color: white; border: none; border-radius: 5px; padding: 10px 20px; cursor: pointer; }
-    .button:hover { background-color: #0056b3; }
-    .warning { background-color: #fff3cd; color: #856404; padding: 10px; border-radius: 5px; }
-    .error { background-color: #f8d7da; color: #721c24; padding: 10px; border-radius: 5px; }
+    .container { max-width: 1200px; margin: auto; }
+    .warning { background-color: #fff3cd; color: #856404; }
+    .error { background-color: #f8d7da; color: #721c24; }
     """
     
-    with gr.Blocks(css=css, title="Advanced Data Processing App") as interface:
-        gr.Markdown("# 🌐 Advanced Data Processing Toolkit")
+    with gr.Blocks(css=css, title="Advanced Text & URL Processor") as interface:
+        gr.Markdown("# 🌐 Advanced URL & Text Processing Toolkit")
         
         with gr.Tab("URL Processing"):
             url_input = gr.Textbox(
@@ -288,7 +279,7 @@ def create_interface():
         
         with gr.Tab("File Input"):
             file_input = gr.File(
-                label="Upload text files or ZIP archives",
+                label="Upload text file or ZIP archive",
                 file_types=[".txt", ".zip", ".md", ".csv", ".json", ".xml"]
             )
 
@@ -309,7 +300,7 @@ def create_interface():
                 placeholder="Paste your JSON data here...",
                 lines=5
             )
-            load_btn = gr.Button("Load Data", variant="primary")
+            load_btn = gr.Button("Load Data")
             chat_output = gr.Textbox(label="Chatbot Response", interactive=False)
 
         process_btn = gr.Button("Process Input", variant="primary")
@@ -320,7 +311,7 @@ def create_interface():
         # Initialize chatbot
         chatbot = Chatbot()
 
-        def process_all_inputs(urls, files, text):
+        def process_all_inputs(urls, file, text):
             """Process all input types with progress tracking"""
             try:
                 processor = URLProcessor()
@@ -338,16 +329,15 @@ def create_interface():
                             content = processor.fetch_content(url)
                             if content:
                                 results.append({
-                                    'source': ' url',
+                                    'source': 'url',
                                     'url': url,
                                     'content': content,
                                     'timestamp': datetime.now().isoformat()
                                 })
                 
                 # Process files
-                if files:
-                    combined_data = file_processor.process_files(files)
-                    results.extend(combined_data)
+                if file:
+                    results.extend(file_processor.process_file(file))
         
                 # Process text input
                 if text:
@@ -405,7 +395,7 @@ def create_interface():
         gr.Markdown("""
         ### Usage Guidelines
         - **URL Processing**: Enter valid HTTP/HTTPS URLs
-        - **File Input**: Upload multiple text files or ZIP archives
+        - **File Input**: Upload text files or ZIP archives
         - **Text Input**: Direct text processing
         - **Chat**: Load your JSON data and ask questions about it
         - Advanced cleaning and validation included
@@ -430,4 +420,4 @@ def main():
     )
 
 if __name__ == "__main__":
-    main()
+    main()