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 )