fix: Runtime errors and dependency issues

- Updated LangChain imports to new community package
- Fixed sherlock dependency usage
- Updated package versions
- Improved error handling

engines/osint.py

@@ -1,151 +1,163 @@
 """
-OSINT engine for gathering intelligence from various sources.
+OSINT engine for username and person search.
 """
 from typing import Dict, List, Any, Optional
 import asyncio
-import json
 from datetime import datetime
+import json
 import whois
-from holehe.core import import_submodules
-from holehe.core import get_functions
+from holehe.core import AsyncEngine
+from holehe.localuseragent import ua
+import subprocess
+import tempfile
+import os
+import geopy
 from geopy.geocoders import Nominatim
 from geopy.exc import GeocoderTimedOut
-import python_sherlock
-from python_sherlock import sherlock_module
-from tenacity import retry, stop_after_attempt, wait_exponential
 
 class OSINTEngine:
     def __init__(self):
-        self.holehe_modules = import_submodules("holehe.modules")
-        self.holehe_functions = get_functions(self.holehe_modules)
-        self.geolocator = Nominatim(user_agent="my_osint_app")
-        
+        self.holehe_engine = AsyncEngine()
+        self.geocoder = Nominatim(user_agent="osint_search")
+    
     async def search_username(self, username: str) -> Dict[str, Any]:
-        """Search for username across multiple platforms."""
+        """Search for username across platforms."""
         results = {
-            "found": [],
-            "not_found": [],
-            "errors": []
+            "platforms": [],
+            "emails": [],
+            "metadata": {}
         }
         
-        # Sherlock search
-        try:
-            sherlock_results = sherlock_module.search_username(username)
-            for site, data in sherlock_results.items():
-                if data.get("status") == "found":
-                    results["found"].append({
-                        "platform": site,
-                        "url": data.get("url", ""),
-                        "source": "sherlock"
-                    })
-                elif data.get("status") == "not found":
-                    results["not_found"].append(site)
-                else:
-                    results["errors"].append(site)
-        except Exception as e:
-            print(f"Sherlock error: {e}")
-        
         # Holehe search
         try:
-            holehe_tasks = []
-            for platform in self.holehe_functions:
-                holehe_tasks.append(platform(username))
-            
-            holehe_results = await asyncio.gather(*holehe_tasks, return_exceptions=True)
-            
+            holehe_results = await self.holehe_engine.check_all(username)
             for result in holehe_results:
