Update app.py

app.py

@@ -6,43 +6,37 @@ import logging
 import mimetypes
 import concurrent.futures
 import string
-import random
-from typing import List, Dict, Optional, Union, Any
+from typing import List, Dict, Optional, Union
 from pathlib import Path
 from urllib.parse import urlparse
 
 import requests
 import validators
 import gradio as gr
-import torch
 import cachetools
 from bs4 import BeautifulSoup
+from fake_useragent import UserAgent
 from ratelimit import limits, sleep_and_retry
 
-# Advanced Logging Configuration
+# Setup logging with detailed configuration
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
     handlers=[
         logging.StreamHandler(),
-        logging.FileHandler('app_advanced.log', encoding='utf-8')
+        logging.FileHandler('app.log', encoding='utf-8')
     ]
 )
+logger = logging.getLogger(__name__)
 
-class AdvancedURLProcessor:
-    """Enhanced URL processing with advanced features"""
+class URLProcessor:
+    """Class to handle URL processing with advanced features"""
     
-    def __init__(
-        self, 
-        timeout: int = 15, 
-        max_retries: int = 3, 
-        concurrent_requests: int = 5,
-        cache_size: int = 100
-    ):
+    def __init__(self, timeout: int = 15, max_retries: int = 3, concurrent_requests: int = 5, cache_size: int = 100):
         self.timeout = timeout
         self.max_retries = max_retries
         self.concurrent_requests = concurrent_requests
-        self.ua = UserAgent()
+        self.ua = UserAgent()  # Initialize UserAgent
         
         # Implement multilevel caching
         self.url_cache = cachetools.LRUCache(maxsize=cache_size)
