Getting no values in person scrapping

[suzhiyam-in]

While scrapping peoples, the result is as follows:

✓ Session loaded
🚀 Scraping: https://www.linkedin.com/in/williamhgates/

============================================================
Person: Unknown
Location: None
About: N/A...

💼 Work Experience (0 positions):

🎓 Education (0 schools):

✓ Done!

You can see that there is no data got from the people scrapping. Company scrapping is working fine. People scrapping is not.

[manofgpa]

Same here. LinkedIn changed the DOM structure. Fixed person.py

`
"""Person/Profile scraper for LinkedIn."""

import logging
from typing import Optional
from urllib.parse import urljoin
from playwright.async_api import Page

from .base import BaseScraper
from ..models import Person, Experience, Education, Accomplishment, Interest, Contact
from ..callbacks import ProgressCallback, SilentCallback
from ..core.exceptions import ScrapingError

logger = logging.getLogger(name)

class PersonScraper(BaseScraper):
"""Async scraper for LinkedIn person profiles."""

def __init__(self, page: Page, callback: Optional[ProgressCallback] = None):
    """
    Initialize person scraper.

    Args:
        page: Playwright page object
        callback: Progress callback
    """
    super().__init__(page, callback)

async def scrape(self, linkedin_url: str) -> Person:
    """
    Scrape a LinkedIn person profile.

    Args:
        linkedin_url: LinkedIn profile URL

    Returns:
        Person object with all scraped data

    Raises:
        AuthenticationError: If not logged in
        ScrapingError: If scraping fails
    """
    await self.callback.on_start("person", linkedin_url)

    try:
        # Navigate to profile first (this loads the page with our session)
        await self.navigate_and_wait(linkedin_url)
        await self.callback.on_progress("Navigated to profile", 10)

        # Now check if logged in
        await self.ensure_logged_in()

        # Wait for main content
        await self.page.wait_for_selector("main", timeout=10000)
        await self.wait_and_focus(1)

        # Get name and location
        name, location = await self._get_name_and_location()
        await self.callback.on_progress(f"Got name: {name}", 20)

        # Check open to work
        open_to_work = await self._check_open_to_work()

        # Get about
        about = await self._get_about()
        await self.callback.on_progress("Got about section", 30)

        # Scroll to load content
        await self.scroll_page_to_half()
        await self.scroll_page_to_bottom(pause_time=0.5, max_scrolls=3)

        # Get experiences
        experiences = await self._get_experiences(linkedin_url)
        await self.callback.on_progress(f"Got {len(experiences)} experiences", 60)

        educations = await self._get_educations(linkedin_url)
        await self.callback.on_progress(f"Got {len(educations)} educations", 50)

        interests = await self._get_interests(linkedin_url)
        await self.callback.on_progress(f"Got {len(interests)} interests", 65)

        accomplishments = await self._get_accomplishments(linkedin_url)
        await self.callback.on_progress(
            f"Got {len(accomplishments)} accomplishments", 85
        )

        contacts = await self._get_contacts(linkedin_url)
        await self.callback.on_progress(f"Got {len(contacts)} contacts", 95)

        person = Person(
            linkedin_url=linkedin_url,
            name=name,
            location=location,
            about=about,
            open_to_work=open_to_work,
            experiences=experiences,
            educations=educations,
            interests=interests,
            accomplishments=accomplishments,
            contacts=contacts,
        )

        await self.callback.on_progress("Scraping complete", 100)
        await self.callback.on_complete("person", person)

        return person

    except Exception as e:
        await self.callback.on_error(e)
        raise ScrapingError(f"Failed to scrape person profile: {e}")

async def _get_name_and_location(self) -> tuple[str, Optional[str]]:
    """Extract name and location from profile topcard."""
    try:
        # The topcard is in [data-view-name="profile-main-level"]
        topcard = self.page.locator('[data-view-name="profile-main-level"]')
        if await topcard.count() == 0:
            # Fallback to first section
            topcard = self.page.locator('main section').first

        # LinkedIn now uses h2 for the profile name
        name_el = topcard.locator('h2').first
        name = "Unknown"
        if await name_el.count() > 0:
            name = (await name_el.text_content() or "Unknown").strip()

        # Location is in a p tag in the same container as the "Contact info" link
        contact_link = topcard.locator('a[href*="overlay/contact-info"]')
        if await contact_link.count() > 0:
            location_container = contact_link.locator('xpath=ancestor::div[1]')
            location_p = location_container.locator('p').first
            if await location_p.count() > 0:
                location = (await location_p.text_content() or "").strip()
                return name, location if location else None

        return name, None
    except Exception as e:
        logger.warning(f"Error getting name/location: {e}")
        return "Unknown", None

async def _check_open_to_work(self) -> bool:
    """Check if profile has open to work badge."""
    try:
        # Look for open to work indicator
        img_title = await self.get_attribute_safe(
            ".pv-top-card-profile-picture img", "title", default=""
        )
        return "#OPEN_TO_WORK" in img_title.upper()
    except:
        return False

async def _get_about(self) -> Optional[str]:
    """Extract about section using data-view-name anchor."""
    try:
        about_section = self.page.locator('[data-view-name="profile-card-about"]')
        if await about_section.count() == 0:
            return None

        # The about text is in a span with data-testid="expandable-text-box"
        text_box = about_section.locator('[data-testid="expandable-text-box"]').first
        if await text_box.count() > 0:
            about_text = await text_box.text_content()
            if about_text:
                # Remove trailing "…\n more" from expandable text
                about_text = about_text.strip()
                for suffix in ["… more", "…\n more", "…more"]:
                    if about_text.endswith(suffix):
                        about_text = about_text[: -len(suffix)].strip()
                return about_text if about_text else None

        # Fallback: get the first p tag after the heading
        about_p = about_section.locator('section p').first
        if await about_p.count() > 0:
            about_text = await about_p.text_content()
            return about_text.strip() if about_text else None

        return None
    except Exception as e:
        logger.debug(f"Error getting about section: {e}")
        return None

async def _get_experiences(self, base_url: str) -> list[Experience]:
    """Extract experiences from the profile page Experience section."""
    experiences = []

    try:
        exp_section = self.page.locator('[data-view-name="profile-card-experience"]')
        if await exp_section.count() == 0:
            return experiences

        # Each experience item is in a div with componentkey starting with "entity-collection-item"
        items = await exp_section.locator('div[componentkey^="entity-collection-item"]').all()

        for item in items:
            try:
                result = await self._parse_experience_card(item)
                if result:
                    if isinstance(result, list):
                        experiences.extend(result)
                    else:
                        experiences.append(result)
            except Exception as e:
                logger.debug(f"Error parsing experience card: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting experiences: {e}")

    return experiences

async def _parse_experience_card(self, item) -> Optional[Experience | list[Experience]]:
    """Parse an experience card from LinkedIn's DOM.

    Handles both single-role and grouped (multi-role at one company) entries.
    Grouped entries have a <ul> with sub-positions.
    """
    try:
        # Check for grouped experience (ul with sub-positions)
        sub_list = item.locator('ul')
        if await sub_list.count() > 0:
            return await self._parse_grouped_experience(item)

        return await self._parse_single_experience(item)

    except Exception as e:
        logger.debug(f"Error parsing experience card: {e}")
        return None

async def _parse_single_experience(self, item) -> Optional[Experience]:
    """Parse a single-role experience card."""
    try:
        links = await item.locator('a').all()
        if len(links) < 2:
            return None

        company_url = await links[0].get_attribute('href')
        detail_link = links[1]

        p_tags = await detail_link.locator('p').all()
        texts = []
        for p in p_tags:
            text = await p.text_content()
            if text and text.strip() and len(text.strip()) < 300:
                texts.append(text.strip())

        if len(texts) < 2:
            return None

        position_title = texts[0]
        company_name = texts[1]
        work_times = texts[2] if len(texts) > 2 else ""
        location = texts[3] if len(texts) > 3 else ""

        from_date, to_date, duration = self._parse_work_times(work_times)

        description = await self._extract_description(item)

        return Experience(
            position_title=position_title,
            institution_name=company_name,
            linkedin_url=company_url,
            from_date=from_date,
            to_date=to_date,
            duration=duration,
            location=location if location else None,
            description=description,
        )

    except Exception as e:
        logger.debug(f"Error parsing single experience: {e}")
        return None

async def _parse_grouped_experience(self, item) -> list[Experience]:
    """Parse a grouped experience (multiple roles at one company)."""
    experiences = []
    try:
        # Get company info from top-level links
        top_links = await item.locator(':scope > div a, :scope > div > div a').all()
        company_url = None
        company_name = None
        location = None

        if top_links:
            company_url = await top_links[0].get_attribute('href')
            if len(top_links) > 1:
                detail_link = top_links[1]
                p_tags = await detail_link.locator('p').all()
                texts = []
                for p in p_tags:
                    text = await p.text_content()
                    if text and text.strip():
                        texts.append(text.strip())
                if texts:
                    company_name = texts[0]
                if len(texts) > 2:
                    location = texts[2]

        # Parse each sub-position from li elements
        sub_items = await item.locator('ul > li').all()
        for sub_item in sub_items:
            try:
                sub_link = sub_item.locator('a').first
                if await sub_link.count() == 0:
                    continue

                p_tags = await sub_link.locator('p').all()
                texts = []
                for p in p_tags:
                    text = await p.text_content()
                    if text and text.strip() and len(text.strip()) < 300:
                        texts.append(text.strip())

                if not texts:
                    continue

                position_title = texts[0]
                work_times = ""
                sub_location = location

                # Remaining p tags could be subtitle, dates, location
                for t in texts[1:]:
                    if self._looks_like_date_range(t):
                        work_times = t
                    elif self._looks_like_location(t):
                        sub_location = t

                from_date, to_date, duration = self._parse_work_times(work_times)
                description = await self._extract_description(sub_item)

                experiences.append(Experience(
                    position_title=position_title,
                    institution_name=company_name or "",
                    linkedin_url=company_url,
                    from_date=from_date,
                    to_date=to_date,
                    duration=duration,
                    location=sub_location,
                    description=description,
                ))

            except Exception as e:
                logger.debug(f"Error parsing sub-position: {e}")
                continue

    except Exception as e:
        logger.debug(f"Error parsing grouped experience: {e}")

    return experiences

@staticmethod
def _looks_like_date_range(text: str) -> bool:
    """Check if text looks like a date range (e.g. 'Jan 2020 - Present · 2 yrs')."""
    months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
              'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
    return (
        any(m in text for m in months)
        or 'Present' in text
        or ('·' in text and any(u in text for u in ['yr', 'mo']))
    )

@staticmethod
def _looks_like_location(text: str) -> bool:
    """Check if text looks like a location."""
    location_indicators = [',', 'Brazil', 'States', 'United', 'Remote',
                           'Area', 'Region', 'City']
    return any(ind in text for ind in location_indicators)

async def _extract_description(self, container) -> Optional[str]:
    """Extract description text from an expandable text box."""
    text_box = container.locator('[data-testid="expandable-text-box"]').first
    if await text_box.count() > 0:
        desc_text = await text_box.text_content()
        if desc_text:
            desc_text = desc_text.strip()
            for suffix in ["… more", "…\n more", "…more"]:
                if desc_text.endswith(suffix):
                    desc_text = desc_text[: -len(suffix)].strip()
            return desc_text if desc_text else None
    return None

def _parse_work_times(
    self, work_times: str
) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """
    Parse work times string into from_date, to_date, duration.

    Examples:
    - "2000 - Present · 26 yrs 1 mo" -> ("2000", "Present", "26 yrs 1 mo")
    - "Jan 2020 - Dec 2022 · 2 yrs" -> ("Jan 2020", "Dec 2022", "2 yrs")
    - "2015 - Present" -> ("2015", "Present", None)
    """
    if not work_times:
        return None, None, None

    try:
        # Split by · to separate date range from duration
        parts = work_times.split("·")
        times = parts[0].strip() if len(parts) > 0 else ""
        duration = parts[1].strip() if len(parts) > 1 else None

        # Parse dates - split by " - " to get from and to
        if " - " in times:
            date_parts = times.split(" - ")
            from_date = date_parts[0].strip()
            to_date = date_parts[1].strip() if len(date_parts) > 1 else ""
        else:
            from_date = times
            to_date = ""

        return from_date, to_date, duration
    except Exception as e:
        logger.debug(f"Error parsing work times '{work_times}': {e}")
        return None, None, None

async def _get_educations(self, base_url: str) -> list[Education]:
    """Extract educations from the profile page Education section."""
    educations = []

    try:
        edu_section = self.page.locator('[data-view-name="profile-card-education"]')
        if await edu_section.count() == 0:
            return educations

        # Education items use componentkey but not with a consistent prefix,
        # so find all direct item containers by looking for the logo link pattern
        items = await edu_section.locator(
            'a[data-view-name="experience-company-logo-click"], '
            'a[href*="/school/"], a[href*="/company/"]'
        ).all()

        # Each logo link's grandparent is the item container
        seen_urls = set()
        for logo_link in items:
            try:
                item_container = logo_link.locator('xpath=ancestor::div[contains(@class, "b902da86")][3]')
                if await item_container.count() == 0:
                    continue

                institution_url = await logo_link.get_attribute('href')
                if institution_url in seen_urls:
                    continue
                seen_urls.add(institution_url)

                # Get the detail link (second a tag in the item)
                all_links = await item_container.locator('a').all()
                detail_link = all_links[1] if len(all_links) > 1 else None
                if not detail_link:
                    continue

                p_tags = await detail_link.locator('p').all()
                texts = []
                for p in p_tags:
                    text = await p.text_content()
                    if text and text.strip() and len(text.strip()) < 300:
                        texts.append(text.strip())

                if not texts:
                    continue

                institution_name = texts[0]
                degree = None
                times = ""

                if len(texts) == 3:
                    degree = texts[1]
                    times = texts[2]
                elif len(texts) == 2:
                    second = texts[1]
                    if any(c.isdigit() for c in second):
                        times = second
                    else:
                        degree = second

                from_date, to_date = self._parse_education_times(times)

                educations.append(Education(
                    institution_name=institution_name,
                    degree=degree.strip() if degree else None,
                    linkedin_url=institution_url,
                    from_date=from_date,
                    to_date=to_date,
                    description=None,
                ))

            except Exception as e:
                logger.debug(f"Error parsing education item: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting educations: {e}")

    return educations

def _parse_education_times(self, times: str) -> tuple[Optional[str], Optional[str]]:
    """
    Parse education times string into from_date, to_date.

    Examples:
    - "1973 - 1977" -> ("1973", "1977")
    - "2015" -> ("2015", "2015")
    - "" -> (None, None)
    """
    if not times:
        return None, None

    try:
        # Split by " - " to get from and to dates
        if " - " in times:
            parts = times.split(" - ")
            from_date = parts[0].strip()
            to_date = parts[1].strip() if len(parts) > 1 else ""
        else:
            # Single year
            from_date = times.strip()
            to_date = times.strip()

        return from_date, to_date
    except Exception as e:
        logger.debug(f"Error parsing education times '{times}': {e}")
        return None, None

async def _get_interests(self, base_url: str) -> list[Interest]:
    """Extract interests from the profile page Interests section."""
    interests = []

    try:
        interests_section = self.page.locator('[data-view-name="profile-card-interests"]')
        if await interests_section.count() == 0:
            return interests

        # Interests use radio buttons or tabs for categories
        tabs = await interests_section.locator('[role="tab"], [role="radio"]').all()

        for tab in tabs:
            try:
                tab_name = await tab.text_content()
                if not tab_name:
                    continue
                tab_name = tab_name.strip()
                category = self._map_interest_tab_to_category(tab_name)

                await tab.click()
                await self.wait_and_focus(0.5)

                # Get interest items in the visible panel
                panel = interests_section.locator('[role="tabpanel"]').first
                if await panel.count() == 0:
                    # Fallback: items may be directly in the section
                    panel = interests_section

                item_links = await panel.locator('a[href*="/company/"], a[href*="/school/"], a[href*="/in/"]').all()
                seen_names = set()
                for link in item_links:
                    href = await link.get_attribute("href")
                    p_tags = await link.locator('p').all()
                    name = None
                    if p_tags:
                        name_text = await p_tags[0].text_content()
                        if name_text:
                            name = name_text.strip()

                    if name and href and name not in seen_names:
                        seen_names.add(name)
                        interests.append(Interest(
                            name=name,
                            category=category,
                            linkedin_url=href,
                        ))

            except Exception as e:
                logger.debug(f"Error processing interest tab: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting interests: {e}")

    return interests

def _map_interest_tab_to_category(self, tab_name: str) -> str:
    tab_lower = tab_name.lower()
    if "compan" in tab_lower:
        return "company"
    elif "group" in tab_lower:
        return "group"
    elif "school" in tab_lower:
        return "school"
    elif "newsletter" in tab_lower:
        return "newsletter"
    elif "voice" in tab_lower or "influencer" in tab_lower:
        return "influencer"
    else:
        return tab_lower

async def _get_accomplishments(self, base_url: str) -> list[Accomplishment]:
    accomplishments = []

    accomplishment_sections = [
        ("certifications", "certification"),
        ("honors", "honor"),
        ("publications", "publication"),
        ("patents", "patent"),
        ("courses", "course"),
        ("projects", "project"),
        ("languages", "language"),
        ("organizations", "organization"),
    ]

    for url_path, category in accomplishment_sections:
        try:
            section_url = urljoin(base_url, f"details/{url_path}/")
            await self.navigate_and_wait(section_url)
            await self.page.wait_for_selector("main", timeout=10000)
            await self.wait_and_focus(1)

            nothing_to_see = await self.page.locator(
                'text="Nothing to see for now"'
            ).count()
            if nothing_to_see > 0:
                continue

            main_list = self.page.locator(
                ".pvs-list__container, main ul, main ol"
            ).first
            if await main_list.count() == 0:
                continue

            items = await main_list.locator(".pvs-list__paged-list-item").all()
            if not items:
                items = await main_list.locator("> li").all()

            seen_titles = set()
            for item in items:
                try:
                    accomplishment = await self._parse_accomplishment_item(
                        item, category
                    )
                    if accomplishment and accomplishment.title not in seen_titles:
                        seen_titles.add(accomplishment.title)
                        accomplishments.append(accomplishment)
                except Exception as e:
                    logger.debug(f"Error parsing {category} item: {e}")
                    continue

        except Exception as e:
            logger.debug(f"Error getting {category}s: {e}")
            continue

    return accomplishments

async def _parse_accomplishment_item(
    self, item, category: str
) -> Optional[Accomplishment]:
    try:
        entity = item.locator(
            'div[data-view-name="profile-component-entity"]'
        ).first
        if await entity.count() > 0:
            spans = await entity.locator('span[aria-hidden="true"]').all()
        else:
            spans = await item.locator('span[aria-hidden="true"]').all()

        title = ""
        issuer = ""
        issued_date = ""
        credential_id = ""

        for i, span in enumerate(spans[:5]):
            text = await span.text_content()
            if not text:
                continue
            text = text.strip()

            if len(text) > 500:
                continue

            if i == 0:
                title = text
            elif "Issued by" in text:
                parts = text.split("·")
                issuer = parts[0].replace("Issued by", "").strip()
                if len(parts) > 1:
                    issued_date = parts[1].strip()
            elif "Issued " in text and not issued_date:
                issued_date = text.replace("Issued ", "")
            elif "Credential ID" in text:
                credential_id = text.replace("Credential ID ", "")
            elif i == 1 and not issuer:
                issuer = text
            elif (
                any(
                    month in text
                    for month in [
                        "Jan",
                        "Feb",
                        "Mar",
                        "Apr",
                        "May",
                        "Jun",
                        "Jul",
                        "Aug",
                        "Sep",
                        "Oct",
                        "Nov",
                        "Dec",
                    ]
                )
                and not issued_date
            ):
                if "·" in text:
                    parts = text.split("·")
                    issued_date = parts[0].strip()
                else:
                    issued_date = text

        link = item.locator('a[href*="credential"], a[href*="verify"]').first
        credential_url = (
            await link.get_attribute("href") if await link.count() > 0 else None
        )

        if not title or len(title) > 200:
            return None

        return Accomplishment(
            category=category,
            title=title,
            issuer=issuer if issuer else None,
            issued_date=issued_date if issued_date else None,
            credential_id=credential_id if credential_id else None,
            credential_url=credential_url,
        )

    except Exception as e:
        logger.debug(f"Error parsing accomplishment: {e}")
        return None

async def _get_contacts(self, base_url: str) -> list[Contact]:
    """Extract contact info from the contact-info overlay dialog."""
    contacts = []

    try:
        contact_url = urljoin(base_url, "overlay/contact-info/")
        await self.navigate_and_wait(contact_url)
        await self.wait_and_focus(1)

        dialog = self.page.locator('dialog, [role="dialog"]').first
        if await dialog.count() == 0:
            logger.warning("Contact info dialog not found")
            return contacts

        contact_sections = await dialog.locator('h3').all()
        
        for section_heading in contact_sections:
            try:
                heading_text = await section_heading.text_content()
                if not heading_text:
                    continue
                heading_text = heading_text.strip().lower()
                
                section_container = section_heading.locator('xpath=ancestor::*[1]')
                if await section_container.count() == 0:
                    continue
                
                contact_type = self._map_contact_heading_to_type(heading_text)
                if not contact_type:
                    continue
                
                links = await section_container.locator('a').all()
                for link in links:
                    href = await link.get_attribute('href')
                    text = await link.text_content()
                    if href and text:
                        text = text.strip()
                        label = None
                        sibling_text = await section_container.locator('span, generic').all()
                        for sib in sibling_text:
                            sib_text = await sib.text_content()
                            if sib_text and sib_text.strip().startswith('(') and sib_text.strip().endswith(')'):
                                label = sib_text.strip()[1:-1]
                                break
                        
                        if contact_type == "linkedin":
                            contacts.append(Contact(type=contact_type, value=href, label=label))
                        elif contact_type == "email" and "mailto:" in href:
                            contacts.append(Contact(type=contact_type, value=href.replace("mailto:", ""), label=label))
                        else:
                            contacts.append(Contact(type=contact_type, value=text, label=label))
                
                if contact_type == "birthday" and not links:
                    birthday_text = await section_container.text_content()
                    if birthday_text:
                        birthday_value = birthday_text.replace(heading_text, "").replace("Birthday", "").strip()
                        if birthday_value:
                            contacts.append(Contact(type="birthday", value=birthday_value))
                
                if contact_type == "phone" and not links:
                    phone_text = await section_container.text_content()
                    if phone_text:
                        phone_value = phone_text.replace(heading_text, "").replace("Phone", "").strip()
                        if phone_value:
                            contacts.append(Contact(type="phone", value=phone_value))
                
                if contact_type == "address" and not links:
                    address_text = await section_container.text_content()
                    if address_text:
                        address_value = address_text.replace(heading_text, "").replace("Address", "").strip()
                        if address_value:
                            contacts.append(Contact(type="address", value=address_value))
                            
            except Exception as e:
                logger.debug(f"Error parsing contact section: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting contacts: {e}")

    return contacts

def _map_contact_heading_to_type(self, heading: str) -> Optional[str]:
    """Map contact section heading to contact type."""
    heading = heading.lower()
    if "profile" in heading:
        return "linkedin"
    elif "website" in heading:
        return "website"
    elif "email" in heading:
        return "email"
    elif "phone" in heading:
        return "phone"
    elif "twitter" in heading or "x.com" in heading:
        return "twitter"
    elif "birthday" in heading:
        return "birthday"
    elif "address" in heading:
        return "address"
    return None

`

