Google Search.

# inverted_index/index_shard.py

"""
Inverted Index Shard Implementation
Demonstrates the core data structures used in search engines.
"""

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from collections import defaultdict
import bisect
import math


@dataclass
class Posting:
    """A single posting entry in the inverted index."""
    doc_id: int
    term_frequency: int
    positions: List[int] = field(default_factory=list)


@dataclass
class PostingsList:
    """List of all postings for a single term."""
    term: str
    doc_frequency: int
    postings: List[Posting] = field(default_factory=list)
    
    def intersect(self, other: 'PostingsList') -> 'PostingsList':
        """
        Intersect two posting lists (for AND queries).
        Uses merge algorithm - O(n + m).
        """
        result = PostingsList(
            term=f"{self.term} AND {other.term}",
            doc_frequency=0,
            postings=[]
        )
        
        i, j = 0, 0
        while i < len(self.postings) and j < len(other.postings):
            if self.postings[i].doc_id == other.postings[j].doc_id:
                combined = Posting(
                    doc_id=self.postings[i].doc_id,
                    term_frequency=self.postings[i].term_frequency + 
                                   other.postings[j].term_frequency,
                    positions=sorted(self.postings[i].positions + 
                                    other.postings[j].positions)
                )
                result.postings.append(combined)
                i += 1
                j += 1
            elif self.postings[i].doc_id < other.postings[j].doc_id:
                i += 1
            else:
                j += 1
        
        result.doc_frequency = len(result.postings)
        return result


class InvertedIndexShard:
    """
    A single shard of the inverted index.
    
    In production:
    - Each shard handles ~1-10M documents
    - Index is memory-mapped for fast access
    - Postings are sorted by doc_id for efficient intersection
    """
    
    def __init__(self, shard_id: int, doc_range: tuple):
        self.shard_id = shard_id
        self.doc_range = doc_range
        self.index: Dict[str, PostingsList] = {}
        self.doc_count = 0
        
    def add_document(self, doc_id: int, tokens: List[str]) -> None:
        """Index a single document."""
        term_positions: Dict[str, List[int]] = defaultdict(list)
        for position, token in enumerate(tokens):
            term_positions[token].append(position)
        
        for term, positions in term_positions.items():
            if term not in self.index:
                self.index[term] = PostingsList(term=term, doc_frequency=0)
            
            posting = Posting(
                doc_id=doc_id,
                term_frequency=len(positions),
                positions=positions
            )
            
            # key= needs Python 3.10+; insort into a list is O(n) per insert
            bisect.insort(
                self.index[term].postings, 
                posting,
                key=lambda p: p.doc_id
            )
            self.index[term].doc_frequency += 1
        
        self.doc_count += 1
    
    def search(self, query_terms: List[str]) -> PostingsList:
        """Search for documents containing all query terms."""
        if not query_terms:
            return PostingsList(term="", doc_frequency=0)
        
        postings_lists = []
        for term in query_terms:
            pl = self.index.get(term)
            if pl is None:
                return PostingsList(term=term, doc_frequency=0)
            postings_lists.append(pl)
        
        # Sort by document frequency (shortest first)
        postings_lists.sort(key=lambda pl: pl.doc_frequency)
        
        # Intersect all posting lists
        result = postings_lists[0]
        for pl in postings_lists[1:]:
            result = result.intersect(pl)
            if result.doc_frequency == 0:
                break
        
        return result
    
    def compute_idf(self, doc_frequency: int) -> float:
        """Compute Inverse Document Frequency."""
        if doc_frequency == 0:
            return 0.0
        return math.log(self.doc_count / doc_frequency)

# ranking/pagerank.py

"""
PageRank Computation at Scale
Demonstrates how to compute PageRank for billions of pages.
"""

from dataclasses import dataclass
from typing import Dict, List


@dataclass
class WebPage:
    """Represents a web page in the link graph."""
    url: str
    outgoing_links: List[str]
    incoming_links: List[str] = None
    pagerank: float = 1.0


