Data Cleaning Pipelines for Web Scraped Data

import re
from html import unescape
from bs4 import BeautifulSoup
import unicodedata

def clean_text(text):
    """Clean text extracted from HTML."""
    if not text:
        return None

    # Remove HTML tags
    text = BeautifulSoup(text, 'html.parser').get_text()

    # Decode HTML entities
    text = unescape(text)

    # Normalize unicode (decompose then compose)
    text = unicodedata.normalize('NFKC', text)

    # Remove control characters except newlines and tabs
    text = ''.join(
        char for char in text
        if unicodedata.category(char) != 'Cc' or char in '\n\t'
    )

    # Normalize whitespace
    text = re.sub(r'[\t ]+', ' ', text)  # Tabs and spaces to single space
    text = re.sub(r'\n\s*\n', '\n\n', text)  # Multiple newlines to double
    text = text.strip()

    return text if text else None

# Example
raw = "<p>  Product   Name™  &nbsp;  </p>"
clean = clean_text(raw)  # "Product Name™"

import ftfy

def fix_encoding(text):
    """Fix common encoding problems like mojibake."""
    if not text:
        return None

    # ftfy fixes most encoding issues automatically
    fixed = ftfy.fix_text(text)

    # Additional fixes for common patterns
    replacements = {
        '’': "'",
        'â€"': '—',
        '•': '•',
        'é': 'é',
        'è': 'è',
    }
    for bad, good in replacements.items():
        fixed = fixed.replace(bad, good)

    return fixed

# Example
broken = "Cafã© Latté"
fixed = fix_encoding(broken)  # "Café Latté"

from price_parser import Price
from decimal import Decimal, InvalidOperation

def parse_price(price_str, default_currency='USD'):
    """Extract numeric price and currency from string."""
    if not price_str:
        return None, None

    # Use price-parser library for robust parsing
    price = Price.fromstring(price_str)

    if price.amount is None:
        return None, None

    return float(price.amount), price.currency or default_currency

# Alternative manual implementation
def parse_price_manual(price_str):
    """Manual price parsing with common formats."""
    import re

    if not price_str:
        return None

    # Remove currency symbols and whitespace
    cleaned = re.sub(r'[^\d.,\-]', '', price_str)

    # Handle European format (1.234,56) vs US format (1,234.56)
    if re.match(r'^\d{1,3}(\.\d{3})*(,\d{2})?$', cleaned):
        # European: 1.234,56 -> 1234.56
        cleaned = cleaned.replace('.', '').replace(',', '.')
    else:
        # US: 1,234.56 -> 1234.56
        cleaned = cleaned.replace(',', '')

    try:
        return float(cleaned)
    except ValueError:
        return None

# Examples
print(parse_price("$19.99"))        # (19.99, 'USD')
print(parse_price("€ 1.234,56"))    # (1234.56, 'EUR')
print(parse_price("19,99 €"))       # (19.99, 'EUR')

from dateutil import parser as date_parser
from datetime import datetime, timedelta
import re

def parse_date(date_str, reference_date=None):
    """Parse various date formats into datetime object."""
    if not date_str:
        return None

    date_str = date_str.strip().lower()
    reference = reference_date or datetime.now()

    # Handle relative dates
    relative_patterns = {
        r'^today$': timedelta(days=0),
        r'^yesterday$': timedelta(days=-1),
        r'^(\d+)\s*days?\s*ago$': lambda m: timedelta(days=-int(m.group(1))),
        r'^(\d+)\s*hours?\s*ago$': lambda m: timedelta(hours=-int(m.group(1))),
        r'^last\s*week$': timedelta(weeks=-1),
        r'^last\s*month$': timedelta(days=-30),
    }

    for pattern, delta in relative_patterns.items():
        match = re.match(pattern, date_str)
        if match:
            if callable(delta):
                delta = delta(match)
            return reference + delta

    # Try standard parsing
    try:
        # dayfirst=True for European dates, fuzzy for partial matches
        return date_parser.parse(date_str, dayfirst=True, fuzzy=True)
    except (ValueError, OverflowError):
        return None

# Examples
print(parse_date("January 15, 2026"))  # datetime(2026, 1, 15)
print(parse_date("15/01/2026"))        # datetime(2026, 1, 15)
print(parse_date("3 days ago"))        # datetime (3 days before now)
print(parse_date("Posted yesterday"))  # datetime (yesterday)

from pydantic import BaseModel, Field, validator, HttpUrl
from typing import Optional
from datetime import datetime
from decimal import Decimal