@@ -50,7 +44,7 @@ class AdvancedURLProcessor:
         
         self.session = requests.Session()
         self.session.headers.update({
-            'User-Agent': self.ua.random,
+            'User -Agent': self.ua.random,  # Use random User-Agent
             'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
             'Accept-Language': 'en-US,en;q=0.5',
             'Connection': 'keep-alive'
@@ -59,13 +53,12 @@ class AdvancedURLProcessor:
     @sleep_and_retry
     @limits(calls=10, period=60)  # Rate limiting: 10 calls per minute
     def validate_url(self, url: str) -> Dict[str, Union[bool, str]]:
-        """Enhanced URL validation with comprehensive checks"""
+        """Validate URL format and accessibility"""
         try:
             # Check cache first
             if url in self.url_cache:
                 return self.url_cache[url]
             
-            # Comprehensive URL validation
             result = urlparse(url)
             validation_result = {
                 'is_valid': False,
@@ -78,23 +71,16 @@ class AdvancedURLProcessor:
                 validation_result['message'] = 'Missing scheme or network location'
                 return validation_result
             
-            # Use validators for additional checks
             if not validators.url(url):
                 validation_result['message'] = 'URL format validation failed'
                 return validation_result
             
             # Perform HEAD request for accessibility
             try:
-                response = self.session.head(
-                    url, 
-                    timeout=self.timeout, 
-                    allow_redirects=True
-                )
-                
+                response = self.session.head(url, timeout=self.timeout, allow_redirects=True)
                 validation_result['is_valid'] = response.status_code in [200, 301, 302]
                 validation_result['status_code'] = response.status_code
                 validation_result['message'] = f"URL is {'valid' if validation_result['is_valid'] else 'invalid'}"
-                
             except requests.RequestException as e:
                 validation_result['message'] = f"Connection error: {str(e)}"
             
@@ -110,306 +96,260 @@ class AdvancedURLProcessor:
             }
 
     @sleep_and_retry
-    @limits(calls=10, period=60)
-    async def fetch_content(self, url: str) -> Optional[str]:
-        """Fetch content from URL with retry mechanism and caching"""
-        try:
-            # Check content cache first
-            if url in self.content_cache:
-                logger.info(f"Cache hit for URL: {url}")
-                return self.content_cache[url]
-
-            for attempt in range(self.max_retries):
-                try:
-                    response = self.session.get(url, timeout=self.timeout)
-                    response.raise_for_status()
-                    content = response.text
-                    
-                    # Cache the content
-                    self.content_cache[url] = content
-                    return content
-                    
-                except requests.RequestException as e:
-                    logger.warning(f"Attempt {attempt + 1}/{self.max_retries} failed for {url}: {str(e)}")
-                    if attempt == self.max_retries - 1:
-                        raise
-                    time.sleep(1)  # Delay between retries
-                    
-        except Exception as e:
-            logger.error(f"Error fetching content from {url}: {e}")
-            return None
-
-class ContentExtractor:
-    """Advanced content extraction and processing"""
-    
-    def __init__(self):
-        self.cleaners = [
-            self._remove_scripts,
-            self._remove_styles,
-            self._remove_special_chars,
-            self._normalize_whitespace
-        ]
+    @limits(calls=20, period=60)  # Refined rate limiting
+    def fetch_content(self, url: str) -> Optional[str]:
+        """Fetch content from URL with retry mechanism"""
+        # Check content cache first
+        if url in self.content_cache:
+            return self.content_cache[url]
         
-    def extract_text(self, html: str, url: str = "") -> Dict[str, Union[str, Dict]]:
-        """Extract and clean text content with metadata"""
-        try:
-            if not html:
-                return {
-                    "success": False,
-                    "content": "",
-                    "metadata": {"error": "Empty HTML content"}
-                }
+        for attempt in range(self.max_retries):
+            try:
+                response = self.session.get(url, timeout=self.timeout)
+                response.raise_for_status()
                 
-            soup = BeautifulSoup(html, 'html.parser')
-            
-            # Extract metadata
-            metadata = self._extract_metadata(soup, url)
-            
-            # Clean content
-            content = self._process_content(soup)
-            
-            return {
-                "success": True,
-                "content": content,
-                "metadata": metadata
-            }
-            
-        except Exception as e:
-            logger.error(f"Content extraction error for {url}: {e}")
-            return {
-                "success": False,
-                "content": "",
-                "metadata": {"error": str(e)}
-            }
-    
-    def _extract_metadata(self, soup: BeautifulSoup, url: str) -> Dict:
-        """Extract page metadata"""
-        metadata = {
-            "title": self._get_title(soup),
-            "description": self._get_meta_description(soup),
-            "keywords": self._get_meta_keywords(soup),
-            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
-            "url": url
-        }
-        return metadata
-    
-    def _process_content(self, soup: BeautifulSoup) -> str:
-        """Process and clean content through multiple passes"""
-        for cleaner in self.cleaners:
-            soup = cleaner(soup)
+                # Use BeautifulSoup for more robust parsing
+                soup = BeautifulSoup(response.text, 'html.parser')
+                
+                # Remove scripts, styles, comments
+                for script in soup(["script", "style"]):
+                    script.decompose()
+                
+                # Extract clean text
+                text = soup.get_text(separator=' ')
+                cleaned_text = self .advanced_text_cleaning(text)
+                
+                # Cache the result
+                self.content_cache[url] = cleaned_text
+                return cleaned_text
+                
+            except requests.RequestException as e:
+                logger.warning(f"Fetch attempt {attempt + 1} failed for {url}: {e}")
+                time.sleep(2 ** attempt)  # Exponential backoff
         
-        # Extract text with preserved structure
-        lines = []
-        for element in soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6']):
-            text = element.get_text(strip=True)
-            if text:
-                lines.append(text)
+        return None
+
+    def advanced_text_cleaning(self, text: str) -> str:
+        """Sophisticated text cleaning and normalization"""
+        if not text:
+            return ""
         
-        return "\n".join(lines)
-    
-    @staticmethod
-    def _remove_scripts(soup: BeautifulSoup) -> BeautifulSoup:
-        for script in soup(["script", "style", "iframe", "noscript"]):
-            script.decompose()
-        return soup
-    
-    @staticmethod
-    def _remove_styles(soup: BeautifulSoup) -> BeautifulSoup:
-        for element in soup.find_all(style=True):
-            del element['style']
-        return soup
-    
-    @staticmethod
-    def _remove_special_chars(soup: BeautifulSoup) -> BeautifulSoup:
-        text = soup.get_text()
-        text = re.sub(r'[^\w\s\.\,\!\?\-]', '', text)
-        new_soup = BeautifulSoup(f"<div>{text}</div>", 'html.parser')
-        return new_soup
-    
-    @staticmethod
-    def _normalize_whitespace(soup: BeautifulSoup) -> BeautifulSoup:
-        text = soup.get_text()
+        # Remove control characters
+        text = re.sub(r'[\x00-\x1F\x7F-\x9F]', '', text)
+        
+        # Normalize Unicode characters
+        text = text.encode('ascii', 'ignore').decode('ascii')
+        
+        # Replace multiple whitespaces
         text = re.sub(r'\s+', ' ', text)
-        new_soup = BeautifulSoup(f"<div>{text}</div>", 'html.parser')
-        return new_soup
-    
-    @staticmethod
-    def _get_title(soup: BeautifulSoup) -> str:
-        title = soup.find('title')
-        return title.get_text(strip=True) if title else ""
-    
-    @staticmethod
-    def _get_meta_description(soup: BeautifulSoup) -> str:
-        meta = soup.find('meta', attrs={'name': 'description'})
-        return meta.get('content', '') if meta else ""
-    
-    @staticmethod
-    def _get_meta_keywords(soup: BeautifulSoup) -> str:
-        meta = soup.find('meta', attrs={'name': 'keywords'})
-        return meta.get('content', '') if meta else ""
+        
+        # Remove HTML entities
+        text = re.sub(r'&[a-zA-Z]+;', '', text)
+        
+        # Normalize quotation marks
+        text = text.replace('"', '"').replace('"', '"')
+        text = text.replace('‘', "'").replace('’', "'")
+        
+        # Remove excessive punctuation
+        text = re.sub(r'([.,!?]){2,}', r'\1', text)
+        
+        return text.strip()
 
-class ContentProcessor:
-    """Main content processing orchestrator"""
-    
-    def __init__(self):
-        self.url_processor = AdvancedURLProcessor()
-        self.content_extractor = ContentExtractor()
-        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+class FileProcessor:
+    """Class to handle file processing"""
     
-    async def process_urls(self, urls: List[str]) -> Dict[str, Any]:
-        """Process multiple URLs concurrently with advanced error handling"""
-        results = {
-            "successful": [],
-            "failed": [],
-            "metadata": {
-                "total_urls": len(urls),
-                "start_time": time.strftime("%Y-%m-%d %H:%M:%S")
-            }
-        }
-        
+    def __init__(self, max_file_size: int = 10 * 1024 * 1024):  # 10MB default
+        self.max_file_size = max_file_size
+        self.supported_text_extensions = {'.txt', '.md', '.csv', '.json', '.xml'}
+
+    def is_text_file(self, filepath: str) -> bool:
+        """Check if file is a text file"""
+        try:
+            mime_type, _ = mimetypes.guess_type(filepath)
+            return mime_type and mime_type.startswith('text/')
+        except Exception:
+            return False
+
+    def process_file(self, file) -> List[Dict]:
+        """Process uploaded file with enhanced error handling"""
+        if not file:
+            return []
+
+        dataset = []
         try:
-            # Validate URLs first
-            valid_urls = []
-            for url in urls:
-                validation_result = self.url_processor.validate_url(url)
-                if validation_result['is_valid']:
-                    valid_urls.append(url)
+            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 []
+
+            with tempfile.TemporaryDirectory() as temp_dir:
+                if zipfile.is_zipfile(file.name):
+                    dataset.extend(self._process_zip_file(file.name, temp_dir))
                 else:
-                    results['failed'].append({
-                        "url": url,
-                        "error": validation_result['message']
-                    })
-            
-            # Process valid URLs concurrently
-            futures = []
-            for url in valid_urls:
-                future = self.executor.submit(self._process_single_url, url)
-                futures.append((url, future))
-            
-            # Collect results
-            for url, future in futures:
-                try:
-                    result = future.result(timeout=30)  # 30-second timeout
-                    if result["success"]:
-                        results["successful"].append(result)
-                    else:
-                        results["failed"].append({
-                            "url": url,
-                            "error": "Processing failed"
-                        })
-                except Exception as e:
-                    logger.error(f"Error processing {url}: {e}")
-                    results["failed"].append({
-                        "url": url,
-                        "error": str(e)
-                    })
-            
-            # Update metadata
-            results["metadata"].update({
-                "end_time": time.strftime("%Y-%m-%d %H:%M:%S"),
-                "successful_count": len(results["successful"]),
-                "failed_count": len(results["failed"])
-            })
-            
-            return results
-            
+                    dataset.extend(self._process_single_file(file))
+
         except Exception as e:
-            logger.error(f"Batch processing error: {e}")
-            raise
+            logger.error(f"Error processing file: {str(e)}")
+            return []
+
+        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:
+            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', errors='ignore') as f:
+                                content = f.read()
+                            if content.strip():
+                                results.append({
+                                    "source": "file",
+                                    "filename": filename,
+                                    "content": content,
+                                    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
+                                })
+                        except Exception as e:
+                            logger.error(f"Error reading file {filename}: {str(e)}")
+        return results
 
-    def _process_single_url(self, url: str) -> Dict:
-        """Process a single URL with comprehensive error handling"""
+    def _process_single_file(self, file) -> List[Dict]:
+        """Process single file"""
+        results = []
         try:
-            response = self.url_processor.session.get(
-                url,
-                timeout=self.url_processor.timeout
-            )
-            response.raise_for_status()
-            
-            result = self.content_extractor.extract_text(
-                response.text,
-                url
-            )
-            
-            result["url"] = url
-            result["timestamp"] = time.strftime("%Y-%m-%d %H:%M:%S")
-            
-            return result
-            
+            content = file.read().decode('utf-8', errors='ignore')
+            if content.strip():
+                results.append({
+                    "source": "file",
+                    "filename": os.path.basename(file.name),
+                    "content": content,
+                    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
+                })
         except Exception as e:
-            logger.error(f"Error processing {url}: {e}")
-            return {
-                "success": False,
-                "url": url,
-                "error": str(e)
-            }
+            logger.error(f"Error processing single file: {str(e)}")
+        return results
 
 def create_interface():
-    """Create Gradio interface with advanced features"""
-    processor = ContentProcessor()
+    """Create a comprehensive Gradio interface with advanced features"""
     
-    with gr.Blocks(title="Advanced URL Content Processor") as interface:
-        gr.Markdown("# Advanced URL Content Processor")
+    css = """
+    .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 Text & URL Processor") as interface:
+        gr.Markdown("# 🌐 Advanced URL & Text Processing Toolkit")
         
-        with gr.Row():
-            with gr.Column():
-                url_input = gr.Textbox(
-                    label="Enter URLs (one per line)",
-                    placeholder="https://example.com\nhttps://example.org",
-                    lines=5
-                )
-                
-                with gr.Row():
-                    process_btn = gr.Button("Process URLs", variant="primary")
-                    clear_btn = gr.Button("Clear")
-            
-            with gr.Column():
-                status_output = gr.JSON(
-                    label="Processing Results",
-                    show_label=True
-                )
+        with gr.Tab("URL Processing"):
+            url_input = gr.Textbox(
+                label="Enter URLs (comma or newline separated)", 
+                lines=5,
+                placeholder="https://example1.com\nhttps://example2.com"
+            )
         
-        gr.Markdown("## Processing Status")
-        with gr.Row():
-            progress_output = gr.Textbox(
-                label="Progress",
-                show_label=True
+        with gr.Tab("File Input"):
+            file_input = gr.File(
+                label="Upload text file or ZIP archive",
+                file_types=[".txt", ".zip", ".md", ".csv", ".json", ".xml"]
+            )
+
+        with gr.Tab("Text Input"):
+            text_input = gr.Textbox(
+                label="Raw Text Input", 
+                lines=5,
+                placeholder="Paste your text here..."
             )
         
-        async def process_urls(urls):
-            if not urls.strip():
-                return {"error": "No URLs provided"}
-            
-            url_list = [url.strip() for url in urls.splitlines() if url.strip()]
-            results = await processor.process_urls(url_list)
-            return results
+        process_btn = gr.Button("Process Input", variant="primary")
+        
+        output_text = gr.Textbox(label="Processing Results", interactive=False)
+        output_file = gr.File(label="Processed Output")
         
-        def clear_inputs():
-            return None, None
+        def process_all_inputs(urls, file, text):
+            """Process all input types with progress tracking"""
+            try:
+                processor = URLProcessor()
+                file_processor = FileProcessor()
+                results = []
+                
+                # Process URLs
+                if urls:
+                    url_list = re.split(r'[,\n]', urls)
+                    url_list = [url.strip() for url in url_list if url.strip()]
+                    
+                    for url in url_list:
+                        validation = processor.validate_url(url)
+                        if validation.get('is_valid'):
+                            content = processor.fetch_content(url)
+                            if content:
+                                results.append({
+                                    'source': 'url',
+                                    'url': url,
+                                    'content': content,
+                                    'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
+                                })
+                
+                # Process files
+                if file:
+                    results.extend(file_processor.process_file(file))
+
+                # Process text input
+                if text:
+                    cleaned_text = processor.advanced_text_cleaning(text)
+                    results.append({
+                        'source': 'direct_input',
+                        'content': cleaned_text,
+                        'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
+                    })
+                
+                # Generate output
+                if results:
+                    output_path = 'processed_data.json'
+                    with open(output_path, 'w', encoding='utf-8') as f:
+                        json.dump(results, f, ensure_ascii=False, indent=2)
+                    
+                    summary = f"Processed {len(results)} items successfully!"
+                    return output_path, summary
+                else:
+                    return None, "No valid content to process."
+            
+            except Exception as e:
+                logger.error(f"Processing error: {e}")
+                return None, f"Error: {str(e)}"
         
         process_btn.click(
-            fn=process_urls,
-            inputs=[url_input],
-            outputs=[status_output]
+            process_all_inputs, 
+            inputs=[url_input, file_input, text_input], 
+            outputs=[output_file, output_text]
         )
         
-        clear_btn.click(
-            fn=clear_inputs,
-            inputs=[],
-            outputs=[url_input, status_output]
-        )
+        gr.Markdown("""
+        ### Usage Guidelines
+        - **URL Processing**: Enter valid HTTP/HTTPS URLs
+        - **File Input**: Upload text files or ZIP archives
+        - **Text Input**: Direct text processing
+        - Advanced cleaning and validation included
+        """)
     
     return interface
 
-if __name__ == "__main__":
-    # Initialize mimetypes
+def main():
+    # Configure system settings
     mimetypes.init()
     
     # Create and launch interface
     interface = create_interface()
     interface.launch(
+        share=True,
         server_name="0.0.0.0",
         server_port=7860,
-        share=False,
         debug=True
-    )
+    )
+
+if __name__ == "__main__":
+    main()