class PageRankComputer:
    """
    Computes PageRank scores for a web graph.
    
    At web scale (a likely design, not Google's published pipeline):
    - Graph has 100+ billion nodes
    - Computation takes days on thousands of machines
    - Uses MapReduce-style distributed iteration
    """
    
    def __init__(self, damping_factor: float = 0.85, max_iterations: int = 100,
                 convergence_threshold: float = 1e-6):
        self.damping_factor = damping_factor
        self.max_iterations = max_iterations
        self.convergence_threshold = convergence_threshold
        self.pages: Dict[str, WebPage] = {}
        self.num_pages = 0
        
    def add_page(self, url: str, outgoing_links: List[str]) -> None:
        """Add a page to the graph."""
        self.pages[url] = WebPage(url=url, outgoing_links=outgoing_links)
        
        for link in outgoing_links:
            if link not in self.pages:
                self.pages[link] = WebPage(url=link, outgoing_links=[])
        
        self.num_pages = len(self.pages)
    
    def build_reverse_links(self) -> None:
        """Build incoming link lists for each page."""
        for page in self.pages.values():
            page.incoming_links = []
        
        for url, page in self.pages.items():
            for link in page.outgoing_links:
                if link in self.pages:
                    self.pages[link].incoming_links.append(url)
    
    def compute(self) -> Dict[str, float]:
        """
        Compute PageRank using power iteration.
        
        Algorithm:
        1. Initialize all pages with equal rank (1/N)
        2. For each iteration:
           - Sum contributions from incoming links
           - Apply damping factor
        3. Repeat until convergence
        """
        if not self.pages:
            return {}
        
        self.build_reverse_links()
        
        # Initialize PageRank values
        initial_rank = 1.0 / self.num_pages
        for page in self.pages.values():
            page.pagerank = initial_rank
        
        # Iterative computation
        for iteration in range(self.max_iterations):
            new_ranks: Dict[str, float] = {}
            
            # Handle dangling pages
            dangling_sum = sum(
                page.pagerank for page in self.pages.values()
                if len(page.outgoing_links) == 0
            )
            dangling_contribution = dangling_sum / self.num_pages
            
            # Compute new rank for each page
            for url, page in self.pages.items():
                new_rank = (1 - self.damping_factor) / self.num_pages
                new_rank += self.damping_factor * dangling_contribution
                
                for source_url in page.incoming_links:
                    source_page = self.pages[source_url]
                    num_outlinks = len(source_page.outgoing_links)
                    if num_outlinks > 0:
                        contribution = source_page.pagerank / num_outlinks
                        new_rank += self.damping_factor * contribution
                
                new_ranks[url] = new_rank
            
            # Check for convergence
            max_diff = max(
                abs(new_ranks[url] - self.pages[url].pagerank)
                for url in self.pages
            )
            
            for url, rank in new_ranks.items():
                self.pages[url].pagerank = rank
            
            if max_diff < self.convergence_threshold:
                print(f"Converged after {iteration + 1} iterations")
                break
        
        return {url: page.pagerank for url, page in self.pages.items()}

# query/scorer.py

"""
BM25 Ranking Algorithm
An improvement over TF-IDF that handles term frequency saturation
and document length normalization.
"""

import math
from typing import Dict, List, Optional


class BM25Scorer:
    """
    BM25 scoring for search relevance.
    
    BM25 is the industry standard for text relevance scoring.
    It improves on TF-IDF by:
    - Saturating term frequency (diminishing returns)
    - Normalizing for document length
    """
    
    K1 = 1.2  # Term frequency saturation parameter
    B = 0.75  # Length normalization parameter
    
    def __init__(self, avg_doc_length: float, total_docs: int):
        self.avg_doc_length = avg_doc_length
        self.total_docs = total_docs
    
    def score(self, doc: Dict, query_terms: List[str]) -> float:
        """
        Compute BM25 score for a document.
        
        Args:
            doc: Document with term_freqs, doc_length, doc_freqs
            query_terms: List of query terms
            
        Returns:
            BM25 score (higher = more relevant)
        """
        score = 0.0
        doc_length = doc.get("doc_length", self.avg_doc_length)
        term_freqs = doc.get("term_freqs", {})
        
        for term in query_terms:
            tf = term_freqs.get(term, 0)
            if tf == 0:
                continue
            
            # IDF component
            df = doc.get("doc_freqs", {}).get(term, 1)
            idf = math.log((self.total_docs - df + 0.5) / (df + 0.5) + 1)
            
            # TF component with saturation
            numerator = tf * (self.K1 + 1)
            denominator = tf + self.K1 * (
                1 - self.B + self.B * (doc_length / self.avg_doc_length)
            )
            
            score += idf * (numerator / denominator)
        
        return score
    
    def compute_freshness(self, last_updated) -> float:
        """Compute freshness score with exponential decay."""
        from datetime import datetime
        
        if last_updated is None:
            return 0.5
        
        age_days = (datetime.utcnow() - last_updated).days
        half_life = 30  # 30-day half-life
        return math.exp(-0.693 * age_days / half_life)