class Product(BaseModel):
    """Validated product schema."""

    name: str = Field(..., min_length=2, max_length=500)
    price: Decimal = Field(..., gt=0, lt=1000000)
    currency: str = Field(default='USD', regex=r'^[A-Z]{3}$')
    url: HttpUrl
    sku: Optional[str] = Field(None, regex=r'^[A-Z0-9-]+$')
    scraped_at: datetime = Field(default_factory=datetime.utcnow)
    in_stock: bool = True

    @validator('name')
    def clean_name(cls, v):
        # Remove excessive whitespace
        return ' '.join(v.split())

    @validator('price', pre=True)
    def parse_price(cls, v):
        if isinstance(v, str):
            # Remove currency symbols
            import re
            cleaned = re.sub(r'[^\d.]', '', v)
            return Decimal(cleaned) if cleaned else None
        return v

    class Config:
        # Allow extra fields but don't include in output
        extra = 'ignore'

# Usage
try:
    product = Product(
        name="  Widget Pro  ",
        price="$29.99",
        url="https://example.com/widget",
        sku="WID-001"
    )
    print(product.dict())
except ValidationError as e:
    print(f"Validation failed: {e}")

class DataValidator:
    """Apply business rules to validate scraped data."""

    def __init__(self):
        self.errors = []
        self.warnings = []

    def validate_product(self, product):
        self.errors = []
        self.warnings = []

        # Price sanity checks
        if product.get('price'):
            price = product['price']
            if price < 0.01:
                self.errors.append(f"Price too low: {price}")
            elif price > 100000:
                self.warnings.append(f"Unusually high price: {price}")

            # Check against historical prices
            historical_avg = self.get_historical_average(product.get('sku'))
            if historical_avg and abs(price - historical_avg) / historical_avg > 0.5:
                self.warnings.append(
                    f"Price changed >50%: {historical_avg} -> {price}"
                )

        # Completeness checks
        required_fields = ['name', 'price', 'url']
        for field in required_fields:
            if not product.get(field):
                self.errors.append(f"Missing required field: {field}")

        # Consistency checks
        if product.get('in_stock') and not product.get('price'):
            self.warnings.append("In stock but no price listed")

        return len(self.errors) == 0

    def get_historical_average(self, sku):
        # Look up in your database
        return None  # Placeholder

import hashlib

def generate_content_hash(item, fields=None):
    """Generate hash from item content for exact deduplication."""
    if fields:
        content = {k: item.get(k) for k in fields}
    else:
        content = item

    # Sort keys for consistent ordering
    content_str = str(sorted(content.items()))
    return hashlib.sha256(content_str.encode()).hexdigest()

class ExactDeduplicator:
    def __init__(self, key_fields=None):
        self.seen_hashes = set()
        self.key_fields = key_fields

    def is_duplicate(self, item):
        item_hash = generate_content_hash(item, self.key_fields)
        if item_hash in self.seen_hashes:
            return True
        self.seen_hashes.add(item_hash)
        return False

# Usage
dedup = ExactDeduplicator(key_fields=['sku', 'url'])
for item in items:
    if not dedup.is_duplicate(item):
        process(item)

from rapidfuzz import fuzz
from collections import defaultdict

class FuzzyDeduplicator:
    """Find near-duplicate items using fuzzy string matching."""

    def __init__(self, threshold=85, key_field='name'):
        self.threshold = threshold
        self.key_field = key_field
        self.items = []

    def find_duplicates(self, item):
        """Find existing items that might be duplicates."""
        item_key = item.get(self.key_field, '')
        duplicates = []

        for existing in self.items:
            existing_key = existing.get(self.key_field, '')
            similarity = fuzz.ratio(item_key.lower(), existing_key.lower())

            if similarity >= self.threshold:
                duplicates.append({
                    'item': existing,
                    'similarity': similarity
                })

        return duplicates

    def add_if_unique(self, item):
        """Add item only if no near-duplicates exist."""
        duplicates = self.find_duplicates(item)
        if not duplicates:
            self.items.append(item)
            return True
        return False

# Example
dedup = FuzzyDeduplicator(threshold=90, key_field='name')

items = [
    {'name': 'Apple iPhone 15 Pro Max 256GB'},
    {'name': 'Apple iPhone 15 Pro Max - 256GB'},  # Duplicate
    {'name': 'Samsung Galaxy S24 Ultra'},
]

for item in items:
    if dedup.add_if_unique(item):
        print(f"Added: {item['name']}")
    else:
        print(f"Duplicate: {item['name']}")

from abc import ABC, abstractmethod
from typing import List, Optional, Any
import logging