[suzhiyam-in]

Thanks man. The code you gave is working. Just a small thing to mention. At line 11, replace this code
logger = logging.getLogger(name)
with this
logger = logging.getLogger(__name__).

logger = logging.getLogger(name) this one doesn't worked for me. It throwed error.
But, logger = logging.getLogger(__name__) this one worked for me.

Also there is some indentation error, as you pasted the code in the comment, the indentation is gone. So copy pasting leads to indentation error.

So, just copy pasting won't work guys. Add a tab space to all the code below the
class PersonScraper(BaseScraper): .

Here is the ready to copy and paste working code.

import logging
from typing import Optional
from urllib.parse import urljoin
from playwright.async_api import Page

from .base import BaseScraper
from ..models import Person, Experience, Education, Accomplishment, Interest, Contact
from ..callbacks import ProgressCallback, SilentCallback
from ..core.exceptions import ScrapingError

logger = logging.getLogger(__name__)

class PersonScraper(BaseScraper):
    """Async scraper for LinkedIn person profiles."""

    def __init__(self, page: Page, callback: Optional[ProgressCallback] = None):
        """
        Initialize person scraper.

        Args:
            page: Playwright page object
            callback: Progress callback
        """
        super().__init__(page, callback)

    async def scrape(self, linkedin_url: str) -> Person:
        """
        Scrape a LinkedIn person profile.

        Args:
            linkedin_url: LinkedIn profile URL

        Returns:
            Person object with all scraped data

        Raises:
            AuthenticationError: If not logged in
            ScrapingError: If scraping fails
        """
        await self.callback.on_start("person", linkedin_url)

        try:
            # Navigate to profile first (this loads the page with our session)
            await self.navigate_and_wait(linkedin_url)
            await self.callback.on_progress("Navigated to profile", 10)

            # Now check if logged in
            await self.ensure_logged_in()

            # Wait for main content
            await self.page.wait_for_selector("main", timeout=10000)
            await self.wait_and_focus(1)

            # Get name and location
            name, location = await self._get_name_and_location()
            await self.callback.on_progress(f"Got name: {name}", 20)

            # Check open to work
            open_to_work = await self._check_open_to_work()

            # Get about
            about = await self._get_about()
            await self.callback.on_progress("Got about section", 30)

            # Scroll to load content
            await self.scroll_page_to_half()
            await self.scroll_page_to_bottom(pause_time=0.5, max_scrolls=3)

            # Get experiences
            experiences = await self._get_experiences(linkedin_url)
            await self.callback.on_progress(f"Got {len(experiences)} experiences", 60)

            educations = await self._get_educations(linkedin_url)
            await self.callback.on_progress(f"Got {len(educations)} educations", 50)

            interests = await self._get_interests(linkedin_url)
            await self.callback.on_progress(f"Got {len(interests)} interests", 65)

            accomplishments = await self._get_accomplishments(linkedin_url)
            await self.callback.on_progress(
                f"Got {len(accomplishments)} accomplishments", 85
            )

            contacts = await self._get_contacts(linkedin_url)
            await self.callback.on_progress(f"Got {len(contacts)} contacts", 95)

            person = Person(
                linkedin_url=linkedin_url,
                name=name,
                location=location,
                about=about,
                open_to_work=open_to_work,
                experiences=experiences,
                educations=educations,
                interests=interests,
                accomplishments=accomplishments,
                contacts=contacts,
            )

            await self.callback.on_progress("Scraping complete", 100)
            await self.callback.on_complete("person", person)

            return person

        except Exception as e:
            await self.callback.on_error(e)
            raise ScrapingError(f"Failed to scrape person profile: {e}")

    async def _get_name_and_location(self) -> tuple[str, Optional[str]]:
        """Extract name and location from profile topcard."""
        try:
            # The topcard is in [data-view-name="profile-main-level"]
            topcard = self.page.locator('[data-view-name="profile-main-level"]')
            if await topcard.count() == 0:
                # Fallback to first section
                topcard = self.page.locator('main section').first

            # LinkedIn now uses h2 for the profile name
            name_el = topcard.locator('h2').first
            name = "Unknown"
            if await name_el.count() > 0:
                name = (await name_el.text_content() or "Unknown").strip()

            # Location is in a p tag in the same container as the "Contact info" link
            contact_link = topcard.locator('a[href*="overlay/contact-info"]')
            if await contact_link.count() > 0:
                location_container = contact_link.locator('xpath=ancestor::div[1]')
                location_p = location_container.locator('p').first
                if await location_p.count() > 0:
                    location = (await location_p.text_content() or "").strip()
                    return name, location if location else None

            return name, None
        except Exception as e:
            logger.warning(f"Error getting name/location: {e}")
            return "Unknown", None

    async def _check_open_to_work(self) -> bool:
        """Check if profile has open to work badge."""
        try:
            # Look for open to work indicator
            img_title = await self.get_attribute_safe(
                ".pv-top-card-profile-picture img", "title", default=""
            )
            return "#OPEN_TO_WORK" in img_title.upper()
        except:
            return False

    async def _get_about(self) -> Optional[str]:
        """Extract about section using data-view-name anchor."""
        try:
            about_section = self.page.locator('[data-view-name="profile-card-about"]')
            if await about_section.count() == 0:
                return None

            # The about text is in a span with data-testid="expandable-text-box"
            text_box = about_section.locator('[data-testid="expandable-text-box"]').first
            if await text_box.count() > 0:
                about_text = await text_box.text_content()
                if about_text:
                    # Remove trailing "…\n more" from expandable text
                    about_text = about_text.strip()
                    for suffix in ["… more", "…\n more", "…more"]:
                        if about_text.endswith(suffix):
                            about_text = about_text[: -len(suffix)].strip()
                    return about_text if about_text else None

            # Fallback: get the first p tag after the heading
            about_p = about_section.locator('section p').first
            if await about_p.count() > 0:
                about_text = await about_p.text_content()
                return about_text.strip() if about_text else None

            return None
        except Exception as e:
            logger.debug(f"Error getting about section: {e}")
            return None

    async def _get_experiences(self, base_url: str) -> list[Experience]:
        """Extract experiences from the profile page Experience section."""
        experiences = []

        try:
            exp_section = self.page.locator('[data-view-name="profile-card-experience"]')
            if await exp_section.count() == 0:
                return experiences

            # Each experience item is in a div with componentkey starting with "entity-collection-item"
            items = await exp_section.locator('div[componentkey^="entity-collection-item"]').all()

            for item in items:
                try:
                    result = await self._parse_experience_card(item)
                    if result:
                        if isinstance(result, list):
                            experiences.extend(result)
                        else:
                            experiences.append(result)
                except Exception as e:
                    logger.debug(f"Error parsing experience card: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting experiences: {e}")

        return experiences

    async def _parse_experience_card(self, item) -> Optional[Experience | list[Experience]]:
        """Parse an experience card from LinkedIn's DOM.

        Handles both single-role and grouped (multi-role at one company) entries.
        Grouped entries have a <ul> with sub-positions.
        """
        try:
            # Check for grouped experience (ul with sub-positions)
            sub_list = item.locator('ul')
            if await sub_list.count() > 0:
                return await self._parse_grouped_experience(item)

            return await self._parse_single_experience(item)

        except Exception as e:
            logger.debug(f"Error parsing experience card: {e}")
            return None

    async def _parse_single_experience(self, item) -> Optional[Experience]:
        """Parse a single-role experience card."""
        try:
            links = await item.locator('a').all()
            if len(links) < 2:
                return None

            company_url = await links[0].get_attribute('href')
            detail_link = links[1]

            p_tags = await detail_link.locator('p').all()
            texts = []
            for p in p_tags:
                text = await p.text_content()
                if text and text.strip() and len(text.strip()) < 300:
                    texts.append(text.strip())

            if len(texts) < 2:
                return None

            position_title = texts[0]
            company_name = texts[1]
            work_times = texts[2] if len(texts) > 2 else ""
            location = texts[3] if len(texts) > 3 else ""

            from_date, to_date, duration = self._parse_work_times(work_times)

            description = await self._extract_description(item)

            return Experience(
                position_title=position_title,
                institution_name=company_name,
                linkedin_url=company_url,
                from_date=from_date,
                to_date=to_date,
                duration=duration,
                location=location if location else None,
                description=description,
            )

        except Exception as e:
            logger.debug(f"Error parsing single experience: {e}")
            return None

    async def _parse_grouped_experience(self, item) -> list[Experience]:
        """Parse a grouped experience (multiple roles at one company)."""
        experiences = []
        try:
            # Get company info from top-level links
            top_links = await item.locator(':scope > div a, :scope > div > div a').all()
            company_url = None
            company_name = None
            location = None

            if top_links:
                company_url = await top_links[0].get_attribute('href')
                if len(top_links) > 1:
                    detail_link = top_links[1]
                    p_tags = await detail_link.locator('p').all()
                    texts = []
                    for p in p_tags:
                        text = await p.text_content()
                        if text and text.strip():
                            texts.append(text.strip())
                    if texts:
                        company_name = texts[0]
                    if len(texts) > 2:
                        location = texts[2]

            # Parse each sub-position from li elements
            sub_items = await item.locator('ul > li').all()
            for sub_item in sub_items:
                try:
                    sub_link = sub_item.locator('a').first
                    if await sub_link.count() == 0:
                        continue

                    p_tags = await sub_link.locator('p').all()
                    texts = []
                    for p in p_tags:
                        text = await p.text_content()
                        if text and text.strip() and len(text.strip()) < 300:
                            texts.append(text.strip())

                    if not texts:
                        continue

                    position_title = texts[0]
                    work_times = ""
                    sub_location = location

                    # Remaining p tags could be subtitle, dates, location
                    for t in texts[1:]:
                        if self._looks_like_date_range(t):
                            work_times = t
                        elif self._looks_like_location(t):
                            sub_location = t

                    from_date, to_date, duration = self._parse_work_times(work_times)
                    description = await self._extract_description(sub_item)

                    experiences.append(Experience(
                        position_title=position_title,
                        institution_name=company_name or "",
                        linkedin_url=company_url,
                        from_date=from_date,
                        to_date=to_date,
                        duration=duration,
                        location=sub_location,
                        description=description,
                    ))

                except Exception as e:
                    logger.debug(f"Error parsing sub-position: {e}")
                    continue

        except Exception as e:
            logger.debug(f"Error parsing grouped experience: {e}")

        return experiences

    @staticmethod
    def _looks_like_date_range(text: str) -> bool:
        """Check if text looks like a date range (e.g. 'Jan 2020 - Present · 2 yrs')."""
        months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
                'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
        return (
            any(m in text for m in months)
            or 'Present' in text
            or ('·' in text and any(u in text for u in ['yr', 'mo']))
        )

    @staticmethod
    def _looks_like_location(text: str) -> bool:
        """Check if text looks like a location."""
        location_indicators = [',', 'Brazil', 'States', 'United', 'Remote',
                            'Area', 'Region', 'City']
        return any(ind in text for ind in location_indicators)

    async def _extract_description(self, container) -> Optional[str]:
        """Extract description text from an expandable text box."""
        text_box = container.locator('[data-testid="expandable-text-box"]').first
        if await text_box.count() > 0:
            desc_text = await text_box.text_content()
            if desc_text:
                desc_text = desc_text.strip()
                for suffix in ["… more", "…\n more", "…more"]:
                    if desc_text.endswith(suffix):
                        desc_text = desc_text[: -len(suffix)].strip()
                return desc_text if desc_text else None
        return None

    def _parse_work_times(
        self, work_times: str
    ) -> tuple[Optional[str], Optional[str], Optional[str]]:
        """
        Parse work times string into from_date, to_date, duration.

        Examples:
        - "2000 - Present · 26 yrs 1 mo" -> ("2000", "Present", "26 yrs 1 mo")
        - "Jan 2020 - Dec 2022 · 2 yrs" -> ("Jan 2020", "Dec 2022", "2 yrs")
        - "2015 - Present" -> ("2015", "Present", None)
        """
        if not work_times:
            return None, None, None

        try:
            # Split by · to separate date range from duration
            parts = work_times.split("·")
            times = parts[0].strip() if len(parts) > 0 else ""
            duration = parts[1].strip() if len(parts) > 1 else None

            # Parse dates - split by " - " to get from and to
            if " - " in times:
                date_parts = times.split(" - ")
                from_date = date_parts[0].strip()
                to_date = date_parts[1].strip() if len(date_parts) > 1 else ""
            else:
                from_date = times
                to_date = ""

            return from_date, to_date, duration
        except Exception as e:
            logger.debug(f"Error parsing work times '{work_times}': {e}")
            return None, None, None

    async def _get_educations(self, base_url: str) -> list[Education]:
        """Extract educations from the profile page Education section."""
        educations = []

        try:
            edu_section = self.page.locator('[data-view-name="profile-card-education"]')
            if await edu_section.count() == 0:
                return educations

            # Education items use componentkey but not with a consistent prefix,
            # so find all direct item containers by looking for the logo link pattern
            items = await edu_section.locator(
                'a[data-view-name="experience-company-logo-click"], '
                'a[href*="/school/"], a[href*="/company/"]'
            ).all()

            # Each logo link's grandparent is the item container
            seen_urls = set()
            for logo_link in items:
                try:
                    item_container = logo_link.locator('xpath=ancestor::div[contains(@class, "b902da86")][3]')
                    if await item_container.count() == 0:
                        continue

                    institution_url = await logo_link.get_attribute('href')
                    if institution_url in seen_urls:
                        continue
                    seen_urls.add(institution_url)

                    # Get the detail link (second a tag in the item)
                    all_links = await item_container.locator('a').all()
                    detail_link = all_links[1] if len(all_links) > 1 else None
                    if not detail_link:
                        continue

                    p_tags = await detail_link.locator('p').all()
                    texts = []
                    for p in p_tags:
                        text = await p.text_content()
                        if text and text.strip() and len(text.strip()) < 300:
                            texts.append(text.strip())

                    if not texts:
                        continue

                    institution_name = texts[0]
                    degree = None
                    times = ""

                    if len(texts) == 3:
                        degree = texts[1]
                        times = texts[2]
                    elif len(texts) == 2:
                        second = texts[1]
                        if any(c.isdigit() for c in second):
                            times = second
                        else:
                            degree = second

                    from_date, to_date = self._parse_education_times(times)

                    educations.append(Education(
                        institution_name=institution_name,
                        degree=degree.strip() if degree else None,
                        linkedin_url=institution_url,
                        from_date=from_date,
                        to_date=to_date,
                        description=None,
                    ))

                except Exception as e:
                    logger.debug(f"Error parsing education item: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting educations: {e}")

        return educations

    def _parse_education_times(self, times: str) -> tuple[Optional[str], Optional[str]]:
        """
        Parse education times string into from_date, to_date.

        Examples:
        - "1973 - 1977" -> ("1973", "1977")
        - "2015" -> ("2015", "2015")
        - "" -> (None, None)
        """
        if not times:
            return None, None

        try:
            # Split by " - " to get from and to dates
            if " - " in times:
                parts = times.split(" - ")
                from_date = parts[0].strip()
                to_date = parts[1].strip() if len(parts) > 1 else ""
            else:
                # Single year
                from_date = times.strip()
                to_date = times.strip()

            return from_date, to_date
        except Exception as e:
            logger.debug(f"Error parsing education times '{times}': {e}")
            return None, None

    async def _get_interests(self, base_url: str) -> list[Interest]:
        """Extract interests from the profile page Interests section."""
        interests = []

        try:
            interests_section = self.page.locator('[data-view-name="profile-card-interests"]')
            if await interests_section.count() == 0:
                return interests

            # Interests use radio buttons or tabs for categories
            tabs = await interests_section.locator('[role="tab"], [role="radio"]').all()

            for tab in tabs:
                try:
                    tab_name = await tab.text_content()
                    if not tab_name:
                        continue
                    tab_name = tab_name.strip()
                    category = self._map_interest_tab_to_category(tab_name)

                    await tab.click()
                    await self.wait_and_focus(0.5)

                    # Get interest items in the visible panel
                    panel = interests_section.locator('[role="tabpanel"]').first
                    if await panel.count() == 0:
                        # Fallback: items may be directly in the section
                        panel = interests_section

                    item_links = await panel.locator('a[href*="/company/"], a[href*="/school/"], a[href*="/in/"]').all()
                    seen_names = set()
                    for link in item_links:
                        href = await link.get_attribute("href")
                        p_tags = await link.locator('p').all()
                        name = None
                        if p_tags:
                            name_text = await p_tags[0].text_content()
                            if name_text:
                                name = name_text.strip()

                        if name and href and name not in seen_names:
                            seen_names.add(name)
                            interests.append(Interest(
                                name=name,
                                category=category,
                                linkedin_url=href,
                            ))

                except Exception as e:
                    logger.debug(f"Error processing interest tab: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting interests: {e}")

        return interests

    def _map_interest_tab_to_category(self, tab_name: str) -> str:
        tab_lower = tab_name.lower()
        if "compan" in tab_lower:
            return "company"
        elif "group" in tab_lower:
            return "group"
        elif "school" in tab_lower:
            return "school"
        elif "newsletter" in tab_lower:
            return "newsletter"
        elif "voice" in tab_lower or "influencer" in tab_lower:
            return "influencer"
        else:
            return tab_lower

    async def _get_accomplishments(self, base_url: str) -> list[Accomplishment]:
        accomplishments = []

        accomplishment_sections = [
            ("certifications", "certification"),
            ("honors", "honor"),
            ("publications", "publication"),
            ("patents", "patent"),
            ("courses", "course"),
            ("projects", "project"),
            ("languages", "language"),
            ("organizations", "organization"),
        ]

        for url_path, category in accomplishment_sections:
            try:
                section_url = urljoin(base_url, f"details/{url_path}/")
                await self.navigate_and_wait(section_url)
                await self.page.wait_for_selector("main", timeout=10000)
                await self.wait_and_focus(1)

                nothing_to_see = await self.page.locator(
                    'text="Nothing to see for now"'
                ).count()
                if nothing_to_see > 0:
                    continue

                main_list = self.page.locator(
                    ".pvs-list__container, main ul, main ol"
                ).first
                if await main_list.count() == 0:
                    continue

                items = await main_list.locator(".pvs-list__paged-list-item").all()
                if not items:
                    items = await main_list.locator("> li").all()

                seen_titles = set()
                for item in items:
                    try:
                        accomplishment = await self._parse_accomplishment_item(
                            item, category
                        )
                        if accomplishment and accomplishment.title not in seen_titles:
                            seen_titles.add(accomplishment.title)
                            accomplishments.append(accomplishment)
                    except Exception as e:
                        logger.debug(f"Error parsing {category} item: {e}")
                        continue

            except Exception as e:
                logger.debug(f"Error getting {category}s: {e}")
                continue

        return accomplishments

    async def _parse_accomplishment_item(
        self, item, category: str
    ) -> Optional[Accomplishment]:
        try:
            entity = item.locator(
                'div[data-view-name="profile-component-entity"]'
            ).first
            if await entity.count() > 0:
                spans = await entity.locator('span[aria-hidden="true"]').all()
            else:
                spans = await item.locator('span[aria-hidden="true"]').all()

            title = ""
            issuer = ""
            issued_date = ""
            credential_id = ""

            for i, span in enumerate(spans[:5]):
                text = await span.text_content()
                if not text:
                    continue
                text = text.strip()

                if len(text) > 500:
                    continue

                if i == 0:
                    title = text
                elif "Issued by" in text:
                    parts = text.split("·")
                    issuer = parts[0].replace("Issued by", "").strip()
                    if len(parts) > 1:
                        issued_date = parts[1].strip()
                elif "Issued " in text and not issued_date:
                    issued_date = text.replace("Issued ", "")
                elif "Credential ID" in text:
                    credential_id = text.replace("Credential ID ", "")
                elif i == 1 and not issuer:
                    issuer = text
                elif (
                    any(
                        month in text
                        for month in [
                            "Jan",
                            "Feb",
                            "Mar",
                            "Apr",
                            "May",
                            "Jun",
                            "Jul",
                            "Aug",
                            "Sep",
                            "Oct",
                            "Nov",
                            "Dec",
                        ]
                    )
                    and not issued_date
                ):
                    if "·" in text:
                        parts = text.split("·")
                        issued_date = parts[0].strip()
                    else:
                        issued_date = text

            link = item.locator('a[href*="credential"], a[href*="verify"]').first
            credential_url = (
                await link.get_attribute("href") if await link.count() > 0 else None
            )

            if not title or len(title) > 200:
                return None

            return Accomplishment(
                category=category,
                title=title,
                issuer=issuer if issuer else None,
                issued_date=issued_date if issued_date else None,
                credential_id=credential_id if credential_id else None,
                credential_url=credential_url,
            )

        except Exception as e:
            logger.debug(f"Error parsing accomplishment: {e}")
            return None

    async def _get_contacts(self, base_url: str) -> list[Contact]:
        """Extract contact info from the contact-info overlay dialog."""
        contacts = []

        try:
            contact_url = urljoin(base_url, "overlay/contact-info/")
            await self.navigate_and_wait(contact_url)
            await self.wait_and_focus(1)

            dialog = self.page.locator('dialog, [role="dialog"]').first
            if await dialog.count() == 0:
                logger.warning("Contact info dialog not found")
                return contacts

            contact_sections = await dialog.locator('h3').all()
            
            for section_heading in contact_sections:
                try:
                    heading_text = await section_heading.text_content()
                    if not heading_text:
                        continue
                    heading_text = heading_text.strip().lower()
                    
                    section_container = section_heading.locator('xpath=ancestor::*[1]')
                    if await section_container.count() == 0:
                        continue
                    
                    contact_type = self._map_contact_heading_to_type(heading_text)
                    if not contact_type:
                        continue
                    
                    links = await section_container.locator('a').all()
                    for link in links:
                        href = await link.get_attribute('href')
                        text = await link.text_content()
                        if href and text:
                            text = text.strip()
                            label = None
                            sibling_text = await section_container.locator('span, generic').all()
                            for sib in sibling_text:
                                sib_text = await sib.text_content()
                                if sib_text and sib_text.strip().startswith('(') and sib_text.strip().endswith(')'):
                                    label = sib_text.strip()[1:-1]
                                    break
                            
                            if contact_type == "linkedin":
                                contacts.append(Contact(type=contact_type, value=href, label=label))
                            elif contact_type == "email" and "mailto:" in href:
                                contacts.append(Contact(type=contact_type, value=href.replace("mailto:", ""), label=label))
                            else:
                                contacts.append(Contact(type=contact_type, value=text, label=label))
                    
                    if contact_type == "birthday" and not links:
                        birthday_text = await section_container.text_content()
                        if birthday_text:
                            birthday_value = birthday_text.replace(heading_text, "").replace("Birthday", "").strip()
                            if birthday_value:
                                contacts.append(Contact(type="birthday", value=birthday_value))
                    
                    if contact_type == "phone" and not links:
                        phone_text = await section_container.text_content()
                        if phone_text:
                            phone_value = phone_text.replace(heading_text, "").replace("Phone", "").strip()
                            if phone_value:
                                contacts.append(Contact(type="phone", value=phone_value))
                    
                    if contact_type == "address" and not links:
                        address_text = await section_container.text_content()
                        if address_text:
                            address_value = address_text.replace(heading_text, "").replace("Address", "").strip()
                            if address_value:
                                contacts.append(Contact(type="address", value=address_value))
                                
                except Exception as e:
                    logger.debug(f"Error parsing contact section: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting contacts: {e}")

        return contacts

    def _map_contact_heading_to_type(self, heading: str) -> Optional[str]:
        """Map contact section heading to contact type."""
        heading = heading.lower()
        if "profile" in heading:
            return "linkedin"
        elif "website" in heading:
            return "website"
        elif "email" in heading:
            return "email"
        elif "phone" in heading:
            return "phone"
        elif "twitter" in heading or "x.com" in heading:
            return "twitter"
        elif "birthday" in heading:
            return "birthday"
        elif "address" in heading:
            return "address"
        return None

