Spaces:
Running
Running
| import json | |
| import os | |
| import torch | |
| import string | |
| import requests | |
| from bs4 import BeautifulSoup | |
| import tempfile | |
| import zipfile | |
| import mimetypes | |
| from tqdm import tqdm | |
| import logging | |
| import gradio as gr | |
| from typing import List, Dict, Union, Optional | |
| from urllib.parse import urlparse | |
| import concurrent.futures | |
| import validators | |
| from pathlib import Path | |
| import re | |
| # Setup logging with more 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') | |
| ] | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class URLProcessor: | |
| """Class to handle URL processing with advanced features""" | |
| def __init__(self, timeout: int = 10, max_retries: int = 3, concurrent_requests: int = 5): | |
| self.timeout = timeout | |
| self.max_retries = max_retries | |
| self.concurrent_requests = concurrent_requests | |
| self.session = requests.Session() | |
| # Add common headers to mimic browser behavior | |
| self.session.headers.update({ | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' | |
| }) | |
| def validate_url(self, url: str) -> bool: | |
| """Validate URL format and accessibility""" | |
| try: | |
| result = urlparse(url) | |
| return all([result.scheme, result.netloc]) and validators.url(url) | |
| except Exception as e: | |
| logger.warning(f"Invalid URL format: {url} - {str(e)}") | |
| return False | |
| def fetch_content(self, url: str) -> Optional[str]: | |
| """Fetch content from URL with retry mechanism""" | |
| for attempt in range(self.max_retries): | |
| try: | |
| response = self.session.get(url, timeout=self.timeout) | |
| response.raise_for_status() | |
| return response.text | |
| except requests.RequestException as e: | |
| logger.error(f"Attempt {attempt + 1}/{self.max_retries} failed for {url}: {str(e)}") | |
| if attempt == self.max_retries - 1: | |
| return None | |
| time.sleep(1) # Delay between retries | |
| def process_urls(self, urls: List[str]) -> List[Dict]: | |
| """Process multiple URLs concurrently""" | |
| valid_urls = [url for url in urls if self.validate_url(url)] | |
| if not valid_urls: | |
| logger.warning("No valid URLs to process") | |
| return [] | |
| results = [] | |
| with concurrent.futures.ThreadPoolExecutor(max_workers=self.concurrent_requests) as executor: | |
| future_to_url = {executor.submit(self.fetch_content, url): url for url in valid_urls} | |
| for future in concurrent.futures.as_completed(future_to_url): | |
| url = future_to_url[future] | |
| try: | |
| html = future.result() | |
| if html: | |
| text = extract_text(html) | |
| if text: | |
| results.append({ | |
| "source": "url", | |
| "url": url, | |
| "content": text, | |
| "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") | |
| }) | |
| else: | |
| logger.warning(f"No text content extracted from {url}") | |
| except Exception as e: | |
| logger.error(f"Error processing {url}: {str(e)}") | |
| return results | |
| def extract_text(html: str) -> str: | |
| """Enhanced text extraction with better cleaning""" | |
| if not html: | |
| return "" | |
| soup = BeautifulSoup(html, 'html.parser') | |
| # Remove unwanted elements | |
| for element in soup(['script', 'style', 'header', 'footer', 'nav']): | |
| element.decompose() | |
| # Extract text with better formatting | |
| text = soup.get_text(separator=' ') | |
| # Clean up the text | |
| lines = (line.strip() for line in text.splitlines()) | |
| chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) | |
| text = ' '.join(chunk for chunk in chunks if chunk) | |
| # Remove excessive whitespace | |
| text = re.sub(r'\s+', ' ', text) | |
| return text.strip() | |
| class FileProcessor: | |
| """Class to handle file processing""" | |
| 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: | |
| 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', 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_file(self, file) -> List[Dict]: | |
| """Process single file""" | |
| results = [] | |
| try: | |
| 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 single file: {str(e)}") | |
| return results | |
| def preprocess_bulk_text(text: str) -> str: | |
| """Enhanced text preprocessing""" | |
| if not text: | |
| return "" | |
| # Normalize line endings | |
| text = text.replace('\r\n', '\n').replace('\r', '\n') | |
| # Define separators | |
| separators = ['\n', ' / ', '/', ';', ' - ', '|', ' '] | |
| # Replace separators with commas if not already comma-separated | |
| if ',' not in text: | |
| for separator in separators: | |
| text = text.replace(separator, ',') | |
| # Handle domain endings | |
| domain_pattern = r'(\.[a-z]{2,})\s+' | |
| text = re.sub(domain_pattern, r'\1,', text) | |
| # Clean up multiple commas and whitespace | |
| text = re.sub(r',+', ',', text) | |
| text = text.strip(',' + string.whitespace) | |
| text = re.sub(r'\s*,\s*', ', ', text) | |
| return text | |
| def create_interface(): | |
| """Create enhanced Gradio interface""" | |
| # Custom CSS for better styling | |
| custom_css = """ | |
| .container { max-width: 1200px; margin: auto; padding: 20px; } | |
| .output-panel { margin-top: 20px; } | |
| .warning { color: #856404; background-color: #fff3cd; padding: 10px; border-radius: 4px; } | |
| .error { color: #721c24; background-color: #f8d7da; padding: 10px; border-radius: 4px; } | |
| """ | |
| with gr.Blocks(css=custom_css) as interface: | |
| gr.Markdown("# Advanced URL and Text Processing Tool") | |
| with gr.Tab("URL Input"): | |
| url_input = gr.Textbox( | |
| label="Enter URLs (comma-separated or one per line)", | |
| placeholder="https://example1.com, https://example2.com", | |
| lines=5 | |
| ) | |
| 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="Enter text directly", | |
| placeholder="Enter your text here...", | |
| lines=5 | |
| ) | |
| # Process button with loading state | |
| process_btn = gr.Button("Process Input", variant="primary") | |
| # Output components | |
| with gr.Row(): | |
| output_file = gr.File(label="Processed Dataset") | |
| output_text = gr.Textbox( | |
| label="Processing Results", | |
| lines=3, | |
| interactive=False | |
| ) | |
| def process_all_inputs(urls, file, text): | |
| """Process all input types with progress tracking""" | |
| try: | |
| dataset = [] | |
| # Process URLs | |
| if urls: | |
| url_processor = URLProcessor() | |
| url_list = [u.strip() for u in urls.split(',') if u.strip()] | |
| dataset.extend(url_processor.process_urls(url_list)) | |
| # Process files | |
| if file: | |
| file_processor = FileProcessor() | |
| dataset.extend(file_processor.process_file(file)) | |
| # Process text input | |
| if text: | |
| processed_text = preprocess_bulk_text(text) | |
| if processed_text: | |
| dataset.append({ | |
| "source": "input", | |
| "content": processed_text, | |
| "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") | |
| }) | |
| if not dataset: | |
| return [None, "No valid data to process. Please check your inputs."] | |
| # Save results | |
| output_file = 'processed_dataset.json' | |
| with open(output_file, 'w', encoding='utf-8') as f: | |
| json.dump(dataset, f, indent=2, ensure_ascii=False) | |
| # Generate summary | |
| summary = f""" | |
| Processing completed successfully! | |
| - URLs processed: {sum(1 for d in dataset if d['source'] == 'url')} | |
| - Files processed: {sum(1 for d in dataset if d['source'] == 'file')} | |
| - Text inputs processed: {sum(1 for d in dataset if d['source'] == 'input')} | |
| """ | |
| return [output_file, summary] | |
| except Exception as e: | |
| error_msg = f"Error during processing: {str(e)}" | |
| logger.error(error_msg) | |
| return [None, error_msg] | |
| # Connect the interface | |
| process_btn.click( | |
| fn=process_all_inputs, | |
| inputs=[url_input, file_input, text_input], | |
| outputs=[output_file, output_text] | |
| ) | |
| # Add comprehensive instructions | |
| gr.Markdown(""" | |
| ## Instructions | |
| 1. **URL Input**: | |
| - Enter URLs separated by commas or new lines | |
| - URLs must start with http:// or https:// | |
| - Invalid URLs will be skipped | |
| 2. **File Input**: | |
| - Upload text files or ZIP archives | |
| - Supported formats: .txt, .zip, .md, .csv, .json, .xml | |
| - Maximum file size: 10MB | |
| 3. **Text Input**: | |
| - Directly enter or paste text | |
| - Text will be automatically formatted | |
| 4. Click 'Process Input' to generate the dataset | |
| The tool will combine all valid inputs into a single JSON dataset file. | |
| """) | |
| return interface | |
| if __name__ == "__main__": | |
| # Initialize mimetypes | |
| mimetypes.init() | |
| # Create and launch the interface | |
| interface = create_interface() | |
| interface.launch( | |
| share=True, | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| debug=True | |
| ) |