spb/uqo-research-db
Public
1# ============================================================2# Project: EDGAR Research Database3# Description: SEC EDGAR Research Infrastructure4#5# Authors:6# - Manel Kammoun — manel.kammoun@uqo.ca7# - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca8# - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca9#10# Institution:11# Université du Québec en Outaouais (UQO)12#13# License:14# Research / Academic Use15# ============================================================1617from __future__ import annotations1819import re20import uuid21from pathlib import Path2223from src.parsing.html_parser import HTMLCleaner24from src.utils.constants import FORM_10K_SECTIONS, FORM_10Q_SECTIONS25from src.utils.logging import get_logger2627logger = get_logger("parsing.sections")2829_10K_ITEM_PATTERN = re.compile(30 r"(?:^|\n)\s*ITEM\s+(\d+[A-B]?)\s*[.\-:—–]?\s*",31 re.IGNORECASE,32)3334_10Q_PART_PATTERN = re.compile(35 r"(?:^|\n)\s*PART\s+(I{1,2})\s*[.\-:—–]?\s*",36 re.IGNORECASE,37)3839_10Q_ITEM_PATTERN = re.compile(40 r"(?:^|\n)\s*ITEM\s+(\d+[A-B]?)\s*[.\-:—–]?\s*",41 re.IGNORECASE,42)434445def _roman_to_int(roman: str) -> int:46 return {"I": 1, "II": 2}.get(roman.upper(), 0)474849class SectionExtractor:50 def __init__(self) -> None:51 self._cleaner = HTMLCleaner()5253 def extract_sections(self, text: str, form_type: str) -> list[dict]:54 form_type = form_type.upper().replace("/A", "").strip()55 if form_type == "10-K":56 return self._extract_10k(text)57 if form_type == "10-Q":58 return self._extract_10q(text)59 if form_type == "8-K":60 return self._extract_8k(text)61 logger.warning("Unsupported form type for section extraction: %s", form_type)62 return self._extract_8k(text)6364 def extract_from_file(self, file_path: Path, form_type: str) -> list[dict]:65 text = self._cleaner.clean_file(Path(file_path))66 if not text:67 return []68 return self.extract_sections(text, form_type)6970 def _extract_10k(self, text: str) -> list[dict]:71 matches = list(_10K_ITEM_PATTERN.finditer(text))72 if not matches:73 logger.info("No Item markers found in 10-K filing")74 return []7576 boundaries: list[tuple[str, int]] = []77 for m in matches:78 item_num = m.group(1).upper()79 if item_num in FORM_10K_SECTIONS:80 boundaries.append((item_num, m.start()))8182 boundaries = self._deduplicate_boundaries(boundaries, text)83 return self._build_sections(boundaries, text, FORM_10K_SECTIONS, "10-K")8485 def _extract_10q(self, text: str) -> list[dict]:86 part_matches = list(_10Q_PART_PATTERN.finditer(text))87 item_matches = list(_10Q_ITEM_PATTERN.finditer(text))8889 if not item_matches:90 logger.info("No Item markers found in 10-Q filing")91 return []9293 part_ranges: dict[int, tuple[int, int]] = {}94 for i, pm in enumerate(part_matches):95 part_num = _roman_to_int(pm.group(1))96 start = pm.start()97 end = part_matches[i + 1].start() if i + 1 < len(part_matches) else len(text)98 part_ranges[part_num] = (start, end)99100 boundaries: list[tuple[str, int]] = []101 for im in item_matches:102 item_num = im.group(1).upper()103 part_num = self._determine_part(im.start(), part_ranges)104 section_key = f"P{part_num}_{item_num}"105 if section_key in FORM_10Q_SECTIONS:106 boundaries.append((section_key, im.start()))107108 boundaries = self._deduplicate_boundaries(boundaries, text)109 return self._build_sections(boundaries, text, FORM_10Q_SECTIONS, "10-Q")110111 def _extract_8k(self, text: str) -> list[dict]:112 if not text.strip():113 return []114 return [115 {116 "section_id": str(uuid.uuid4()),117 "section_name": "full_body",118 "section_order": 0,119 "item_number": "N/A",120 "text": text.strip(),121 "word_count": len(text.split()),122 "char_count": len(text),123 }124 ]125126 def _determine_part(127 self, position: int, part_ranges: dict[int, tuple[int, int]]128 ) -> int:129 for part_num, (start, end) in part_ranges.items():130 if start <= position < end:131 return part_num132 if part_ranges:133 return min(part_ranges.keys())134 return 1135136 def _deduplicate_boundaries(137 self, boundaries: list[tuple[str, int]], text: str138 ) -> list[tuple[str, int]]:139 if not boundaries:140 return boundaries141142 seen: dict[str, list[tuple[str, int]]] = {}143 for key, pos in boundaries:144 seen.setdefault(key, []).append((key, pos))145146 result: list[tuple[str, int]] = []147 for key, entries in seen.items():148 if len(entries) == 1:149 result.append(entries[0])150 continue151 # When duplicate markers exist, pick the one followed by the most152 # substantive text (heuristic: the last occurrence is typically the153 # real section start in SEC filings with a table-of-contents up top).154 result.append(entries[-1])155156 result.sort(key=lambda x: x[1])157 return result158159 def _build_sections(160 self,161 boundaries: list[tuple[str, int]],162 text: str,163 section_map: dict[str, str],164 form_type: str,165 ) -> list[dict]:166 sections: list[dict] = []167 for i, (item_key, start) in enumerate(boundaries):168 end = boundaries[i + 1][1] if i + 1 < len(boundaries) else len(text)169 section_text = text[start:end].strip()170171 header_match = re.match(172 r"(?:ITEM|PART)\s+\S+\s*[.\-:—–]?\s*\S*\s*\n?",173 section_text,174 re.IGNORECASE,175 )176 if header_match:177 section_text = section_text[header_match.end():].strip()178179 if len(section_text.split()) < 20:180 logger.debug(181 "Skipping near-empty section %s in %s", item_key, form_type182 )183 continue184185 sections.append(186 {187 "section_id": str(uuid.uuid4()),188 "section_name": section_map.get(item_key, item_key),189 "section_order": len(sections),190 "item_number": item_key,191 "text": section_text,192 "word_count": len(section_text.split()),193 "char_count": len(section_text),194 }195 )196197 return sections198