app.py

import json
import sys
import os
import re
import time
import logging
import mimetypes
import tempfile
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse
from typing import List, Dict, Tuple, Union, Optional
import requests
import validators
import gradio as gr
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from cleantext import clean
import qrcode
import zipfile

# 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.log', encoding='utf-8')
    ])
logger = logging.getLogger(__name__)


class URLProcessor:
    def __init__(self):
        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',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        })

    def advanced_text_cleaning(self, text: str) -> str:
        """Robust text cleaning with version compatibility"""
        try:
            cleaned_text = clean(
                text,
                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)  # Remove control characters
            text = text.encode('ascii', 'ignore').decode('ascii')  # Remove non-ASCII characters
            text = re.sub(r'\s+', ' ', text)  # Normalize whitespace
            return text.strip()

    def validate_url(self, url: str) -> Dict:
        """Validate URL format and accessibility"""
        try:
            if not validators.url(url):
                return {'is_valid': False, 'message': 'Invalid URL format'}

            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]:
        """Universal content fetcher with special case handling"""
        try:
            # Google Drive document handling
            if 'drive.google.com' in url:
                return self._handle_google_drive(url)

            # Google Calendar ICS handling
            if 'calendar.google.com' in url and 'ical' in url:
                return self._handle_google_calendar(url)

            # Standard HTML processing
            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]:
        """Process Google Drive file links"""
        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
    
            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()
    
            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]:
        """Process Google Calendar ICS feeds"""
        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]:
        """Standard HTML content processing"""
        try:
            response = self.session.get(url, timeout=self.timeout)
            response.raise_for_status()

            soup = BeautifulSoup(response.text, 'html.parser')

            # Remove unwanted elements
            for element in soup(['script', 'style', 'nav', 'footer', 'header', 'meta', 'link']):
                element.decompose()
                
            # Remove login walls and overlays common on social media sites
            for element in soup.select('.login-wall, .signup-wall, .overlay, .modal, [role="dialog"], [aria-modal="true"]'):
                element.decompose()
                
            # Remove specific elements for known sites
            if 'facebook.com' in url:
                for element in soup.select('[data-testid="cookie-policy-manage-dialog"], [role="banner"], [role="complementary"]'):
                    element.decompose()
            elif 'instagram.com' in url or 'twitter.com' in url or 'x.com' in url:
                for element in soup.select('[role="presentation"], [role="banner"], [role="complementary"]'):
                    element.decompose()

            # Extract content using a general approach
            # First try to find main content containers
            main_content = soup.find('main') or soup.find('article') or soup.find('div', class_=lambda c: c and any(term in c for term in ['content', 'main', 'body', 'post', 'feed', 'timeline']))
            
            # If no specific container found, fall back to body
            if not main_content or not main_content.get_text(strip=True):
                logger.info(f"No main content container found for {url}, using body")
                main_content = soup.body if soup.body else soup
                
            # Extract text with proper spacing
            text_content = main_content.get_text(separator='\n', strip=True)
            
            # If content is too short, try a more aggressive approach to get all visible text
            if len(text_content) < 100:
                logger.info(f"Content too short for {url}, using all visible text")
                visible_text = []
                for element in soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li', 'span', 'div']):
                    if element.get_text(strip=True):
                        visible_text.append(element.get_text(strip=True))
                text_content = '\n'.join(visible_text)

            # Clean and structure content
            cleaned_content = self.advanced_text_cleaning(text_content)

            return {
                'content': cleaned_content,
                'content_type': response.headers.get('Content-Type', ''),
                'timestamp': datetime.now().isoformat(),
                'url': url  # Add the URL to the returned data for reference
            }
        except Exception as e:
            logger.error(f"HTML processing failed for {url}: {e}")
            return None


