import asyncio import time from typing import List, Dict, Any from datetime import datetime import logging from urllib.parse import quote_plus from ...shared.selenium_base.selenium_manager import SeleniumManager from ...shared.kafka_client.kafka_client import SearchKafkaClient from ...shared.common_models.search_models import SearchResult, SearchResponse from ..core.config import settings logger = logging.getLogger(__name__) class SocialMediaSearchService: def __init__(self): self.kafka_client = SearchKafkaClient( bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS, consumer_group=settings.KAFKA_CONSUMER_GROUP ) async def process_search_task(self, message: Dict[str, Any]): """Process a social media search task""" try: task_id = message.get("id") keywords = message.get("keywords", []) logger.info(f"Processing social media search task: {task_id}") start_time = time.time() # Search multiple platforms all_results = [] # Search Twitter/X (if accessible) twitter_results = await self.search_twitter(keywords) all_results.extend(twitter_results) # Search LinkedIn (public posts) linkedin_results = await self.search_linkedin(keywords) all_results.extend(linkedin_results) # Search Instagram (public posts) instagram_results = await self.search_instagram(keywords) all_results.extend(instagram_results) processing_time = time.time() - start_time response = SearchResponse( task_id=task_id, platform="social_media", keywords=keywords, results=all_results, total_results=len(all_results), processing_time=processing_time, timestamp=datetime.utcnow() ) await self.kafka_client.send_search_result( task_id=task_id, platform="social_media", results=response.dict() ) logger.info(f"Social media search completed: {task_id}, found {len(all_results)} results") except Exception as e: logger.error(f"Error processing social media search task: {e}") await self.kafka_client.send_error_result( task_id=message.get("id", "unknown"), platform="social_media", error=str(e) ) async def search_twitter(self, keywords: List[str], max_results: int = 10) -> List[SearchResult]: """Search Twitter/X (limited due to API restrictions)""" results = [] try: async with SeleniumManager(headless=True).driver_context() as selenium: query = " ".join(keywords) encoded_query = quote_plus(query) # Note: Twitter search without login is very limited search_url = f"https://twitter.com/search?q={encoded_query}&src=typed_query" logger.info(f"Searching Twitter for: {query}") await selenium.navigate_to(search_url) await selenium.wait_for_page_load() await asyncio.sleep(3) # Twitter often requires login, so this might not work tweet_elements = await selenium.find_elements("[data-testid='tweet']") for element in tweet_elements[:max_results]: try: result = await self._extract_twitter_result(element) if result: results.append(result) except Exception as e: logger.warning(f"Error extracting Twitter result: {e}") continue except Exception as e: logger.warning(f"Error searching Twitter: {e}") return results async def search_linkedin(self, keywords: List[str], max_results: int = 10) -> List[SearchResult]: """Search LinkedIn public posts""" results = [] try: async with SeleniumManager(headless=True).driver_context() as selenium: query = " ".join(keywords) encoded_query = quote_plus(query) # LinkedIn public search search_url = f"https://www.linkedin.com/search/results/content/?keywords={encoded_query}" logger.info(f"Searching LinkedIn for: {query}") await selenium.navigate_to(search_url) await selenium.wait_for_page_load() await asyncio.sleep(3) # LinkedIn often requires login for detailed results post_elements = await selenium.find_elements(".feed-shared-update-v2") for element in post_elements[:max_results]: try: result = await self._extract_linkedin_result(element) if result: results.append(result) except Exception as e: logger.warning(f"Error extracting LinkedIn result: {e}") continue except Exception as e: logger.warning(f"Error searching LinkedIn: {e}") return results async def search_instagram(self, keywords: List[str], max_results: int = 10) -> List[SearchResult]: """Search Instagram public posts""" results = [] try: async with SeleniumManager(headless=True).driver_context() as selenium: query = " ".join(keywords) encoded_query = quote_plus(query) # Instagram hashtag search (public) hashtag = keywords[0] if keywords else "general" search_url = f"https://www.instagram.com/explore/tags/{hashtag}/" logger.info(f"Searching Instagram for hashtag: {hashtag}") await selenium.navigate_to(search_url) await selenium.wait_for_page_load() await asyncio.sleep(3) # Instagram posts post_elements = await selenium.find_elements("article a") for element in post_elements[:max_results]: try: result = await self._extract_instagram_result(element, hashtag) if result: results.append(result) except Exception as e: logger.warning(f"Error extracting Instagram result: {e}") continue except Exception as e: logger.warning(f"Error searching Instagram: {e}") return results async def _extract_twitter_result(self, element) -> SearchResult: """Extract Twitter result""" try: # Extract tweet text text_element = element.find_element("css selector", "[data-testid='tweetText']") content = text_element.text if text_element else "" # Extract author author_element = element.find_element("css selector", "[data-testid='User-Names'] a") author = author_element.text if author_element else "" # Extract URL url_element = element.find_element("css selector", "a[href*='/status/']") url = url_element.get_attribute("href") if url_element else "" return SearchResult( title=f"Tweet by {author}", url=url, description=content, snippet=content[:200] + "..." if len(content) > 200 else content, author=author, source="twitter", metadata={ "platform": "twitter", "extracted_at": datetime.utcnow().isoformat() } ) except Exception as e: logger.warning(f"Error extracting Twitter result: {e}") return None async def _extract_linkedin_result(self, element) -> SearchResult: """Extract LinkedIn result""" try: # Extract post content content_element = element.find_element("css selector", ".feed-shared-text") content = content_element.text if content_element else "" # Extract author author_element = element.find_element("css selector", ".feed-shared-actor__name") author = author_element.text if author_element else "" return SearchResult( title=f"LinkedIn post by {author}", url="", # LinkedIn URLs are complex description=content, snippet=content[:200] + "..." if len(content) > 200 else content, author=author, source="linkedin", metadata={ "platform": "linkedin", "extracted_at": datetime.utcnow().isoformat() } ) except Exception as e: logger.warning(f"Error extracting LinkedIn result: {e}") return None async def _extract_instagram_result(self, element, hashtag: str) -> SearchResult: """Extract Instagram result""" try: url = element.get_attribute("href") if element else "" if url and not url.startswith("http"): url = f"https://www.instagram.com{url}" return SearchResult( title=f"Instagram post - #{hashtag}", url=url, description=f"Instagram post tagged with #{hashtag}", snippet=f"Instagram post tagged with #{hashtag}", source="instagram", metadata={ "platform": "instagram", "hashtag": hashtag, "extracted_at": datetime.utcnow().isoformat() } ) except Exception as e: logger.warning(f"Error extracting Instagram result: {e}") return None