move scraper modules to internals

Author: ks6088ts

.env.template

@@ -63,3 +63,6 @@ PDF_LOADER_DATA_DIR_PATH="./data"
 ## OpenTelemetry Settings
 OTEL_SERVICE_NAME="template-langgraph"
 OTEL_COLLECTOR_ENDPOINT="http://localhost:4317"
+
+## Scraper Settings
+SCRAPER_TYPE="mock" # Options: "mock", "httpx", "youtube_transcript"

scripts/agent_operator.py

@@ -18,15 +18,11 @@
     AgentState,
     Article,
 )
-from template_langgraph.agents.news_summarizer_agent.scrapers import (
-    BaseScraper,
-    HttpxScraper,
-    YouTubeTranscriptScraper,
-)
 from template_langgraph.agents.news_summarizer_agent.summarizers import (
     LlmSummarizer,
 )
 from template_langgraph.agents.task_decomposer_agent.agent import graph as task_decomposer_agent_graph
+from template_langgraph.internals.scrapers import get_scraper
 from template_langgraph.loggers import get_logger
 
 # Initialize the Typer application
@@ -56,18 +52,6 @@ def get_agent_graph(name: str):
         raise ValueError(f"Unknown agent name: {name}")
 
 
-def get_scraper(scraper_type: str) -> BaseScraper:
-    scraper = None
-    if scraper_type == "Httpx":
-        scraper = HttpxScraper()
-    elif scraper_type == "YouTubeTranscript":
-        scraper = YouTubeTranscriptScraper()
-
-    if not scraper:
-        raise ValueError(f"Unknown scraper type: {scraper_type}")
-    return scraper
-
-
 @app.command()
 def png(
     name: str = typer.Option(
@@ -159,12 +143,6 @@ def news_summarizer_agent(
         "-u",
         help="Comma-separated list of URLs to summarize",
     ),
-    scraper: str = typer.Option(
-        "Httpx",  # YouTubeTranscript
-        "--scraper",
-        "-s",
-        help="Scraper to use for fetching content",
-    ),
     verbose: bool = typer.Option(
         False,
         "--verbose",
@@ -178,7 +156,7 @@ def news_summarizer_agent(
 
     graph = NewsSummarizerAgent(
         notifier=MockNotifier(),
-        scraper=get_scraper(scraper),
+        scraper=get_scraper(),
         summarizer=LlmSummarizer(),
     ).create_graph()
     for event in graph.stream(

template_langgraph/agents/news_summarizer_agent/agent.py

@@ -8,16 +8,16 @@
     StructuredArticle,
     SummarizeWebContentState,
 )
-from template_langgraph.agents.news_summarizer_agent.scrapers import (
-    BaseScraper,
-    HttpxScraper,
-    MockScraper,
-)
 from template_langgraph.agents.news_summarizer_agent.summarizers import (
     BaseSummarizer,
     LlmSummarizer,
     MockSummarizer,
 )
+from template_langgraph.internals.scrapers import (
+    BaseScraper,
+    HttpxScraper,
+    MockScraper,
+)
 from template_langgraph.llms.azure_openais import AzureOpenAiWrapper
 from template_langgraph.loggers import get_logger
 

template_langgraph/agents/news_summarizer_agent/scrapers.py renamed to template_langgraph/internals/scrapers.py

@@ -8,15 +8,40 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from enum import Enum
+from functools import lru_cache
 
 import httpx
+from pydantic_settings import BaseSettings, SettingsConfigDict
 from youtube_transcript_api import YouTubeTranscriptApi
 
 from template_langgraph.loggers import get_logger
 
 logger = get_logger(__name__)
 
 
+class ScraperType(str, Enum):
+    MOCK = "mock"
+    HTTPX = "httpx"
+    YOUTUBE_TRANSCRIPT = "youtube_transcript"
+
+
+class Settings(BaseSettings):
+    scraper_type: ScraperType = ScraperType.MOCK
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_ignore_empty=True,
+        extra="ignore",
+    )
+
+
+@lru_cache
+def get_scraper_settings() -> Settings:
+    """Get scraper settings."""
+    return Settings()
+
+
 class BaseScraper(ABC):
     """Abstract base scraper.
 
@@ -60,6 +85,7 @@ class YouTubeTranscriptScraper(BaseScraper):
     """YouTube transcript scraper."""
 
     def scrape(self, url: str) -> str:
+        logger.info(f"Fetching YouTube transcript for URL: {url}")
         video_id = url.split("v=")[-1].split("&")[0]
         transcript = YouTubeTranscriptApi().fetch(
             video_id=video_id,
@@ -69,9 +95,15 @@ def scrape(self, url: str) -> str:
         return " ".join(text_list)
 
 
-__all__ = [
-    "BaseScraper",
-    "MockScraper",
-    "HttpxScraper",
-    "YouTubeTranscriptScraper",
-]
+def get_scraper(settings: Settings = None) -> BaseScraper:
+    if settings is None:
+        settings = get_scraper_settings()
+
+    if settings.scraper_type == ScraperType.MOCK:
+        return MockScraper()
+    elif settings.scraper_type == ScraperType.HTTPX:
+        return HttpxScraper()
+    elif settings.scraper_type == ScraperType.YOUTUBE_TRANSCRIPT:
+        return YouTubeTranscriptScraper()
+    else:
+        raise ValueError(f"Unknown scraper type: {settings.scraper_type}")