| import re |
| from pathlib import Path |
| from typing import Any, Dict, List |
|
|
| from schemas import ( |
| AITutorDocument, |
| Activity, |
| DocumentMetadata, |
| Figure, |
| Question, |
| Section, |
| SourceReference, |
| Table, |
| TextBlock, |
| ) |
|
|
|
|
| class DocumentNormalizer: |
|
|
| def normalize( |
| self, |
| docling_document: Any, |
| filename: str, |
| raw_json: Dict[str, Any], |
| ) -> AITutorDocument: |
|
|
| document_id = Path(filename).stem |
|
|
| raw = raw_json |
|
|
| pages = raw.get("pages", {}) |
| texts = raw.get("texts", []) |
| pictures = raw.get("pictures", []) |
| tables = raw.get("tables", []) |
|
|
| markdown = docling_document.export_to_markdown() |
|
|
| title = self._detect_title(markdown) |
|
|
| metadata = DocumentMetadata( |
| id=document_id, |
| filename=filename, |
| format=Path(filename).suffix.lower().replace(".", ""), |
| title=title, |
| language="English", |
| page_count=len(pages), |
| ) |
|
|
| sections = self._extract_sections( |
| markdown, |
| document_id, |
| ) |
|
|
| text_blocks = self._extract_text_blocks( |
| texts, |
| document_id, |
| ) |
|
|
| normalized_tables = self._extract_tables( |
| docling_document, |
| tables, |
| document_id, |
| ) |
|
|
| figures = self._extract_figures( |
| pictures, |
| document_id, |
| ) |
|
|
| activities = self._extract_activities( |
| markdown, |
| document_id, |
| ) |
|
|
| questions = self._extract_questions( |
| markdown, |
| document_id, |
| ) |
|
|
| statistics = { |
| "pages": len(pages), |
| "text_elements": len(texts), |
| "pictures": len(pictures), |
| "tables": len(tables), |
| "sections": len(sections), |
| "activities": len(activities), |
| "questions": len(questions), |
| } |
|
|
| return AITutorDocument( |
| document=metadata, |
| sections=sections, |
| text_blocks=text_blocks, |
| tables=normalized_tables, |
| figures=figures, |
| activities=activities, |
| questions=questions, |
| statistics=statistics, |
| raw_docling=raw_json, |
| ) |
|
|
| |
| |
| |
|
|
| def _detect_title(self, markdown: str) -> str: |
|
|
| lines = [ |
| line.strip() |
| for line in markdown.splitlines() |
| if line.strip() |
| ] |
|
|
| for line in lines[:20]: |
|
|
| clean = re.sub(r"<!--.*?-->", "", line).strip() |
|
|
| if len(clean) > 10: |
| return clean |
|
|
| return "" |
|
|
| |
| |
| |
|
|
| def _extract_sections( |
| self, |
| markdown: str, |
| document_id: str, |
| ) -> List[Section]: |
|
|
| sections = [] |
|
|
| lines = markdown.splitlines() |
|
|
| section_pattern = re.compile( |
| r"^#{1,6}\s+(.+)$" |
| ) |
|
|
| current_id = None |
|
|
| for index, line in enumerate(lines): |
|
|
| match = section_pattern.match(line.strip()) |
|
|
| if not match: |
| continue |
|
|
| title = match.group(1).strip() |
|
|
| section_match = re.match( |
| r"^(\d+(?:\.\d+)*)\s+(.+)", |
| title, |
| ) |
|
|
| if section_match: |
|
|
| section_number = section_match.group(1) |
|
|
| section_title = section_match.group(2) |
|
|
| level = section_number.count(".") + 1 |
|
|
| section_id = ( |
| f"{document_id}-section-{section_number}" |
| ) |
|
|
| section = Section( |
| id=section_id, |
| title=section_title, |
| level=level, |
| ) |
|
|
| sections.append(section) |
|
|
| current_id = section_id |
|
|
| return sections |
|
|
| |
| |
| |
|
|
| def _extract_text_blocks( |
| self, |
| texts: List[Dict[str, Any]], |
| document_id: str, |
| ) -> List[TextBlock]: |
|
|
| result = [] |
|
|
| for index, item in enumerate(texts): |
|
|
| text = item.get("text", "") |
|
|
| if not text: |
| continue |
|
|
| text_type = ( |
| item.get("label") |
| or item.get("type") |
| or "paragraph" |
| ) |
|
|
| result.append( |
| TextBlock( |
| id=f"{document_id}-text-{index + 1}", |
| text=text, |
| type=str(text_type), |
| ) |
| ) |
|
|
| return result |
|
|
| |
| |
| |
|
|
| def _extract_tables( |
| self, |
| docling_document: Any, |
| tables: List[Dict[str, Any]], |
| document_id: str, |
| ) -> List[Table]: |
|
|
| result = [] |
|
|
| for index, table in enumerate(tables): |
|
|
| markdown = None |
|
|
| try: |
| table_obj = None |
|
|
| table_ref = table.get("self_ref") |
|
|
| if table_ref: |
| table_obj = docling_document.get_ref( |
| table_ref |
| ) |
|
|
| if table_obj: |
| markdown = table_obj.export_to_markdown() |
|
|
| except Exception: |
| markdown = None |
|
|
| result.append( |
| Table( |
| id=f"{document_id}-table-{index + 1}", |
| markdown=markdown, |
| ) |
| ) |
|
|
| return result |
|
|
| |
| |
| |
|
|
| def _extract_figures( |
| self, |
| pictures: List[Dict[str, Any]], |
| document_id: str, |
| ) -> List[Figure]: |
|
|
| result = [] |
|
|
| for index, picture in enumerate(pictures): |
|
|
| caption = picture.get("caption") |
|
|
| if isinstance(caption, dict): |
| caption = caption.get("text") |
|
|
| result.append( |
| Figure( |
| id=f"{document_id}-figure-{index + 1}", |
| caption=caption, |
| ) |
| ) |
|
|
| return result |
|
|
| |
| |
| |
|
|
| def _extract_activities( |
| self, |
| markdown: str, |
| document_id: str, |
| ) -> List[Activity]: |
|
|
| result = [] |
|
|
| pattern = re.compile( |
| r"(?:^|\n)##?\s*(Activity\s+\d+(?:\.\d+)?)" |
| r"(?::\s*([^\n]+))?", |
| re.IGNORECASE, |
| ) |
|
|
| matches = list(pattern.finditer(markdown)) |
|
|
| for index, match in enumerate(matches): |
|
|
| activity_number = match.group(1) |
| title = match.group(2) or "Activity" |
|
|
| start = match.end() |
|
|
| if index + 1 < len(matches): |
| end = matches[index + 1].start() |
| else: |
| end = len(markdown) |
|
|
| content = markdown[start:end].strip() |
|
|
| result.append( |
| Activity( |
| id=f"{document_id}-{activity_number.lower().replace(' ', '-')}", |
| title=f"{activity_number}: {title}", |
| content=[content[:5000]], |
| ) |
| ) |
|
|
| return result |
|
|
| |
| |
| |
|
|
| def _extract_questions( |
| self, |
| markdown: str, |
| document_id: str, |
| ) -> List[Question]: |
|
|
| result = [] |
|
|
| lines = markdown.splitlines() |
|
|
| question_number = 0 |
|
|
| for line in lines: |
|
|
| text = line.strip() |
|
|
| if not text: |
| continue |
|
|
| if re.match( |
| r"^\d+[\.\)]\s+.+\?$", |
| text, |
| ): |
|
|
| question_number += 1 |
|
|
| result.append( |
| Question( |
| id=f"{document_id}-question-{question_number}", |
| question=re.sub( |
| r"^\d+[\.\)]\s*", |
| "", |
| text, |
| ), |
| ) |
| ) |
|
|
| return result |
