Update app.py

app.py

@@ -1,282 +1,428 @@
+"""
+Advanced URL & Text Processing Suite - Main Application
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A sophisticated Gradio interface for URL processing, file manipulation, and QR operations.
+Features a modern, responsive UI with real-time feedback and advanced processing capabilities.
+"""
+
+import gradio as gr
+import logging
 import json
 import os
-import re
-import time
-import logging
-import mimetypes
-import concurrent.futures
-import string
-from typing import List, Dict, Optional, Union
+import sys
+from datetime import datetime
 from pathlib import Path
-from urllib.parse import urlparse
+from typing import Dict, List, Optional, Union, Any, Tuple
 
-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 url_processor import URLProcessor
+from file_processor import FileProcessor
+from qr_processor import QRProcessor
 
-# Advanced Logging Configuration
+# Configure logging
 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')
-    ]
+    format='%(asctime)s.%(msecs)03d [%(levelname)s] %(name)s - %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
 )
 logger = logging.getLogger(__name__)
 
-class AdvancedURLProcessor:
-    """Enhanced URL processing with advanced features"""
+# Modern UI Configuration
+THEME = gr.themes.Soft(
+    primary_hue="indigo",
+    secondary_hue="blue",
+    neutral_hue="slate",
+    spacing_size=gr.themes.sizes.spacing_md,
+    radius_size=gr.themes.sizes.radius_md,
+    text_size=gr.themes.sizes.text_md,
+).set(
+    body_background_fill="*background_fill_secondary",
+    button_primary_background_fill="*primary_500",
+    button_primary_background_fill_hover="*primary_600",
+    button_primary_text_color="white",
+    button_primary_border_color="*primary_500",
+    button_secondary_background_fill="*secondary_500",
+    button_secondary_background_fill_hover="*secondary_600",
+    button_secondary_text_color="white",
+    button_secondary_border_color="*secondary_500",
+)
+
+# Custom CSS for enhanced visual appeal
+CUSTOM_CSS = """
+.container {
+    max-width: 1200px !important;
+    margin: auto !important;
+    padding: 2rem !important;
+}
+
+.header {
+    text-align: center;
+    margin-bottom: 2rem;
+    padding: 1rem;
+    background: linear-gradient(135deg, #6366f1 0%, #2563eb 100%);
+    border-radius: 1rem;
+    color: white;
+}
+
+.status-bar {
+    background: #f8fafc;
+    border-radius: 0.5rem;
+    padding: 0.5rem;
+    margin: 1rem 0;
+    border: 1px solid #e2e8f0;
+}
+
+.success-message {
+    color: #059669;
+    background: #ecfdf5;
+    border: 1px solid #059669;
+    border-radius: 0.5rem;
+    padding: 1rem;
+    margin: 1rem 0;
+}
+
+.error-message {
+    color: #dc2626;
+    background: #fef2f2;
+    border: 1px solid #dc2626;
+    border-radius: 0.5rem;
+    padding: 1rem;
+    margin: 1rem 0;
+}
+
+.tab-nav {
+    background: #f8fafc;
+    border-radius: 0.5rem;
+    padding: 0.5rem;
+    margin-bottom: 1rem;
+}
+
+.tab-nav button {
+    transition: all 0.3s ease;
+}
+
+.tab-nav button:hover {
+    transform: translateY(-1px);
+}
+
+.json-editor {
+    font-family: 'JetBrains Mono', monospace;
+    background: #1e293b;
+    color: #e2e8f0;
+    border-radius: 0.5rem;
+    padding: 1rem;
+}
+
+.file-upload {
+    border: 2px dashed #e2e8f0;
+    border-radius: 1rem;
+    padding: 2rem;
+    text-align: center;
+    transition: all 0.3s ease;
+}
+
+.file-upload:hover {
+    border-color: #6366f1;
+    background: #f8fafc;
+}
+
+@keyframes pulse {
+    0% { transform: scale(1); }
+    50% { transform: scale(1.05); }
+    100% { transform: scale(1); }
+}
+
+.processing {
+    animation: pulse 2s infinite;
+}
+"""
+
+class AdvancedInterface:
+    """Advanced Gradio interface with sophisticated processing capabilities"""
     