logger = logging.getLogger(__name__)

class PipelineStep(ABC):
    """Base class for pipeline steps."""

    @abstractmethod
    def process(self, item: dict) -> Optional[dict]:
        """Process item. Return None to drop it."""
        pass

    @property
    def name(self):
        return self.__class__.__name__


class CleaningPipeline:
    """Orchestrate data cleaning steps."""

    def __init__(self, steps: List[PipelineStep]):
        self.steps = steps
        self.stats = {
            'processed': 0,
            'passed': 0,
            'dropped': 0,
            'dropped_by': {}
        }

    def process(self, item: dict) -> Optional[dict]:
        self.stats['processed'] += 1

        for step in self.steps:
            try:
                item = step.process(item)
                if item is None:
                    self.stats['dropped'] += 1
                    self.stats['dropped_by'][step.name] = \
                        self.stats['dropped_by'].get(step.name, 0) + 1
                    logger.debug(f"Item dropped by {step.name}")
                    return None
            except Exception as e:
                logger.error(f"Error in {step.name}: {e}")
                self.stats['dropped'] += 1
                return None

        self.stats['passed'] += 1
        return item

    def process_batch(self, items: List[dict]) -> List[dict]:
        return [
            result for item in items
            if (result := self.process(item)) is not None
        ]


# Example steps
class TextCleaningStep(PipelineStep):
    def __init__(self, fields: List[str]):
        self.fields = fields

    def process(self, item):
        for field in self.fields:
            if field in item and item[field]:
                item[field] = clean_text(item[field])
        return item


class PriceParsingStep(PipelineStep):
    def process(self, item):
        if 'price_raw' in item:
            price, currency = parse_price(item['price_raw'])
            item['price'] = price
            item['currency'] = currency
            del item['price_raw']
        return item


class ValidationStep(PipelineStep):
    def __init__(self, schema_class):
        self.schema_class = schema_class

    def process(self, item):
        try:
            validated = self.schema_class(**item)
            return validated.dict()
        except ValidationError:
            return None


class DeduplicationStep(PipelineStep):
    def __init__(self, key_fields):
        self.deduplicator = ExactDeduplicator(key_fields)

    def process(self, item):
        if self.deduplicator.is_duplicate(item):
            return None
        return item


# Build pipeline
pipeline = CleaningPipeline([
    TextCleaningStep(['name', 'description']),
    PriceParsingStep(),
    ValidationStep(Product),
    DeduplicationStep(['sku']),
])

# Process data
clean_items = pipeline.process_batch(raw_items)
print(f"Pipeline stats: {pipeline.stats}")

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List
import statistics

@dataclass
class QualityMetrics:
    """Track data quality metrics over time."""

    timestamp: datetime = field(default_factory=datetime.utcnow)
    total_items: int = 0
    valid_items: int = 0
    field_completeness: Dict[str, float] = field(default_factory=dict)
    field_validity: Dict[str, float] = field(default_factory=dict)
    duplicate_rate: float = 0.0
    anomalies: List[str] = field(default_factory=list)

    @property
    def overall_quality_score(self) -> float:
        if self.total_items == 0:
            return 0.0
        validity_rate = self.valid_items / self.total_items
        completeness = statistics.mean(self.field_completeness.values()) if self.field_completeness else 1.0
        return (validity_rate * 0.5 + completeness * 0.3 + (1 - self.duplicate_rate) * 0.2)


class QualityMonitor:
    """Monitor and alert on data quality issues."""

    def __init__(self, alert_threshold=0.8):
        self.alert_threshold = alert_threshold
        self.history: List[QualityMetrics] = []

    def analyze_batch(self, items: List[dict], required_fields: List[str]) -> QualityMetrics:
        metrics = QualityMetrics(total_items=len(items))

        # Field completeness
        for field in required_fields:
            present = sum(1 for item in items if item.get(field))
            metrics.field_completeness[field] = present / len(items) if items else 0

        # Detect anomalies
        if metrics.overall_quality_score < self.alert_threshold:
            metrics.anomalies.append(
                f"Quality score {metrics.overall_quality_score:.2f} below threshold"
            )

        # Compare with historical baseline
        if self.history:
            avg_historical = statistics.mean(m.overall_quality_score for m in self.history[-10:])
            if metrics.overall_quality_score < avg_historical * 0.9:
                metrics.anomalies.append(
                    f"Quality dropped 10%+ from historical average ({avg_historical:.2f})"
                )

        self.history.append(metrics)
        return metrics

    def should_alert(self, metrics: QualityMetrics) -> bool:
        return len(metrics.anomalies) > 0