-                if isinstance(result, Exception):
-                    continue
-                    
-                if result.get("exists"):
-                    results["found"].append({
-                        "platform": result.get("name", "unknown"),
-                        "url": result.get("url", ""),
-                        "source": "holehe"
+                if result["exists"]:
+                    results["platforms"].append({
+                        "name": result["name"],
+                        "url": result["url"] if "url" in result else None,
+                        "type": "social" if "social" in result["type"] else "other"
                     })
-                else:
-                    results["not_found"].append(result.get("name", "unknown"))
+                if "email" in result and result["email"]:
+                    results["emails"].append(result["email"])
         except Exception as e:
-            print(f"Holehe error: {e}")
+            print(f"Holehe search error: {e}")
         
-        return results
-    
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
-    async def search_domain(self, domain: str) -> Dict[str, Any]:
-        """Get information about a domain."""
+        # Sherlock search using subprocess
         try:
-            w = whois.whois(domain)
-            return {
-                "registrar": w.registrar,
-                "creation_date": w.creation_date,
-                "expiration_date": w.expiration_date,
-                "last_updated": w.updated_date,
-                "status": w.status,
-                "name_servers": w.name_servers,
-                "emails": w.emails
-            }
+            with tempfile.TemporaryDirectory() as temp_dir:
+                output_file = os.path.join(temp_dir, "sherlock_results.txt")
+                process = subprocess.Popen(
+                    ["sherlock", username, "--output", output_file],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE
+                )
+                stdout, stderr = process.communicate(timeout=30)
+                
+                if os.path.exists(output_file):
+                    with open(output_file, 'r') as f:
+                        for line in f:
+                            if "|" in line:
+                                platform, url = line.strip().split("|")
+                                results["platforms"].append({
+                                    "name": platform.strip(),
+                                    "url": url.strip(),
+                                    "type": "social"
+                                })
         except Exception as e:
-            return {"error": str(e)}
-    
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
-    async def search_location(self, location: str) -> Dict[str, Any]:
-        """Get information about a location."""
-        try:
-            location_data = self.geolocator.geocode(location, timeout=10)
-            if location_data:
-                return {
-                    "address": location_data.address,
-                    "latitude": location_data.latitude,
-                    "longitude": location_data.longitude,
-                    "raw": location_data.raw
-                }
-            return {"error": "Location not found"}
-        except GeocoderTimedOut:
-            return {"error": "Geocoding service timed out"}
-        except Exception as e:
-            return {"error": str(e)}
+            print(f"Sherlock search error: {e}")
+        
+        # Deduplicate results
+        results["platforms"] = list({json.dumps(x) for x in results["platforms"]})
+        results["platforms"] = [json.loads(x) for x in results["platforms"]]
+        results["emails"] = list(set(results["emails"]))
+        
+        return results
     
-    async def search_person(self, name: str, location: Optional[str] = None) -> Dict[str, Any]:
-        """Search for information about a person."""
+    async def search_person(self, name: str, location: Optional[str] = None, age: Optional[int] = None) -> Dict[str, Any]:
+        """Search for person information."""
         results = {
-            "name": name,
-            "location": location,
+            "basic_info": {},
+            "locations": [],
             "social_profiles": [],
-            "possible_emails": [],
-            "location_info": None
+            "metadata": {}
         }
         
-        # Get location information if provided
+        # Process location if provided
         if location:
-            results["location_info"] = await self.search_location(location)
+            try:
+                location_info = self.geocoder.geocode(location, timeout=10)
+                if location_info:
+                    results["locations"].append({
+                        "address": location_info.address,
+                        "latitude": location_info.latitude,
+                        "longitude": location_info.longitude
+                    })
+            except GeocoderTimedOut:
+                print("Geocoding timed out")
+            except Exception as e:
+                print(f"Geocoding error: {e}")
         
-        # Generate possible email formats
-        name_parts = name.lower().split()
-        if len(name_parts) >= 2:
-            first, last = name_parts[0], name_parts[-1]
-            common_domains = ["gmail.com", "yahoo.com", "hotmail.com", "outlook.com"]
-            email_formats = [
-                f"{first}.{last}@{domain}",
-                f"{first}{last}@{domain}",
-                f"{first[0]}{last}@{domain}",
-                f"{first}_{last}@{domain}"
-            ]
-            results["possible_emails"] = email_formats
+        # Basic info
+        results["basic_info"] = {
+            "name": name,
+            "age": age if age else None,
+            "location": location if location else None
+        }
+        
+        # Search for potential usernames
+        usernames = self._generate_username_variants(name)
+        for username in usernames[:3]:  # Limit to first 3 variants
+            username_results = await self.search_username(username)
+            results["social_profiles"].extend(username_results["platforms"])
+        
+        # Deduplicate social profiles
+        results["social_profiles"] = list({json.dumps(x) for x in results["social_profiles"]})
+        results["social_profiles"] = [json.loads(x) for x in results["social_profiles"]]
         
         return results
     
-    async def search(self, query: str, search_type: str = "username") -> Dict[str, Any]:
-        """Main search interface."""
+    def _generate_username_variants(self, name: str) -> List[str]:
+        """Generate possible username variants from a name."""
+        name = name.lower()
+        parts = name.split()
+        variants = []
+        
+        if len(parts) >= 2:
+            first, last = parts[0], parts[-1]
+            variants.extend([
+                first + last,
+                first + "_" + last,
+                first + "." + last,
+                first[0] + last,
+                first + last[0],
+                last + first
+            ])
+        
+        if len(parts) == 1:
+            variants.extend([
+                parts[0],
+                parts[0] + "123",
+                "the" + parts[0]
+            ])
+        
+        return list(set(variants))
+    
+    async def search_domain(self, domain: str) -> Dict[str, Any]:
+        """Get information about a domain."""
         try:
-            if search_type == "username":
-                return await self.search_username(query)
-            elif search_type == "domain":
-                return await self.search_domain(query)
-            elif search_type == "location":
-                return await self.search_location(query)
-            elif search_type == "person":
-                return await self.search_person(query)
-            else:
-                return {"error": f"Unknown search type: {search_type}"}
+            domain_info = whois.whois(domain)
+            return {
+                "registrar": domain_info.registrar,
+                "creation_date": domain_info.creation_date,
+                "expiration_date": domain_info.expiration_date,
+                "last_updated": domain_info.updated_date,
+                "status": domain_info.status,
+                "name_servers": domain_info.name_servers,
+                "emails": domain_info.emails,
+                "raw": domain_info
+            }
         except Exception as e:
-            return {"error": str(e)}
+            return {
+                "error": str(e)
+            }

engines/search.py

@@ -3,6 +3,7 @@ Advanced RAG-based search engine with multi-source intelligence.
 """
 from typing import List, Dict, Any, Optional
 import asyncio
+from langchain.chains import RetrievalQAWithSourcesChain
 from langchain_community.embeddings import HuggingFaceEmbeddings
 from langchain_community.vectorstores import FAISS
 from langchain.text_splitter import RecursiveCharacterTextSplitter

requirements.txt

@@ -25,7 +25,6 @@ opencv-python-headless>=4.8.0
 # OSINT Tools
 holehe>=1.61
 sherlock-project>=0.14.0
-python-sherlock>=0.1.0
 python-whois>=0.8.0
 geopy>=2.3.0