# crawler/web_crawler.py

"""
Distributed Web Crawler
Demonstrates key crawling concepts:
- Politeness (rate limiting per domain)
- Priority-based URL frontier
- Parallel fetching with async IO
"""

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set
from datetime import datetime
from urllib.parse import urlparse, urljoin
import asyncio
import heapq


@dataclass
class CrawlURL:
    """A URL in the crawl frontier."""
    url: str
    priority: float
    last_crawl: Optional[datetime] = None
    depth: int = 0
    
    def __lt__(self, other):
        return self.priority > other.priority


class URLFrontier:
    """
    Priority queue of URLs to crawl.
    
    Key features:
    - Priority-based ordering
    - Deduplication
    - Per-domain queuing for politeness
    """
    
    def __init__(self, max_size: int = 10_000_000):
        self.max_size = max_size
        self.queue: List[CrawlURL] = []
        self.seen_urls: Set[str] = set()
        self.domain_last_access: Dict[str, datetime] = {}
        
    def add(self, url: str, priority: float = 0.5, depth: int = 0) -> bool:
        """Add URL to frontier if not seen."""
        url = url.lower().rstrip("/").split("#")[0]
        
        if url in self.seen_urls:
            return False
        
        if len(self.seen_urls) >= self.max_size:
            return False
        
        crawl_url = CrawlURL(url=url, priority=priority, depth=depth)
        heapq.heappush(self.queue, crawl_url)
        self.seen_urls.add(url)
        return True
    
    def get_next(self) -> Optional[CrawlURL]:
        """Get next URL respecting politeness constraints."""
        if not self.queue:
            return None
        
        now = datetime.utcnow()
        deferred: List[CrawlURL] = []  # blocked by politeness; re-queue after the scan
        result: Optional[CrawlURL] = None
        
        while self.queue:
            crawl_url = heapq.heappop(self.queue)
            domain = urlparse(crawl_url.url).netloc
            
            last_access = self.domain_last_access.get(domain)
            if last_access and (now - last_access).total_seconds() < 1.0:
                deferred.append(crawl_url)
                continue
            
            self.domain_last_access[domain] = now
            result = crawl_url
            break
        
        for item in deferred:
            heapq.heappush(self.queue, item)
        return result


class WebCrawler:
    """
    Distributed web crawler with async IO.
    """
    
    def __init__(self, num_workers: int = 100, max_pages: int = 1_000_000):
        self.num_workers = num_workers
        self.max_pages = max_pages
        self.frontier = URLFrontier()
        self.pages_crawled = 0
    
    def add_seeds(self, seed_urls: List[str]) -> None:
        """Add seed URLs to start crawling."""
        for url in seed_urls:
            self.frontier.add(url, priority=1.0, depth=0)
    
    async def crawl(self) -> None:
        """Start crawling with multiple workers."""
        workers = [
            asyncio.create_task(self._worker(i))
            for i in range(self.num_workers)
        ]
        await asyncio.gather(*workers)
    
    async def _worker(self, worker_id: int) -> None:
        """Single crawler worker."""
        import aiohttp
        
        async with aiohttp.ClientSession(
            headers={"User-Agent": "GoogleBot/2.1"}
        ) as session:
            while self.pages_crawled < self.max_pages:
                crawl_url = self.frontier.get_next()
                if crawl_url is None:
                    await asyncio.sleep(0.1)
                    continue
                
                try:
                    async with session.get(
                        crawl_url.url,
                        timeout=aiohttp.ClientTimeout(total=30)
                    ) as response:
                        if response.status == 200:
                            content = await response.text()
                            links = self._extract_links(crawl_url.url, content)
                            
                            for link in links:
                                self.frontier.add(
                                    link, priority=0.5,
                                    depth=crawl_url.depth + 1
                                )
                            
                            self.pages_crawled += 1
                except Exception:
                    pass
    
    def _extract_links(self, base_url: str, html: str) -> List[str]:
        """Extract links from HTML content."""
        import re
        links = []
        href_pattern = re.compile(r'href=["\']([^"\']+)["\']')
        
        for match in href_pattern.finditer(html):
            href = match.group(1)
            if href.startswith(("javascript:", "mailto:", "#")):
                continue
            
            absolute_url = urljoin(base_url, href)
            if absolute_url.startswith(("http://", "https://")):
                links.append(absolute_url)
        
        return links