Update app.py

app.py

@@ -6,357 +6,277 @@ import logging
 import mimetypes
 import concurrent.futures
 import string
-import zipfile
-import tempfile
-from datetime import datetime
 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
-from cleantext import clean
-
 
-# Setup logging with detailed configuration
+# Advanced Logging Configuration
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
     handlers=[
         logging.StreamHandler(),
-        logging.FileHandler('app.log', encoding='utf-8')
+        logging.FileHandler('app_advanced.log', encoding='utf-8')
     ]
 )
 logger = logging.getLogger(__name__)
 
-class URLProcessor:
-    def __init__(self):
+class AdvancedURLProcessor:
+    """Enhanced URL processing with advanced features"""
+    
+    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()
+        
+        # Implement multilevel caching
+        self.url_cache = cachetools.LRUCache(maxsize=cache_size)
+        self.content_cache = cachetools.TTLCache(maxsize=cache_size, ttl=3600)  # 1-hour cache
+        
         self.session = requests.Session()
-        self.timeout = 10  # seconds
         self.session.headers.update({
-            'User-Agent': UserAgent().random,
-            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
+            'User-Agent': self.ua.random,
+            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
             'Accept-Language': 'en-US,en;q=0.5',
-            'Accept-Encoding': 'gzip, deflate, br',
-            'Connection': 'keep-alive',
-            'Upgrade-Insecure-Requests': '1'
+            'Connection': 'keep-alive'
         })
 
