app.py

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
    )