import re from typing import List, Dict, Any, Optional, Tuple from pilotcore.chunking.base import BaseChunker from pilotcore.chunking.recursive import RecursiveCharacterChunker # Comprehensive catalog of semantic section categories SEMANTIC_SECTIONS: List[Tuple[str, re.Pattern]] = [ # Resumes & CVs ("experience", re.compile(r"^(?:work\s+|professional\s+|relevant\s+|industry\s+)?experience|employment(?:\s+history)?|work\s+history|career\s+history", re.IGNORECASE)), ("education", re.compile(r"^education(?:al\s+background)?|academic(?:\s+background|\s+qualifications)?|qualifications|degrees", re.IGNORECASE)), ("projects", re.compile(r"^(?:featured\s+|key\s+|technical\s+|academic\s+|personal\s+)?projects", re.IGNORECASE)), ("skills", re.compile(r"^(?:technical\s+|core\s+|key\s+)?skills|skills\s*&?\s*(?:expertise|competencies|tools)|technical\s+expertise|technologies", re.IGNORECASE)), ("certifications", re.compile(r"^certifications?|licenses(?:\s*&?\s*certifications?)?|courses|accreditations", re.IGNORECASE)), ("achievements", re.compile(r"^achievements?|awards(?:\s*&?\s*honors)?|honors|accomplishments", re.IGNORECASE)), ("publications", re.compile(r"^publications?|research(?:\s+papers?)?|patents", re.IGNORECASE)), ("leadership", re.compile(r"^positions?\s+of\s+responsibility|leadership(?:\s+experience)?|volunteering|extracurricular(?:\s+activities)?", re.IGNORECASE)), ("summary", re.compile(r"^(?:executive\s+|professional\s+|career\s+)?summary|about(?:\s+me)?|profile|objective", re.IGNORECASE)), ("contact", re.compile(r"^contact(?:\s+info(?:rmation)?)?|personal\s+details|portfolio", re.IGNORECASE)), # Technical Reports, Papers & Documentation ("abstract", re.compile(r"^abstract", re.IGNORECASE)), ("introduction", re.compile(r"^introduction|overview|background(?:\s+and\s+motivation)?", re.IGNORECASE)), ("architecture", re.compile(r"^(?:system\s+)?architecture|system\s+design|framework\s+overview|architectural\s+overview", re.IGNORECASE)), ("methodology", re.compile(r"^methodology|methods|proposed\s+(?:method|approach|system)|experimental\s+setup", re.IGNORECASE)), ("implementation", re.compile(r"^implementation(?:\s+details)?|technical\s+implementation|pipeline\s+design", re.IGNORECASE)), ("experiments", re.compile(r"^experiments|benchmarks?|evaluations?|performance\s+analysis", re.IGNORECASE)), ("results", re.compile(r"^results(?:\s+and\s+discussion)?|findings|key\s+metrics", re.IGNORECASE)), ("discussion", re.compile(r"^discussion|analysis|tradeoffs|limitations", re.IGNORECASE)), ("related_work", re.compile(r"^related\s+work|literature\s+review|prior\s+art", re.IGNORECASE)), ("conclusion", re.compile(r"^conclusion(?:\s+and\s+future\s+work)?|summary\s+and\s+conclusions?|future\s+work", re.IGNORECASE)), ("references", re.compile(r"^references|bibliography", re.IGNORECASE)), ("appendix", re.compile(r"^appendix(?:\s+[A-Z0-9]+)?|supplementary\s+material", re.IGNORECASE)), # Business & Financial ("financials", re.compile(r"^(?:financial\s+performance|revenue(?:\s+breakdown)?|q[1-4]\s+(?:financials|results|performance))", re.IGNORECASE)), ("recommendations", re.compile(r"^(?:strategic\s+)?recommendations?|action\s+items|next\s+steps", re.IGNORECASE)), ] # Patterns for sub-items: Bullet projects/roles (e.g. "• DevDiscuss GitHub | Live" or "OCR Engineer Sep 2023 - Aug 2024") DATE_RANGE_PATTERN = re.compile(r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec|20\d\d)\b.*?(?:present|\d{4})", re.IGNORECASE) BULLET_SUBHEADER_PATTERN = re.compile(r"^[\s•\-\*]+([A-Za-z0-9_\-\s]+?)(?:\s+(?:GitHub|Live|Coursera|\d{4}))?$", re.MULTILINE) class StructureAwareChunker(BaseChunker): """ Intelligent Structure-Aware Chunker. Extracts semantic document hierarchies across: 1. Markdown `#` to `######` headers & HTML `

