app2.py

import json
import os
import re
import logging
import mimetypes
import time
import io
from selenium import webdriver
from selenium.webdriver.common.by import By
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
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from ratelimit import limits, sleep_and_retry
from cleantext import clean
import qrcode
from PIL import Image
from pyzbar.pyzbar import decode
import nest_asyncio
nest_asyncio.apply()
import aiohttp

# Setup logging
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__)

# Ensure output directories exist
Path('output/qr_codes').mkdir(parents=True, exist_ok=True)

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,
                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)  # 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:
            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]:
        """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')
            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)
            
            return {
                'content': cleaned_content,
                'content_type': response.headers.get('Content-Type', ''),
                'timestamp': datetime.now().isoformat()
            }
        except Exception as e:
            logger.error(f"HTML processing failed: {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: 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', 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]:
        """Process a single file"""
        try:
            file_stat = os.stat(file.name)
            
            if file_stat.st_size > 100 * 1024 * 1024:  # 100MB
                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)  # First 1MB
                    content += "\n...[Content truncated due to large file size]...\n"
                    
                    f.seek(max(0, file_stat.st_size - 1 * 1024 * 1024))
                    content += f.read()  # Last 1MB
            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 clean_json(data: Union[str, Dict]) -> Optional[Dict]:
    """Clean and validate JSON data"""
    try:
        if isinstance(data, str):
            data = data.strip()
            data = json.loads(data)
        
        cleaned = json.loads(json.dumps(data))
        return cleaned
    except json.JSONDecodeError as e:
        logger.error(f"JSON cleaning error: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error while cleaning JSON: {e}")
        return None

def generate_qr_code(data: Union[str, Dict], combined: bool = True) -> List[str]:
    """Generate QR code(s) from data"""
    try:
        output_dir = Path('output/qr_codes')
        output_dir.mkdir(parents=True, exist_ok=True)
        
        if combined:
            cleaned_data = clean_json(data)
            if cleaned_data:
                qr = qrcode.QRCode(
                    version=None,
                    error_correction=qrcode.constants.ERROR_CORRECT_L,
                    box_size=10,
                    border=4,
                )
                json_str = json.dumps(cleaned_data, ensure_ascii=False)
                qr.add_data(json_str)
                qr.make(fit=True)
                
                img = qr.make_image(fill_color="black", back_color="white")
                output_path = output_dir / f'combined_qr_{int(time.time())}.png'
                img.save(str(output_path))
                return [str(output_path)]
        else:
            if isinstance(data, list):
                paths = []
                for idx, item in enumerate(data):
                    cleaned_item = clean_json(item)
                    if cleaned_item:
                        qr = qrcode.QRCode(
                            version=None,
                            error_correction=qrcode.constants.ERROR_CORRECT_L,
                            box_size=10,
                            border=4,
                        )
                        json_str = json.dumps(cleaned_item, ensure_ascii=False)
                        qr.add_data(json_str)
                        qr.make(fit=True)
                        
                        img = qr.make_image(fill_color="black", back_color="white")
                        output_path = output_dir / f'item_{idx}_qr_{int(time.time())}.png'
                        img.save(str(output_path))
                        paths.append(str(output_path))
                return paths
            else:
                cleaned_item = clean_json(data)
                if cleaned_item:
                    qr = qrcode.QRCode(
                        version=None,
                        error_correction=qrcode.constants.ERROR_CORRECT_L,
                        box_size=10,
                        border=4,
                    )
                    json_str = json.dumps(cleaned_item, ensure_ascii=False)
                    qr.add_data(json_str)
                    qr.make(fit=True)
                    
                    img = qr.make_image(fill_color="black", back_color="white")
                    output_path = output_dir / f'single_qr_{int(time.time())}.png'
                    img.save(str(output_path))
                    return [str(output_path)]
        
            return []
    except Exception as e:
        logger.error(f"QR generation error: {e}")
        return []

def decode_qr_code(image_path: str) -> str:
    """Decode QR code from an image file"""
    try:
        img = Image.open(image_path)
        decoded_objects = decode(qr_img)
        if decoded_objects:
            return decoded_objects[0].data.decode('utf-8')
        raise ValueError("Unable to decode QR code")
    except Exception as e:
        logger.error(f"QR decoding error: {e}")
        return None

def datachat_trained(data_input: str, query: str) -> str:
    """Handle trained data interaction logic"""
    data = clean_json(data_input)
    if not data:
        return "Invalid JSON data provided."
    return f"[Trained Mode]\nData: {json.dumps(data, indent=2)}\nQuery: {query}"

def datachat_simple(data_input: str, query: str) -> str:
    """Handle simple chat interaction logic"""
    data = clean_json(data_input)
    if not data:
        return "Invalid JSON data provided."
    return f"[Chat Mode]\nData: {json.dumps(data, indent=2)}\nQuestion: {query}"

def datachat_interface(mode: str, data_source: str, json_input: str, qr_image: str, query: str) -> str:
    """Interface for DataChat functionality"""
    data = None
    if data_source == "JSON Input":
        data = json_input
    elif data_source == "QR Code":
        try:
            decoded_data = decode_qr_code(qr_image)
            data = decoded_data
        except Exception as e:
            return f"Invalid QR code data provided: {e}"
    else:
        return "No valid data source selected."

    if mode == "Trained with Data":
        return datachat_trained(data, query)
    elif mode == "Chat about Data":
        return datachat_simple(data, query)
    else:
        return "Invalid mode selected."

def create_interface():
    """Create a comprehensive Gradio interface with advanced features"""
    css = """
    .container { max-width: 1200px; margin: auto; }
    .warning { background-color: #fff3cd; color: #856404; padding: 10px; border-radius: 4px; }
    .error { background-color: #f8d7da; color: #721c24; padding: 10px; border-radius: 4px; }
    .success { background-color: #d4edda; color: #155724; padding: 10px; border-radius: 4px; }
    """
    
    with gr.Blocks(css=css, title="Advanced Data Processor & QR Code Generator") as interface:
        gr.Markdown("# 🌐 Advanced Data Processing & QR Code Generator")
        
        with gr.Tab("DataChat"):
            mode = gr.Radio(["Trained with Data", "Chat about Data"], label="Mode")
            data_source = gr.Radio(["JSON Input", "QR Code"], label="Data Source")
            json_input = gr.Textbox(lines=8, label="JSON Data")
            qr_image = gr.Image(label="QR Code Image", type="filepath")
            query = gr.Textbox(label="Query")

            submit_btn = gr.Button("Submit")
            output = gr.Textbox(label="Response")

            submit_btn.click(datachat_interface, [mode, data_source, json_input, qr_image, query], output)

        with gr.Tab("QR Generator"):
            qr_input = gr.Textbox(lines=8, label="Input JSON for QR")
            generate_btn = gr.Button("Generate QR")
            qr_output = gr.Image(label="Generated QR Code")

            def generate_qr(json_data):
                data = clean_json(json_data)
                if data:
                    return generate_qr_code(data)
                return None

            generate_btn.click(generate_qr, qr_input, qr_output)

    return interface

def main():
    mimetypes.init()
    Path('output/qr_codes').mkdir(parents=True, exist_ok=True)
    interface = create_interface()
    interface.launch(
        server_name="0.0.0.0",
        server_port=8000,
        show_error=True,
        share=False,
        inbrowser=True,
        debug=True
    )

if __name__ == "__main__":
    main()