initial commit

scraper/__init__.py

@@ -0,0 +1 @@
+# Initialize scraper package

scraper/client.py

@@ -0,0 +1,59 @@
+import httpx
+import asyncio
+from loguru import logger
+from typing import Optional
+
+class GuardianClient:
+    def __init__(self, concurrency_limit: int = 5):
+        self.headers = {
+            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3",
+            "Accept-Language": "en-US,en;q=0.9",
+            "Accept-Encoding": "gzip, deflate, br",
+            "Connection": "keep-alive",
+            "Upgrade-Insecure-Requests": "1",
+            "Referer": "https://www.google.com/",
+        }
+        self.semaphore = asyncio.Semaphore(concurrency_limit)
+        self.client = httpx.AsyncClient(
+            headers=self.headers,
+            follow_redirects=True,
+            timeout=httpx.Timeout(10.0, read=30.0),
+            http2=True
+        )
+
+    async def fetch(self, url: str, retries: int = 3) -> Optional[str]:
+        async with self.semaphore:
+            for attempt in range(retries):
+                try:
+                    logger.debug(f"Fetching {url} (Attempt {attempt + 1}/{retries})")
+                    response = await self.client.get(url)
+
+                    if response.status_code == 200:
+                        return response.text
+
+                    if response.status_code == 429:
+                        wait_time = (attempt + 1) * 2
+                        logger.warning(f"Rate limited (429) on {url}. Waiting {wait_time}s...")
+                        await asyncio.sleep(wait_time)
+                        continue
+
+                    if 500 <= response.status_code < 600:
+                        wait_time = (attempt + 1) * 2
+                        logger.warning(f"Server error ({response.status_code}) on {url}. Waiting {wait_time}s...")
+                        await asyncio.sleep(wait_time)
+                        continue
+
+                    logger.error(f"Failed to fetch {url}: HTTP {response.status_code}")
+                    return None
+
+                except httpx.RequestError as e:
+                    logger.error(f"Request error fetching {url}: {e}")
+                    if attempt == retries - 1:
+                        return None
+                    await asyncio.sleep((attempt + 1) * 2)
+
+            return None
+
+    async def close(self):
+        await self.client.aclose()

scraper/models.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel, Field
+from datetime import datetime
+from typing import Optional
+
+class Article(BaseModel):
+    headline: str = Field(..., description="The headline of the article")
+    url: str = Field(..., description="The URL of the article")
+    body: str = Field(..., description="The main body text of the article")
+    scraped_at: datetime = Field(default_factory=datetime.utcnow, description="Timestamp of when the article was scraped")
+
+class ScrapeResult(BaseModel):
+    articles: list[Article]
+    total_count: int
+    scraped_at: datetime = Field(default_factory=datetime.utcnow)

scraper/parser.py

@@ -0,0 +1,91 @@
+from bs4 import BeautifulSoup
+from typing import List, Tuple, Optional
+import re
+
+class GuardianParser:
+    # Regular expression to identify article URLs
+    # Guardian articles typically follow /category/article-slug
+    # Refined pattern: articles usually have more segments or a specific slug format
+    # Avoid matching simple category paths like /world/middleeast
+    ARTICLE_URL_PATTERN = re.compile(r'^/(world|politics|sport|business|culture|lifeandstyle|opinion|environment|science|tech)/[a-zA-Z0-9-]+(/[a-zA-Z0-9-]+)*$')
+
+    @staticmethod
+    def extract_article_links(html: str) -> List[Tuple[str, str]]:
+        """
+        Extracts headlines and URLs from the home page using the data-link-name attribute.
+        """
+        soup = BeautifulSoup(html, 'html.parser')
+        articles = []
+
+        # Target links with data-link-name="article"
+        for a in soup.select('a[data-link-name="article"]'):
+            href = a.get('href', '')
+            if not href:
+                continue
+
+            # Clean href (remove query params)
+            href = href.split('?')[0]
+
+            # Handle absolute and relative URLs
+            if href.startswith('http'):
+                full_url = href
+            elif href.startswith('/'):
+                full_url = f"https://www.theguardian.com{href}"
+            else:
+                continue
+
+            headline = a.get_text(strip=True)
+            if headline:
+                articles.append((headline, full_url))
+
+        # De-duplicate while preserving order
+        seen = set()
+        unique_articles = []
+        for headline, url in articles:
+            if url not in seen:
+                unique_articles.append((headline, url))
+                seen.add(url)
+
+        return unique_articles
+
+
+        # De-duplicate while preserving order
+        seen = set()
+        unique_articles = []
+        for headline, url in articles:
+            if url not in seen:
+                unique_articles.append((headline, url))
+                seen.add(url)
+
+        return unique_articles
+
+    @staticmethod
+    def extract_article_body(html: str) -> Optional[str]:
+        """
+        Extracts the main body text from an article page.
+        """
+        soup = BeautifulSoup(html, 'html.parser')
+
+        # Primary target: data-gu-context="body"
+        body_div = soup.find('div', {'data-gu-context': 'body'})
+        if body_div:
+            # Remove script and style elements
+            for script_or_style in body_div(['script', 'style']):
+                script_or_style.decompose()
+            return body_div.get_text(separator='\n', strip=True)
+
+        # Fallback 1: <article> tag
+        article_tag = soup.find('article')
+        if article_tag:
+            for script_or_style in article_tag(['script', 'style']):
+                script_or_style.decompose()
+            return article_tag.get_text(separator='\n', strip=True)
+
+        # Fallback 2: Main content area
+        main_content = soup.find('main')
+        if main_content:
+            for script_or_style in main_content(['script', 'style']):
+                script_or_style.decompose()
+            return main_content.get_text(separator='\n', strip=True)
+
+        return None

scraper/utils.py

@@ -0,0 +1,16 @@
+import json
+import os
+from .models import ScrapeResult
+
+def save_results_to_json(result: ScrapeResult, file_path: str):
+    """
+    Saves the scrape result to a JSON file.
+    """
+    # Ensure the directory exists
+    os.makedirs(os.path.dirname(file_path), exist_ok=True)
+
+    # Convert Pydantic model to dict
+    data = result.model_dump(mode='json')
+
+    with open(file_path, 'w', encoding='utf-8') as f:
+        json.dump(data, f, indent=4, ensure_ascii=False)