`-`

` tags 2. Resume & CV structural sections (Experience, Projects, Education, Skills, etc.) 3. Technical paper / report structures (Abstract, Methods, Results, Conclusion, etc.) 4. Individual subsection blocks (Projects, Roles with dates, and Companies) Generates rich contextual breadcrumb paths: [Section > Subsection / Role (Details)] {chunk_content} """ MD_HEADER_PATTERN = re.compile(r"^(#{1,6})\s+(.+)$", re.MULTILINE) HTML_HEADER_PATTERN = re.compile(r"]*)?>(.*?)", re.IGNORECASE) def chunk( self, text: str, chunk_size: int = 500, overlap: int = 80, **kwargs: Any, ) -> List[Dict[str, Any]]: if not text or not text.strip(): return [] # 1. Parse document into structural blocks sections = self._parse_document_structure(text) if not sections: fallback = RecursiveCharacterChunker() return fallback.chunk(text, chunk_size=chunk_size, overlap=overlap) # 2. Process sections and sub-split oversized sections while preserving breadcrumbs recursive_splitter = RecursiveCharacterChunker() final_chunks = [] for sec in sections: content = sec["content"].strip() if not content: continue header_path = sec["header_path"] header_level = sec["level"] section_title = sec["title"] section_type = sec["type"] subsection = sec.get("subsection") # Format chunk header: [Path] breadcrumb_header = f"[{header_path}]" if header_path else "" if len(content) <= chunk_size: formatted_text = ( f"{breadcrumb_header}\n\n{content}" if breadcrumb_header else content ) final_chunks.append({ "text": formatted_text, "metadata": { "section_title": section_title, "section_type": section_type, "subsection": subsection, "header_path": header_path, "structure_type": sec.get("structure_type", "section"), "level": header_level, }, }) else: # Sub-split oversized section content sub_splits = recursive_splitter.chunk( content, chunk_size=chunk_size, overlap=overlap, ) for sub_idx, sub in enumerate(sub_splits): sub_text = sub["text"].strip() if not sub_text: continue part_header = ( f"[{header_path} - Part {sub_idx + 1}]" if header_path else "" ) formatted_text = ( f"{part_header}\n\n{sub_text}" if part_header else sub_text ) final_chunks.append({ "text": formatted_text, "metadata": { "section_title": section_title, "section_type": section_type, "subsection": subsection, "header_path": header_path, "structure_type": sec.get("structure_type", "section"), "level": header_level, "sub_chunk_index": sub_idx, }, }) return final_chunks def _parse_document_structure(self, text: str) -> List[Dict[str, Any]]: lines = text.splitlines(keepends=True) sections = [] current_major_title: Optional[str] = None current_major_type: str = "general" current_major_level: int = 1 current_sub_title: Optional[str] = None current_sub_level: int = 2 current_lines: List[str] = [] def flush_current_section(): nonlocal current_lines, current_major_title, current_sub_title, current_major_type, current_major_level, current_sub_level if not current_lines: return content_str = "".join(current_lines).strip() if not content_str: current_lines = [] return # Determine title & path if current_major_title and current_sub_title: path = f"{current_major_title} > {current_sub_title}" title = current_major_title level = current_sub_level elif current_major_title: path = current_major_title title = current_major_title level = current_major_level else: path = "" title = "Overview" level = 1 sections.append({ "title": title, "type": current_major_type, "subsection": current_sub_title, "header_path": path, "level": level, "structure_type": "semantic_section", "content": content_str, }) current_lines = [] for line in lines: trimmed = line.strip() if not trimmed: current_lines.append(line) continue # 1. Check for Markdown Header md_match = self.MD_HEADER_PATTERN.match(trimmed) if md_match: flush_current_section() level = len(md_match.group(1)) title = md_match.group(2).strip() if level <= 2: current_major_title = title current_major_type = self._classify_title(title) current_major_level = level current_sub_title = None else: current_sub_title = title current_sub_level = level current_lines.append(line) continue # 2. Check for HTML Header html_match = self.HTML_HEADER_PATTERN.search(trimmed) if html_match: flush_current_section() level = int(html_match.group(1)) title = re.sub(r"<[^>]+>", "", html_match.group(2)).strip() if level <= 2: current_major_title = title current_major_type = self._classify_title(title) current_major_level = level current_sub_title = None else: current_sub_title = title current_sub_level = level current_lines.append(line) continue # 3. Check for Semantic Domain Heading (Resume / Paper / Document Sections) sem_title, sem_type = self._detect_semantic_heading(trimmed) if sem_title: flush_current_section() current_major_title = sem_title current_major_type = sem_type current_major_level = 1 current_sub_title = None current_lines.append(line) continue # 4. Check for Sub-item / Role / Project within major section if current_major_title and current_major_type in ["projects", "experience", "leadership", "education"]: sub_header = self._detect_sub_item(trimmed) if sub_header: flush_current_section() current_sub_title = sub_header current_sub_level = 2 current_lines.append(line) continue current_lines.append(line) flush_current_section() return sections @staticmethod def _detect_semantic_heading(line: str) -> Tuple[Optional[str], str]: """ Detect standalone section headers like 'Experience', 'Education', 'Projects', 'Technical Skills'. """ clean = line.rstrip(":").strip() words = clean.split() if not words or len(words) > 6: return None, "general" # Check against semantic catalog for sec_type, pattern in SEMANTIC_SECTIONS: if pattern.fullmatch(clean) or (len(words) <= 3 and pattern.match(clean)): # Return normalized Title Case return clean.title() if not clean.isupper() else clean.title(), sec_type # Check for standalone ALL-CAPS headers (e.g. 'TECHNICAL SKILLS', 'WORK EXPERIENCE') if len(clean) >= 3 and clean.isupper() and clean.replace(" ", "").isalpha() and len(words) <= 4: return clean.title(), "section" return None, "general" @staticmethod def _detect_sub_item(line: str) -> Optional[str]: """ Detect specific project names or role headings like: - '• DevDiscuss GitHub | Live' - '• OCR Engineer Sep 2023 - Aug 2024' - 'Intern Feb 2022 - Mar 2022' - 'Heart Disease RAG Assistant' """ clean = line.strip().lstrip("•-*").strip() if not clean or len(clean) > 80: return None # Check if line contains a role with date range: "OCR Engineer Sep 2023 - Aug 2024" if DATE_RANGE_PATTERN.search(clean) and len(clean.split()) <= 10: return clean # Check bullet project headers: "• DevDiscuss" or "• DevDiscuss GitHub | Live" if line.strip().startswith(("•", "-", "*")) and len(clean.split()) <= 8: # Clean off trailing links like GitHub | Live cleaned_name = re.sub(r"\s+(?:GitHub|Live|Demo|Link|Coursera)(?:\s*\|\s*(?:GitHub|Live|Demo|Link))?", "", clean, flags=re.IGNORECASE).strip() if cleaned_name and len(cleaned_name) >= 3: return cleaned_name return None @staticmethod def _classify_title(title: str) -> str: clean = title.lower().strip() for sec_type, pattern in SEMANTIC_SECTIONS: if pattern.search(clean): return sec_type return "general"