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