-    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()
+    def __init__(self):
+        """Initialize processors and interface components"""
+        self.url_processor = URLProcessor()
+        self.file_processor = FileProcessor()
+        self.qr_processor = QRProcessor()
+        self.processing_history: List[Dict] = []
         
-        # 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.session.headers.update({
-            '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',
-            'Connection': 'keep-alive'
-        })
+    def process_urls(self, urls: str, mode: str) -> Tuple[str, str, str]:
+        """Process URLs with advanced error handling and result formatting"""
+        try:
+            # Split and clean URLs
+            url_list = [url.strip() for url in urls.split('\n') if url.strip()]
+            
+            if not url_list:
+                return "", "⚠️ No valid URLs provided", ""
+            
+            # Process URLs
+            results = self.url_processor.process_urls(url_list, mode=mode)
+            
+            # Format results
+            formatted_results = json.dumps(results, indent=2)
+            summary = self._generate_summary(results)
+            
+            # Update history
+            self.processing_history.append({
+                'timestamp': datetime.now().isoformat(),
+                'type': 'url_processing',
+                'input_count': len(url_list),
+                'success_count': len([r for r in results if 'error' not in r]),
+                'results': results
+            })
+            
+            return formatted_results, f"✅ Successfully processed {len(url_list)} URLs", summary
+            
+        except Exception as e:
+            logger.error(f"URL processing error: {e}", exc_info=True)
+            return "", f"❌ Error: {str(e)}", ""
 