-    def advanced_text_cleaning(self, text: str) -> str:
+    @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"""
         try:
-            cleaned_text = clean(
-                text,
-                fix_unicode=True,
-                to_ascii=True,
-                lower=True,
-                no_line_breaks=True,
-                no_urls=True,
-                no_emails=True,
-                no_phone_numbers=True,
-                no_numbers=False,
-                no_digits=False,
-                no_currency_symbols=True,
-                no_punct=False
-            ).strip()
-            return cleaned_text
-        except Exception as e:
-            logger.warning(f"Text cleaning error: {e}. Using fallback method.")
-            text = re.sub(r'[\x00-\x1F\x7F-\x9F]', '', text)
-            text = text.encode('ascii', 'ignore').decode('ascii')
-            text = re.sub(r'\s+', ' ', text)
-            return text.strip()
-
-    def validate_url(self, url: str) -> Dict:
-        try:
-            if not validators.url(url):
-                return {'is_valid': False, 'message': 'Invalid URL format'}
+            # Check cache first
+            if url in self.url_cache:
+                return self.url_cache[url]
             
-            response = self.session.head(url, timeout=self.timeout)
-            response.raise_for_status()
-            return {'is_valid': True, 'message': 'URL is valid and accessible'}
-        except Exception as e:
-            return {'is_valid': False, 'message': f'URL validation failed: {str(e)}'}
-
-    def fetch_content(self, url: str) -> Optional[Dict]:
-        try:
-            if 'drive.google.com' in url:
-                return self._handle_google_drive(url)
-            if 'calendar.google.com' in url and 'ical' in url:
-                return self._handle_google_calendar(url)
-            return self._fetch_html_content(url)
-        except Exception as e:
-            logger.error(f"Content fetch failed: {e}")
-            return None
-
-    def _handle_google_drive(self, url: str) -> Optional[Dict]:
-        try:
-            file_id = re.search(r'/file/d/([a-zA-Z0-9_-]+)', url)
-            if not file_id:
-                logger.error(f"Invalid Google Drive URL: {url}")
-                return None
+            # Comprehensive URL validation
+            result = urlparse(url)
+            validation_result = {
+                'is_valid': False,
+                'message': 'Invalid URL',
+                'scheme': result.scheme,
+                'netloc': result.netloc
+            }
             
-            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()
+            if not all([result.scheme, result.netloc]):
+                validation_result['message'] = 'Missing scheme or network location'
+                return validation_result
             
-            return {
-                'content': response.text,
-                'content_type': response.headers.get('Content-Type', ''),
-                'timestamp': datetime.now().isoformat()
-            }
-        except Exception as e:
-            logger.error(f"Google Drive processing failed: {e}")
-            return None
-
-    def _handle_google_calendar(self, url: str) -> Optional[Dict]:
-        try:
-            response = self.session.get(url, timeout=self.timeout)
-            response.raise_for_status()
-            return {
-                'content': response.text,
-                'content_type': 'text/calendar',
-                'timestamp': datetime.now().isoformat()
-            }
-        except Exception as e:
-            logger.error(f"Calendar fetch failed: {e}")
-            return None
-
-    def _fetch_html_content(self, url: str) -> Optional[Dict]:
-        try:
-            response = self.session.get(url, timeout=self.timeout)
-            response.raise_for_status()
+            # 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
+                )
+                
+                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)}"
             
-            soup = BeautifulSoup(response.text, 'html.parser')
-            for element in soup(['script', 'style', 'nav', 'footer', 'header', 'meta', 'link']):
-                element.decompose()
-            main_content = soup.find('main') or soup.find('article') or soup.body
-            if main_content is None:
-                logger.warning(f"No main content found for URL: {url}")
-                return {
-                    'content': '',
-                    'content_type': response.headers.get('Content-Type', ''),
-                    'timestamp': datetime.now().isoformat()
-                }
-            text_content = main_content.get_text(separator='\n', strip=True)
-            cleaned_content = self.advanced_text_cleaning(text_content)
+            # Cache the result
+            self.url_cache[url] = validation_result
+            return validation_result
+        
+        except Exception as e:
+            logger.error(f"Unexpected error validating URL {url}: {e}")
             return {
-                'content': cleaned_content,
-                'content_type': response.headers.get('Content-Type', ''),
-                'timestamp': datetime.now().isoformat()
+                'is_valid': False, 
+                'message': f"Unexpected validation error: {str(e)}"
             }
-        except Exception as e:
-            logger.error(f"HTML processing failed: {e}")
-            return None
-
-class FileProcessor:
-    def __init__(self, max_file_size: int = 2 * 1024 * 1024 * 1024):
-        self.max_file_size = max_file_size
-        self.supported_text_extensions = {'.txt', '.md', '.csv', '.json', '.xml'}
-
-    def is_text_file(self, filepath: str) -> bool:
-        try:
-            mime_type, _ = mimetypes.guess_type(filepath)
-            return (mime_type and mime_type.startswith('text/')) or \
-                   (os.path.splitext(filepath)[1].lower() in self.supported_text_extensions)
-        except Exception:
-            return False
-
-    def process_file(self, file) -> List[Dict]:
-        if not file:
-            return []
-
-        dataset = []
-        try:
-            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:
-                    dataset.extend(self._process_single_file(file))
-
-        except Exception as e:
-            logger.error(f"Error processing file: {str(e)}")
-            return []
-
-        return dataset
-
-    def _process_zip_file(self, zip_path: str, temp_dir: str) -> List[Dict]:
-        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', 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) -> List[Dict]:
-        try:
-            file_stat = os.stat(file.name)
-            if file_stat.st_size > 100 * 1024 * 1024:
-                logger.info(f"Processing large file: {file.name} ({file_stat.st_size} bytes)")
-                content = ""
-                with open(file.name, 'r', encoding='utf-8', errors='ignore') as f:
-                    content = f.read(1 * 1024 * 1024)
-                    content += "\n...[Content truncated due to large file size]...\n"
-                    f.seek(max(0, file_stat.st_size - 1 * 1024 * 1024))
-                    content += f.read()
-            else:
-                with open(file.name, 'r', encoding='utf-8', errors='ignore') as f:
-                    content = f.read()
-
-            return [{
-                'source': 'file',
-                'filename': os.path.basename(file.name),
-                'file_size': file_stat.st_size,
-                '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,
-                'timestamp': datetime.now().isoformat()
-            }]
-        except Exception as e:
-            logger.error(f"File processing error: {e}")
-            return []
-
-def generate_qr(json_data):
-    if json_data:
-        qr = qrcode.make(json_data)
-        qr_path = f"output/qr_code_{int(time.time())}.png"
-        qr.save(qr_path)
-        return qr_path
-    return None
-
-def create_interface():
+    def advanced_text_cleaning(self, text: str) -> str:
+        """Sophisticated text cleaning and normalization"""
+        if not text:
+            return ""
+        
+        # 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)
+        
+        # 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()
+
+    @sleep_and_retry
+    @limits(calls=20, period=60)  # Refined rate limiting
+    def fetch_content(self, url: str) -> Optional[str]:
+        """Advanced content fetching with multiple safeguards"""
+        # Check content cache first
+        if url in self.content_cache:
+            return self.content_cache[url]
+        
+        for attempt in range(self.max_retries):
+            try:
+                response = self.session.get(
+                    url, 
+                    timeout=self.timeout, 
+                    headers={'User-Agent': self.ua.random}
+                )
+                response.raise_for_status()
+                
+                # 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
+        
+        return None
+
+def create_advanced_interface():
+    """Create a comprehensive Gradio interface with advanced features"""
+    
     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.Tab("URL Processing"):
             url_input = gr.Textbox(
                 label="Enter URLs (comma or newline separated)", 
                 lines=5,
                 placeholder="https://example1.com\nhttps://example2.com"
             )
-
-        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..."
             )
-
+        
         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 process_all_inputs(urls, file, text):
+        
+        def process_input(urls, text):
             try:
-                processor = URLProcessor()
-                file_processor = FileProcessor()
+                processor = AdvancedURLProcessor()
                 results = []
-
+                
+                # Process URLs
                 if urls:
-                    url_list = re.split(r'[\,\n]', 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': datetime.now().isoformat()
+                                    'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
                                 })
-
-                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': datetime.now().isoformat()
+                        'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
                     })
-
+                
+                # Generate output
                 if results:
-                    output_dir = Path('output') / datetime.now().strftime('%Y-%m-%d')
-                    output_dir.mkdir(parents=True, exist_ok=True)
-                    output_path = output_dir / f'processed_{int(time.time())}.json'
-
+                    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 str(output_path), summary
+                    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(
-            process_all_inputs, 
-            inputs=[url_input, file_input, text_input], 
+            process_input, 
+            inputs=[url_input, text_input], 
             outputs=[output_file, output_text]
         )
-
+        
         gr.Markdown("""
         ### Usage Guidelines
-        - **URL Processing**: Enter valid HTTP/HTTPS URLs
-        - **File Input**: Upload text files or ZIP archives
-        - **Text Input**: Direct text processing
+        - URL Processing: Enter valid HTTP/HTTPS URLs
+        - Text Input: Direct text processing
         - Advanced cleaning and validation included
         """)
-
+    
     return interface
 
 def main():
+    # Configure system settings
     mimetypes.init()
-    interface = create_interface()
+    
+    # Create and launch interface
+    interface = create_advanced_interface()
     interface.launch(
+        share=True,
         server_name="0.0.0.0",
         server_port=7860,
-        show_error=True,
-        share=False,
-        inbrowser=True,
         debug=True
     )
 
 if __name__ == "__main__":
-    main()
+    main()