Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import pandas as pd | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from playwright.async_api import async_playwright | |
| import asyncio | |
| import re | |
| import time | |
| import random | |
| from urllib.parse import urljoin, urlparse | |
| from fake_useragent import UserAgent | |
| import spacy | |
| import json | |
| import os | |
| from typing import List, Dict, Tuple, Optional | |
| import logging | |
| # Setup logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Initialize components | |
| ua = UserAgent() | |
| # Load spaCy model for German NER | |
| try: | |
| nlp = spacy.load("de_core_news_sm") | |
| except OSError: | |
| logger.error("German spaCy model not found. Install with: python -m spacy download de_core_news_sm") | |
| nlp = None | |
| class ImprintScraper: | |
| def __init__(self): | |
| self.session = requests.Session() | |
| self.session.headers.update({ | |
| 'User-Agent': ua.random, | |
| 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', | |
| 'Accept-Language': 'de-DE,de;q=0.9,en;q=0.8', | |
| 'Accept-Encoding': 'gzip, deflate, br', | |
| 'DNT': '1', | |
| 'Connection': 'keep-alive', | |
| 'Upgrade-Insecure-Requests': '1' | |
| }) | |
| # Regex patterns | |
| self.email_pattern = re.compile(r'[\w\.-]+@[\w\.-]+\.\w+') | |
| self.phone_pattern = re.compile(r'(?:\+?\d{1,3})?[\s\-]?(?:\(?\d+\)?[\s\-]?)+') | |
| # Keywords for imprint pages | |
| self.imprint_keywords = [ | |
| 'impressum', 'imprint', 'kontakt', 'contact', 'about', | |
| 'über uns', 'rechtliches', 'legal', 'datenschutz' | |
| ] | |
| # Owner keywords (German focus) | |
| self.owner_keywords = [ | |
| 'inhaber', 'geschäftsführer', 'geschäftsführerin', 'vertretungsberechtigt', | |
| 'owner', 'ceo', 'managing director', 'eigentümer', 'betreiber', | |
| 'verantwortlich', 'ansprechpartner' | |
| ] | |
| def clean_phone(self, phone: str) -> str: | |
| """Clean and format phone number""" | |
| if not phone: | |
| return "" | |
| # Remove extra whitespace and common separators | |
| phone = re.sub(r'[\s\-\(\)]+', '', phone.strip()) | |
| # Remove common prefixes like "Tel:" or "Telefon:" | |
| phone = re.sub(r'^(tel|telefon|phone|fon)[:.]?', '', phone.lower()) | |
| # Ensure it looks like a valid phone number | |
| if len(phone) < 6 or len(phone) > 20: | |
| return "" | |
| return phone | |
| def extract_emails(self, text: str) -> List[str]: | |
| """Extract email addresses from text""" | |
| emails = self.email_pattern.findall(text.lower()) | |
| # Filter out common false positives | |
| filtered_emails = [] | |
| for email in emails: | |
| if not any(exclude in email for exclude in ['example.', 'test.', 'placeholder']): | |
| filtered_emails.append(email) | |
| return list(set(filtered_emails)) | |
| def extract_phones(self, text: str) -> List[str]: | |
| """Extract phone numbers from text""" | |
| # Find potential phone numbers | |
| potential_phones = self.phone_pattern.findall(text) | |
| cleaned_phones = [] | |
| for phone in potential_phones: | |
| cleaned = self.clean_phone(phone) | |
| if cleaned and len(cleaned) >= 6: | |
| cleaned_phones.append(cleaned) | |
| return list(set(cleaned_phones)) | |
| def extract_owner_with_spacy(self, text: str) -> Optional[str]: | |
| """Extract owner name using spaCy NER""" | |
| if not nlp: | |
| return None | |
| doc = nlp(text) | |
| # Look for person entities near owner keywords | |
| text_lower = text.lower() | |
| best_owner = None | |
| min_distance = float('inf') | |
| for ent in doc.ents: | |
| if ent.label_ == "PERSON" and len(ent.text.strip()) > 1: | |
| # Find the closest owner keyword | |
| for keyword in self.owner_keywords: | |
| keyword_pos = text_lower.find(keyword) | |
| if keyword_pos != -1: | |
| distance = abs(ent.start_char - keyword_pos) | |
| if distance < min_distance and distance < 200: # Within 200 characters | |
| min_distance = distance | |
| best_owner = ent.text.strip() | |
| return best_owner | |
| def extract_owner_with_regex(self, text: str) -> Optional[str]: | |
| """Extract owner name using regex patterns""" | |
| text_lower = text.lower() | |
| for keyword in self.owner_keywords: | |
| # Look for pattern: "Keyword: Name" or "Keyword Name" | |
| patterns = [ | |
| rf'{keyword}[:.]?\s*([A-ZÄÖÜ][a-zäöüß]+(?:\s+[A-ZÄÖÜ][a-zäöüß]+)*)', | |
| rf'({keyword})\s*[:.]?\s*([A-ZÄÖÜ][a-zäöüß]+(?:\s+[A-ZÄÖÜ][a-zäöüß]+)*)', | |
| ] | |
| for pattern in patterns: | |
| matches = re.search(pattern, text, re.IGNORECASE | re.MULTILINE) | |
| if matches: | |
| # Return the captured name part | |
| name = matches.group(-1).strip() | |
| if len(name) > 1 and not any(skip in name.lower() for skip in ['gmbh', 'ag', 'kg', 'ltd', 'inc']): | |
| return name | |
| return None | |
| def find_imprint_links(self, soup: BeautifulSoup, base_url: str) -> List[str]: | |
| """Find links that might lead to imprint pages""" | |
| imprint_links = [] | |
| for link in soup.find_all('a', href=True): | |
| href = link.get('href', '').lower() | |
| text = link.get_text().lower().strip() | |
| # Check if link text or href contains imprint keywords | |
| for keyword in self.imprint_keywords: | |
| if keyword in text or keyword in href: | |
| full_url = urljoin(base_url, link['href']) | |
| imprint_links.append(full_url) | |
| break | |
| return list(set(imprint_links)) | |
| async def scrape_with_playwright(self, url: str) -> Tuple[str, str]: | |
| """Scrape page content using Playwright for JavaScript rendering""" | |
| try: | |
| async with async_playwright() as p: | |
| browser = await p.chromium.launch(headless=True) | |
| context = await browser.new_context( | |
| user_agent=ua.random, | |
| viewport={'width': 1920, 'height': 1080} | |
| ) | |
| page = await context.new_page() | |
| # Set timeout and navigate | |
| await page.goto(url, timeout=30000, wait_until='domcontentloaded') | |
| await page.wait_for_timeout(2000) # Wait for JS to execute | |
| content = await page.content() | |
| title = await page.title() | |
| await browser.close() | |
| return content, title | |
| except Exception as e: | |
| logger.error(f"Playwright scraping failed for {url}: {str(e)}") | |
| return "", "" | |
| def scrape_with_requests(self, url: str) -> Tuple[str, str]: | |
| """Scrape page content using requests (static)""" | |
| try: | |
| # Rotate user agent | |
| self.session.headers['User-Agent'] = ua.random | |
| response = self.session.get(url, timeout=10, allow_redirects=True) | |
| response.raise_for_status() | |
| soup = BeautifulSoup(response.content, 'html.parser') | |
| # Remove script and style elements | |
| for script in soup(["script", "style"]): | |
| script.decompose() | |
| text = soup.get_text() | |
| title = soup.title.string if soup.title else "" | |
| return text, title | |
| except Exception as e: | |
| logger.error(f"Requests scraping failed for {url}: {str(e)}") | |
| return "", "" | |
| async def extract_contact_info(self, url: str) -> Dict[str, str]: | |
| """Main extraction function for a single URL""" | |
| result = { | |
| 'website': url, | |
| 'imprint_url': url, | |
| 'email': '', | |
| 'phone': '', | |
| 'owner': '', | |
| 'status': 'Processing...' | |
| } | |
| try: | |
| # Step 1: Try with requests first (faster) | |
| logger.info(f"Scraping {url} with requests...") | |
| content, title = self.scrape_with_requests(url) | |
| if not content: | |
| # Step 2: Fallback to Playwright | |
| logger.info(f"Falling back to Playwright for {url}...") | |
| content, title = await self.scrape_with_playwright(url) | |
| if not content: | |
| result['status'] = 'Failed to load page' | |
| return result | |
| # Step 3: Try to find imprint page if this is homepage | |
| soup = BeautifulSoup(content, 'html.parser') | |
| imprint_links = self.find_imprint_links(soup, url) | |
| best_content = content | |
| best_url = url | |
| # Check imprint pages for better information | |
| for imprint_url in imprint_links[:3]: # Check max 3 imprint links | |
| try: | |
| imp_content, imp_title = self.scrape_with_requests(imprint_url) | |
| if not imp_content: | |
| imp_content, imp_title = await self.scrape_with_playwright(imprint_url) | |
| if imp_content and len(imp_content) > len(best_content) * 0.1: | |
| # Use imprint page if it has substantial content | |
| best_content = imp_content | |
| best_url = imprint_url | |
| result['imprint_url'] = imprint_url | |
| break | |
| time.sleep(random.uniform(1, 3)) # Random delay | |
| except Exception as e: | |
| logger.warning(f"Failed to check imprint link {imprint_url}: {str(e)}") | |
| continue | |
| # Step 4: Extract information | |
| emails = self.extract_emails(best_content) | |
| phones = self.extract_phones(best_content) | |
| # Owner extraction with multiple methods | |
| owner = self.extract_owner_with_spacy(best_content) | |
| if not owner: | |
| owner = self.extract_owner_with_regex(best_content) | |
| # Set results | |
| result['email'] = emails[0] if emails else '' | |
| result['phone'] = phones[0] if phones else '' | |
| result['owner'] = owner if owner else '' | |
| result['status'] = 'Success' | |
| if not any([result['email'], result['phone'], result['owner']]): | |
| result['status'] = 'No contact info found' | |
| except Exception as e: | |
| logger.error(f"Error processing {url}: {str(e)}") | |
| result['status'] = f'Error: {str(e)}' | |
| return result | |
| async def process_urls(urls: List[str], progress_callback=None) -> pd.DataFrame: | |
| """Process multiple URLs and return results as DataFrame""" | |
| scraper = ImprintScraper() | |
| results = [] | |
| for i, url in enumerate(urls): | |
| if not url.strip(): | |
| continue | |
| # Ensure URL has protocol | |
| if not url.startswith(('http://', 'https://')): | |
| url = 'https://' + url.strip() | |
| if progress_callback: | |
| progress_callback(f"Processing {i+1}/{len(urls)}: {url}") | |
| result = await scraper.extract_contact_info(url) | |
| results.append(result) | |
| # Random delay between requests | |
| if i < len(urls) - 1: | |
| await asyncio.sleep(random.uniform(2, 5)) | |
| return pd.DataFrame(results) | |
| def parse_input_urls(text_input: str, file_input) -> List[str]: | |
| """Parse URLs from text input or uploaded file""" | |
| urls = [] | |
| # Parse text input | |
| if text_input.strip(): | |
| urls.extend([url.strip() for url in text_input.strip().split('\n') if url.strip()]) | |
| # Parse file input | |
| if file_input: | |
| try: | |
| if file_input.name.endswith('.csv'): | |
| df = pd.read_csv(file_input.name) | |
| # Try to find URL column | |
| for col in df.columns: | |
| if 'url' in col.lower() or 'website' in col.lower() or 'domain' in col.lower(): | |
| urls.extend(df[col].dropna().astype(str).tolist()) | |
| break | |
| else: | |
| # If no URL column found, use first column | |
| urls.extend(df.iloc[:, 0].dropna().astype(str).tolist()) | |
| else: | |
| # Assume text file with one URL per line | |
| with open(file_input.name, 'r', encoding='utf-8') as f: | |
| urls.extend([url.strip() for url in f.readlines() if url.strip()]) | |
| except Exception as e: | |
| logger.error(f"Error reading file: {str(e)}") | |
| return list(set(urls)) # Remove duplicates | |
| async def scrape_websites(text_input: str, file_input, progress=gr.Progress()) -> Tuple[pd.DataFrame, str]: | |
| """Main function called by Gradio interface""" | |
| # Parse input URLs | |
| urls = parse_input_urls(text_input, file_input) | |
| if not urls: | |
| return pd.DataFrame(), "No URLs provided. Please enter URLs in the text box or upload a file." | |
| progress(0, desc="Starting scraper...") | |
| def update_progress(msg): | |
| progress(0.5, desc=msg) | |
| # Process URLs | |
| try: | |
| results_df = await process_urls(urls, progress_callback=update_progress) | |
| # Save results to CSV | |
| output_file = "scraped_contacts.csv" | |
| results_df.to_csv(output_file, index=False, encoding='utf-8') | |
| progress(1.0, desc="Scraping complete!") | |
| success_count = len(results_df[results_df['status'] == 'Success']) | |
| total_count = len(results_df) | |
| summary = f"Processed {total_count} URLs. Successfully extracted data from {success_count} sites ({success_count/total_count*100:.1f}% success rate)." | |
| return results_df, summary | |
| except Exception as e: | |
| error_msg = f"Error during scraping: {str(e)}" | |
| logger.error(error_msg) | |
| return pd.DataFrame(), error_msg | |
| # Create Gradio interface | |
| def create_interface(): | |
| with gr.Blocks(title="🇩🇪 DACH Store Imprint Scraper") as app: | |
| gr.Markdown(""" | |
| # 🇩🇪 DACH Store Imprint Scraper | |
| Extract contact details from legally required imprint pages in Germany, Austria, and Switzerland. | |
| **Features:** | |
| - 📧 Email extraction | |
| - 📞 Phone number extraction | |
| - 👤 Owner name extraction using AI and NER | |
| - 🔄 Automatic imprint page detection | |
| - 🎭 Anti-blocking measures with user agent rotation | |
| - ⚡ JavaScript rendering fallback with Playwright | |
| """) | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("### Input URLs") | |
| text_input = gr.Textbox( | |
| label="URLs (one per line)", | |
| placeholder="https://example-shop.de\nhttps://another-store.at\nhttps://swiss-shop.ch", | |
| lines=10, | |
| max_lines=20 | |
| ) | |
| file_input = gr.File( | |
| label="Or upload CSV file with URLs", | |
| file_types=['.csv', '.txt'] | |
| ) | |
| scrape_btn = gr.Button("🚀 Start Scraping", variant="primary", size="lg") | |
| with gr.Row(): | |
| status_text = gr.Textbox(label="Status", interactive=False) | |
| with gr.Row(): | |
| results_df = gr.Dataframe( | |
| label="Extraction Results", | |
| headers=["Website", "Imprint URL", "Email", "Phone", "Owner", "Status"], | |
| interactive=False | |
| ) | |
| with gr.Row(): | |
| download_btn = gr.DownloadButton( | |
| "📥 Download Results CSV", | |
| value="scraped_contacts.csv", | |
| visible=False | |
| ) | |
| # Event handlers | |
| scrape_btn.click( | |
| fn=scrape_websites, | |
| inputs=[text_input, file_input], | |
| outputs=[results_df, status_text] | |
| ).then( | |
| fn=lambda df: gr.DownloadButton(visible=len(df) > 0), | |
| inputs=[results_df], | |
| outputs=[download_btn] | |
| ) | |
| return app | |
| if __name__ == "__main__": | |
| # Create and launch the application | |
| app = create_interface() | |
| app.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=True | |
| ) |