|
|
""" |
|
|
src/utils/tool_factory.py |
|
|
Tool Factory Pattern for Parallel Agent Execution |
|
|
|
|
|
This module provides a factory pattern for creating independent tool instances |
|
|
for each agent, enabling safe parallel execution without shared state issues. |
|
|
|
|
|
Usage: |
|
|
from src.utils.tool_factory import create_tool_set |
|
|
|
|
|
class MyAgentNode: |
|
|
def __init__(self): |
|
|
# Each agent gets its own private tool set |
|
|
self.tools = create_tool_set() |
|
|
|
|
|
def some_method(self, state): |
|
|
twitter_tool = self.tools.get("scrape_twitter") |
|
|
result = twitter_tool.invoke({"query": "..."}) |
|
|
""" |
|
|
|
|
|
from typing import Dict, Any, Optional, List |
|
|
import logging |
|
|
|
|
|
logger = logging.getLogger("Roger.tool_factory") |
|
|
|
|
|
|
|
|
class ToolSet: |
|
|
""" |
|
|
Encapsulates a complete set of independent tool instances for an agent. |
|
|
|
|
|
Each ToolSet instance contains its own copy of all tools, ensuring |
|
|
that parallel agents don't share state or create race conditions. |
|
|
|
|
|
Thread Safety: |
|
|
Each ToolSet is independent. Multiple agents can safely use |
|
|
their own ToolSet instances in parallel without conflicts. |
|
|
|
|
|
Example: |
|
|
agent1_tools = ToolSet() |
|
|
agent2_tools = ToolSet() |
|
|
|
|
|
# These are independent instances - no shared state |
|
|
agent1_tools.get("scrape_twitter").invoke({...}) |
|
|
agent2_tools.get("scrape_twitter").invoke({...}) # Safe to run in parallel |
|
|
""" |
|
|
|
|
|
def __init__(self, include_profile_scrapers: bool = True): |
|
|
""" |
|
|
Initialize a new ToolSet with fresh tool instances. |
|
|
|
|
|
Args: |
|
|
include_profile_scrapers: Whether to include profile-based scrapers |
|
|
(Twitter profile, LinkedIn profile, etc.) |
|
|
""" |
|
|
self._tools: Dict[str, Any] = {} |
|
|
self._include_profile_scrapers = include_profile_scrapers |
|
|
self._create_tools() |
|
|
logger.debug(f"ToolSet created with {len(self._tools)} tools") |
|
|
|
|
|
def get(self, tool_name: str) -> Optional[Any]: |
|
|
""" |
|
|
Get a tool by name. |
|
|
|
|
|
Args: |
|
|
tool_name: Name of the tool (e.g., "scrape_twitter", "scrape_reddit") |
|
|
|
|
|
Returns: |
|
|
Tool instance if found, None otherwise |
|
|
""" |
|
|
return self._tools.get(tool_name) |
|
|
|
|
|
def as_dict(self) -> Dict[str, Any]: |
|
|
""" |
|
|
Get all tools as a dictionary. |
|
|
|
|
|
Returns: |
|
|
Dictionary mapping tool names to tool instances |
|
|
""" |
|
|
return self._tools.copy() |
|
|
|
|
|
def list_tools(self) -> List[str]: |
|
|
""" |
|
|
List all available tool names. |
|
|
|
|
|
Returns: |
|
|
List of tool names in this ToolSet |
|
|
""" |
|
|
return list(self._tools.keys()) |
|
|
|
|
|
def _create_tools(self) -> None: |
|
|
""" |
|
|
Create fresh instances of all tools. |
|
|
|
|
|
This method imports and creates new tool instances, ensuring |
|
|
each ToolSet has its own independent copies. |
|
|
""" |
|
|
from langchain_core.tools import tool |
|
|
import json |
|
|
from datetime import datetime |
|
|
|
|
|
|
|
|
|
|
|
from src.utils.utils import ( |
|
|
scrape_reddit_impl, |
|
|
scrape_local_news_impl, |
|
|
scrape_cse_stock_impl, |
|
|
scrape_government_gazette_impl, |
|
|
scrape_parliament_minutes_impl, |
|
|
scrape_train_schedule_impl, |
|
|
PLAYWRIGHT_AVAILABLE, |
|
|
ensure_playwright, |
|
|
load_playwright_storage_state_path, |
|
|
create_or_restore_playwright_session, |
|
|
clean_twitter_text, |
|
|
extract_twitter_timestamp, |
|
|
clean_fb_text, |
|
|
clean_linkedin_text, |
|
|
extract_media_id_instagram, |
|
|
fetch_caption_via_private_api, |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_reddit( |
|
|
keywords: List[str], limit: int = 20, subreddit: Optional[str] = None |
|
|
): |
|
|
""" |
|
|
Scrape Reddit for posts matching specific keywords. |
|
|
Optionally restrict to a specific subreddit. |
|
|
""" |
|
|
data = scrape_reddit_impl( |
|
|
keywords=keywords, limit=limit, subreddit=subreddit |
|
|
) |
|
|
return json.dumps(data, default=str) |
|
|
|
|
|
self._tools["scrape_reddit"] = scrape_reddit |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_local_news( |
|
|
keywords: Optional[List[str]] = None, max_articles: int = 30 |
|
|
): |
|
|
""" |
|
|
Scrape local Sri Lankan news from Daily Mirror, Daily FT, and News First. |
|
|
""" |
|
|
data = scrape_local_news_impl(keywords=keywords, max_articles=max_articles) |
|
|
return json.dumps(data, default=str) |
|
|
|
|
|
self._tools["scrape_local_news"] = scrape_local_news |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_cse_stock_data( |
|
|
symbol: str = "ASPI", period: str = "1d", interval: str = "1h" |
|
|
): |
|
|
""" |
|
|
Fetch Colombo Stock Exchange data using yfinance. |
|
|
""" |
|
|
data = scrape_cse_stock_impl( |
|
|
symbol=symbol, period=period, interval=interval |
|
|
) |
|
|
return json.dumps(data, default=str) |
|
|
|
|
|
self._tools["scrape_cse_stock_data"] = scrape_cse_stock_data |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_government_gazette( |
|
|
keywords: Optional[List[str]] = None, max_items: int = 15 |
|
|
): |
|
|
""" |
|
|
Scrape latest government gazettes from gazette.lk. |
|
|
""" |
|
|
data = scrape_government_gazette_impl( |
|
|
keywords=keywords, max_items=max_items |
|
|
) |
|
|
return json.dumps(data, default=str) |
|
|
|
|
|
self._tools["scrape_government_gazette"] = scrape_government_gazette |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_parliament_minutes( |
|
|
keywords: Optional[List[str]] = None, max_items: int = 20 |
|
|
): |
|
|
""" |
|
|
Scrape parliament Hansard and minutes from parliament.lk. |
|
|
""" |
|
|
data = scrape_parliament_minutes_impl( |
|
|
keywords=keywords, max_items=max_items |
|
|
) |
|
|
return json.dumps(data, default=str) |
|
|
|
|
|
self._tools["scrape_parliament_minutes"] = scrape_parliament_minutes |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_train_schedule( |
|
|
from_station: Optional[str] = None, |
|
|
to_station: Optional[str] = None, |
|
|
keyword: Optional[str] = None, |
|
|
max_items: int = 30, |
|
|
): |
|
|
""" |
|
|
Scrape train schedules from railway.gov.lk. |
|
|
""" |
|
|
data = scrape_train_schedule_impl( |
|
|
from_station=from_station, |
|
|
to_station=to_station, |
|
|
keyword=keyword, |
|
|
max_items=max_items, |
|
|
) |
|
|
return json.dumps(data, default=str) |
|
|
|
|
|
self._tools["scrape_train_schedule"] = scrape_train_schedule |
|
|
|
|
|
|
|
|
@tool |
|
|
def think_tool(thought: str) -> str: |
|
|
""" |
|
|
Use this tool to think through complex problems step-by-step. |
|
|
Write out your reasoning process here before taking action. |
|
|
""" |
|
|
return f"Thought recorded: {thought}" |
|
|
|
|
|
self._tools["think_tool"] = think_tool |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if PLAYWRIGHT_AVAILABLE: |
|
|
self._create_playwright_tools() |
|
|
else: |
|
|
logger.warning( |
|
|
"Playwright not available - social media tools will be limited" |
|
|
) |
|
|
self._create_fallback_social_tools() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if self._include_profile_scrapers: |
|
|
self._create_profile_scraper_tools() |
|
|
|
|
|
def _create_playwright_tools(self) -> None: |
|
|
"""Create Playwright-based social media tools.""" |
|
|
from langchain_core.tools import tool |
|
|
import json |
|
|
import os |
|
|
import time |
|
|
import random |
|
|
import re |
|
|
from datetime import datetime |
|
|
from urllib.parse import quote_plus |
|
|
from playwright.sync_api import sync_playwright |
|
|
|
|
|
from src.utils.utils import ( |
|
|
ensure_playwright, |
|
|
load_playwright_storage_state_path, |
|
|
clean_twitter_text, |
|
|
extract_twitter_timestamp, |
|
|
clean_fb_text, |
|
|
clean_linkedin_text, |
|
|
extract_media_id_instagram, |
|
|
fetch_caption_via_private_api, |
|
|
) |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_twitter(query: str = "Sri Lanka", max_items: int = 20): |
|
|
""" |
|
|
Twitter scraper - extracts actual tweet text, author, and metadata using Playwright session. |
|
|
Requires a valid Twitter session file. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
|
|
|
site = "twitter" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
|
|
|
if not session_path: |
|
|
alt_paths = [ |
|
|
os.path.join( |
|
|
os.getcwd(), "src", "utils", ".sessions", "tw_state.json" |
|
|
), |
|
|
os.path.join(os.getcwd(), ".sessions", "tw_state.json"), |
|
|
os.path.join(os.getcwd(), "tw_state.json"), |
|
|
] |
|
|
for path in alt_paths: |
|
|
if os.path.exists(path): |
|
|
session_path = path |
|
|
break |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps( |
|
|
{ |
|
|
"error": "No Twitter session found", |
|
|
"solution": "Run the Twitter session manager to create a session", |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch( |
|
|
headless=True, |
|
|
args=[ |
|
|
"--disable-blink-features=AutomationControlled", |
|
|
"--no-sandbox", |
|
|
"--disable-dev-shm-usage", |
|
|
], |
|
|
) |
|
|
|
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
viewport={"width": 1280, "height": 720}, |
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
|
|
) |
|
|
|
|
|
context.add_init_script( |
|
|
""" |
|
|
Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); |
|
|
window.chrome = {runtime: {}}; |
|
|
""" |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
|
|
|
search_urls = [ |
|
|
f"https://x.com/search?q={quote_plus(query)}&src=typed_query&f=live", |
|
|
f"https://x.com/search?q={quote_plus(query)}&src=typed_query", |
|
|
] |
|
|
|
|
|
success = False |
|
|
for url in search_urls: |
|
|
try: |
|
|
page.goto(url, timeout=60000, wait_until="domcontentloaded") |
|
|
time.sleep(5) |
|
|
|
|
|
|
|
|
popup_selectors = [ |
|
|
"[data-testid='app-bar-close']", |
|
|
"[aria-label='Close']", |
|
|
"button:has-text('Not now')", |
|
|
] |
|
|
for selector in popup_selectors: |
|
|
try: |
|
|
if ( |
|
|
page.locator(selector).count() > 0 |
|
|
and page.locator(selector).first.is_visible() |
|
|
): |
|
|
page.locator(selector).first.click() |
|
|
time.sleep(1) |
|
|
except: |
|
|
pass |
|
|
|
|
|
try: |
|
|
page.wait_for_selector( |
|
|
"article[data-testid='tweet']", timeout=15000 |
|
|
) |
|
|
success = True |
|
|
break |
|
|
except: |
|
|
continue |
|
|
except: |
|
|
continue |
|
|
|
|
|
if not success or "login" in page.url: |
|
|
return json.dumps( |
|
|
{"error": "Session invalid or tweets not found"}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
|
|
|
seen = set() |
|
|
scroll_attempts = 0 |
|
|
max_scroll_attempts = 15 |
|
|
|
|
|
TWEET_SELECTOR = "article[data-testid='tweet']" |
|
|
TEXT_SELECTOR = "div[data-testid='tweetText']" |
|
|
USER_SELECTOR = "div[data-testid='User-Name']" |
|
|
|
|
|
while ( |
|
|
len(results) < max_items |
|
|
and scroll_attempts < max_scroll_attempts |
|
|
): |
|
|
scroll_attempts += 1 |
|
|
|
|
|
|
|
|
try: |
|
|
show_more_buttons = page.locator( |
|
|
"[data-testid='tweet-text-show-more-link']" |
|
|
).all() |
|
|
for button in show_more_buttons: |
|
|
if button.is_visible(): |
|
|
try: |
|
|
button.click() |
|
|
time.sleep(0.3) |
|
|
except: |
|
|
pass |
|
|
except: |
|
|
pass |
|
|
|
|
|
tweets = page.locator(TWEET_SELECTOR).all() |
|
|
new_tweets_found = 0 |
|
|
|
|
|
for tweet in tweets: |
|
|
if len(results) >= max_items: |
|
|
break |
|
|
|
|
|
try: |
|
|
tweet.scroll_into_view_if_needed() |
|
|
time.sleep(0.1) |
|
|
|
|
|
if ( |
|
|
tweet.locator("span:has-text('Promoted')").count() |
|
|
> 0 |
|
|
or tweet.locator("span:has-text('Ad')").count() > 0 |
|
|
): |
|
|
continue |
|
|
|
|
|
text_content = "" |
|
|
text_element = tweet.locator(TEXT_SELECTOR).first |
|
|
if text_element.count() > 0: |
|
|
text_content = text_element.inner_text() |
|
|
|
|
|
cleaned_text = clean_twitter_text(text_content) |
|
|
|
|
|
user_info = "Unknown" |
|
|
user_element = tweet.locator(USER_SELECTOR).first |
|
|
if user_element.count() > 0: |
|
|
user_text = user_element.inner_text() |
|
|
user_info = user_text.split("\n")[0].strip() |
|
|
|
|
|
timestamp = extract_twitter_timestamp(tweet) |
|
|
|
|
|
text_key = cleaned_text[:50] if cleaned_text else "" |
|
|
unique_key = f"{user_info}_{text_key}" |
|
|
|
|
|
if ( |
|
|
cleaned_text |
|
|
and len(cleaned_text) > 20 |
|
|
and unique_key not in seen |
|
|
and not any( |
|
|
word in cleaned_text.lower() |
|
|
for word in ["promoted", "advertisement"] |
|
|
) |
|
|
): |
|
|
|
|
|
seen.add(unique_key) |
|
|
results.append( |
|
|
{ |
|
|
"source": "Twitter", |
|
|
"poster": user_info, |
|
|
"text": cleaned_text, |
|
|
"timestamp": timestamp, |
|
|
"url": "https://x.com", |
|
|
} |
|
|
) |
|
|
new_tweets_found += 1 |
|
|
except: |
|
|
continue |
|
|
|
|
|
if len(results) < max_items: |
|
|
page.evaluate( |
|
|
"window.scrollTo(0, document.documentElement.scrollHeight)" |
|
|
) |
|
|
time.sleep(random.uniform(2, 3)) |
|
|
|
|
|
if new_tweets_found == 0: |
|
|
scroll_attempts += 1 |
|
|
|
|
|
browser.close() |
|
|
|
|
|
return json.dumps( |
|
|
{ |
|
|
"source": "Twitter", |
|
|
"query": query, |
|
|
"results": results, |
|
|
"total_found": len(results), |
|
|
"fetched_at": datetime.utcnow().isoformat(), |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_twitter"] = scrape_twitter |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_linkedin(keywords: Optional[List[str]] = None, max_items: int = 10): |
|
|
""" |
|
|
LinkedIn search using Playwright session. |
|
|
Requires environment variables: LINKEDIN_USER, LINKEDIN_PASSWORD (if creating session). |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "linkedin" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No LinkedIn session found"}, default=str) |
|
|
|
|
|
keyword = " ".join(keywords) if keywords else "Sri Lanka" |
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
|
|
no_viewport=True, |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
url = f"https://www.linkedin.com/search/results/content/?keywords={keyword.replace(' ', '%20')}" |
|
|
|
|
|
try: |
|
|
page.goto(url, timeout=60000, wait_until="domcontentloaded") |
|
|
except: |
|
|
pass |
|
|
|
|
|
page.wait_for_timeout(random.randint(4000, 7000)) |
|
|
|
|
|
try: |
|
|
if ( |
|
|
page.locator("a[href*='login']").is_visible() |
|
|
or "auth_wall" in page.url |
|
|
): |
|
|
return json.dumps({"error": "Session invalid"}) |
|
|
except: |
|
|
pass |
|
|
|
|
|
seen = set() |
|
|
no_new_data_count = 0 |
|
|
previous_height = 0 |
|
|
|
|
|
POST_SELECTOR = "div.feed-shared-update-v2, li.artdeco-card" |
|
|
TEXT_SELECTOR = ( |
|
|
"div.update-components-text span.break-words, span.break-words" |
|
|
) |
|
|
POSTER_SELECTOR = ( |
|
|
"span.update-components-actor__name span[dir='ltr']" |
|
|
) |
|
|
|
|
|
while len(results) < max_items: |
|
|
try: |
|
|
see_more_buttons = page.locator( |
|
|
"button.feed-shared-inline-show-more-text__see-more-less-toggle" |
|
|
).all() |
|
|
for btn in see_more_buttons: |
|
|
if btn.is_visible(): |
|
|
try: |
|
|
btn.click(timeout=500) |
|
|
except: |
|
|
pass |
|
|
except: |
|
|
pass |
|
|
|
|
|
posts = page.locator(POST_SELECTOR).all() |
|
|
|
|
|
for post in posts: |
|
|
if len(results) >= max_items: |
|
|
break |
|
|
try: |
|
|
post.scroll_into_view_if_needed() |
|
|
raw_text = "" |
|
|
text_el = post.locator(TEXT_SELECTOR).first |
|
|
if text_el.is_visible(): |
|
|
raw_text = text_el.inner_text() |
|
|
|
|
|
cleaned_text = clean_linkedin_text(raw_text) |
|
|
poster_name = "(Unknown)" |
|
|
poster_el = post.locator(POSTER_SELECTOR).first |
|
|
if poster_el.is_visible(): |
|
|
poster_name = poster_el.inner_text().strip() |
|
|
|
|
|
key = f"{poster_name[:20]}::{cleaned_text[:30]}" |
|
|
if ( |
|
|
cleaned_text |
|
|
and len(cleaned_text) > 20 |
|
|
and key not in seen |
|
|
): |
|
|
seen.add(key) |
|
|
results.append( |
|
|
{ |
|
|
"source": "LinkedIn", |
|
|
"poster": poster_name, |
|
|
"text": cleaned_text, |
|
|
"url": "https://www.linkedin.com", |
|
|
} |
|
|
) |
|
|
except: |
|
|
continue |
|
|
|
|
|
page.evaluate("window.scrollTo(0, document.body.scrollHeight)") |
|
|
page.wait_for_timeout(random.randint(2000, 4000)) |
|
|
|
|
|
new_height = page.evaluate("document.body.scrollHeight") |
|
|
if new_height == previous_height: |
|
|
no_new_data_count += 1 |
|
|
if no_new_data_count > 3: |
|
|
break |
|
|
else: |
|
|
no_new_data_count = 0 |
|
|
previous_height = new_height |
|
|
|
|
|
browser.close() |
|
|
return json.dumps( |
|
|
{"site": "LinkedIn", "results": results}, default=str |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}) |
|
|
|
|
|
self._tools["scrape_linkedin"] = scrape_linkedin |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_facebook(keywords: Optional[List[str]] = None, max_items: int = 10): |
|
|
""" |
|
|
Facebook scraper using Playwright session (Desktop). |
|
|
Extracts posts from keyword search with poster names and text. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "facebook" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
alt_paths = [ |
|
|
os.path.join( |
|
|
os.getcwd(), "src", "utils", ".sessions", "fb_state.json" |
|
|
), |
|
|
os.path.join(os.getcwd(), ".sessions", "fb_state.json"), |
|
|
] |
|
|
for path in alt_paths: |
|
|
if os.path.exists(path): |
|
|
session_path = path |
|
|
break |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No Facebook session found"}, default=str) |
|
|
|
|
|
keyword = " ".join(keywords) if keywords else "Sri Lanka" |
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
|
|
viewport={"width": 1400, "height": 900}, |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
search_url = f"https://www.facebook.com/search/posts?q={keyword.replace(' ', '%20')}" |
|
|
|
|
|
page.goto(search_url, timeout=120000) |
|
|
time.sleep(5) |
|
|
|
|
|
seen = set() |
|
|
stuck = 0 |
|
|
last_scroll = 0 |
|
|
|
|
|
MESSAGE_SELECTOR = "div[data-ad-preview='message']" |
|
|
|
|
|
POSTER_SELECTORS = [ |
|
|
"h3 strong a span", |
|
|
"h3 strong span", |
|
|
"strong a span", |
|
|
"a[role='link'] span", |
|
|
] |
|
|
|
|
|
def extract_poster(post): |
|
|
parent = post.locator( |
|
|
"xpath=ancestor::div[contains(@class, 'x1yztbdb')][1]" |
|
|
) |
|
|
for selector in POSTER_SELECTORS: |
|
|
try: |
|
|
el = parent.locator(selector).first |
|
|
if el and el.count() > 0: |
|
|
name = el.inner_text().strip() |
|
|
if name and name != "Facebook" and len(name) > 1: |
|
|
return name |
|
|
except: |
|
|
pass |
|
|
return "(Unknown)" |
|
|
|
|
|
while len(results) < max_items: |
|
|
posts = page.locator(MESSAGE_SELECTOR).all() |
|
|
|
|
|
for post in posts: |
|
|
try: |
|
|
raw = post.inner_text().strip() |
|
|
cleaned = clean_fb_text(raw) |
|
|
poster = extract_poster(post) |
|
|
|
|
|
if cleaned and len(cleaned) > 30: |
|
|
key = poster + "::" + cleaned |
|
|
if key not in seen: |
|
|
seen.add(key) |
|
|
results.append( |
|
|
{ |
|
|
"source": "Facebook", |
|
|
"poster": poster, |
|
|
"text": cleaned, |
|
|
"url": "https://www.facebook.com", |
|
|
} |
|
|
) |
|
|
|
|
|
if len(results) >= max_items: |
|
|
break |
|
|
except: |
|
|
pass |
|
|
|
|
|
page.evaluate("window.scrollBy(0, 2300)") |
|
|
time.sleep(1.2) |
|
|
|
|
|
new_scroll = page.evaluate("window.scrollY") |
|
|
stuck = stuck + 1 if new_scroll == last_scroll else 0 |
|
|
last_scroll = new_scroll |
|
|
|
|
|
if stuck >= 3: |
|
|
break |
|
|
|
|
|
browser.close() |
|
|
return json.dumps( |
|
|
{"site": "Facebook", "results": results[:max_items]}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_facebook"] = scrape_facebook |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_instagram(keywords: Optional[List[str]] = None, max_items: int = 15): |
|
|
""" |
|
|
Instagram scraper using Playwright session. |
|
|
Scrapes posts from hashtag search and extracts captions. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "instagram" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
alt_paths = [ |
|
|
os.path.join( |
|
|
os.getcwd(), "src", "utils", ".sessions", "ig_state.json" |
|
|
), |
|
|
os.path.join(os.getcwd(), ".sessions", "ig_state.json"), |
|
|
] |
|
|
for path in alt_paths: |
|
|
if os.path.exists(path): |
|
|
session_path = path |
|
|
break |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No Instagram session found"}, default=str) |
|
|
|
|
|
keyword = " ".join(keywords) if keywords else "srilanka" |
|
|
keyword = keyword.replace(" ", "") |
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
user_agent="Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15", |
|
|
viewport={"width": 430, "height": 932}, |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
url = f"https://www.instagram.com/explore/tags/{keyword}/" |
|
|
|
|
|
page.goto(url, timeout=120000) |
|
|
page.wait_for_timeout(4000) |
|
|
|
|
|
for _ in range(12): |
|
|
page.mouse.wheel(0, 2500) |
|
|
page.wait_for_timeout(1500) |
|
|
|
|
|
anchors = page.locator("a[href*='/p/'], a[href*='/reel/']").all() |
|
|
links = [] |
|
|
|
|
|
for a in anchors: |
|
|
href = a.get_attribute("href") |
|
|
if href: |
|
|
full = "https://www.instagram.com" + href |
|
|
links.append(full) |
|
|
if len(links) >= max_items: |
|
|
break |
|
|
|
|
|
for link in links: |
|
|
page.goto(link, timeout=120000) |
|
|
page.wait_for_timeout(2000) |
|
|
|
|
|
media_id = extract_media_id_instagram(page) |
|
|
caption = fetch_caption_via_private_api(page, media_id) |
|
|
|
|
|
if not caption: |
|
|
try: |
|
|
caption = ( |
|
|
page.locator("article h1, article span") |
|
|
.first.inner_text() |
|
|
.strip() |
|
|
) |
|
|
except: |
|
|
caption = None |
|
|
|
|
|
if caption: |
|
|
results.append( |
|
|
{ |
|
|
"source": "Instagram", |
|
|
"text": caption, |
|
|
"url": link, |
|
|
"poster": "(Instagram User)", |
|
|
} |
|
|
) |
|
|
|
|
|
browser.close() |
|
|
return json.dumps( |
|
|
{"site": "Instagram", "results": results}, default=str |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_instagram"] = scrape_instagram |
|
|
|
|
|
def _create_fallback_social_tools(self) -> None: |
|
|
"""Create fallback tools when Playwright is not available.""" |
|
|
from langchain_core.tools import tool |
|
|
import json |
|
|
|
|
|
@tool |
|
|
def scrape_twitter(query: str = "Sri Lanka", max_items: int = 20): |
|
|
"""Twitter scraper (requires Playwright).""" |
|
|
return json.dumps( |
|
|
{"error": "Playwright not available for Twitter scraping"} |
|
|
) |
|
|
|
|
|
@tool |
|
|
def scrape_linkedin(keywords: Optional[List[str]] = None, max_items: int = 10): |
|
|
"""LinkedIn scraper (requires Playwright).""" |
|
|
return json.dumps( |
|
|
{"error": "Playwright not available for LinkedIn scraping"} |
|
|
) |
|
|
|
|
|
@tool |
|
|
def scrape_facebook(keywords: Optional[List[str]] = None, max_items: int = 10): |
|
|
"""Facebook scraper (requires Playwright).""" |
|
|
return json.dumps( |
|
|
{"error": "Playwright not available for Facebook scraping"} |
|
|
) |
|
|
|
|
|
@tool |
|
|
def scrape_instagram(keywords: Optional[List[str]] = None, max_items: int = 15): |
|
|
"""Instagram scraper (requires Playwright).""" |
|
|
return json.dumps( |
|
|
{"error": "Playwright not available for Instagram scraping"} |
|
|
) |
|
|
|
|
|
self._tools["scrape_twitter"] = scrape_twitter |
|
|
self._tools["scrape_linkedin"] = scrape_linkedin |
|
|
self._tools["scrape_facebook"] = scrape_facebook |
|
|
self._tools["scrape_instagram"] = scrape_instagram |
|
|
|
|
|
def _create_profile_scraper_tools(self) -> None: |
|
|
"""Create profile-based scraper tools for competitive intelligence.""" |
|
|
from langchain_core.tools import tool |
|
|
import json |
|
|
import os |
|
|
import time |
|
|
import random |
|
|
import re |
|
|
from datetime import datetime |
|
|
|
|
|
from src.utils.utils import ( |
|
|
PLAYWRIGHT_AVAILABLE, |
|
|
ensure_playwright, |
|
|
load_playwright_storage_state_path, |
|
|
clean_twitter_text, |
|
|
extract_twitter_timestamp, |
|
|
clean_fb_text, |
|
|
extract_media_id_instagram, |
|
|
fetch_caption_via_private_api, |
|
|
) |
|
|
|
|
|
if not PLAYWRIGHT_AVAILABLE: |
|
|
return |
|
|
|
|
|
from playwright.sync_api import sync_playwright |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_twitter_profile(username: str, max_items: int = 20): |
|
|
""" |
|
|
Twitter PROFILE scraper - targets a specific user's timeline. |
|
|
Perfect for monitoring competitor accounts, influencers, or business profiles. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "twitter" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
alt_paths = [ |
|
|
os.path.join( |
|
|
os.getcwd(), "src", "utils", ".sessions", "tw_state.json" |
|
|
), |
|
|
os.path.join(os.getcwd(), ".sessions", "tw_state.json"), |
|
|
] |
|
|
for path in alt_paths: |
|
|
if os.path.exists(path): |
|
|
session_path = path |
|
|
break |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No Twitter session found"}, default=str) |
|
|
|
|
|
results = [] |
|
|
username = username.lstrip("@") |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True, args=["--no-sandbox"]) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
viewport={"width": 1280, "height": 720}, |
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
profile_url = f"https://x.com/{username}" |
|
|
|
|
|
try: |
|
|
page.goto( |
|
|
profile_url, timeout=60000, wait_until="domcontentloaded" |
|
|
) |
|
|
time.sleep(5) |
|
|
|
|
|
try: |
|
|
page.wait_for_selector( |
|
|
"article[data-testid='tweet']", timeout=15000 |
|
|
) |
|
|
except: |
|
|
return json.dumps( |
|
|
{"error": f"Profile not found or private: @{username}"} |
|
|
) |
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}) |
|
|
|
|
|
if "login" in page.url: |
|
|
return json.dumps({"error": "Session expired"}) |
|
|
|
|
|
seen = set() |
|
|
scroll_attempts = 0 |
|
|
|
|
|
while len(results) < max_items and scroll_attempts < 10: |
|
|
scroll_attempts += 1 |
|
|
|
|
|
tweets = page.locator("article[data-testid='tweet']").all() |
|
|
|
|
|
for tweet in tweets: |
|
|
if len(results) >= max_items: |
|
|
break |
|
|
|
|
|
try: |
|
|
tweet.scroll_into_view_if_needed() |
|
|
|
|
|
if ( |
|
|
tweet.locator("span:has-text('Promoted')").count() |
|
|
> 0 |
|
|
): |
|
|
continue |
|
|
|
|
|
text_content = "" |
|
|
text_element = tweet.locator( |
|
|
"div[data-testid='tweetText']" |
|
|
).first |
|
|
if text_element.count() > 0: |
|
|
text_content = text_element.inner_text() |
|
|
|
|
|
cleaned_text = clean_twitter_text(text_content) |
|
|
timestamp = extract_twitter_timestamp(tweet) |
|
|
|
|
|
|
|
|
likes = 0 |
|
|
try: |
|
|
like_button = tweet.locator("[data-testid='like']") |
|
|
if like_button.count() > 0: |
|
|
like_text = ( |
|
|
like_button.first.get_attribute( |
|
|
"aria-label" |
|
|
) |
|
|
or "" |
|
|
) |
|
|
like_match = re.search(r"(\d+)", like_text) |
|
|
if like_match: |
|
|
likes = int(like_match.group(1)) |
|
|
except: |
|
|
pass |
|
|
|
|
|
text_key = cleaned_text[:50] if cleaned_text else "" |
|
|
unique_key = f"{username}_{text_key}_{timestamp}" |
|
|
|
|
|
if ( |
|
|
cleaned_text |
|
|
and len(cleaned_text) > 20 |
|
|
and unique_key not in seen |
|
|
): |
|
|
seen.add(unique_key) |
|
|
results.append( |
|
|
{ |
|
|
"source": "Twitter", |
|
|
"poster": f"@{username}", |
|
|
"text": cleaned_text, |
|
|
"timestamp": timestamp, |
|
|
"url": profile_url, |
|
|
"likes": likes, |
|
|
} |
|
|
) |
|
|
except: |
|
|
continue |
|
|
|
|
|
if len(results) < max_items: |
|
|
page.evaluate( |
|
|
"window.scrollTo(0, document.documentElement.scrollHeight)" |
|
|
) |
|
|
time.sleep(random.uniform(2, 3)) |
|
|
|
|
|
browser.close() |
|
|
|
|
|
return json.dumps( |
|
|
{ |
|
|
"site": "Twitter Profile", |
|
|
"username": username, |
|
|
"results": results, |
|
|
"total_found": len(results), |
|
|
"fetched_at": datetime.utcnow().isoformat(), |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_twitter_profile"] = scrape_twitter_profile |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_facebook_profile(profile_url: str, max_items: int = 10): |
|
|
""" |
|
|
Facebook PROFILE scraper - monitors a specific page or user profile. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "facebook" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No Facebook session found"}, default=str) |
|
|
|
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
|
|
viewport={"width": 1400, "height": 900}, |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
page.goto(profile_url, timeout=120000) |
|
|
time.sleep(5) |
|
|
|
|
|
if "login" in page.url: |
|
|
return json.dumps({"error": "Session expired"}) |
|
|
|
|
|
seen = set() |
|
|
stuck = 0 |
|
|
last_scroll = 0 |
|
|
|
|
|
MESSAGE_SELECTOR = "div[data-ad-preview='message']" |
|
|
|
|
|
while len(results) < max_items: |
|
|
posts = page.locator(MESSAGE_SELECTOR).all() |
|
|
|
|
|
for post in posts: |
|
|
try: |
|
|
raw = post.inner_text().strip() |
|
|
cleaned = clean_fb_text(raw) |
|
|
|
|
|
if ( |
|
|
cleaned |
|
|
and len(cleaned) > 30 |
|
|
and cleaned not in seen |
|
|
): |
|
|
seen.add(cleaned) |
|
|
results.append( |
|
|
{ |
|
|
"source": "Facebook", |
|
|
"text": cleaned, |
|
|
"url": profile_url, |
|
|
} |
|
|
) |
|
|
|
|
|
if len(results) >= max_items: |
|
|
break |
|
|
except: |
|
|
pass |
|
|
|
|
|
page.evaluate("window.scrollBy(0, 2300)") |
|
|
time.sleep(1.5) |
|
|
|
|
|
new_scroll = page.evaluate("window.scrollY") |
|
|
stuck = stuck + 1 if new_scroll == last_scroll else 0 |
|
|
last_scroll = new_scroll |
|
|
|
|
|
if stuck >= 3: |
|
|
break |
|
|
|
|
|
browser.close() |
|
|
return json.dumps( |
|
|
{ |
|
|
"site": "Facebook Profile", |
|
|
"profile_url": profile_url, |
|
|
"results": results[:max_items], |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_facebook_profile"] = scrape_facebook_profile |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_instagram_profile(username: str, max_items: int = 15): |
|
|
""" |
|
|
Instagram PROFILE scraper - monitors a specific user's profile. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "instagram" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No Instagram session found"}, default=str) |
|
|
|
|
|
username = username.lstrip("@") |
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
user_agent="Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15", |
|
|
viewport={"width": 430, "height": 932}, |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
url = f"https://www.instagram.com/{username}/" |
|
|
|
|
|
page.goto(url, timeout=120000) |
|
|
page.wait_for_timeout(4000) |
|
|
|
|
|
if "login" in page.url: |
|
|
return json.dumps({"error": "Session expired"}) |
|
|
|
|
|
for _ in range(8): |
|
|
page.mouse.wheel(0, 2500) |
|
|
page.wait_for_timeout(1500) |
|
|
|
|
|
anchors = page.locator("a[href*='/p/'], a[href*='/reel/']").all() |
|
|
links = [] |
|
|
|
|
|
for a in anchors: |
|
|
href = a.get_attribute("href") |
|
|
if href: |
|
|
full = "https://www.instagram.com" + href |
|
|
links.append(full) |
|
|
if len(links) >= max_items: |
|
|
break |
|
|
|
|
|
for link in links: |
|
|
page.goto(link, timeout=120000) |
|
|
page.wait_for_timeout(2000) |
|
|
|
|
|
media_id = extract_media_id_instagram(page) |
|
|
caption = fetch_caption_via_private_api(page, media_id) |
|
|
|
|
|
if not caption: |
|
|
try: |
|
|
caption = ( |
|
|
page.locator("article h1, article span") |
|
|
.first.inner_text() |
|
|
.strip() |
|
|
) |
|
|
except: |
|
|
caption = None |
|
|
|
|
|
if caption: |
|
|
results.append( |
|
|
{ |
|
|
"source": "Instagram", |
|
|
"poster": f"@{username}", |
|
|
"text": caption, |
|
|
"url": link, |
|
|
} |
|
|
) |
|
|
|
|
|
browser.close() |
|
|
return json.dumps( |
|
|
{ |
|
|
"site": "Instagram Profile", |
|
|
"username": username, |
|
|
"results": results, |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_instagram_profile"] = scrape_instagram_profile |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_linkedin_profile(company_or_username: str, max_items: int = 10): |
|
|
""" |
|
|
LinkedIn PROFILE scraper - monitors a company or user profile. |
|
|
""" |
|
|
ensure_playwright() |
|
|
|
|
|
site = "linkedin" |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir="src/utils/.sessions" |
|
|
) |
|
|
if not session_path: |
|
|
session_path = load_playwright_storage_state_path( |
|
|
site, out_dir=".sessions" |
|
|
) |
|
|
|
|
|
if not session_path: |
|
|
return json.dumps({"error": "No LinkedIn session found"}, default=str) |
|
|
|
|
|
results = [] |
|
|
|
|
|
try: |
|
|
with sync_playwright() as p: |
|
|
browser = p.chromium.launch(headless=True) |
|
|
context = browser.new_context( |
|
|
storage_state=session_path, |
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
|
|
viewport={"width": 1400, "height": 900}, |
|
|
) |
|
|
|
|
|
page = context.new_page() |
|
|
|
|
|
if not company_or_username.startswith("http"): |
|
|
if "company/" in company_or_username: |
|
|
profile_url = f"https://www.linkedin.com/company/{company_or_username.replace('company/', '')}" |
|
|
else: |
|
|
profile_url = ( |
|
|
f"https://www.linkedin.com/in/{company_or_username}" |
|
|
) |
|
|
else: |
|
|
profile_url = company_or_username |
|
|
|
|
|
page.goto(profile_url, timeout=120000) |
|
|
page.wait_for_timeout(5000) |
|
|
|
|
|
if "login" in page.url or "authwall" in page.url: |
|
|
return json.dumps({"error": "Session expired"}) |
|
|
|
|
|
|
|
|
try: |
|
|
posts_tab = page.locator("a:has-text('Posts')").first |
|
|
if posts_tab.is_visible(): |
|
|
posts_tab.click() |
|
|
page.wait_for_timeout(3000) |
|
|
except: |
|
|
pass |
|
|
|
|
|
seen = set() |
|
|
no_new_data_count = 0 |
|
|
previous_height = 0 |
|
|
|
|
|
while len(results) < max_items and no_new_data_count < 3: |
|
|
posts = page.locator("div.feed-shared-update-v2").all() |
|
|
|
|
|
for post in posts: |
|
|
if len(results) >= max_items: |
|
|
break |
|
|
try: |
|
|
post.scroll_into_view_if_needed() |
|
|
text_el = post.locator("span.break-words").first |
|
|
if text_el.is_visible(): |
|
|
raw_text = text_el.inner_text() |
|
|
|
|
|
from src.utils.utils import clean_linkedin_text |
|
|
|
|
|
cleaned = clean_linkedin_text(raw_text) |
|
|
|
|
|
if ( |
|
|
cleaned |
|
|
and len(cleaned) > 20 |
|
|
and cleaned[:50] not in seen |
|
|
): |
|
|
seen.add(cleaned[:50]) |
|
|
results.append( |
|
|
{ |
|
|
"source": "LinkedIn", |
|
|
"text": cleaned, |
|
|
"url": profile_url, |
|
|
} |
|
|
) |
|
|
except: |
|
|
continue |
|
|
|
|
|
page.evaluate("window.scrollTo(0, document.body.scrollHeight)") |
|
|
page.wait_for_timeout(random.randint(2000, 4000)) |
|
|
|
|
|
new_height = page.evaluate("document.body.scrollHeight") |
|
|
if new_height == previous_height: |
|
|
no_new_data_count += 1 |
|
|
else: |
|
|
no_new_data_count = 0 |
|
|
previous_height = new_height |
|
|
|
|
|
browser.close() |
|
|
return json.dumps( |
|
|
{ |
|
|
"site": "LinkedIn Profile", |
|
|
"profile": company_or_username, |
|
|
"results": results, |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}, default=str) |
|
|
|
|
|
self._tools["scrape_linkedin_profile"] = scrape_linkedin_profile |
|
|
|
|
|
|
|
|
@tool |
|
|
def scrape_product_reviews( |
|
|
product_keyword: str, |
|
|
platforms: Optional[List[str]] = None, |
|
|
max_items: int = 10, |
|
|
): |
|
|
""" |
|
|
Multi-platform product review aggregator for competitive intelligence. |
|
|
""" |
|
|
if platforms is None: |
|
|
platforms = ["reddit", "twitter"] |
|
|
|
|
|
all_reviews = [] |
|
|
|
|
|
|
|
|
if "reddit" in platforms: |
|
|
try: |
|
|
reddit_tool = self._tools.get("scrape_reddit") |
|
|
if reddit_tool: |
|
|
reddit_data = reddit_tool.invoke( |
|
|
{ |
|
|
"keywords": [ |
|
|
f"{product_keyword} review", |
|
|
product_keyword, |
|
|
], |
|
|
"limit": max_items, |
|
|
} |
|
|
) |
|
|
|
|
|
reddit_results = ( |
|
|
json.loads(reddit_data) |
|
|
if isinstance(reddit_data, str) |
|
|
else reddit_data |
|
|
) |
|
|
for item in reddit_results: |
|
|
if isinstance(item, dict): |
|
|
all_reviews.append( |
|
|
{ |
|
|
"platform": "Reddit", |
|
|
"text": item.get("title", "") |
|
|
+ " " |
|
|
+ item.get("selftext", ""), |
|
|
"url": item.get("url", ""), |
|
|
} |
|
|
) |
|
|
except: |
|
|
pass |
|
|
|
|
|
|
|
|
if "twitter" in platforms: |
|
|
try: |
|
|
twitter_tool = self._tools.get("scrape_twitter") |
|
|
if twitter_tool: |
|
|
twitter_data = twitter_tool.invoke( |
|
|
{ |
|
|
"query": f"{product_keyword} review", |
|
|
"max_items": max_items, |
|
|
} |
|
|
) |
|
|
|
|
|
twitter_results = ( |
|
|
json.loads(twitter_data) |
|
|
if isinstance(twitter_data, str) |
|
|
else twitter_data |
|
|
) |
|
|
if ( |
|
|
isinstance(twitter_results, dict) |
|
|
and "results" in twitter_results |
|
|
): |
|
|
for item in twitter_results["results"]: |
|
|
all_reviews.append( |
|
|
{ |
|
|
"platform": "Twitter", |
|
|
"text": item.get("text", ""), |
|
|
"url": item.get("url", ""), |
|
|
} |
|
|
) |
|
|
except: |
|
|
pass |
|
|
|
|
|
return json.dumps( |
|
|
{ |
|
|
"product": product_keyword, |
|
|
"total_reviews": len(all_reviews), |
|
|
"reviews": all_reviews, |
|
|
"platforms_searched": platforms, |
|
|
}, |
|
|
default=str, |
|
|
) |
|
|
|
|
|
self._tools["scrape_product_reviews"] = scrape_product_reviews |
|
|
|
|
|
|
|
|
def create_tool_set(include_profile_scrapers: bool = True) -> ToolSet: |
|
|
""" |
|
|
Factory function to create a new ToolSet with independent tool instances. |
|
|
|
|
|
This is the primary entry point for creating tools for an agent. |
|
|
Each call creates a completely independent set of tools. |
|
|
|
|
|
Args: |
|
|
include_profile_scrapers: Whether to include profile-based scrapers |
|
|
|
|
|
Returns: |
|
|
A new ToolSet instance with fresh tool instances |
|
|
|
|
|
Example: |
|
|
# In an agent node |
|
|
class MyAgentNode: |
|
|
def __init__(self): |
|
|
self.tools = create_tool_set() |
|
|
|
|
|
def process(self, state): |
|
|
twitter = self.tools.get("scrape_twitter") |
|
|
result = twitter.invoke({"query": "..."}) |
|
|
""" |
|
|
return ToolSet(include_profile_scrapers=include_profile_scrapers) |
|
|
|
|
|
|
|
|
|
|
|
__all__ = [ |
|
|
"ToolSet", |
|
|
"create_tool_set", |
|
|
] |
|
|
|