class FileProcessor:
    """Class to handle file processing"""

    def __init__(self, max_file_size: int = 2 * 1024 * 1024 * 1024):  # 2GB 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/')) or \
                   (os.path.splitext(filepath)[1].lower() in self.supported_text_extensions)
        except Exception:
            return False

    def process_file(self, file) -> List[Dict]:
        """Process uploaded file with enhanced error handling"""
        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, temp_dir):
        """Extract and process files within a ZIP archive."""
        result = []
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            zip_ref.extractall(temp_dir)
            for extracted_file in os.listdir(temp_dir):
                extracted_file_path = os.path.join(temp_dir, extracted_file)
                if os.path.isfile(extracted_file_path):
                    with open(extracted_file_path, 'r', encoding='utf-8', errors='ignore') as f:
                        result.append({
                            'source': 'file_from_zip',
                            'filename': extracted_file,
                            'content': f.read(),
                            'timestamp': datetime.now().isoformat()
                        })
        return result

    def _process_single_file(self, file) -> List[Dict]:
        try:
            file_stat = os.stat(file.name)

            # For very large files, read in chunks and summarize
            if file_stat.st_size > 100 * 1024 * 1024:  # 100MB
                logger.info(f"Processing large file: {file.name} ({file_stat.st_size} bytes)")

                # Read first and last  1MB for extremely large files
                content = ""
                with open(file.name, 'r', encoding='utf-8', errors='ignore') as f:
                    content = f.read(1 * 1024 * 1024)  # First 1MB
                    content += "\n...[Content truncated due to large file size]...\n"

                    # Seek to the last 1MB
                    f.seek(max(0, file_stat.st_size - 1 * 1024 * 1024))
                    content += f.read()  # Last 1MB
            else:
                # Regular file processing
                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 []

# Move process_all_inputs outside of the FileProcessor class
def process_all_inputs(urls, file, text, notes):
    """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': datetime.now().isoformat()
                        })

        # 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': datetime.now().isoformat()
            })

        # 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'

            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!"
            json_data = json.dumps(results, indent=2)  # Prepare JSON for QR code
            return str(output_path), summary, json_data  # Return JSON for editor
        else:
            return None, "No valid content to process.", ""

    except Exception as e:
        logger.error(f"Processing error: {e}")
        return None, f"Error: {str(e)}", ""

# Also move generate_qr_code outside of the FileProcessor class
def generate_qr_code(json_data):
    """Generate QR code from JSON data and return the file path."""
    if json_data:
        return generate_qr(json_data)

# Move generate_qr outside of the FileProcessor class as well
def generate_qr(json_data):
    """Generate QR code from JSON data and return the file path."""
    try:
        # Try first with automatic version selection
        qr = qrcode.QRCode(
            error_correction=qrcode.constants.ERROR_CORRECT_L,
            box_size=10,
            border=4,
        )
        qr.add_data(json_data)
        qr.make(fit=True)
        
        img = qr.make_image(fill_color="black", back_color="white")
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
        img.save(temp_file.name)
        return temp_file.name
    except Exception as e:
        # If the data is too large for a QR code
        logger.error(f"QR generation error: {e}")
        
        # Create a simple QR with error message
        qr = qrcode.QRCode(
            version=1,
            error_correction=qrcode.constants.ERROR_CORRECT_L,
            box_size=10,
            border=4,
        )
        qr.add_data("Error: Data too large for QR code")
        qr.make(fit=True)
        
        img = qr.make_image(fill_color="black", back_color="white")
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
        img.save(temp_file.name)
        return temp_file.name

def create_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..."
            )

        with gr.Tab("JSON Editor"):
            json_editor = gr.Textbox(
                label="JSON Editor",
                lines=20,
                placeholder="View and edit your JSON data here...",
                interactive=True,
                elem_id="json-editor"  # Optional: for custom styling
            )

        with gr.Tab("Scratchpad"):
            scratchpad = gr.Textbox(
                label="Scratchpad",
                lines=10,
                placeholder="Quick notes or text collections...",
                interactive=True
            )

        process_btn = gr.Button("Process Input", variant="primary")
        qr_btn = gr.Button("Generate QR Code", variant="secondary")

        output_text = gr.Textbox(label="Processing Results", interactive=False)
        output_file = gr.File(label="Processed Output")
        qr_output = gr.Image(label="QR Code", type="filepath")  # To display the generated QR code

        process_btn.click(
            process_all_inputs,
            inputs=[url_input, file_input, text_input, scratchpad],
            outputs=[output_file, output_text, json_editor]  # Update outputs to include JSON editor
        )

        qr_btn.click(
            generate_qr_code,
            inputs=json_editor,
            outputs=qr_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
    - **JSON Editor**: View and edit your JSON data
    - **Scratchpad**: Quick notes or text collections
    - Advanced cleaning and validation included
    """)
    return interface

def main():
    # Configure system settings
    mimetypes.init()

    # Create and launch interface
    interface = create_interface()

    # Launch with proper configuration
    interface.launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True,
        share=False,
        inbrowser=True,
        debug=True
    )

if __name__ == "__main__":
    main()