from crewai.tools import BaseTool
from pydantic import BaseModel, Field, ConfigDict
from typing import List, Optional
from apify import Actor
from src.tools.base import RunApifyActor
class TwitterScraperInput(BaseModel):
"""Input schema for TwitterScraper tool."""
searchTerms: Optional[List[str]] = Field(
description="Search terms to find tweets containing these terms. Alternative to using Twitter URLs.",
default=None
)
sort: Optional[str] = Field(
description="How to sort the returned tweets. Setting to 'Latest' yields more results.",
default="Top",
enum=["Top", "Latest"]
)
start: Optional[str] = Field(
description="Scrape tweets starting from this date (format: YYYY-MM-DD)",
default=None
)
end: Optional[str] = Field(
description="Scrape tweets until this date (format: YYYY-MM-DD)",
default=None
)
class TwitterScraperTool(BaseTool):
name: str = "Twitter Scraper"
description: str = "Tool for scraping Twitter content with configurable parameters"
args_schema: type[BaseModel] = TwitterScraperInput
actor: Actor = Field(description="Apify Actor instance")
model_config = ConfigDict(arbitrary_types_allowed=True)
def _run(
self,
searchTerms: Optional[List[str]] = None,
sort: Optional[str] = "latest",
start: Optional[str] = None,
end: Optional[str] = None
) -> str:
run_inputs = {}
if searchTerms:
run_inputs["searchTerms"] = searchTerms
if sort:
run_inputs["sort"] = sort
run_inputs["maxItems"] = 5
if start:
run_inputs["start"] = start
if end:
run_inputs["end"] = end
run_actor = RunApifyActor(self.actor)
dataset = run_actor._run("apidojo/twitter-scraper-lite", run_inputs)
return dataset