[shubham-fluxon]

Wondering how often does linkedin update the dom structure?

[diligentxcao]

Thanks man. The code you gave is working. Just a small thing to mention. At line 11, replace this code谢谢兄弟。你给的代码正在工作。只是想提一下。在第11行，替换该代码 logger = logging.getLogger(name)logger = logging.getLogger（name） with this  用这个 logger = logging.getLogger(__name__).logger = logging.getLogger（__name__）。

logger = logging.getLogger(name) this one doesn't worked for me. It throwed error.logger = logging.getLogger（name） 这个对我没用。它报错了。 But, logger = logging.getLogger(__name__) this one worked for me.但是，logger = logging.getLogger（__name__）这个方法对我来说很有效。

Also there is some indentation error, as you pasted the code in the comment, the indentation is gone. So copy pasting leads to indentation error.还有缩进错误，当你把代码粘贴到注释里时，缩进已经消失了。所以复制粘贴会导致缩进错误。

So, just copy pasting won't work guys. Add a tab space to all the code below the所以，单纯复制粘贴是行不通的，伙计们。在所有代码下方添加制表符空格 class PersonScraper(BaseScraper): .类 PersonScraper（BaseScraper）：

Here is the ready to copy and paste working code.这里有现成的可复制粘贴可用代码。