-    @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"""
+    def process_file(self, file: gr.File) -> Tuple[str, str, Optional[Path]]:
+        """Process uploaded file with comprehensive error handling"""
         try:
-            # Check cache first
-            if url in self.url_cache:
-                return self.url_cache[url]
+            if not file:
+                return "", "⚠️ No file provided", None
             
-            # Comprehensive URL validation
-            result = urlparse(url)
-            validation_result = {
-                'is_valid': False,
-                'message': 'Invalid URL',
-                'scheme': result.scheme,
-                'netloc': result.netloc
-            }
+            results = self.file_processor.process_file(file)
             
-            if not all([result.scheme, result.netloc]):
-                validation_result['message'] = 'Missing scheme or network location'
-                return validation_result
+            # Generate output file
+            output_path = Path('processed_output.json')
+            with open(output_path, 'w') as f:
+                json.dump(results, f, indent=2)
             
-            # Use validators for additional checks
-            if not validators.url(url):
-                validation_result['message'] = 'URL format validation failed'
-                return validation_result
+            # Update history
+            self.processing_history.append({
+                'timestamp': datetime.now().isoformat(),
+                'type': 'file_processing',
+                'filename': file.name,
+                'results': results
+            })
             
-            # 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)}"
+            return (
+                json.dumps(results, indent=2),
+                f"✅ Successfully processed {file.name}",
+                output_path
+            )
             
-            # 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 {
-                'is_valid': False, 
-                'message': f"Unexpected validation error: {str(e)}"
-            }
+            logger.error(f"File processing error: {e}", exc_info=True)
+            return "", f"❌ Error: {str(e)}", None
 
-    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()
+    def generate_qr(self, data: str, size: int = 10) -> Tuple[Optional[Path], str]:
+        """Generate QR code with advanced options"""
+        try:
+            if not data:
+                return None, "⚠️ No data provided"
+            
+            result = self.qr_processor.generate_qr_code(data, size=size)
+            
+            if 'error' in result:
+                return None, f"❌ Error: {result['error']}"
+            
+            # Update history
+            self.processing_history.append({
+                'timestamp': datetime.now().isoformat(),
+                'type': 'qr_generation',
+                'data_length': len(data),
+                'output_path': result['output_path']
+            })
+            
+            return result['output_path'], "✅ QR code generated successfully"
+            
+        except Exception as e:
+            logger.error(f"QR generation error: {e}", exc_info=True)
+            return None, f"❌ Error: {str(e)}"
 
-    @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]
+    def _generate_summary(self, results: List[Dict]) -> str:
+        """Generate a human-readable summary of processing results"""
+        success_count = len([r for r in results if 'error' not in r])
+        error_count = len(results) - success_count
         
-        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
+        summary = [
+            "📊 Processing Summary",
+            f"Total URLs: {len(results)}",
+            f"Successful: {success_count}",
+            f"Failed: {error_count}",
+            "",
+            "🔍 Details:"
+        ]
         
-        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")
+        for result in results:
+            if 'error' in result:
+                summary.append(f"❌ {result['url']}: {result['error']}")
+            else:
+                summary.append(f"✅ {result['url']}: Successfully processed")
         
-        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"
-            )
+        return '\n'.join(summary)
+
+    def create_interface(self) -> gr.Blocks:
+        """Create a sophisticated Gradio interface"""
         
-        with gr.Tab("Text Input"):
-            text_input = gr.Textbox(
-                label="Raw Text Input", 
-                lines=5,
-                placeholder="Paste your text here..."
+        with gr.Blocks(theme=THEME, css=CUSTOM_CSS) as interface:
+            # Header
+            gr.Markdown(
+                """
+                <div class="header">
+                    <h1>🌐 Advanced URL & Text Processing Suite</h1>
+                    <p>Enterprise-grade toolkit for URL processing, file manipulation, and QR operations</p>
+                </div>
+                """
             )
-        
-        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_input(urls, text):
-            try:
-                processor = AdvancedURLProcessor()
-                results = []
-                
-                # Process URLs
-                if urls:
-                    url_list = re.split(r'[,\n]', urls)
-                    url_list = [url.strip() for url in url_list if url.strip()]
+            
+            # Main tabs
+            with gr.Tabs(elem_classes="tab-nav") as tabs:
+                # URL Processing Tab
+                with gr.Tab("🔗 URL Processing", id=1):
+                    with gr.Row():
+                        with gr.Column(scale=2):
+                            url_input = gr.Textbox(
+                                label="URLs",
+                                placeholder="Enter URLs (one per line)",
+                                lines=5
+                            )
+                            mode = gr.Radio(
+                                choices=["basic", "interactive", "deep"],
+                                value="basic",
+                                label="Processing Mode"
+                            )
+                            process_btn = gr.Button("🚀 Process URLs", variant="primary")
+                        
+                        with gr.Column(scale=1):
+                            gr.Markdown(
+                                """
+                                ### 📝 Instructions
+                                1. Enter URLs (one per line)
+                                2. Select processing mode:
+                                   - Basic: Simple content fetch
+                                   - Interactive: Handle dynamic content
+                                   - Deep: Follow links and analyze deeply
+                                3. Click Process to start
+                                """
+                            )
                     
-                    for url in url_list:
-                        validation = processor.validate_url(url)
-                        if validation.get('is_valid'):
-                            content = processor.fetch_content(url)
-                            if content:
-                                results.append({
-                                    'url': url,
-                                    'content': content,
-                                    'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
-                                })
-                
-                # 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")
-                    })
+                    with gr.Row():
+                        status_output = gr.Textbox(label="Status", interactive=False)
+                        clear_btn = gr.Button("🗑️ Clear", variant="secondary")
+                    
+                    with gr.Accordion("Results", open=False):
+                        json_output = gr.JSON(label="Detailed Results")
+                        summary_output = gr.Textbox(
+                            label="Summary",
+                            interactive=False,
+                            lines=10
+                        )
                 
-                # 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)
+                # File Processing Tab
+                with gr.Tab("📁 File Processing", id=2):
+                    with gr.Row():
+                        with gr.Column(scale=2):
+                            file_input = gr.File(
+                                label="Upload File",
+                                file_types=[".txt", ".pdf", ".zip", ".tar.gz"]
+                            )
+                            process_file_btn = gr.Button("📥 Process File", variant="primary")
+                        
+                        with gr.Column(scale=1):
+                            gr.Markdown(
+                                """
+                                ### 📝 Supported Formats
+                                - Text files (.txt)
+                                - PDF documents (.pdf)
+                                - Archives (.zip, .tar.gz)
+                                - Images (.jpg, .png)
+                                - And more...
+                                """
+                            )
                     
-                    summary = f"Processed {len(results)} items successfully!"
-                    return output_path, summary
-                else:
-                    return None, "No valid content to process."
+                    with gr.Row():
+                        file_status = gr.Textbox(label="Status", interactive=False)
+                        file_output = gr.File(label="Processed Output")
+                    
+                    with gr.Accordion("Results", open=False):
+                        file_json_output = gr.JSON(label="Detailed Results")
+                
+                # QR Code Tab
+                with gr.Tab("📱 QR Code", id=3):
+                    with gr.Row():
+                        with gr.Column(scale=2):
+                            qr_input = gr.Textbox(
+                                label="Data",
+                                placeholder="Enter data to encode",
+                                lines=3
+                            )
+                            qr_size = gr.Slider(
+                                minimum=5,
+                                maximum=50,
+                                value=10,
+                                step=1,
+                                label="QR Code Size"
+                            )
+                            generate_qr_btn = gr.Button("✨ Generate QR", variant="primary")
+                        
+                        with gr.Column(scale=1):
+                            qr_output = gr.Image(label="Generated QR Code")
+                            qr_status = gr.Textbox(label="Status", interactive=False)
             
-            except Exception as e:
-                logger.error(f"Processing error: {e}")
-                return None, f"Error: {str(e)}"
-        
-        process_btn.click(
-            process_input, 
-            inputs=[url_input, text_input], 
-            outputs=[output_file, output_text]
-        )
+            # Footer
+            gr.Markdown(
+                """
+                <div style="text-align: center; margin-top: 2rem; padding: 1rem; background: #f8fafc; border-radius: 0.5rem;">
+                    <p>Made with ❤️ by Advanced URL Processing Team</p>
+                    <p style="font-size: 0.8rem;">Version 1.0.0</p>
+                </div>
+                """
+            )
+            
+            # Event handlers
+            process_btn.click(
+                fn=self.process_urls,
+                inputs=[url_input, mode],
+                outputs=[json_output, status_output, summary_output]
+            )
+            
+            clear_btn.click(
+                fn=lambda: ("", "", ""),
+                inputs=[],
+                outputs=[url_input, status_output, summary_output]
+            )
+            
+            process_file_btn.click(
+                fn=self.process_file,
+                inputs=[file_input],
+                outputs=[file_json_output, file_status, file_output]
+            )
+            
+            generate_qr_btn.click(
+                fn=self.generate_qr,
+                inputs=[qr_input, qr_size],
+                outputs=[qr_output, qr_status]
+            )
         
-        gr.Markdown("""
-        ### Usage Guidelines
-        - URL Processing: Enter valid HTTP/HTTPS URLs
-        - Text Input: Direct text processing
-        - Advanced cleaning and validation included
-        """)
-    
-    return interface
+        return interface
 
 def main():
-    # Configure system settings
-    mimetypes.init()
-    
-    # Create and launch interface
-    interface = create_advanced_interface()
-    interface.launch(
-        share=True,
-        server_name="0.0.0.0",
-        server_port=7860,
-        debug=True
-    )
+    """Main entry point"""
+    try:
+        # Initialize interface
+        app = AdvancedInterface()
+        interface = app.create_interface()
+        
+        # Launch with optimal settings
+        interface.launch(
+            server_name="0.0.0.0",
+            server_port=8000,
+            share=False,
+            debug=True,
+            enable_queue=True,
+            max_threads=40,
+            auth=None,  # Add authentication if needed
+            ssl_keyfile=None,  # Add SSL if needed
+            ssl_certfile=None
+        )
+    except Exception as e:
+        logger.error(f"Application startup error: {e}", exc_info=True)
+        sys.exit(1)
 
 if __name__ == "__main__":
     main()