Spaces:
Paused
Paused
| import os | |
| import logging | |
| import time | |
| import json | |
| import asyncio | |
| from typing import Dict, Optional, Any, List | |
| from apify_client import ApifyClient | |
| from pydantic import ValidationError | |
| from ..core.models import ProfileData | |
| logger = logging.getLogger(__name__) | |
| class LinkedInScraper: | |
| """Handles LinkedIn profile scraping using Apify.""" | |
| def __init__(self): | |
| self.api_token = os.getenv("APIFY_API_TOKEN") | |
| self.client = ApifyClient(self.api_token) if self.api_token else None | |
| self.actor_id = "2SyF0bVxmgGr8IVCZ" # ID for "Mass Linkedin Profile Scraper with Email" | |
| if not self.client or not self.actor_id: | |
| logger.error("Apify client or Actor ID is not configured. Scraping will fail.") | |
| self.client = None # Ensure client is None if config is incomplete | |
| def _validate_profile_url(self, url: str) -> bool: | |
| """Validate LinkedIn profile URL format.""" | |
| url = url.strip().lower() | |
| return "linkedin.com/in/" in url | |
| def _clean_and_fix_url(self, url: str) -> str: | |
| """Clean and fix LinkedIn URL to ensure it's in the correct format.""" | |
| url = url.strip() | |
| if not url.startswith(("http://", "https://")): | |
| url = "https://" + url | |
| if "/in/" in url: | |
| url = url.split("?")[0] | |
| if not url.endswith("/"): | |
| url += "/" | |
| return url | |
| def _clean_profile_data(self, raw_data: Dict[str, Any]) -> Dict[str, Any]: | |
| """Clean and structure raw profile data based on observed dev_fusion/Linkedin-Profile-Scraper output.""" | |
| logger.debug(f"Cleaning data. Raw keys: {list(raw_data.keys())}") | |
| cleaned = { | |
| "full_name": raw_data.get("fullName", ""), | |
| "headline": raw_data.get("headline", ""), | |
| "about": raw_data.get("about", ""), # Use 'about' key from user JSON | |
| "location": raw_data.get("addressWithCountry", raw_data.get("location", "")), # Use specific address key | |
| "profile_url": raw_data.get("linkedinUrl", raw_data.get("profileUrl", "")), # Use 'linkedinUrl' key | |
| "experience": [], | |
| "education": [], | |
| "skills": [], | |
| "recommendations": [], # Keep as not present in user JSON | |
| "accomplishments": [], # Process projects, certs etc. if needed | |
| "raw_data": raw_data | |
| } | |
| # --- Process Experience --- | |
| experiences_raw = raw_data.get("experiences", []) | |
| logger.debug(f"Found {len(experiences_raw)} raw experience entries.") | |
| if isinstance(experiences_raw, list): | |
| for exp in experiences_raw: | |
| if isinstance(exp, dict): | |
| title = exp.get("title", "") | |
| subtitle = exp.get("subtitle", "") | |
| company_name = subtitle.split('·')[0].strip() if '·' in subtitle else subtitle | |
| caption = exp.get("caption", "") | |
| date_range_str = caption.split('·')[0].strip() if '·' in caption else caption | |
| duration_str = caption.split('·')[1].strip() if '·' in caption and len(caption.split('·')) > 1 else "" | |
| start_date = date_range_str.split('-')[0].strip() if '-' in date_range_str else date_range_str | |
| end_date = date_range_str.split('-')[1].strip() if '-' in date_range_str and len(date_range_str.split('-')) > 1 else "Present" | |
| description = "" | |
| sub_components = exp.get("subComponents", []) | |
| if sub_components and isinstance(sub_components, list) and isinstance(sub_components[0], dict): | |
| desc_list = sub_components[0].get("description", []) | |
| if isinstance(desc_list, list): | |
| text_component = next((item for item in desc_list if isinstance(item, dict) and item.get("type") == "textComponent"), None) | |
| if text_component: | |
| description = text_component.get("text", "") | |
| cleaned["experience"].append({ | |
| "title": title, | |
| "company": company_name, | |
| "location": exp.get("metadata", ""), | |
| "start_date": start_date, | |
| "end_date": end_date, | |
| "description": description.strip(), | |
| "duration": duration_str | |
| }) | |
| logger.info(f"Processed {len(cleaned['experience'])} experience entries.") | |
| # --- Process Education --- | |
| educations_raw = raw_data.get("educations", []) | |
| logger.debug(f"Found {len(educations_raw)} raw education entries.") | |
| if isinstance(educations_raw, list): | |
| for edu in educations_raw: | |
| if isinstance(edu, dict): | |
| school_name = edu.get("title", "") | |
| subtitle = edu.get("subtitle", "") | |
| degree = subtitle.split(',')[0].strip() if ',' in subtitle else subtitle | |
| field = subtitle.split(',')[1].strip() if ',' in subtitle and len(subtitle.split(',')) > 1 else "" | |
| caption = edu.get("caption", "") | |
| start_date = caption.split('-')[0].strip() if '-' in caption else caption | |
| end_date = caption.split('-')[1].strip() if '-' in caption and len(caption.split('-')) > 1 else "" | |
| description = "" | |
| sub_components = edu.get("subComponents", []) | |
| if sub_components and isinstance(sub_components, list) and isinstance(sub_components[0], dict): | |
| desc_list = sub_components[0].get("description", []) | |
| if isinstance(desc_list, list): | |
| text_component = next((item for item in desc_list if isinstance(item, dict) and item.get("type") == "textComponent"), None) | |
| if text_component: | |
| description = text_component.get("text", "") | |
| cleaned["education"].append({ | |
| "school": school_name, | |
| "degree": degree, | |
| "field": field, | |
| "start_date": start_date, | |
| "end_date": end_date, | |
| "description": description.strip() | |
| }) | |
| logger.info(f"Processed {len(cleaned['education'])} education entries.") | |
| # --- Process Skills --- | |
| skills_raw = raw_data.get("skills", []) | |
| logger.debug(f"Found {len(skills_raw)} raw skill entries.") | |
| processed_skills = [] | |
| if isinstance(skills_raw, list): | |
| for skill_item in skills_raw: | |
| if isinstance(skill_item, dict): | |
| name = skill_item.get("title") | |
| if name and isinstance(name, str): | |
| processed_skills.append(name) | |
| cleaned["skills"] = list(set(processed_skills)) | |
| logger.info(f"Processed {len(cleaned['skills'])} skill entries.") | |
| # --- Process Accomplishments (Example: Projects) --- | |
| projects_raw = raw_data.get("projects", []) | |
| if isinstance(projects_raw, list): | |
| project_items = [] | |
| for proj in projects_raw: | |
| if isinstance(proj, dict): | |
| title = proj.get("title") | |
| if title: project_items.append(title) | |
| if project_items: | |
| cleaned["accomplishments"].append({"type": "Projects", "items": project_items}) | |
| logger.info(f"Processed {len(project_items)} projects.") | |
| # --- Process Accomplishments (Example: Certifications) --- | |
| certs_raw = raw_data.get("licenseAndCertificates", []) | |
| if isinstance(certs_raw, list): | |
| cert_items = [] | |
| for cert in certs_raw: | |
| if isinstance(cert, dict): | |
| title = cert.get("title") | |
| if title: cert_items.append(title) | |
| if cert_items: | |
| cleaned["accomplishments"].append({"type": "Certifications", "items": cert_items}) | |
| logger.info(f"Processed {len(cert_items)} certifications.") | |
| logger.debug(f"Final cleaned data structure prepared for ProfileData model.") | |
| return cleaned | |
| async def scrape_profile(self, profile_url: str, max_retries: int = 3, retry_delay_secs: int = 2) -> Optional[ProfileData]: | |
| """ | |
| Scrape LinkedIn profile data using Apify with retries. | |
| Returns ProfileData object if successful, None otherwise. | |
| Args: | |
| profile_url: LinkedIn profile URL to scrape. | |
| max_retries: Maximum number of attempts for the API call. | |
| retry_delay_secs: Delay between retries in seconds. | |
| Returns: | |
| ProfileData object if successful, None if validation fails or API fails after retries. | |
| """ | |
| if not self.client: | |
| logger.error("Apify client not initialized. Cannot scrape.") | |
| return None | |
| if not self._validate_profile_url(profile_url): | |
| logger.error(f"Invalid LinkedIn URL format: {profile_url}") | |
| return None | |
| profile_url = self._clean_and_fix_url(profile_url) | |
| last_exception = None | |
| for attempt in range(max_retries): | |
| logger.info(f"Attempt {attempt + 1}/{max_retries} to scrape profile data for: {profile_url}") | |
| try: | |
| run_input = {"profileUrls": [profile_url]} | |
| logger.info(f"Running Apify actor '{self.actor_id}' with input: {run_input}") | |
| # Run the actor call within the try block | |
| run = self.client.actor(self.actor_id).call(run_input=run_input, wait_secs=120) | |
| dataset_id = run.get("defaultDatasetId") | |
| if not dataset_id: | |
| # Treat no dataset ID as a potentially recoverable error | |
| logger.warning(f"Attempt {attempt + 1}: No dataset ID returned from Apify actor run.") | |
| last_exception = Exception("No dataset ID returned from Apify run") | |
| # Continue to next retry after delay | |
| if attempt < max_retries - 1: | |
| time.sleep(retry_delay_secs) | |
| continue | |
| logger.info(f"Attempt {attempt + 1}: Retrieving results from dataset: {dataset_id}") | |
| dataset_items = self.client.dataset(dataset_id).list_items().items | |
| if not dataset_items: | |
| # Treat no items as potentially recoverable | |
| logger.warning(f"Attempt {attempt + 1}: No items found in dataset.") | |
| last_exception = Exception("No items found in Apify dataset") | |
| # Continue to next retry after delay | |
| if attempt < max_retries - 1: | |
| time.sleep(retry_delay_secs) | |
| continue | |
| profile_data_raw = dataset_items[0] | |
| logger.info(f"Attempt {attempt + 1}: Successfully retrieved data from Apify.") | |
| cleaned_data = self._clean_profile_data(profile_data_raw) | |
| # Validate data - if validation fails, it's unlikely to succeed on retry | |
| try: | |
| real_profile_data = ProfileData(**cleaned_data) | |
| logger.info(f"Attempt {attempt + 1}: Successfully processed and validated profile data.") | |
| return real_profile_data # SUCCESS! | |
| except ValidationError as ve: | |
| logger.error(f"Validation error processing scraped data (attempt {attempt + 1}): {ve}. Aborting retries.") | |
| return None # Data structure error, unlikely to be fixed by retry | |
| except Exception as e: | |
| last_exception = e | |
| logger.error(f"Error during Apify scraping attempt {attempt + 1}/{max_retries}: {e}") | |
| if attempt < max_retries - 1: | |
| logger.info(f"Waiting {retry_delay_secs} seconds before next attempt...") | |
| await asyncio.sleep(retry_delay_secs) # Use asyncio.sleep for async function | |
| # If loop finishes without returning, all retries failed | |
| logger.error(f"Apify scraping failed after {max_retries} attempts. Last error: {last_exception}") | |
| return None | |