import logging
from typing import Optional
from urllib.parse import urljoin
from playwright.async_api import Page

from .base import BaseScraper
from ..models import Person, Experience, Education, Accomplishment, Interest, Contact
from ..callbacks import ProgressCallback, SilentCallback
from ..core.exceptions import ScrapingError

logger = logging.getLogger(__name__)

class PersonScraper(BaseScraper):
    """Async scraper for LinkedIn person profiles."""

    def __init__(self, page: Page, callback: Optional[ProgressCallback] = None):
        """
        Initialize person scraper.

        Args:
            page: Playwright page object
            callback: Progress callback
        """
        super().__init__(page, callback)

    async def scrape(self, linkedin_url: str) -> Person:
        """
        Scrape a LinkedIn person profile.

        Args:
            linkedin_url: LinkedIn profile URL

        Returns:
            Person object with all scraped data

        Raises:
            AuthenticationError: If not logged in
            ScrapingError: If scraping fails
        """
        await self.callback.on_start("person", linkedin_url)

        try:
            # Navigate to profile first (this loads the page with our session)
            await self.navigate_and_wait(linkedin_url)
            await self.callback.on_progress("Navigated to profile", 10)

            # Now check if logged in
            await self.ensure_logged_in()

            # Wait for main content
            await self.page.wait_for_selector("main", timeout=10000)
            await self.wait_and_focus(1)

            # Get name and location
            name, location = await self._get_name_and_location()
            await self.callback.on_progress(f"Got name: {name}", 20)

            # Check open to work
            open_to_work = await self._check_open_to_work()

            # Get about
            about = await self._get_about()
            await self.callback.on_progress("Got about section", 30)

            # Scroll to load content
            await self.scroll_page_to_half()
            await self.scroll_page_to_bottom(pause_time=0.5, max_scrolls=3)

            # Get experiences
            experiences = await self._get_experiences(linkedin_url)
            await self.callback.on_progress(f"Got {len(experiences)} experiences", 60)

            educations = await self._get_educations(linkedin_url)
            await self.callback.on_progress(f"Got {len(educations)} educations", 50)

            interests = await self._get_interests(linkedin_url)
            await self.callback.on_progress(f"Got {len(interests)} interests", 65)

            accomplishments = await self._get_accomplishments(linkedin_url)
            await self.callback.on_progress(
                f"Got {len(accomplishments)} accomplishments", 85
            )

            contacts = await self._get_contacts(linkedin_url)
            await self.callback.on_progress(f"Got {len(contacts)} contacts", 95)

            person = Person(
                linkedin_url=linkedin_url,
                name=name,
                location=location,
                about=about,
                open_to_work=open_to_work,
                experiences=experiences,
                educations=educations,
                interests=interests,
                accomplishments=accomplishments,
                contacts=contacts,
            )

            await self.callback.on_progress("Scraping complete", 100)
            await self.callback.on_complete("person", person)

            return person

        except Exception as e:
            await self.callback.on_error(e)
            raise ScrapingError(f"Failed to scrape person profile: {e}")

    async def _get_name_and_location(self) -> tuple[str, Optional[str]]:
        """Extract name and location from profile topcard."""
        try:
            # The topcard is in [data-view-name="profile-main-level"]
            topcard = self.page.locator('[data-view-name="profile-main-level"]')
            if await topcard.count() == 0:
                # Fallback to first section
                topcard = self.page.locator('main section').first

            # LinkedIn now uses h2 for the profile name
            name_el = topcard.locator('h2').first
            name = "Unknown"
            if await name_el.count() > 0:
                name = (await name_el.text_content() or "Unknown").strip()

            # Location is in a p tag in the same container as the "Contact info" link
            contact_link = topcard.locator('a[href*="overlay/contact-info"]')
            if await contact_link.count() > 0:
                location_container = contact_link.locator('xpath=ancestor::div[1]')
                location_p = location_container.locator('p').first
                if await location_p.count() > 0:
                    location = (await location_p.text_content() or "").strip()
                    return name, location if location else None

            return name, None
        except Exception as e:
            logger.warning(f"Error getting name/location: {e}")
            return "Unknown", None

    async def _check_open_to_work(self) -> bool:
        """Check if profile has open to work badge."""
        try:
            # Look for open to work indicator
            img_title = await self.get_attribute_safe(
                ".pv-top-card-profile-picture img", "title", default=""
            )
            return "#OPEN_TO_WORK" in img_title.upper()
        except:
            return False

    async def _get_about(self) -> Optional[str]:
        """Extract about section using data-view-name anchor."""
        try:
            about_section = self.page.locator('[data-view-name="profile-card-about"]')
            if await about_section.count() == 0:
                return None

            # The about text is in a span with data-testid="expandable-text-box"
            text_box = about_section.locator('[data-testid="expandable-text-box"]').first
            if await text_box.count() > 0:
                about_text = await text_box.text_content()
                if about_text:
                    # Remove trailing "…\n more" from expandable text
                    about_text = about_text.strip()
                    for suffix in ["… more", "…\n more", "…more"]:
                        if about_text.endswith(suffix):
                            about_text = about_text[: -len(suffix)].strip()
                    return about_text if about_text else None

            # Fallback: get the first p tag after the heading
            about_p = about_section.locator('section p').first
            if await about_p.count() > 0:
                about_text = await about_p.text_content()
                return about_text.strip() if about_text else None

            return None
        except Exception as e:
            logger.debug(f"Error getting about section: {e}")
            return None

    async def _get_experiences(self, base_url: str) -> list[Experience]:
        """Extract experiences from the profile page Experience section."""
        experiences = []

        try:
            exp_section = self.page.locator('[data-view-name="profile-card-experience"]')
            if await exp_section.count() == 0:
                return experiences

            # Each experience item is in a div with componentkey starting with "entity-collection-item"
            items = await exp_section.locator('div[componentkey^="entity-collection-item"]').all()

            for item in items:
                try:
                    result = await self._parse_experience_card(item)
                    if result:
                        if isinstance(result, list):
                            experiences.extend(result)
                        else:
                            experiences.append(result)
                except Exception as e:
                    logger.debug(f"Error parsing experience card: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting experiences: {e}")

        return experiences

    async def _parse_experience_card(self, item) -> Optional[Experience | list[Experience]]:
        """Parse an experience card from LinkedIn's DOM.

        Handles both single-role and grouped (multi-role at one company) entries.
        Grouped entries have a <ul> with sub-positions.
        """
        try:
            # Check for grouped experience (ul with sub-positions)
            sub_list = item.locator('ul')
            if await sub_list.count() > 0:
                return await self._parse_grouped_experience(item)

            return await self._parse_single_experience(item)

        except Exception as e:
            logger.debug(f"Error parsing experience card: {e}")
            return None

    async def _parse_single_experience(self, item) -> Optional[Experience]:
        """Parse a single-role experience card."""
        try:
            links = await item.locator('a').all()
            if len(links) < 2:
                return None

            company_url = await links[0].get_attribute('href')
            detail_link = links[1]

            p_tags = await detail_link.locator('p').all()
            texts = []
            for p in p_tags:
                text = await p.text_content()
                if text and text.strip() and len(text.strip()) < 300:
                    texts.append(text.strip())

            if len(texts) < 2:
                return None

            position_title = texts[0]
            company_name = texts[1]
            work_times = texts[2] if len(texts) > 2 else ""
            location = texts[3] if len(texts) > 3 else ""

            from_date, to_date, duration = self._parse_work_times(work_times)

            description = await self._extract_description(item)

            return Experience(
                position_title=position_title,
                institution_name=company_name,
                linkedin_url=company_url,
                from_date=from_date,
                to_date=to_date,
                duration=duration,
                location=location if location else None,
                description=description,
            )

        except Exception as e:
            logger.debug(f"Error parsing single experience: {e}")
            return None

    async def _parse_grouped_experience(self, item) -> list[Experience]:
        """Parse a grouped experience (multiple roles at one company)."""
        experiences = []
        try:
            # Get company info from top-level links
            top_links = await item.locator(':scope > div a, :scope > div > div a').all()
            company_url = None
            company_name = None
            location = None

            if top_links:
                company_url = await top_links[0].get_attribute('href')
                if len(top_links) > 1:
                    detail_link = top_links[1]
                    p_tags = await detail_link.locator('p').all()
                    texts = []
                    for p in p_tags:
                        text = await p.text_content()
                        if text and text.strip():
                            texts.append(text.strip())
                    if texts:
                        company_name = texts[0]
                    if len(texts) > 2:
                        location = texts[2]

            # Parse each sub-position from li elements
            sub_items = await item.locator('ul > li').all()
            for sub_item in sub_items:
                try:
                    sub_link = sub_item.locator('a').first
                    if await sub_link.count() == 0:
                        continue

                    p_tags = await sub_link.locator('p').all()
                    texts = []
                    for p in p_tags:
                        text = await p.text_content()
                        if text and text.strip() and len(text.strip()) < 300:
                            texts.append(text.strip())

                    if not texts:
                        continue

                    position_title = texts[0]
                    work_times = ""
                    sub_location = location

                    # Remaining p tags could be subtitle, dates, location
                    for t in texts[1:]:
                        if self._looks_like_date_range(t):
                            work_times = t
                        elif self._looks_like_location(t):
                            sub_location = t

                    from_date, to_date, duration = self._parse_work_times(work_times)
                    description = await self._extract_description(sub_item)

                    experiences.append(Experience(
                        position_title=position_title,
                        institution_name=company_name or "",
                        linkedin_url=company_url,
                        from_date=from_date,
                        to_date=to_date,
                        duration=duration,
                        location=sub_location,
                        description=description,
                    ))

                except Exception as e:
                    logger.debug(f"Error parsing sub-position: {e}")
                    continue

        except Exception as e:
            logger.debug(f"Error parsing grouped experience: {e}")

        return experiences

    @staticmethod
    def _looks_like_date_range(text: str) -> bool:
        """Check if text looks like a date range (e.g. 'Jan 2020 - Present · 2 yrs')."""
        months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
                'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
        return (
            any(m in text for m in months)
            or 'Present' in text
            or ('·' in text and any(u in text for u in ['yr', 'mo']))
        )

    @staticmethod
    def _looks_like_location(text: str) -> bool:
        """Check if text looks like a location."""
        location_indicators = [',', 'Brazil', 'States', 'United', 'Remote',
                            'Area', 'Region', 'City']
        return any(ind in text for ind in location_indicators)

    async def _extract_description(self, container) -> Optional[str]:
        """Extract description text from an expandable text box."""
        text_box = container.locator('[data-testid="expandable-text-box"]').first
        if await text_box.count() > 0:
            desc_text = await text_box.text_content()
            if desc_text:
                desc_text = desc_text.strip()
                for suffix in ["… more", "…\n more", "…more"]:
                    if desc_text.endswith(suffix):
                        desc_text = desc_text[: -len(suffix)].strip()
                return desc_text if desc_text else None
        return None

    def _parse_work_times(
        self, work_times: str
    ) -> tuple[Optional[str], Optional[str], Optional[str]]:
        """
        Parse work times string into from_date, to_date, duration.

        Examples:
        - "2000 - Present · 26 yrs 1 mo" -> ("2000", "Present", "26 yrs 1 mo")
        - "Jan 2020 - Dec 2022 · 2 yrs" -> ("Jan 2020", "Dec 2022", "2 yrs")
        - "2015 - Present" -> ("2015", "Present", None)
        """
        if not work_times:
            return None, None, None

        try:
            # Split by · to separate date range from duration
            parts = work_times.split("·")
            times = parts[0].strip() if len(parts) > 0 else ""
            duration = parts[1].strip() if len(parts) > 1 else None

            # Parse dates - split by " - " to get from and to
            if " - " in times:
                date_parts = times.split(" - ")
                from_date = date_parts[0].strip()
                to_date = date_parts[1].strip() if len(date_parts) > 1 else ""
            else:
                from_date = times
                to_date = ""

            return from_date, to_date, duration
        except Exception as e:
            logger.debug(f"Error parsing work times '{work_times}': {e}")
            return None, None, None

    async def _get_educations(self, base_url: str) -> list[Education]:
        """Extract educations from the profile page Education section."""
        educations = []

        try:
            edu_section = self.page.locator('[data-view-name="profile-card-education"]')
            if await edu_section.count() == 0:
                return educations

            # Education items use componentkey but not with a consistent prefix,
            # so find all direct item containers by looking for the logo link pattern
            items = await edu_section.locator(
                'a[data-view-name="experience-company-logo-click"], '
                'a[href*="/school/"], a[href*="/company/"]'
            ).all()

            # Each logo link's grandparent is the item container
            seen_urls = set()
            for logo_link in items:
                try:
                    item_container = logo_link.locator('xpath=ancestor::div[contains(@class, "b902da86")][3]')
                    if await item_container.count() == 0:
                        continue

                    institution_url = await logo_link.get_attribute('href')
                    if institution_url in seen_urls:
                        continue
                    seen_urls.add(institution_url)

                    # Get the detail link (second a tag in the item)
                    all_links = await item_container.locator('a').all()
                    detail_link = all_links[1] if len(all_links) > 1 else None
                    if not detail_link:
                        continue

                    p_tags = await detail_link.locator('p').all()
                    texts = []
                    for p in p_tags:
                        text = await p.text_content()
                        if text and text.strip() and len(text.strip()) < 300:
                            texts.append(text.strip())

                    if not texts:
                        continue

                    institution_name = texts[0]
                    degree = None
                    times = ""

                    if len(texts) == 3:
                        degree = texts[1]
                        times = texts[2]
                    elif len(texts) == 2:
                        second = texts[1]
                        if any(c.isdigit() for c in second):
                            times = second
                        else:
                            degree = second

                    from_date, to_date = self._parse_education_times(times)

                    educations.append(Education(
                        institution_name=institution_name,
                        degree=degree.strip() if degree else None,
                        linkedin_url=institution_url,
                        from_date=from_date,
                        to_date=to_date,
                        description=None,
                    ))

                except Exception as e:
                    logger.debug(f"Error parsing education item: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting educations: {e}")

        return educations

    def _parse_education_times(self, times: str) -> tuple[Optional[str], Optional[str]]:
        """
        Parse education times string into from_date, to_date.

        Examples:
        - "1973 - 1977" -> ("1973", "1977")
        - "2015" -> ("2015", "2015")
        - "" -> (None, None)
        """
        if not times:
            return None, None

        try:
            # Split by " - " to get from and to dates
            if " - " in times:
                parts = times.split(" - ")
                from_date = parts[0].strip()
                to_date = parts[1].strip() if len(parts) > 1 else ""
            else:
                # Single year
                from_date = times.strip()
                to_date = times.strip()

            return from_date, to_date
        except Exception as e:
            logger.debug(f"Error parsing education times '{times}': {e}")
            return None, None

    async def _get_interests(self, base_url: str) -> list[Interest]:
        """Extract interests from the profile page Interests section."""
        interests = []

        try:
            interests_section = self.page.locator('[data-view-name="profile-card-interests"]')
            if await interests_section.count() == 0:
                return interests

            # Interests use radio buttons or tabs for categories
            tabs = await interests_section.locator('[role="tab"], [role="radio"]').all()

            for tab in tabs:
                try:
                    tab_name = await tab.text_content()
                    if not tab_name:
                        continue
                    tab_name = tab_name.strip()
                    category = self._map_interest_tab_to_category(tab_name)

                    await tab.click()
                    await self.wait_and_focus(0.5)

                    # Get interest items in the visible panel
                    panel = interests_section.locator('[role="tabpanel"]').first
                    if await panel.count() == 0:
                        # Fallback: items may be directly in the section
                        panel = interests_section

                    item_links = await panel.locator('a[href*="/company/"], a[href*="/school/"], a[href*="/in/"]').all()
                    seen_names = set()
                    for link in item_links:
                        href = await link.get_attribute("href")
                        p_tags = await link.locator('p').all()
                        name = None
                        if p_tags:
                            name_text = await p_tags[0].text_content()
                            if name_text:
                                name = name_text.strip()

                        if name and href and name not in seen_names:
                            seen_names.add(name)
                            interests.append(Interest(
                                name=name,
                                category=category,
                                linkedin_url=href,
                            ))

                except Exception as e:
                    logger.debug(f"Error processing interest tab: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting interests: {e}")

        return interests

    def _map_interest_tab_to_category(self, tab_name: str) -> str:
        tab_lower = tab_name.lower()
        if "compan" in tab_lower:
            return "company"
        elif "group" in tab_lower:
            return "group"
        elif "school" in tab_lower:
            return "school"
        elif "newsletter" in tab_lower:
            return "newsletter"
        elif "voice" in tab_lower or "influencer" in tab_lower:
            return "influencer"
        else:
            return tab_lower

    async def _get_accomplishments(self, base_url: str) -> list[Accomplishment]:
        accomplishments = []

        accomplishment_sections = [
            ("certifications", "certification"),
            ("honors", "honor"),
            ("publications", "publication"),
            ("patents", "patent"),
            ("courses", "course"),
            ("projects", "project"),
            ("languages", "language"),
            ("organizations", "organization"),
        ]

        for url_path, category in accomplishment_sections:
            try:
                section_url = urljoin(base_url, f"details/{url_path}/")
                await self.navigate_and_wait(section_url)
                await self.page.wait_for_selector("main", timeout=10000)
                await self.wait_and_focus(1)

                nothing_to_see = await self.page.locator(
                    'text="Nothing to see for now"'
                ).count()
                if nothing_to_see > 0:
                    continue

                main_list = self.page.locator(
                    ".pvs-list__container, main ul, main ol"
                ).first
                if await main_list.count() == 0:
                    continue

                items = await main_list.locator(".pvs-list__paged-list-item").all()
                if not items:
                    items = await main_list.locator("> li").all()

                seen_titles = set()
                for item in items:
                    try:
                        accomplishment = await self._parse_accomplishment_item(
                            item, category
                        )
                        if accomplishment and accomplishment.title not in seen_titles:
                            seen_titles.add(accomplishment.title)
                            accomplishments.append(accomplishment)
                    except Exception as e:
                        logger.debug(f"Error parsing {category} item: {e}")
                        continue

            except Exception as e:
                logger.debug(f"Error getting {category}s: {e}")
                continue

        return accomplishments

    async def _parse_accomplishment_item(
        self, item, category: str
    ) -> Optional[Accomplishment]:
        try:
            entity = item.locator(
                'div[data-view-name="profile-component-entity"]'
            ).first
            if await entity.count() > 0:
                spans = await entity.locator('span[aria-hidden="true"]').all()
            else:
                spans = await item.locator('span[aria-hidden="true"]').all()

            title = ""
            issuer = ""
            issued_date = ""
            credential_id = ""

            for i, span in enumerate(spans[:5]):
                text = await span.text_content()
                if not text:
                    continue
                text = text.strip()

                if len(text) > 500:
                    continue

                if i == 0:
                    title = text
                elif "Issued by" in text:
                    parts = text.split("·")
                    issuer = parts[0].replace("Issued by", "").strip()
                    if len(parts) > 1:
                        issued_date = parts[1].strip()
                elif "Issued " in text and not issued_date:
                    issued_date = text.replace("Issued ", "")
                elif "Credential ID" in text:
                    credential_id = text.replace("Credential ID ", "")
                elif i == 1 and not issuer:
                    issuer = text
                elif (
                    any(
                        month in text
                        for month in [
                            "Jan",
                            "Feb",
                            "Mar",
                            "Apr",
                            "May",
                            "Jun",
                            "Jul",
                            "Aug",
                            "Sep",
                            "Oct",
                            "Nov",
                            "Dec",
                        ]
                    )
                    and not issued_date
                ):
                    if "·" in text:
                        parts = text.split("·")
                        issued_date = parts[0].strip()
                    else:
                        issued_date = text

            link = item.locator('a[href*="credential"], a[href*="verify"]').first
            credential_url = (
                await link.get_attribute("href") if await link.count() > 0 else None
            )

            if not title or len(title) > 200:
                return None

            return Accomplishment(
                category=category,
                title=title,
                issuer=issuer if issuer else None,
                issued_date=issued_date if issued_date else None,
                credential_id=credential_id if credential_id else None,
                credential_url=credential_url,
            )

        except Exception as e:
            logger.debug(f"Error parsing accomplishment: {e}")
            return None

    async def _get_contacts(self, base_url: str) -> list[Contact]:
        """Extract contact info from the contact-info overlay dialog."""
        contacts = []

        try:
            contact_url = urljoin(base_url, "overlay/contact-info/")
            await self.navigate_and_wait(contact_url)
            await self.wait_and_focus(1)

            dialog = self.page.locator('dialog, [role="dialog"]').first
            if await dialog.count() == 0:
                logger.warning("Contact info dialog not found")
                return contacts

            contact_sections = await dialog.locator('h3').all()
            
            for section_heading in contact_sections:
                try:
                    heading_text = await section_heading.text_content()
                    if not heading_text:
                        continue
                    heading_text = heading_text.strip().lower()
                    
                    section_container = section_heading.locator('xpath=ancestor::*[1]')
                    if await section_container.count() == 0:
                        continue
                    
                    contact_type = self._map_contact_heading_to_type(heading_text)
                    if not contact_type:
                        continue
                    
                    links = await section_container.locator('a').all()
                    for link in links:
                        href = await link.get_attribute('href')
                        text = await link.text_content()
                        if href and text:
                            text = text.strip()
                            label = None
                            sibling_text = await section_container.locator('span, generic').all()
                            for sib in sibling_text:
                                sib_text = await sib.text_content()
                                if sib_text and sib_text.strip().startswith('(') and sib_text.strip().endswith(')'):
                                    label = sib_text.strip()[1:-1]
                                    break
                            
                            if contact_type == "linkedin":
                                contacts.append(Contact(type=contact_type, value=href, label=label))
                            elif contact_type == "email" and "mailto:" in href:
                                contacts.append(Contact(type=contact_type, value=href.replace("mailto:", ""), label=label))
                            else:
                                contacts.append(Contact(type=contact_type, value=text, label=label))
                    
                    if contact_type == "birthday" and not links:
                        birthday_text = await section_container.text_content()
                        if birthday_text:
                            birthday_value = birthday_text.replace(heading_text, "").replace("Birthday", "").strip()
                            if birthday_value:
                                contacts.append(Contact(type="birthday", value=birthday_value))
                    
                    if contact_type == "phone" and not links:
                        phone_text = await section_container.text_content()
                        if phone_text:
                            phone_value = phone_text.replace(heading_text, "").replace("Phone", "").strip()
                            if phone_value:
                                contacts.append(Contact(type="phone", value=phone_value))
                    
                    if contact_type == "address" and not links:
                        address_text = await section_container.text_content()
                        if address_text:
                            address_value = address_text.replace(heading_text, "").replace("Address", "").strip()
                            if address_value:
                                contacts.append(Contact(type="address", value=address_value))
                                
                except Exception as e:
                    logger.debug(f"Error parsing contact section: {e}")
                    continue

        except Exception as e:
            logger.warning(f"Error getting contacts: {e}")

        return contacts

    def _map_contact_heading_to_type(self, heading: str) -> Optional[str]:
        """Map contact section heading to contact type."""
        heading = heading.lower()
        if "profile" in heading:
            return "linkedin"
        elif "website" in heading:
            return "website"
        elif "email" in heading:
            return "email"
        elif "phone" in heading:
            return "phone"
        elif "twitter" in heading or "x.com" in heading:
            return "twitter"
        elif "birthday" in heading:
            return "birthday"
        elif "address" in heading:
            return "address"
        return None

