FEAT: Add LinkedIn data retrieval and resolve test failures for data agents

Author: FardinHash

outputs/reports/Alibaba_analysis_report.pdf

@@ -3,23 +3,53 @@
 
 class GoogleSearchAgent:
     def get_google_data(self, query):
+        """
+        Retrieve summarized data from Google search results.
+        """
         url = f"https://www.google.com/search?q={query}"
         headers = {"User-Agent": "Mozilla/5.0"}
         try:
             response = requests.get(url, headers=headers, timeout=10)
             response.raise_for_status()
             soup = BeautifulSoup(response.text, "html.parser")
-            results = [a.text for a in soup.select("a")]
-            return {"google_results": results[:5]}  
+            results = [a.text for a in soup.select("a") if a.text.strip()]
+            return {"google_summary": " ".join(results[:5])}  
         except requests.exceptions.RequestException as e:
             return {"error": f"Google Search error: {e}"}
 
 class WikipediaAgent:
     def get_wikipedia_data(self, query):
+        """
+        Retrieve summarized data from Wikipedia.
+        """
         url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{query}"
         try:
             response = requests.get(url, timeout=10)
             response.raise_for_status()
-            return response.json()  
+            data = response.json()
+            return {"wikipedia_summary": data.get("extract", "No summary available")}
         except requests.exceptions.RequestException as e:
             return {"error": f"Wikipedia API error: {e}"}
+
+class LinkedInSearchAgent:
+    def get_linkedin_data(self, query):
+        """
+        Retrieve company-related information from LinkedIn.
+        """
+        url = f"https://www.linkedin.com/search/results/companies/?keywords={query}"
+        headers = {"User-Agent": "Mozilla/5.0"}
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+            response.raise_for_status()
+            soup = BeautifulSoup(response.text, "html.parser")
+
+            results = [
+                item.text.strip()
+                for item in soup.select("div.entity-result__primary-subtitle")
+            ]
+            if results:
+                return {"linkedin_summary": " ".join(results[:3])}
+            else:
+                return {"linkedin_summary": "No data available"}
+        except requests.exceptions.RequestException as e:
+            return {"error": f"LinkedIn Search error: {e}"}

src/agents/data_retrieval.py

@@ -11,16 +11,23 @@ def __init__(self):
 
     def analyze(self, data, company_name):
         openai.api_key = self.api_key
+
+        google_summary = data.get("google_summary", "No data available")
+        wikipedia_summary = data.get("wikipedia_summary", "No data available")
+        linkedin_summary = data.get("linkedin_summary", "No data available")
+        
         prompt = (
-            f"Perform a detailed SWOT analysis for the company '{company_name}' based on the following information:\n\n"
-            f"Google Summary:\n{data.get('google_summary', 'No data available')}\n\n"
-            f"Wikipedia Summary:\n{data.get('wikipedia_summary', 'No data available')}\n\n"
+            f"Perform a detailed SWOT analysis for the company '{company_name}' using the following data:\n\n"
+            f"Google Summary:\n{google_summary}\n\n"
+            f"Wikipedia Summary:\n{wikipedia_summary}\n\n"
+            f"LinkedIn Summary:\n{linkedin_summary}\n\n"
             "Provide the analysis in the format:\n"
             "Strengths:\n- Point 1\n- Point 2\n\n"
             "Weaknesses:\n- Point 1\n- Point 2\n\n"
             "Opportunities:\n- Point 1\n- Point 2\n\n"
             "Threats:\n- Point 1\n- Point 2"
         )
+
         try:
             response = openai.ChatCompletion.create(
                 model="gpt-4",

src/agents/swot_analysis.py

@@ -1,4 +1,4 @@
-from agents.data_retrieval import GoogleSearchAgent, WikipediaAgent
+from agents.data_retrieval import GoogleSearchAgent, WikipediaAgent, LinkedInSearchAgent
 from utils.normalization import normalize_data
 from agents.nlp_processing import NLPProcessingAgent
 from agents.swot_analysis import SWOTAnalysisAgent
@@ -8,6 +8,7 @@ class Orchestrator:
     def __init__(self):
         self.google_agent = GoogleSearchAgent()
         self.wikipedia_agent = WikipediaAgent()
+        self.linkedin_agent = LinkedInSearchAgent()
         self.nlp_agent = NLPProcessingAgent()
         self.swot_agent = SWOTAnalysisAgent()
         self.report_agent = ReportGenerator()
@@ -16,9 +17,10 @@ async def run(self, query):
         # Step 1: Retrieve data from alternative sources
         google_data = self.google_agent.get_google_data(query)
         wikipedia_data = self.wikipedia_agent.get_wikipedia_data(query)
+        linkedin_data = self.linkedin_agent.get_linkedin_data(query)
 
         # Step 2: Normalize data
-        raw_data = {"google": google_data, "wikipedia": wikipedia_data}
+        raw_data = {"google": google_data, "wikipedia": wikipedia_data, "linkedin": linkedin_data,}
         normalized_data = normalize_data(raw_data)
 
         # Step 3: Process data

src/orchestrator.py

@@ -6,14 +6,20 @@ def normalize_data(raw_data):
 
     google_data = raw_data.get("google", {})
     if "error" not in google_data:
-        normalized_data["google_summary"] = google_data.get("google_results", [])
+        normalized_data["google_summary"] = google_data.get("google_summary", [])
     else:
         normalized_data["google_summary"] = google_data.get("error")
 
     wikipedia_data = raw_data.get("wikipedia", {})
     if "error" not in wikipedia_data:
-        normalized_data["wikipedia_summary"] = wikipedia_data.get("extract", "No summary available")
+        normalized_data["wikipedia_summary"] = wikipedia_data.get("wikipedia_summary", "No summary available")
     else:
         normalized_data["wikipedia_summary"] = wikipedia_data.get("error")
 
+    linkedin_data = raw_data.get("linkedin", {})
+    if "error" not in linkedin_data:
+        normalized_data["linkedin_summary"] = linkedin_data.get("linkedin_summary", "No data available")
+    else:
+        normalized_data["linkedin_summary"] = linkedin_data.get("error")
+
     return normalized_data

src/utils/normalization.py

@@ -1,13 +1,19 @@
-from src.agents.data_retrieval import GoogleSearchAgent, WikipediaAgent
+from src.agents.data_retrieval import GoogleSearchAgent, WikipediaAgent, LinkedInSearchAgent
 
 def test_google_data_retrieval():
     agent = GoogleSearchAgent()
     result = agent.get_google_data("OpenAI")
-    assert "google_results" in result
-    assert len(result["google_results"]) > 0
+    assert "google_summary" in result
+    assert len(result["google_summary"]) > 0
 
 def test_wikipedia_data_retrieval():
     agent = WikipediaAgent()
     result = agent.get_wikipedia_data("OpenAI")
-    assert "extract" in result
-    assert result["extract"] != ""
+    assert "wikipedia_summary" in result
+    assert len(result["wikipedia_summary"]) > 0
+
+def test_linkedin_data_retrieval():
+    agent = LinkedInSearchAgent()
+    result = agent.get_linkedin_data("OpenAI")
+    assert "linkedin_summary" in result
+    assert len(result["linkedin_summary"]) > 0