Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import requests | |
| import re | |
| import logging | |
| import json | |
| from selenium import webdriver | |
| from selenium.webdriver.chrome.options import Options | |
| from PIL import Image | |
| import io | |
| import zipfile | |
| import os | |
| import datetime | |
| from urllib.parse import urlparse | |
| from bs4 import BeautifulSoup | |
| import tempfile | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s') | |
| def sanitize_filename(filename): | |
| return re.sub(r'[<>:"/\\|?*\n]+', '_', filename) | |
| def validate_url(url): | |
| """Validate if the URL is properly formatted.""" | |
| try: | |
| result = urlparse(url) | |
| return all([result.scheme, result.netloc]) | |
| except: | |
| return False | |
| def get_latest_data(url): | |
| """Get the latest HTML content of a webpage.""" | |
| try: | |
| response = requests.get(url, timeout=10) | |
| return response.text | |
| except Exception as e: | |
| logging.error(f"Error fetching latest data from {url}: {str(e)}") | |
| return None | |
| def compare_html(old_html, new_html): | |
| """Compare two HTML contents to detect changes.""" | |
| if not old_html or not new_html: | |
| return False | |
| return old_html.strip() != new_html.strip() | |
| def compare_screenshot(old_screenshot, new_screenshot): | |
| """Compare two screenshots to detect changes.""" | |
| try: | |
| if not old_screenshot or not new_screenshot: | |
| return False | |
| old_img = Image.open(io.BytesIO(old_screenshot)) | |
| new_img = Image.open(io.BytesIO(new_screenshot)) | |
| return not (old_img == new_img) | |
| except Exception as e: | |
| logging.error(f"Error comparing screenshots: {str(e)}") | |
| return False | |
| def alert_changes(url, change_type): | |
| """Log detected changes.""" | |
| timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| logging.warning(f"[{timestamp}] Changes detected at {url}: {change_type}") | |
| return f"[{timestamp}] {change_type}" | |
| def extract_links_from_page(url): | |
| """Extract all links from a webpage.""" | |
| try: | |
| response = requests.get(url, timeout=10) | |
| soup = BeautifulSoup(response.text, 'html.parser') | |
| links = [a['href'] for a in soup.find_all('a', href=True)] | |
| return links | |
| except Exception as e: | |
| logging.error(f"Error extracting links from {url}: {str(e)}") | |
| return [] | |
| def take_screenshot(url): | |
| """Take a screenshot of a webpage.""" | |
| try: | |
| chrome_options = Options() | |
| chrome_options.add_argument("--headless") | |
| chrome_options.add_argument("--no-sandbox") | |
| chrome_options.add_argument("--disable-dev-shm-usage") | |
| chrome_options.add_argument("--window-size=1920,1080") | |
| driver = webdriver.Chrome(options=chrome_options) | |
| driver.get(url) | |
| screenshot = driver.get_screenshot_as_png() | |
| driver.quit() | |
| image = Image.open(io.BytesIO(screenshot)) | |
| max_size = (1024, 1024) | |
| image.thumbnail(max_size, Image.LANCZOS) | |
| img_byte_arr = io.BytesIO() | |
| image.save(img_byte_arr, format='PNG') | |
| return img_byte_arr.getvalue() | |
| except Exception as e: | |
| logging.error(f"Screenshot error for {url}: {str(e)}") | |
| return None | |
| def is_webpage(url): | |
| """Check if the URL points to a webpage (HTML).""" | |
| try: | |
| response = requests.head(url, timeout=10) | |
| content_type = response.headers.get('Content-Type', '').lower() | |
| return 'text/html' in content_type | |
| except Exception as e: | |
| logging.error(f"Error checking content type for {url}: {str(e)}") | |
| return False | |
| def crawl_url(url, depth, max_depth, visited=None): | |
| """Recursively crawl a URL up to a specified depth.""" | |
| if visited is None: | |
| visited = set() | |
| if depth > max_depth or url in visited: | |
| return [] | |
| visited.add(url) | |
| screenshots = [] | |
| if is_webpage(url): | |
| links = extract_links_from_page(url) | |
| screenshot = take_screenshot(url) | |
| if screenshot: | |
| screenshots.append((url, screenshot)) | |
| if depth < max_depth: | |
| for link in links: | |
| if not link.startswith(('http://', 'https://')): | |
| link = f"https://{link}" | |
| screenshots.extend(crawl_url(link, depth + 1, max_depth, visited)) | |
| else: | |
| logging.info(f"Skipping non-webpage content: {url}") | |
| return screenshots | |
| def process_urls(url_input, bulk_toggle, action_radio, max_urls, crawl_depth, progress=gr.Progress()): | |
| """Process URLs with crawl depth and change detection.""" | |
| # Validate URLs first | |
| urls = re.split(r'[,\n]+', url_input.strip()) if bulk_toggle else [url_input] | |
| urls = [url.strip() for url in urls if url.strip()] | |
| urls = urls[:int(max_urls)] | |
| # Validate all URLs | |
| invalid_urls = [url for url in urls if not validate_url(url)] | |
| if invalid_urls: | |
| return None, json.dumps({"error": f"Invalid URLs detected: {', '.join(invalid_urls)}"}, indent=2) | |
| scraped_data = [] | |
| screenshots = [] | |
| changes_log = [] | |
| # Initialize progress tracking | |
| total_urls = len(urls) | |
| progress(0) | |
| # Directory to store scraped data | |
| data_dir = 'scraped_data' | |
| os.makedirs(data_dir, exist_ok=True) | |
| # Process each URL | |
| for idx, url in enumerate(urls): | |
| if not url.startswith(('http://', 'https://')): | |
| url = f'https://{url}' | |
| # Sanitize URL for file naming | |
| sanitized_url = sanitize_filename(url) | |
| # Check for changes | |
| old_html_path = os.path.join(data_dir, f"{sanitized_url}_html.txt") | |
| old_screenshot_path = os.path.join(data_dir, f"{sanitized_url}_screenshot.png") | |
| # Fetch latest data | |
| latest_html = get_latest_data(url) | |
| latest_screenshot = take_screenshot(url) | |
| # Compare with previous data if available | |
| if os.path.exists(old_html_path): | |
| with open(old_html_path, 'r', encoding='utf-8') as f: | |
| old_html = f.read() | |
| if compare_html(old_html, latest_html): | |
| changes_log.append(alert_changes(url, "HTML content has changed")) | |
| if os.path.exists(old_screenshot_path): | |
| with open(old_screenshot_path, 'rb') as f: | |
| old_screenshot = f.read() | |
| if latest_screenshot and compare_screenshot(old_screenshot, latest_screenshot): | |
| changes_log.append(alert_changes(url, "Visual content has changed")) | |
| # Store latest data | |
| if latest_html: | |
| with open(old_html_path, 'w', encoding='utf-8') as f: | |
| f.write(latest_html) | |
| if latest_screenshot: | |
| with open(old_screenshot_path, 'wb') as f: | |
| f.write(latest_screenshot) | |
| # Prepare output data | |
| if action_radio in ['Scrape data', 'Both']: | |
| scraped_data.append({ | |
| 'url': url, | |
| 'content': latest_html, | |
| 'timestamp': datetime.datetime.now().isoformat(), | |
| 'changes_detected': changes_log | |
| }) | |
| if action_radio in ['Capture image', 'Both']: | |
| crawled_screenshots = crawl_url(url, depth=1, max_depth=int(crawl_depth)) | |
| screenshots.extend(crawled_screenshots) | |
| # Update progress | |
| progress((idx + 1) / total_urls) | |
| # Create a temporary file to store the ZIP | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as tmp_file: | |
| with zipfile.ZipFile(tmp_file, 'w', zipfile.ZIP_DEFLATED) as zipf: | |
| # Add screenshots to ZIP | |
| for screenshot_url, screenshot_data in screenshots: | |
| sanitized_screenshot_url = sanitize_filename(screenshot_url) | |
| filename = f"{sanitized_screenshot_url}.png" | |
| zipf.writestr(filename, screenshot_data) | |
| # Add scraped data and changes log to ZIP | |
| if scraped_data: | |
| data_to_save = { | |
| 'scraped_data': scraped_data, | |
| 'changes_log': changes_log, | |
| 'timestamp': datetime.datetime.now().isoformat() | |
| } | |
| zipf.writestr('data.json', json.dumps(data_to_save, indent=2)) | |
| # Get the path to the temporary file | |
| zip_file_path = tmp_file.name | |
| # Prepare display data | |
| display_data = { | |
| 'scraped_urls': len(scraped_data), | |
| 'screenshots_taken': len(screenshots), | |
| 'changes_detected': changes_log | |
| } | |
| # Return the path to the temporary ZIP file and display data | |
| return zip_file_path, json.dumps(display_data, indent=2) | |
| def create_interface(): | |
| """Create the Gradio interface.""" | |
| with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| gr.Markdown( | |
| """ | |
| # Smart Web Scraper with Change Detection | |
| Monitor and capture changes in web content automatically. | |
| """ | |
| ) | |
| with gr.Tabs(): | |
| with gr.Tab("URL Scrape/Screenshot"): | |
| url_input = gr.Textbox( | |
| label="Enter URL(s)", | |
| value="https://example.com", | |
| placeholder="Enter single URL or multiple URLs separated by commas" | |
| ) | |
| with gr.Row(): | |
| bulk_toggle = gr.Checkbox(label="Bulk URLs", value=False) | |
| action_radio = gr.Radio( | |
| ["Scrape data", "Capture image", "Both"], | |
| label="Select Action", | |
| value="Both" | |
| ) | |
| with gr.Row(): | |
| max_urls = gr.Slider( | |
| minimum=1, | |
| maximum=20, | |
| value=5, | |
| step=1, | |
| label="Max URLs to process" | |
| ) | |
| crawl_depth = gr.Slider( | |
| minimum=1, | |
| maximum=3, | |
| value=1, | |
| step=1, | |
| label="Crawl Depth" | |
| ) | |
| process_button = gr.Button("Process URLs", variant="primary") | |
| with gr.Column(): | |
| screenshot_zip = gr.File(label="Download Results") # Removed file_name | |
| scraped_data_output = gr.JSON(label="Results Summary") | |
| process_button.click( | |
| fn=process_urls, | |
| inputs=[ | |
| url_input, | |
| bulk_toggle, | |
| action_radio, | |
| max_urls, | |
| crawl_depth | |
| ], | |
| outputs=[ | |
| screenshot_zip, | |
| scraped_data_output | |
| ], | |
| show_progress=True | |
| ) | |
| gr.Markdown( | |
| """ | |
| ### Features | |
| - Bulk URL processing | |
| - Screenshot capture | |
| - Content change detection | |
| - Recursive crawling | |
| - Automatic data storage | |
| """ | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| demo = create_interface() | |
| demo.launch() |