jpmschweitzer/core-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

Remove obsolete web scraper module

Browse Source 
Web scraping functionality is no longer needed. Removes:
- src/web_scraper/ directory (6 files)
- /web-scraper/scrape endpoint from tools controller
- References from health controller endpoints list

Tools controller now only contains DNS lookup functionality.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-17 16:28:10 +01:00
co-authored by Claude Opus 4.5
parent c45fdcb528
commit 33f31b09a3
8 changed files with 1 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/controllers/health_controller.py
+1 -1
View File
@@ -61,7 +61,7 @@ class HealthController(BaseController):
"chat_completions": "/v1/chat/completions", "chat_completions": "/v1/chat/completions",
"models": "/v1/models", "models": "/v1/models",
"conversations": "/v1/conversations", "conversations": "/v1/conversations",
"web_scraper": "/web-scraper/scrape", "dns_lookup": "/tools/dns/lookup",
"infrastructure": "/infrastructure", "infrastructure": "/infrastructure",
"health": "/health", "health": "/health",
"health_full": "/health/full" "health_full": "/health/full"
src/controllers/tools_controller.py
-66
View File
@@ -2,16 +2,12 @@
Tools Controller Tools Controller
Provides utility tool endpoints including: Provides utility tool endpoints including:
- Web scraping and content extraction
- DNS lookups - DNS lookups
""" """
from fastapi import APIRouter, HTTPException, status from fastapi import APIRouter, HTTPException, status
from src.controllers.base import BaseController from src.controllers.base import BaseController
from src.logging_config import get_logger from src.logging_config import get_logger
from src.web_scraper.schemas import WebScraperRequest, WebScraperResponse
from src.web_scraper.service import WebScraperService
from src.web_scraper.exceptions import FetchError, ScrapingError
from src.dns.schemas import DNSLookupRequest, DNSLookupResponse from src.dns.schemas import DNSLookupRequest, DNSLookupResponse
from src.dns.service import DNSService from src.dns.service import DNSService
from src.dns.exceptions import DNSQueryError from src.dns.exceptions import DNSQueryError
@@ -24,79 +20,17 @@ class ToolsController(BaseController):
Controller for utility tools Controller for utility tools
Provides endpoints for: Provides endpoints for:
- Web scraping and content extraction
- DNS lookups - DNS lookups
""" """
def __init__(self): def __init__(self):
super().__init__(prefix="/tools", tags=["Tools"]) super().__init__(prefix="/tools", tags=["Tools"])
# Initialize services (could be dependency injected for testing)
self.scraper_service = WebScraperService()
self.dns_service = DNSService() self.dns_service = DNSService()
def create_router(self) -> APIRouter: def create_router(self) -> APIRouter:
"""Create and configure the router""" """Create and configure the router"""
router = APIRouter(prefix=self.prefix, tags=self.tags) router = APIRouter(prefix=self.prefix, tags=self.tags)
@router.post(
"/scrape",
response_model=WebScraperResponse,
status_code=status.HTTP_200_OK,
summary="Scrape website content",
description="""
Scrape and extract main content from a website.
Uses trafilatura for intelligent content extraction (articles, blog posts, documentation),
with BeautifulSoup as fallback. Perfect for feeding webpage content to LLMs.
**Features:**
- Intelligent main content extraction
- Removes navigation, ads, footers
- Optional link extraction
- Configurable content length limits
**Rate Limiting:** None (internal network use only)
"""
)
async def scrape_website(request: WebScraperRequest) -> WebScraperResponse:
"""
Scrape a website and extract its main content
Args:
request: Scraping request with URL and options
Returns:
Extracted content with metadata
Raises:
HTTPException: 400 for fetch errors, 500 for processing errors
"""
try:
logger.info(f"Received scrape request for: {request.url}")
result = await self.scraper_service.scrape_url(request)
return result
except FetchError as e:
logger.warning(f"Fetch failed: {str(e)}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Failed to fetch URL: {str(e)}"
)
except ScrapingError as e:
logger.error(f"Scraping failed: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to extract content: {str(e)}"
)
except Exception as e:
logger.error(f"Unexpected error: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="An unexpected error occurred"
)
@router.post( @router.post(
"/dns/lookup", "/dns/lookup",
response_model=DNSLookupResponse, response_model=DNSLookupResponse,
src/web_scraper/__init__.py
-13
View File
@@ -1,13 +0,0 @@
"""
Web scraper module for extracting content from websites
"""
from src.web_scraper.router import router
from src.web_scraper.schemas import WebScraperRequest, WebScraperResponse
from src.web_scraper.service import WebScraperService
__all__ = [
"router",
"WebScraperRequest",
"WebScraperResponse",
"WebScraperService",
]
src/web_scraper/config.py
-32
View File
@@ -1,32 +0,0 @@
"""
Configuration for web scraper module
"""
from pydantic_settings import BaseSettings
from functools import lru_cache
class WebScraperSettings(BaseSettings):
"""Web scraper specific settings"""
# HTTP client configuration
request_timeout: int = 30
max_redirects: int = 5
user_agent: str = "Mozilla/5.0 (compatible; CoreCode/1.0)"
# Content extraction
default_max_length: int = 10000
max_links_to_extract: int = 50
# Rate limiting (future use)
rate_limit_enabled: bool = False
requests_per_minute: int = 60
class Config:
env_prefix = "WEB_SCRAPER_"
case_sensitive = False
@lru_cache()
def get_web_scraper_settings() -> WebScraperSettings:
"""Cached web scraper settings instance"""
return WebScraperSettings()
src/web_scraper/exceptions.py
-18
View File
@@ -1,18 +0,0 @@
"""
Custom exceptions for web scraper module
"""
class WebScraperException(Exception):
"""Base exception for web scraper module"""
pass
class FetchError(WebScraperException):
"""Raised when URL fetch fails"""
pass
class ScrapingError(WebScraperException):
"""Raised when content extraction fails"""
pass
src/web_scraper/router.py
-79
View File
@@ -1,79 +0,0 @@
"""
API routes for web scraper module
"""
from fastapi import APIRouter, HTTPException, status
from src.logging_config import get_logger
from src.web_scraper.schemas import WebScraperRequest, WebScraperResponse
from src.web_scraper.service import WebScraperService
from src.web_scraper.exceptions import FetchError, ScrapingError
logger = get_logger(__name__)
router = APIRouter(
prefix="/web-scraper",
tags=["Web Scraper"]
)
# Initialize service (could be dependency injected for testing)
scraper_service = WebScraperService()
@router.post(
"/scrape",
response_model=WebScraperResponse,
status_code=status.HTTP_200_OK,
summary="Scrape website content",
description="""
Scrape and extract main content from a website.
Uses trafilatura for intelligent content extraction (articles, blog posts, documentation),
with BeautifulSoup as fallback. Perfect for feeding webpage content to LLMs.
**Features:**
- Intelligent main content extraction
- Removes navigation, ads, footers
- Optional link extraction
- Configurable content length limits
**Rate Limiting:** None (internal network use only)
"""
)
async def scrape_website(request: WebScraperRequest) -> WebScraperResponse:
"""
Scrape a website and extract its main content
Args:
request: Scraping request with URL and options
Returns:
Extracted content with metadata
Raises:
HTTPException: 400 for fetch errors, 500 for processing errors
"""
try:
logger.info(f"Received scrape request for: {request.url}")
result = await scraper_service.scrape_url(request)
return result
except FetchError as e:
logger.warning(f"Fetch failed: {str(e)}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Failed to fetch URL: {str(e)}"
)
except ScrapingError as e:
logger.error(f"Scraping failed: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to extract content: {str(e)}"
)
except Exception as e:
logger.error(f"Unexpected error: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="An unexpected error occurred"
)
src/web_scraper/schemas.py
-69
View File
@@ -1,69 +0,0 @@
"""
Pydantic schemas for web scraper module
"""
from pydantic import HttpUrl, Field
from typing import Optional
from datetime import datetime
from src.base_schema import BaseSchema
class WebScraperRequest(BaseSchema):
"""Request model for web scraping"""
url: HttpUrl = Field(
...,
description="The URL to scrape",
examples=["https://example.com/article"]
)
extract_main_content: bool = Field(
default=True,
description="Use intelligent content extraction (trafilatura) vs raw HTML parsing"
)
include_links: bool = Field(
default=False,
description="Include list of links found on the page"
)
max_length: Optional[int] = Field(
default=10000,
ge=100,
le=100000,
description="Maximum content length to return (100-100000 chars)"
)
class WebScraperResponse(BaseSchema):
"""Response model for web scraping"""
url: str = Field(
...,
description="The scraped URL"
)
title: Optional[str] = Field(
default=None,
description="Page title extracted from <title> tag"
)
content: str = Field(
...,
description="Extracted page content"
)
extracted_at: datetime = Field(
...,
description="UTC timestamp when content was extracted"
)
content_length: int = Field(
...,
ge=0,
description="Length of extracted content in characters"
)
links: Optional[list[str]] = Field(
default=None,
description="List of HTTP(S) links found on the page (max 50)"
)
src/web_scraper/service.py
-213
View File
@@ -1,213 +0,0 @@
"""
Business logic for web scraper module
"""
import httpx
from bs4 import BeautifulSoup
import trafilatura
from datetime import datetime, timezone
from typing import Optional
from src.logging_config import get_logger
from src.web_scraper.config import get_web_scraper_settings
from src.web_scraper.schemas import WebScraperRequest, WebScraperResponse
from src.web_scraper.exceptions import ScrapingError, FetchError
logger = get_logger(__name__)
class WebScraperService:
"""Service class for web scraping operations"""
def __init__(self):
self.settings = get_web_scraper_settings()
async def scrape_url(self, request: WebScraperRequest) -> WebScraperResponse:
"""
Scrape and extract content from a URL
Args:
request: Scraping request parameters
Returns:
Extracted content with metadata
Raises:
FetchError: If URL cannot be fetched
ScrapingError: If content extraction fails
"""
url_str = str(request.url)
logger.info(f"Starting scrape for URL: {url_str}")
try:
# Fetch the webpage
html_content = await self._fetch_url(url_str)
# Extract content based on settings
if request.extract_main_content:
content = self._extract_main_content(html_content, request.include_links)
else:
content = self._extract_basic_content(html_content)
# Extract metadata
title = self._extract_title(html_content)
links = self._extract_links(html_content) if request.include_links else None
# Clean and truncate content
content = self._clean_content(content)
if request.max_length and len(content) > request.max_length:
content = content[:request.max_length] + "\n\n[Content truncated...]"
logger.debug(f"Content truncated to {request.max_length} characters")
logger.info(f"Successfully scraped {len(content)} characters from {url_str}")
return WebScraperResponse(
url=url_str,
title=title,
content=content,
extracted_at=datetime.now(timezone.utc),
content_length=len(content),
links=links
)
except FetchError:
raise
except Exception as e:
logger.error(f"Scraping failed for {url_str}: {str(e)}", exc_info=True)
raise ScrapingError(f"Failed to scrape content: {str(e)}")
async def _fetch_url(self, url: str) -> str:
"""
Fetch HTML content from URL
Args:
url: URL to fetch
Returns:
HTML content as string
Raises:
FetchError: If fetch fails
"""
try:
async with httpx.AsyncClient(
timeout=self.settings.request_timeout,
follow_redirects=True,
max_redirects=self.settings.max_redirects
) as client:
logger.debug(f"Fetching URL: {url}")
response = await client.get(
url,
headers={"User-Agent": self.settings.user_agent}
)
response.raise_for_status()
logger.debug(f"Fetched {len(response.text)} bytes from {url}")
return response.text
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error {e.response.status_code} for {url}")
raise FetchError(f"HTTP {e.response.status_code}: {e.response.reason_phrase}")
except httpx.RequestError as e:
logger.error(f"Request error for {url}: {str(e)}")
raise FetchError(f"Failed to fetch URL: {str(e)}")
def _extract_main_content(self, html: str, include_links: bool = False) -> str:
"""
Extract main content using trafilatura (intelligent extraction)
Args:
html: Raw HTML content
include_links: Whether to preserve links in output
Returns:
Extracted content
"""
logger.debug("Extracting main content with trafilatura")
content = trafilatura.extract(
html,
include_links=include_links,
include_images=False,
output_format='txt',
no_fallback=False
)
# Fallback to BeautifulSoup if trafilatura fails
if not content:
logger.debug("Trafilatura extraction failed, falling back to BeautifulSoup")
content = self._extract_basic_content(html)
return content
def _extract_basic_content(self, html: str) -> str:
"""
Extract content using basic BeautifulSoup parsing
Args:
html: Raw HTML content
Returns:
Extracted text content
"""
logger.debug("Extracting content with BeautifulSoup")
soup = BeautifulSoup(html, 'html.parser')
# Remove unwanted elements
for element in soup(["script", "style", "nav", "footer", "header", "aside"]):
element.decompose()
# Extract text
text = soup.get_text(separator='\n', strip=True)
return text
def _extract_title(self, html: str) -> Optional[str]:
"""
Extract page title from HTML
Args:
html: Raw HTML content
Returns:
Page title or None
"""
soup = BeautifulSoup(html, 'html.parser')
title = soup.title.string if soup.title else None
if title:
title = title.strip()
logger.debug(f"Extracted title: {title}")
return title
def _extract_links(self, html: str) -> list[str]:
"""
Extract HTTP(S) links from HTML
Args:
html: Raw HTML content
Returns:
List of absolute HTTP(S) URLs
"""
soup = BeautifulSoup(html, 'html.parser')
links = [
a.get('href')
for a in soup.find_all('a', href=True)
if a.get('href', '').startswith('http')
]
# Limit number of links
links = links[:self.settings.max_links_to_extract]
logger.debug(f"Extracted {len(links)} links")
return links
def _clean_content(self, content: str) -> str:
"""
Clean and normalize extracted content
Args:
content: Raw extracted content
Returns:
Cleaned content
"""
# Remove empty lines and normalize whitespace
lines = [line.strip() for line in content.split('\n') if line.strip()]
cleaned = '\n'.join(lines)
return cleaned