""" Multi-Detector File Parser for AES-Feedback. Detects document structure using multiple strategies: - TableDetector: DOCX tables with Soal/Jawaban columns - SectionMarkerDetector: [URAIAN], [JAWABAN], SOAL:, JAWAB: markers - NumberedListDetector: numbered Soal-Jawaban pairs (1. 2. 3.) - KeywordDetector: heuristic keyword matching - SingleFallback: treat entire file as one essay Each detector returns confidence (0.0-1.0) + extracted pairs. Detector with highest confidence >= 0.7 determines multi-qa mode. """ import re from typing import List, Optional, Tuple class QAPair: """A single soal-jawaban pair.""" def __init__(self, soal: str, jawaban: str): self.soal = soal.strip() self.jawaban = jawaban.strip() def __repr__(self): return f"QAPair(soal={self.soal[:40]}..., jawaban={self.jawaban[:40]}...)" class DetectionResult: """Result from a single detector.""" def __init__(self, strategy: str, confidence: float, pairs: List[QAPair]): self.strategy = strategy self.confidence = confidence self.pairs = pairs def __repr__(self): return f"DetectionResult(strategy={self.strategy}, confidence={self.confidence}, pairs={len(self.pairs)})" class DocumentParser: """Extract text from .docx and .pdf files.""" @staticmethod def extract_text_from_docx(path: str) -> Tuple[str, Optional[object]]: """Extract full text + doc object from .docx.""" from docx import Document doc = Document(path) full_text = "\n".join(p.text for p in doc.paragraphs) return full_text, doc @staticmethod def extract_text_from_pdf(path: str) -> Tuple[str, None]: """Extract full text from .pdf.""" import fitz doc = fitz.open(path) full_text = "\n".join(page.get_text() for page in doc) doc.close() return full_text, None @staticmethod def extract_text(path: str) -> Tuple[str, Optional[object]]: """Auto-detect file type and extract text.""" if path.lower().endswith(".docx"): return DocumentParser.extract_text_from_docx(path) elif path.lower().endswith(".pdf"): return DocumentParser.extract_text_from_pdf(path) else: raise ValueError(f"Unsupported file format: {path}. Only .docx and .pdf are supported.") class StructureDetector: """ Run all detectors and select the best result. """ def __init__(self, text: str, doc: Optional[object] = None): self.text = text self.doc = doc # python-docx Document object (for table detection) def detect(self) -> DetectionResult: detectors = [ ("table", self._detect_table), ("section_marker", self._detect_section_markers), ("numbered_list", self._detect_numbered_list), ("keyword", self._detect_keyword), ] best = DetectionResult("single", 0.0, self._single_fallback()) for name, detector_fn in detectors: try: result = detector_fn() if result and result.confidence > best.confidence: best = result except Exception: continue # If confidence >= 0.7, use multi-qa. Otherwise return single fallback. if best.confidence >= 0.7: return best return DetectionResult("single", 0.0, self._single_fallback()) def _single_fallback(self) -> List[QAPair]: """Treat entire text as one essay.""" return [QAPair("", self.text)] def _detect_table(self) -> Optional[DetectionResult]: """Detect Soal-Jawaban pairs from DOCX tables.""" if self.doc is None: return None pairs = [] for table in self.doc.tables: rows = table.rows if len(rows) < 2: continue headers = [cell.text.strip().lower() for cell in rows[0].cells] has_soal = any("soal" in h for h in headers) has_jawaban = any("jawaban" in h for h in headers) if not (has_soal and has_jawaban): continue # Find column indices soal_idx = next(i for i, h in enumerate(headers) if "soal" in h) jawaban_idx = next(i for i, h in enumerate(headers) if "jawaban" in h) for row in rows[1:]: cells = row.cells if soal_idx < len(cells) and jawaban_idx < len(cells): soal = cells[soal_idx].text.strip() jawaban = cells[jawaban_idx].text.strip() if soal or jawaban: pairs.append(QAPair(soal, jawaban)) if pairs: confidence = min(1.0, 0.7 + 0.05 * len(pairs)) return DetectionResult("table", confidence, pairs) return None def _detect_section_markers(self) -> Optional[DetectionResult]: """Detect pairs using section markers like [URAIAN], [JAWABAN], SOAL:, JAWAB:.""" lines = self.text.strip().split("\n") sections = [] current_section = None current_content = [] section_patterns = [ (r"\[URAIAN\]", "soal"), (r"\[JAWABAN\]", "jawaban"), (r"^SOAL\s*\d*\s*:", "soal"), (r"^JAWAB\s*\d*\s*:", "jawaban"), (r"^PERTANYAAN\s*\d*\s*:", "soal"), (r"^PERTANYAAN\s*\d*\s*\.", "soal"), ] for line in lines: matched = False for pattern, section_type in section_patterns: if re.search(pattern, line.strip(), re.IGNORECASE): if current_section is not None: sections.append((current_section, "\n".join(current_content).strip())) current_section = section_type # Remove the marker from content cleaned = re.sub(pattern, "", line, flags=re.IGNORECASE).strip() current_content = [cleaned] if cleaned else [] matched = True break if not matched and current_section is not None: current_content.append(line) if current_section is not None: sections.append((current_section, "\n".join(current_content).strip())) if not sections: return None soals = [content for sec_type, content in sections if sec_type == "soal"] jawabans = [content for sec_type, content in sections if sec_type == "jawaban"] # If single combined section for each, try splitting by numbered items if len(soals) == 1 and len(jawabans) == 1: soal_items = self._split_numbered_items(soals[0]) jawaban_items = self._split_numbered_items(jawabans[0]) if len(soal_items) == len(jawaban_items) and len(soal_items) >= 2: pairs = [QAPair(soal_items[i], jawaban_items[i]) for i in range(len(soal_items))] confidence = 0.7 + 0.05 * min(len(pairs), 4) return DetectionResult("section_marker", min(confidence, 1.0), pairs) pairs = [] for i, jawaban in enumerate(jawabans): soal = soals[i] if i < len(soals) else "" pairs.append(QAPair(soal, jawaban)) if pairs: confidence = 0.7 + 0.05 * min(len(pairs), 4) return DetectionResult("section_marker", min(confidence, 1.0), pairs) return None def _split_numbered_items(self, text: str) -> List[str]: """Split text by numbered lines. Handles '1. text', '1) text', '1\\ntext'.""" lines = text.strip().split("\n") items = [] current = None started = False for line in lines: stripped = line.strip() if not stripped: continue if re.match(r"^(?:soal|jawaban|pertanyaan)s?\s*$", stripped, re.IGNORECASE): continue m = re.match(r"^(\d+)\s*$", stripped) if m: if current is not None: items.append("\n".join(current).strip()) current = [] started = True continue m = re.match(r"^(\d+)\s*[\.\)]\s*(.+)", stripped) if m: if current is not None: items.append("\n".join(current).strip()) current = [m.group(2).strip()] started = True continue if started: if current is not None: current.append(line) if current is not None: items.append("\n".join(current).strip()) return items def _detect_numbered_list(self) -> Optional[DetectionResult]: """Detect numbered Soal-Jawaban pairs like 'Soal 1', 'Soal 2', etc.""" lines = self.text.strip().split("\n") pairs = [] # Try to detect alternating soals/jawabans soal_pattern = re.compile(r"^(?:Soal|Pertanyaan|SOAL|PERTANYAAN)\s*(\d+)\s*[\.:]?\s*(.*)", re.IGNORECASE) jawaban_pattern = re.compile(r"^(?:Jawaban|JAWABAN|JAWAB)\s*(\d+)\s*[\.:]?\s*(.*)", re.IGNORECASE) current_soal = {} current_jawaban = {} for line in lines: stripped = line.strip() if not stripped: continue m = soal_pattern.match(stripped) if m: idx = int(m.group(1)) current_soal[idx] = m.group(2).strip() continue m = jawaban_pattern.match(stripped) if m: idx = int(m.group(1)) current_jawaban[idx] = m.group(2).strip() continue # Check for simple numbered list: "1. text" or "1) text" m = re.match(r"^(\d+)\s*[\.\)]\s*(.+)", stripped) if m: idx = int(m.group(1)) content = m.group(2).strip() # Try to classify as soal or jawaban by proximity or content if idx not in current_soal and idx not in current_jawaban: # First pass: tentatively store both possibilities current_soal[idx] = content # Try to match by index all_indices = sorted(set(list(current_soal.keys()) + list(current_jawaban.keys()))) for idx in all_indices: soal = current_soal.get(idx, "") jawaban = current_jawaban.get(idx, "") if soal or jawaban: pairs.append(QAPair(soal, jawaban)) if pairs and len(pairs) >= 2: confidence = 0.65 + 0.05 * min(len(pairs), 5) return DetectionResult("numbered_list", min(confidence, 1.0), pairs) return None def _detect_keyword(self) -> Optional[DetectionResult]: """Detect pairs using heuristic keyword matching.""" text_lower = self.text.lower() lines = self.text.strip().split("\n") pairs = [] # Find soal lines and jawaban lines by keyword density soal_keywords = {"soal", "pertanyaan", "uraian", "deskripsi", "soal:"} jawaban_keywords = {"jawaban", "jawab", "jawaban:", "jawab:"} current_soal = "" current_jawaban = "" mode = None for line in lines: stripped = line.strip() if not stripped: continue line_lower = stripped.lower() is_soal_line = any(kw in line_lower for kw in soal_keywords) is_jawaban_line = any(kw in line_lower for kw in jawaban_keywords) if is_soal_line and len(stripped) < 100: if current_jawaban and current_soal: pairs.append(QAPair(current_soal, current_jawaban)) current_soal = stripped current_jawaban = "" mode = "soal" elif is_jawaban_line and len(stripped) < 100: if current_soal and not current_jawaban: current_jawaban = "" elif current_jawaban: pairs.append(QAPair(current_soal, current_jawaban)) current_soal = "" current_jawaban = "" mode = "jawaban" elif mode == "soal": current_soal += "\n" + stripped elif mode == "jawaban": current_jawaban += "\n" + stripped if current_soal or current_jawaban: pairs.append(QAPair(current_soal, current_jawaban)) if pairs and len(pairs) >= 2: confidence = 0.6 + 0.05 * min(len(pairs), 4) return DetectionResult("keyword", min(confidence, 1.0), pairs) return None @staticmethod def parse(path: str) -> DetectionResult: """ High-level API: parse a .docx or .pdf file and return detection result. Args: path: Path to .docx or .pdf file Returns: DetectionResult with strategy, confidence, and extracted QAPairs """ text, doc = DocumentParser.extract_text(path) detector = StructureDetector(text, doc) return detector.detect()