Same here. LinkedIn changed the DOM structure. Fixed person.py

` """Person/Profile scraper for LinkedIn."""

import logging from typing import Optional from urllib.parse import urljoin from playwright.async_api import Page

from .base import BaseScraper from ..models import Person, Experience, Education, Accomplishment, Interest, Contact from ..callbacks import ProgressCallback, SilentCallback from ..core.exceptions import ScrapingError

logger = logging.getLogger(name)

class PersonScraper(BaseScraper): """Async scraper for LinkedIn person profiles."""

def __init__(self, page: Page, callback: Optional[ProgressCallback] = None):
    """
    Initialize person scraper.

    Args:
        page: Playwright page object
        callback: Progress callback
    """
    super().__init__(page, callback)

async def scrape(self, linkedin_url: str) -> Person:
    """
    Scrape a LinkedIn person profile.

    Args:
        linkedin_url: LinkedIn profile URL

    Returns:
        Person object with all scraped data

    Raises:
        AuthenticationError: If not logged in
        ScrapingError: If scraping fails
    """
    await self.callback.on_start("person", linkedin_url)

    try:
        # Navigate to profile first (this loads the page with our session)
        await self.navigate_and_wait(linkedin_url)
        await self.callback.on_progress("Navigated to profile", 10)

        # Now check if logged in
        await self.ensure_logged_in()

        # Wait for main content
        await self.page.wait_for_selector("main", timeout=10000)
        await self.wait_and_focus(1)

        # Get name and location
        name, location = await self._get_name_and_location()
        await self.callback.on_progress(f"Got name: {name}", 20)

        # Check open to work
        open_to_work = await self._check_open_to_work()

        # Get about
        about = await self._get_about()
        await self.callback.on_progress("Got about section", 30)

        # Scroll to load content
        await self.scroll_page_to_half()
        await self.scroll_page_to_bottom(pause_time=0.5, max_scrolls=3)

        # Get experiences
        experiences = await self._get_experiences(linkedin_url)
        await self.callback.on_progress(f"Got {len(experiences)} experiences", 60)

        educations = await self._get_educations(linkedin_url)
        await self.callback.on_progress(f"Got {len(educations)} educations", 50)

        interests = await self._get_interests(linkedin_url)
        await self.callback.on_progress(f"Got {len(interests)} interests", 65)

        accomplishments = await self._get_accomplishments(linkedin_url)
        await self.callback.on_progress(
            f"Got {len(accomplishments)} accomplishments", 85
        )

        contacts = await self._get_contacts(linkedin_url)
        await self.callback.on_progress(f"Got {len(contacts)} contacts", 95)

        person = Person(
            linkedin_url=linkedin_url,
            name=name,
            location=location,
            about=about,
            open_to_work=open_to_work,
            experiences=experiences,
            educations=educations,
            interests=interests,
            accomplishments=accomplishments,
            contacts=contacts,
        )

        await self.callback.on_progress("Scraping complete", 100)
        await self.callback.on_complete("person", person)

        return person

    except Exception as e:
        await self.callback.on_error(e)
        raise ScrapingError(f"Failed to scrape person profile: {e}")

async def _get_name_and_location(self) -> tuple[str, Optional[str]]:
    """Extract name and location from profile topcard."""
    try:
        # The topcard is in [data-view-name="profile-main-level"]
        topcard = self.page.locator('[data-view-name="profile-main-level"]')
        if await topcard.count() == 0:
            # Fallback to first section
            topcard = self.page.locator('main section').first

        # LinkedIn now uses h2 for the profile name
        name_el = topcard.locator('h2').first
        name = "Unknown"
        if await name_el.count() > 0:
            name = (await name_el.text_content() or "Unknown").strip()

        # Location is in a p tag in the same container as the "Contact info" link
        contact_link = topcard.locator('a[href*="overlay/contact-info"]')
        if await contact_link.count() > 0:
            location_container = contact_link.locator('xpath=ancestor::div[1]')
            location_p = location_container.locator('p').first
            if await location_p.count() > 0:
                location = (await location_p.text_content() or "").strip()
                return name, location if location else None

        return name, None
    except Exception as e:
        logger.warning(f"Error getting name/location: {e}")
        return "Unknown", None

async def _check_open_to_work(self) -> bool:
    """Check if profile has open to work badge."""
    try:
        # Look for open to work indicator
        img_title = await self.get_attribute_safe(
            ".pv-top-card-profile-picture img", "title", default=""
        )
        return "#OPEN_TO_WORK" in img_title.upper()
    except:
        return False

async def _get_about(self) -> Optional[str]:
    """Extract about section using data-view-name anchor."""
    try:
        about_section = self.page.locator('[data-view-name="profile-card-about"]')
        if await about_section.count() == 0:
            return None

        # The about text is in a span with data-testid="expandable-text-box"
        text_box = about_section.locator('[data-testid="expandable-text-box"]').first
        if await text_box.count() > 0:
            about_text = await text_box.text_content()
            if about_text:
                # Remove trailing "…\n more" from expandable text
                about_text = about_text.strip()
                for suffix in ["… more", "…\n more", "…more"]:
                    if about_text.endswith(suffix):
                        about_text = about_text[: -len(suffix)].strip()
                return about_text if about_text else None

        # Fallback: get the first p tag after the heading
        about_p = about_section.locator('section p').first
        if await about_p.count() > 0:
            about_text = await about_p.text_content()
            return about_text.strip() if about_text else None

        return None
    except Exception as e:
        logger.debug(f"Error getting about section: {e}")
        return None

async def _get_experiences(self, base_url: str) -> list[Experience]:
    """Extract experiences from the profile page Experience section."""
    experiences = []

    try:
        exp_section = self.page.locator('[data-view-name="profile-card-experience"]')
        if await exp_section.count() == 0:
            return experiences

        # Each experience item is in a div with componentkey starting with "entity-collection-item"
        items = await exp_section.locator('div[componentkey^="entity-collection-item"]').all()

        for item in items:
            try:
                result = await self._parse_experience_card(item)
                if result:
                    if isinstance(result, list):
                        experiences.extend(result)
                    else:
                        experiences.append(result)
            except Exception as e:
                logger.debug(f"Error parsing experience card: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting experiences: {e}")

    return experiences

async def _parse_experience_card(self, item) -> Optional[Experience | list[Experience]]:
    """Parse an experience card from LinkedIn's DOM.

    Handles both single-role and grouped (multi-role at one company) entries.
    Grouped entries have a <ul> with sub-positions.
    """
    try:
        # Check for grouped experience (ul with sub-positions)
        sub_list = item.locator('ul')
        if await sub_list.count() > 0:
            return await self._parse_grouped_experience(item)

        return await self._parse_single_experience(item)

    except Exception as e:
        logger.debug(f"Error parsing experience card: {e}")
        return None

async def _parse_single_experience(self, item) -> Optional[Experience]:
    """Parse a single-role experience card."""
    try:
        links = await item.locator('a').all()
        if len(links) < 2:
            return None

        company_url = await links[0].get_attribute('href')
        detail_link = links[1]

        p_tags = await detail_link.locator('p').all()
        texts = []
        for p in p_tags:
            text = await p.text_content()
            if text and text.strip() and len(text.strip()) < 300:
                texts.append(text.strip())

        if len(texts) < 2:
            return None

        position_title = texts[0]
        company_name = texts[1]
        work_times = texts[2] if len(texts) > 2 else ""
        location = texts[3] if len(texts) > 3 else ""

        from_date, to_date, duration = self._parse_work_times(work_times)

        description = await self._extract_description(item)

        return Experience(
            position_title=position_title,
            institution_name=company_name,
            linkedin_url=company_url,
            from_date=from_date,
            to_date=to_date,
            duration=duration,
            location=location if location else None,
            description=description,
        )

    except Exception as e:
        logger.debug(f"Error parsing single experience: {e}")
        return None

async def _parse_grouped_experience(self, item) -> list[Experience]:
    """Parse a grouped experience (multiple roles at one company)."""
    experiences = []
    try:
        # Get company info from top-level links
        top_links = await item.locator(':scope > div a, :scope > div > div a').all()
        company_url = None
        company_name = None
        location = None

        if top_links:
            company_url = await top_links[0].get_attribute('href')
            if len(top_links) > 1:
                detail_link = top_links[1]
                p_tags = await detail_link.locator('p').all()
                texts = []
                for p in p_tags:
                    text = await p.text_content()
                    if text and text.strip():
                        texts.append(text.strip())
                if texts:
                    company_name = texts[0]
                if len(texts) > 2:
                    location = texts[2]

        # Parse each sub-position from li elements
        sub_items = await item.locator('ul > li').all()
        for sub_item in sub_items:
            try:
                sub_link = sub_item.locator('a').first
                if await sub_link.count() == 0:
                    continue

                p_tags = await sub_link.locator('p').all()
                texts = []
                for p in p_tags:
                    text = await p.text_content()
                    if text and text.strip() and len(text.strip()) < 300:
                        texts.append(text.strip())

                if not texts:
                    continue

                position_title = texts[0]
                work_times = ""
                sub_location = location

                # Remaining p tags could be subtitle, dates, location
                for t in texts[1:]:
                    if self._looks_like_date_range(t):
                        work_times = t
                    elif self._looks_like_location(t):
                        sub_location = t

                from_date, to_date, duration = self._parse_work_times(work_times)
                description = await self._extract_description(sub_item)

                experiences.append(Experience(
                    position_title=position_title,
                    institution_name=company_name or "",
                    linkedin_url=company_url,
                    from_date=from_date,
                    to_date=to_date,
                    duration=duration,
                    location=sub_location,
                    description=description,
                ))

            except Exception as e:
                logger.debug(f"Error parsing sub-position: {e}")
                continue

    except Exception as e:
        logger.debug(f"Error parsing grouped experience: {e}")

    return experiences

@staticmethod
def _looks_like_date_range(text: str) -> bool:
    """Check if text looks like a date range (e.g. 'Jan 2020 - Present · 2 yrs')."""
    months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
              'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
    return (
        any(m in text for m in months)
        or 'Present' in text
        or ('·' in text and any(u in text for u in ['yr', 'mo']))
    )

@staticmethod
def _looks_like_location(text: str) -> bool:
    """Check if text looks like a location."""
    location_indicators = [',', 'Brazil', 'States', 'United', 'Remote',
                           'Area', 'Region', 'City']
    return any(ind in text for ind in location_indicators)

async def _extract_description(self, container) -> Optional[str]:
    """Extract description text from an expandable text box."""
    text_box = container.locator('[data-testid="expandable-text-box"]').first
    if await text_box.count() > 0:
        desc_text = await text_box.text_content()
        if desc_text:
            desc_text = desc_text.strip()
            for suffix in ["… more", "…\n more", "…more"]:
                if desc_text.endswith(suffix):
                    desc_text = desc_text[: -len(suffix)].strip()
            return desc_text if desc_text else None
    return None

def _parse_work_times(
    self, work_times: str
) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """
    Parse work times string into from_date, to_date, duration.

    Examples:
    - "2000 - Present · 26 yrs 1 mo" -> ("2000", "Present", "26 yrs 1 mo")
    - "Jan 2020 - Dec 2022 · 2 yrs" -> ("Jan 2020", "Dec 2022", "2 yrs")
    - "2015 - Present" -> ("2015", "Present", None)
    """
    if not work_times:
        return None, None, None

    try:
        # Split by · to separate date range from duration
        parts = work_times.split("·")
        times = parts[0].strip() if len(parts) > 0 else ""
        duration = parts[1].strip() if len(parts) > 1 else None

        # Parse dates - split by " - " to get from and to
        if " - " in times:
            date_parts = times.split(" - ")
            from_date = date_parts[0].strip()
            to_date = date_parts[1].strip() if len(date_parts) > 1 else ""
        else:
            from_date = times
            to_date = ""

        return from_date, to_date, duration
    except Exception as e:
        logger.debug(f"Error parsing work times '{work_times}': {e}")
        return None, None, None

async def _get_educations(self, base_url: str) -> list[Education]:
    """Extract educations from the profile page Education section."""
    educations = []

    try:
        edu_section = self.page.locator('[data-view-name="profile-card-education"]')
        if await edu_section.count() == 0:
            return educations

        # Education items use componentkey but not with a consistent prefix,
        # so find all direct item containers by looking for the logo link pattern
        items = await edu_section.locator(
            'a[data-view-name="experience-company-logo-click"], '
            'a[href*="/school/"], a[href*="/company/"]'
        ).all()

        # Each logo link's grandparent is the item container
        seen_urls = set()
        for logo_link in items:
            try:
                item_container = logo_link.locator('xpath=ancestor::div[contains(@class, "b902da86")][3]')
                if await item_container.count() == 0:
                    continue

                institution_url = await logo_link.get_attribute('href')
                if institution_url in seen_urls:
                    continue
                seen_urls.add(institution_url)

                # Get the detail link (second a tag in the item)
                all_links = await item_container.locator('a').all()
                detail_link = all_links[1] if len(all_links) > 1 else None
                if not detail_link:
                    continue

                p_tags = await detail_link.locator('p').all()
                texts = []
                for p in p_tags:
                    text = await p.text_content()
                    if text and text.strip() and len(text.strip()) < 300:
                        texts.append(text.strip())

                if not texts:
                    continue

                institution_name = texts[0]
                degree = None
                times = ""

                if len(texts) == 3:
                    degree = texts[1]
                    times = texts[2]
                elif len(texts) == 2:
                    second = texts[1]
                    if any(c.isdigit() for c in second):
                        times = second
                    else:
                        degree = second

                from_date, to_date = self._parse_education_times(times)

                educations.append(Education(
                    institution_name=institution_name,
                    degree=degree.strip() if degree else None,
                    linkedin_url=institution_url,
                    from_date=from_date,
                    to_date=to_date,
                    description=None,
                ))

            except Exception as e:
                logger.debug(f"Error parsing education item: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting educations: {e}")

    return educations

def _parse_education_times(self, times: str) -> tuple[Optional[str], Optional[str]]:
    """
    Parse education times string into from_date, to_date.

    Examples:
    - "1973 - 1977" -> ("1973", "1977")
    - "2015" -> ("2015", "2015")
    - "" -> (None, None)
    """
    if not times:
        return None, None

    try:
        # Split by " - " to get from and to dates
        if " - " in times:
            parts = times.split(" - ")
            from_date = parts[0].strip()
            to_date = parts[1].strip() if len(parts) > 1 else ""
        else:
            # Single year
            from_date = times.strip()
            to_date = times.strip()

        return from_date, to_date
    except Exception as e:
        logger.debug(f"Error parsing education times '{times}': {e}")
        return None, None

async def _get_interests(self, base_url: str) -> list[Interest]:
    """Extract interests from the profile page Interests section."""
    interests = []

    try:
        interests_section = self.page.locator('[data-view-name="profile-card-interests"]')
        if await interests_section.count() == 0:
            return interests

        # Interests use radio buttons or tabs for categories
        tabs = await interests_section.locator('[role="tab"], [role="radio"]').all()

        for tab in tabs:
            try:
                tab_name = await tab.text_content()
                if not tab_name:
                    continue
                tab_name = tab_name.strip()
                category = self._map_interest_tab_to_category(tab_name)

                await tab.click()
                await self.wait_and_focus(0.5)

                # Get interest items in the visible panel
                panel = interests_section.locator('[role="tabpanel"]').first
                if await panel.count() == 0:
                    # Fallback: items may be directly in the section
                    panel = interests_section

                item_links = await panel.locator('a[href*="/company/"], a[href*="/school/"], a[href*="/in/"]').all()
                seen_names = set()
                for link in item_links:
                    href = await link.get_attribute("href")
                    p_tags = await link.locator('p').all()
                    name = None
                    if p_tags:
                        name_text = await p_tags[0].text_content()
                        if name_text:
                            name = name_text.strip()

                    if name and href and name not in seen_names:
                        seen_names.add(name)
                        interests.append(Interest(
                            name=name,
                            category=category,
                            linkedin_url=href,
                        ))

            except Exception as e:
                logger.debug(f"Error processing interest tab: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting interests: {e}")

    return interests

def _map_interest_tab_to_category(self, tab_name: str) -> str:
    tab_lower = tab_name.lower()
    if "compan" in tab_lower:
        return "company"
    elif "group" in tab_lower:
        return "group"
    elif "school" in tab_lower:
        return "school"
    elif "newsletter" in tab_lower:
        return "newsletter"
    elif "voice" in tab_lower or "influencer" in tab_lower:
        return "influencer"
    else:
        return tab_lower

async def _get_accomplishments(self, base_url: str) -> list[Accomplishment]:
    accomplishments = []

    accomplishment_sections = [
        ("certifications", "certification"),
        ("honors", "honor"),
        ("publications", "publication"),
        ("patents", "patent"),
        ("courses", "course"),
        ("projects", "project"),
        ("languages", "language"),
        ("organizations", "organization"),
    ]

    for url_path, category in accomplishment_sections:
        try:
            section_url = urljoin(base_url, f"details/{url_path}/")
            await self.navigate_and_wait(section_url)
            await self.page.wait_for_selector("main", timeout=10000)
            await self.wait_and_focus(1)

            nothing_to_see = await self.page.locator(
                'text="Nothing to see for now"'
            ).count()
            if nothing_to_see > 0:
                continue

            main_list = self.page.locator(
                ".pvs-list__container, main ul, main ol"
            ).first
            if await main_list.count() == 0:
                continue

            items = await main_list.locator(".pvs-list__paged-list-item").all()
            if not items:
                items = await main_list.locator("> li").all()

            seen_titles = set()
            for item in items:
                try:
                    accomplishment = await self._parse_accomplishment_item(
                        item, category
                    )
                    if accomplishment and accomplishment.title not in seen_titles:
                        seen_titles.add(accomplishment.title)
                        accomplishments.append(accomplishment)
                except Exception as e:
                    logger.debug(f"Error parsing {category} item: {e}")
                    continue

        except Exception as e:
            logger.debug(f"Error getting {category}s: {e}")
            continue

    return accomplishments

async def _parse_accomplishment_item(
    self, item, category: str
) -> Optional[Accomplishment]:
    try:
        entity = item.locator(
            'div[data-view-name="profile-component-entity"]'
        ).first
        if await entity.count() > 0:
            spans = await entity.locator('span[aria-hidden="true"]').all()
        else:
            spans = await item.locator('span[aria-hidden="true"]').all()

        title = ""
        issuer = ""
        issued_date = ""
        credential_id = ""

        for i, span in enumerate(spans[:5]):
            text = await span.text_content()
            if not text:
                continue
            text = text.strip()

            if len(text) > 500:
                continue

            if i == 0:
                title = text
            elif "Issued by" in text:
                parts = text.split("·")
                issuer = parts[0].replace("Issued by", "").strip()
                if len(parts) > 1:
                    issued_date = parts[1].strip()
            elif "Issued " in text and not issued_date:
                issued_date = text.replace("Issued ", "")
            elif "Credential ID" in text:
                credential_id = text.replace("Credential ID ", "")
            elif i == 1 and not issuer:
                issuer = text
            elif (
                any(
                    month in text
                    for month in [
                        "Jan",
                        "Feb",
                        "Mar",
                        "Apr",
                        "May",
                        "Jun",
                        "Jul",
                        "Aug",
                        "Sep",
                        "Oct",
                        "Nov",
                        "Dec",
                    ]
                )
                and not issued_date
            ):
                if "·" in text:
                    parts = text.split("·")
                    issued_date = parts[0].strip()
                else:
                    issued_date = text

        link = item.locator('a[href*="credential"], a[href*="verify"]').first
        credential_url = (
            await link.get_attribute("href") if await link.count() > 0 else None
        )

        if not title or len(title) > 200:
            return None

        return Accomplishment(
            category=category,
            title=title,
            issuer=issuer if issuer else None,
            issued_date=issued_date if issued_date else None,
            credential_id=credential_id if credential_id else None,
            credential_url=credential_url,
        )

    except Exception as e:
        logger.debug(f"Error parsing accomplishment: {e}")
        return None

async def _get_contacts(self, base_url: str) -> list[Contact]:
    """Extract contact info from the contact-info overlay dialog."""
    contacts = []

    try:
        contact_url = urljoin(base_url, "overlay/contact-info/")
        await self.navigate_and_wait(contact_url)
        await self.wait_and_focus(1)

        dialog = self.page.locator('dialog, [role="dialog"]').first
        if await dialog.count() == 0:
            logger.warning("Contact info dialog not found")
            return contacts

        contact_sections = await dialog.locator('h3').all()
        
        for section_heading in contact_sections:
            try:
                heading_text = await section_heading.text_content()
                if not heading_text:
                    continue
                heading_text = heading_text.strip().lower()
                
                section_container = section_heading.locator('xpath=ancestor::*[1]')
                if await section_container.count() == 0:
                    continue
                
                contact_type = self._map_contact_heading_to_type(heading_text)
                if not contact_type:
                    continue
                
                links = await section_container.locator('a').all()
                for link in links:
                    href = await link.get_attribute('href')
                    text = await link.text_content()
                    if href and text:
                        text = text.strip()
                        label = None
                        sibling_text = await section_container.locator('span, generic').all()
                        for sib in sibling_text:
                            sib_text = await sib.text_content()
                            if sib_text and sib_text.strip().startswith('(') and sib_text.strip().endswith(')'):
                                label = sib_text.strip()[1:-1]
                                break
                        
                        if contact_type == "linkedin":
                            contacts.append(Contact(type=contact_type, value=href, label=label))
                        elif contact_type == "email" and "mailto:" in href:
                            contacts.append(Contact(type=contact_type, value=href.replace("mailto:", ""), label=label))
                        else:
                            contacts.append(Contact(type=contact_type, value=text, label=label))
                
                if contact_type == "birthday" and not links:
                    birthday_text = await section_container.text_content()
                    if birthday_text:
                        birthday_value = birthday_text.replace(heading_text, "").replace("Birthday", "").strip()
                        if birthday_value:
                            contacts.append(Contact(type="birthday", value=birthday_value))
                
                if contact_type == "phone" and not links:
                    phone_text = await section_container.text_content()
                    if phone_text:
                        phone_value = phone_text.replace(heading_text, "").replace("Phone", "").strip()
                        if phone_value:
                            contacts.append(Contact(type="phone", value=phone_value))
                
                if contact_type == "address" and not links:
                    address_text = await section_container.text_content()
                    if address_text:
                        address_value = address_text.replace(heading_text, "").replace("Address", "").strip()
                        if address_value:
                            contacts.append(Contact(type="address", value=address_value))
                            
            except Exception as e:
                logger.debug(f"Error parsing contact section: {e}")
                continue

    except Exception as e:
        logger.warning(f"Error getting contacts: {e}")

    return contacts

def _map_contact_heading_to_type(self, heading: str) -> Optional[str]:
    """Map contact section heading to contact type."""
    heading = heading.lower()
    if "profile" in heading:
        return "linkedin"
    elif "website" in heading:
        return "website"
    elif "email" in heading:
        return "email"
    elif "phone" in heading:
        return "phone"
    elif "twitter" in heading or "x.com" in heading:
        return "twitter"
    elif "birthday" in heading:
        return "birthday"
    elif "address" in heading:
        return "address"
    return None

`

Hello, I am using the modified code you provided, but still cannot obtain the information. Could you please tell me if there is a latest solution? thank you.The specific details are as follows:

✓ Session loaded
🚀 Scraping: https://www.linkedin.com/in/williamhgates/

============================================================
Name: Bill Gates
Location: None
About: N/A
Experiences: 0
Education: 0

✓ Done!

[suzhiyam-in]

Aah man! Seems like linkedin again changed the dom structure. We can't do anything man. Either we need to constantly change our code depending on the changed linkedin dom, or, I don't know what to do man.