src/ingest/parser.py

"""Ingestion components for parsing resumes and persisting structured ATS artifacts."""

import re
import unicodedata
from dataclasses import dataclass
from pathlib import Path

import pymupdf
import pymupdf.layout  # noqa: F401  # activates pymupdf-layout enhancements for pymupdf4llm
import pymupdf4llm

from src.ingest.entities import HeadingSpan, ParsedResume, SectionItem

SECTION_MAPPING = {
    # -------------------
    # SUMMARY
    # -------------------
    "summary": "summary",
    "professional summary": "summary",
    "executive summary": "summary",
    "profile": "summary",
    "professional profile": "summary",
    "about": "summary",
    "about me": "summary",
    "sobre mi": "summary",
    "sobre mí": "summary",
    "career summary": "summary",
    "objective": "summary",
    "career objective": "summary",
    "personal statement": "summary",
    "overview": "summary",
    # -------------------
    # EXPERIENCE
    # -------------------
    "experience": "experience",
    "professional experience": "experience",
    "work experience": "experience",
    "employment history": "experience",
    "work history": "experience",
    "career history": "experience",
    "employment": "experience",
    "professional background": "experience",
    "relevant experience": "experience",
    "industry experience": "experience",
    "internship experience": "experience",
    "internships": "experience",
    "positions held": "experience",
    "experiencia": "experience",
    "experiencia profesional": "experience",
    "experiencia laboral": "experience",
    # -------------------
    # EDUCATION
    # -------------------
    "education": "education",
    "academic background": "education",
    "academic history": "education",
    "academic experience": "education",
    "qualifications": "education",
    "academic qualifications": "education",
    "degrees": "education",
    "degree": "education",
    "studies": "education",
    "formal education": "education",
    "courses": "education",
    "coursework": "education",
    "relevant coursework": "education",
    "training": "education",
    "educacion": "education",
    "educación": "education",
    "formacion": "education",
    "formación": "education",
    # -------------------
    # SKILLS
    # -------------------
    "skills": "skills",
    "technical skills": "skills",
    "core skills": "skills",
    "key skills": "skills",
    "professional skills": "skills",
    "hard skills": "skills",
    "soft skills": "skills",
    "competencies": "skills",
    "core competencies": "skills",
    "expertise": "skills",
    "technical expertise": "skills",
    "technologies": "skills",
    "tech stack": "skills",
    "tools": "skills",
    "informatica": "skills",
    "informática": "skills",
    "habilidades": "skills",
    "competencias": "skills",
    "aptitudes": "skills",
    # -------------------
    # PROJECTS
    # -------------------
    "projects": "projects",
    "personal projects": "projects",
    "academic projects": "projects",
    "professional projects": "projects",
    "selected projects": "projects",
    "key projects": "projects",
    "portfolio": "projects",
    "research projects": "projects",
    "proyectos": "projects",
    # -------------------
    # CERTIFICATIONS
    # -------------------
    "certifications": "certifications",
    "certification": "certifications",
    "licenses": "certifications",
    "licenses and certifications": "certifications",
    "professional certifications": "certifications",
    "credentials": "certifications",
    "accreditations": "certifications",
    "certificaciones": "certifications",
    "licencias": "certifications",
    # -------------------
    # CONTACT
    # -------------------
    "contact": "contact",
    "contact information": "contact",
    "personal information": "contact",
    "personal details": "contact",
    "contact details": "contact",
    "get in touch": "contact",
    "contact me": "contact",
    "contacto": "contact",
    "informacion de contacto": "contact",
    "información de contacto": "contact",
    "idiomas": "skills",
    "languages": "skills",
    "publications": "projects",
    "publicaciones": "projects",
}


@dataclass
class PDFResumeParser:
    """Data model for pdfresumeparser values."""

    parser_version: str = "stage3.v1"

    def parse(self, path: Path) -> ParsedResume:
        """Runs parse logic.

        Args:
            path (Path): Filesystem path of the file being parsed or ingested.

        Returns:
            ParsedResume: Return value for this function.
        """
        markdown = self.extract_markdown(path)
        return self.parse_markdown(markdown=markdown, source_file=str(path))

    def parse_markdown(self, markdown: str, source_file: str) -> ParsedResume:
        """Parses input content into the normalized structure expected by ingestion logic.

        Args:
            markdown (str): Markdown document emitted by PDF extraction.
            source_file (str): Source file path string stored for idempotency checks.

        Returns:
            ParsedResume: Return value for this function.
        """
        clean_markdown = self._preclean_markdown(markdown)
        clean_text, _ = self.clean_resume_blocks(clean_markdown)
        links = self.extract_links(clean_markdown)
        spans = self._find_heading_spans(clean_markdown)
        for i, span in enumerate(spans):
            spans[i] = HeadingSpan(
                raw_heading=span.raw_heading,
                title=self._map_heading_to_section(span.raw_heading),
                start_line=span.start_line,
                end_line=span.end_line,
            )
        spans = self._absorb_generals_into_single_line_sections(spans)
        sections, section_items = self._extract_sections_and_items(clean_markdown, spans)
        language = self.detect_language(clean_markdown)

        return ParsedResume(
            source_file=source_file,
            raw_text=markdown,
            clean_text=clean_text,
            links=links,
            sections=sections,
            section_items=section_items,
            language=language,
            parser_version=self.parser_version,
        )

    def extract_markdown(self, path: Path) -> str:
        """Extracts structured information from parsed or raw resume content.

        Args:
            path (Path): Filesystem path of the PDF or source file being processed.

        Returns:
            str: Normalized string result produced by this helper.

        Raises:
            Exception: Propagates validation or runtime errors encountered by this operation.
        """
        if not path.exists():
            raise FileNotFoundError(f"Resume not found: {path}")
        doc = pymupdf.open(path)
        try:
            return pymupdf4llm.to_markdown(doc, show_progress=False, use_ocr=True, force_ocr=False)
        except RuntimeError as exc:
            if "Tesseract" not in str(exc):
                raise
            return pymupdf4llm.to_markdown(doc, show_progress=False, use_ocr=False, force_ocr=False)

    def split_by_blocks(self, text: str) -> list[str]:
        """Runs split by blocks logic.

        Args:
            text (str): Text input being parsed, normalized, or scored.

        Returns:
            list[str]: Ordered list produced by this operation.
        """
        blocks = re.split(r"\n\n", text)
        cleaned: list[str] = []
        for block in blocks:
            normalized = re.sub(r"#+\s", "", block).lstrip("\n").strip()
            if normalized:
                cleaned.append(normalized)
        return cleaned

    def clean_resume_blocks(self, text: str) -> tuple[str, list[str]]:
        """Runs clean resume blocks logic.

        Args:
            text (str): Text input being parsed, normalized, or scored.

        Returns:
            tuple[str, list[str]]: Tuple containing the values produced by this operation.
        """
        extracted_links: list[str] = []
        unique_blocks: list[str] = []
        seen_blocks: set[str] = set()

        for block in self.split_by_blocks(text):
            links = re.findall(r"https?://[^\s\)\]]+", block)
            extracted_links.extend(links)

            if re.match(r"^[\-\s]+$", block):
                continue

            cleaned_block = re.sub(r"https?://[^\s\)\]]+", "", block).strip()
            cleaned_block = re.sub(r"\[([^\[\]]+)\]\s*\(\s*\)", r"\1", cleaned_block).strip()
            normalized_block = " ".join(cleaned_block.splitlines()).strip()

            if normalized_block and normalized_block not in seen_blocks:
                seen_blocks.add(normalized_block)
                unique_blocks.append(normalized_block)

        text_out = "\n".join(unique_blocks)
        unique_links = sorted(set(extracted_links))
        return text_out, unique_links

    def extract_links(self, text: str) -> list[str]:
        """Extracts structured information from parsed or raw resume content.

        Args:
            text (str): Raw text content being normalized, parsed, or scored.

        Returns:
            list[str]: List of normalized string values.
        """
        links = re.findall(r"https?://[^\s\)\]]+", text)
        return sorted(set(links))

    def extract_sections(
        self, markdown: str, spans: list[HeadingSpan] | None = None
    ) -> dict[str, str]:
        """Extracts structured data from raw resume or markdown input.

        Args:
            markdown (str): Markdown document emitted by PDF extraction.
            spans (list[HeadingSpan] | None): Detected heading spans used to split sections.

        Returns:
            dict[str, str]: Return value for this function.
        """
        if spans is None:
            mapped_spans: list[HeadingSpan] = []
            for span in self._find_heading_spans(markdown):
                mapped_spans.append(
                    HeadingSpan(
                        raw_heading=span.raw_heading,
                        title=self._map_heading_to_section(span.raw_heading),
                        start_line=span.start_line,
                        end_line=span.end_line,
                    )
                )
            spans = self._absorb_generals_into_single_line_sections(mapped_spans)

        sections, _ = self._extract_sections_and_items(markdown, spans)
        return sections

    def _extract_sections_and_items(
        self, markdown: str, spans: list[HeadingSpan]
    ) -> tuple[dict[str, str], list[SectionItem]]:
        """Helper that handles extract sections and items.

        Args:
            markdown (str): Markdown document emitted by PDF extraction.
            spans (list[HeadingSpan]): Detected heading spans used to split sections.

        Returns:
            tuple[dict[str, str], list[SectionItem]]: Tuple containing the values produced by this operation.
        """
        lines = markdown.splitlines()
        sections: dict[str, str] = {}
        items: list[SectionItem] = []

        for span in spans:
            if span.start_line >= len(lines):
                continue

            content_lines = lines[span.start_line + 1 : span.end_line + 1]
            content = "\n".join(line.strip() for line in content_lines if line.strip()).strip()
            if not content:
                continue

            if span.title in sections:
                sections[span.title] = f"{sections[span.title]}\n\n{content}"
            else:
                sections[span.title] = content

            items.append(
                SectionItem(
                    raw_heading=span.raw_heading,
                    normalized_type=span.title,
                    content=content,
                    confidence=1.0 if span.title != "general" else 0.5,
                    signals=self._build_section_signals(
                        normalized_type=span.title,
                        raw_heading=span.raw_heading,
                        content=content,
                    ),
                )
            )

        if not sections:
            fallback = markdown.strip()
            if fallback:
                sections["general"] = fallback
                items.append(
                    SectionItem(
                        raw_heading="",
                        normalized_type="general",
                        content=fallback,
                        confidence=0.3,
                        signals=self._build_section_signals(
                            normalized_type="general",
                            raw_heading="",
                            content=fallback,
                        ),
                    )
                )

        return sections, items

    def detect_language(self, text: str) -> str:
        """Runs detect language logic.

        Args:
            text (str): Text input being parsed, normalized, or scored.

        Returns:
            str: Normalized string result.
        """
        lowered = text.lower()
        english_markers = ["experience", "education", "skills", "university"]
        spanish_markers = ["experiencia", "educación", "habilidades", "universidad"]

        english_score = sum(1 for marker in english_markers if marker in lowered)
        spanish_score = sum(1 for marker in spanish_markers if marker in lowered)

        if english_score == 0 and spanish_score == 0:
            return "unknown"
        if english_score >= spanish_score:
            return "en"
        return "es"

    def _remove_omitted_pictures(self, markdown: str) -> str:
        """Helper that handles remove omitted pictures.

        Args:
            markdown (str): Markdown document emitted by PDF extraction.

        Returns:
            str: Normalized string result.
        """
        return re.sub(r"\*\*==>.*?<==\*\*", "", markdown, flags=re.DOTALL)

    def _remove_encoding_artifacts(self, markdown: str) -> str:
        """
        Removes common encoding mismatch artifacts such as
        the Unicode replacement character (�).
        """
        return markdown.replace("\ufffd", "")

    def _clean_markdown_table_artifacts(self, text: str) -> str:
        """
        Cleans flattened markdown tables into readable plain text.
        """

        lines = text.splitlines()
        cleaned_lines = []

        for line in lines:
            line = line.strip()

            # Skip separator rows like |---|---|
            if re.fullmatch(r"\|?\s*-+\s*(\|\s*-+\s*)+\|?", line):
                continue

            # Remove leading/trailing pipes
            line = line.strip("|")

            # Replace remaining pipes with a readable separator
            line = re.sub(r"\s*\|\s*", " - ", line)

            # Remove accidental double separators
            line = re.sub(r"-\s*-", "-", line)

            # Remove trailing double pipes
            line = re.sub(r"\|\|+$", "", line)

            line = line.strip()

            if line:
                cleaned_lines.append(line)

        return "\n".join(cleaned_lines)

    def _remove_all_bullet_chars(self, text: str) -> str:
        """Helper that handles remove all bullet chars.

        Args:
            text (str): Text input being parsed, normalized, or scored.

        Returns:
            str: Normalized string result.
        """
        bullet_chars = r"[\u2022\u25AA\u25E6\u2023\u00B7]"
        return re.sub(bullet_chars, "", text)

    def _remove_dotted_leaders(self, text: str) -> str:
        """
        Removes long sequences of spaced dots like:
        . . . . . . . . . .
        but keeps normal sentence punctuation.
        """
        # Match: dot + space repeated at least 3 times
        pattern = r"(?:\.\s*){3,}"
        return re.sub(pattern, "", text)

    def _preclean_markdown(self, markdown: str) -> str:
        """Helper that handles preclean markdown.

        Args:
            markdown (str): Markdown document emitted by PDF extraction.

        Returns:
            str: Normalized string result.
        """
        clean_markdown = self._remove_omitted_pictures(markdown)
        clean_markdown = self._remove_encoding_artifacts(clean_markdown)
        clean_markdown = self._clean_markdown_table_artifacts(clean_markdown)
        clean_markdown = self._remove_all_bullet_chars(clean_markdown)
        clean_markdown = self._remove_dotted_leaders(clean_markdown)
        return clean_markdown

    def _find_heading_spans(self, markdown: str) -> list[HeadingSpan]:
        """Helper that handles find heading spans.

        Args:
            markdown (str): Markdown document emitted by PDF extraction.

        Returns:
            list[HeadingSpan]: Ordered list produced by this operation.
        """
        lines = markdown.splitlines()
        heading_pattern = re.compile(r"^(#{1,6})\s+(.*)")

        spans: list[HeadingSpan] = []
        current_span: HeadingSpan | None = None

        for i, line in enumerate(lines):
            match = heading_pattern.match(line)

            if match:
                # Close previous span
                if current_span is not None:
                    current_span.end_line = i - 1
                    spans.append(current_span)

                # Start new span
                title = match.group(2).strip()
                current_span = HeadingSpan(
                    raw_heading=title,
                    title=title,
                    start_line=i,
                    end_line=-1,  # temporary placeholder
                )

        # Close last span
        if current_span is not None:
            current_span.end_line = len(lines) - 1
            spans.append(current_span)

        return spans

    def _map_heading_to_section(self, title: str) -> str:
        """Helper that handles map heading to section.

        Args:
            title (str): Title text rendered in card/table output.

        Returns:
            str: Normalized string result.
        """
        normalized = self._normalize_heading_text(title)
        normalized = " ".join(normalized.split())

        for key, value in SECTION_MAPPING.items():
            if key in normalized:
                return value

        return "general"

    def _normalize_heading_text(self, title: str) -> str:
        """Helper that handles normalize heading text.

        Args:
            title (str): Title text rendered in card/table output.

        Returns:
            str: Normalized string result.
        """
        no_markdown = re.sub(r"[*_`~]+", " ", title)
        folded = unicodedata.normalize("NFKD", no_markdown)
        folded = "".join(ch for ch in folded if not unicodedata.combining(ch))
        return re.sub(r"[^a-z0-9\s]+", " ", folded.lower())

    def _build_section_signals(
        self,
        *,
        normalized_type: str,
        raw_heading: str,
        content: str,
    ) -> dict:
        """Helper that handles build section signals.

        Args:
            normalized_type (str): Normalized section type generated by heading mapping.
            raw_heading (str): Original heading text before normalization.
            content (str): Section body content associated with the heading.

        Returns:
            dict: Return value for this function.
        """
        flags: list[str] = []
        heading_mapped_to_general = bool(raw_heading.strip()) and normalized_type == "general"
        if heading_mapped_to_general:
            flags.append("heading_unknown")
        if len(content.split()) < 8:
            flags.append("short_content")
        if self._looks_like_contact_block(content):
            flags.append("looks_like_contact_block")

        recat = self._suggest_recategorization(
            normalized_type=normalized_type,
            content=content,
            has_contact_hint="looks_like_contact_block" in flags,
        )

        confidence_inputs = {
            "word_count": len(content.split()),
            "heading_mapped_to_general": heading_mapped_to_general,
        }

        return {
            "diagnostic_flags": flags,
            "confidence_inputs": confidence_inputs,
            "recategorization_candidate": recat,
        }

    def _looks_like_contact_block(self, content: str) -> bool:
        """Helper that handles looks like contact block.

        Args:
            content (str): Section body content associated with the heading.

        Returns:
            bool: True when the condition is satisfied; otherwise False.
        """
        lowered = content.lower()
        has_email = bool(re.search(r"[a-z0-9._%+\-]+@[a-z0-9.\-]+\.[a-z]{2,}", lowered))
        has_phone = bool(re.search(r"(?:\+?\d[\d\s().\-/]{6,}\d)", content))
        return has_email or has_phone

    def _suggest_recategorization(
        self,
        *,
        normalized_type: str,
        content: str,
        has_contact_hint: bool,
    ) -> dict | None:
        """Helper that handles suggest recategorization.

        Args:
            normalized_type (str): Normalized section type generated by heading mapping.
            content (str): Section body content associated with the heading.
            has_contact_hint (bool): Whether contact-pattern signals were detected in content.

        Returns:
            dict | None: Return value for this function.
        """
        lowered = content.lower()
        if normalized_type != "general":
            return None

        if has_contact_hint:
            return {"section_type": "contact", "confidence": 0.8}

        keyword_buckets = {
            "skills": ["python", "sql", "java", "skills", "technologies", "stack"],
            "experience": ["experience", "responsible", "led", "worked", "managed"],
            "contact": ["email", "phone", "linkedin", "github"],
        }
        for target, keywords in keyword_buckets.items():
            hits = sum(1 for keyword in keywords if keyword in lowered)
            if hits >= 2:
                return {"section_type": target, "confidence": 0.65}
        return None

    def _absorb_generals_into_single_line_sections(
        self,
        spans: list[HeadingSpan],
    ) -> list[HeadingSpan]:
        """
        If a non-'general' section has only one line (start_line == end_line),
        absorb consecutive following 'general' sections into it.
        """

        result: list[HeadingSpan] = []
        i = 0

        while i < len(spans):
            current = spans[i]

            # Only apply rule to non-general single-line sections
            if current.title != "general" and current.start_line == current.end_line:
                j = i + 1

                # Absorb consecutive general sections
                while j < len(spans) and spans[j].title == "general":
                    current.end_line = spans[j].end_line
                    j += 1

                result.append(current)
                i = j  # Skip absorbed spans

            else:
                result.append(current)
                i += 1

        return result