"""insurance_crawler.py — 보험사 공개 데이터 크롤러 프로토타입. Scrapling Smart Matching으로 구조 변경에 강인한 크롤링. crawl_utils.py의 ProxyRotator/html_to_markdown/clean_html 활용. 주의사항: - 합법적 공개 데이터(보험사 공시 페이지 등)만을 대상으로 합니다. - 실제 크롤링 전 반드시 대상 사이트의 robots.txt를 확인하고 준수해야 합니다. - 본 모듈은 프로토타입이며 외부 네트워크 호출은 사용처에서 관리합니다. """ from typing import Any, Optional from crawl_utils import ProxyRotator, clean_html, html_to_markdown from scrapling.parser import Selector, Selectors class InsuranceCrawler: """보험사 공시 페이지 데이터를 추출하는 크롤러 프로토타입. Scrapling의 Smart Matching(auto_save/adaptive/find_similar)을 활용하여 웹사이트 구조 변경에도 데이터 추출을 유지한다. robots.txt 준수: 실제 크롤링 시 반드시 대상 사이트의 robots.txt를 확인하고 허용된 경로만 크롤링해야 합니다. """ def __init__( self, adaptive: bool = True, proxy_list: Optional[list[str]] = None, ) -> None: """InsuranceCrawler 초기화. Args: adaptive: Smart Matching 활성화 여부 (True면 auto_save/adaptive 사용) proxy_list: 프록시 URL 리스트 (없으면 직접 연결) """ self.adaptive = adaptive self.proxy_rotator: Optional[ProxyRotator] = ProxyRotator(proxy_list) if proxy_list else None def parse(self, html: str, url: str = "") -> Selector: """HTML을 Smart Matching 지원 Selector로 파싱. Args: html: 파싱할 HTML 문자열 url: 원본 URL (Smart Matching 식별자로 활용) Returns: Selector 인스턴스 (adaptive=self.adaptive 설정) """ return Selector(html, url=url, adaptive=self.adaptive) def extract_with_selector( self, page: Selector, css_selector: str, identifier: str = "", fields: Optional[dict[str, str]] = None, ) -> list[dict[str, Optional[str]]]: """CSS 셀렉터 기반 데이터 추출. Smart Matching으로 auto_save/adaptive 적용. Args: page: parse()로 생성한 Selector css_selector: 항목 컨테이너 CSS 셀렉터 (예: ".product") identifier: Smart Matching 저장 식별자 (예: "insurance_product") 빈 문자열이면 css_selector를 식별자로 사용 fields: 필드명→서브셀렉터 매핑 (예: {"name": ".name", "price": ".price"}) None이면 컨테이너 텍스트만 추출 Returns: [{"name": "보험A", "price": "10000"}, ...] 형태의 딕셔너리 리스트 """ # identifier 미지정 시 css_selector를 식별자로 사용 ident = identifier if identifier else css_selector # Smart Matching 옵션 적용 elements: Selectors = page.css( css_selector, auto_save=self.adaptive, adaptive=self.adaptive, identifier=ident, ) if not elements: return [] if fields is None: # fields 미지정: 컨테이너 전체 텍스트를 'text' 키로 추출 result: list[dict[str, Optional[str]]] = [] for elem in elements: text_val = elem.css("::text").get() if text_val is None: # 자식 텍스트가 여러 개인 경우 getall로 합치기 texts = elem.css("::text").getall() text_val = " ".join(t.strip() for t in texts if t.strip()) or None result.append({"text": text_val}) return result return self._extract_fields(elements, fields) def extract_similar( self, page: Selector, reference_selector: str, fields: Optional[dict[str, str]] = None, threshold: float = 0.2, ) -> list[dict[str, Optional[str]]]: """find_similar()로 반복 구조 데이터 자동 추출. 기준 요소 하나를 찾고, 구조적으로 유사한 요소를 모두 찾아 데이터 추출. Args: page: Selector 인스턴스 reference_selector: 기준 요소의 CSS 셀렉터 (첫 번째 매칭 요소 기준) fields: 필드명→서브셀렉터 매핑 threshold: 유사도 임계값 (0.0~1.0) Returns: 기준 요소 + 유사 요소에서 추출한 딕셔너리 리스트 """ # 기준 요소 탐색 ref_element: Optional[Selector] = page.css(reference_selector).first if ref_element is None: return [] # find_similar로 유사 구조 요소 탐색 (기준 요소 제외 유사 요소) similar_elements: Selectors = ref_element.find_similar(similarity_threshold=threshold) # 기준 요소 + 유사 요소 결합 all_elements: list[Selector] = [ref_element] + list(similar_elements) if fields is None: # fields 미지정: 텍스트만 추출 result: list[dict[str, Optional[str]]] = [] for elem in all_elements: texts = elem.css("::text").getall() text_val: Optional[str] = " ".join(t.strip() for t in texts if t.strip()) or None result.append({"text": text_val}) return result return self._extract_fields(all_elements, fields) def extract_table( self, page: Selector, table_selector: str = "table", header_selector: str = "th", row_selector: str = "tr", cell_selector: str = "td", ) -> list[dict[str, str]]: """HTML 테이블 데이터를 딕셔너리 리스트로 추출. 첫 행(th)을 헤더로, 나머지 행(td)을 데이터로 변환. Args: page: Selector 인스턴스 table_selector: 테이블 CSS 셀렉터 (기본값: "table") header_selector: 헤더 셀 CSS 셀렉터 (기본값: "th") row_selector: 행 CSS 셀렉터 (기본값: "tr") cell_selector: 데이터 셀 CSS 셀렉터 (기본값: "td") Returns: [{"보험명": "화재보험", "보험료": "50000"}, ...] 형태의 딕셔너리 리스트 헤더 없으면 col_0, col_1 등 자동 명명 """ table: Optional[Selector] = page.css(table_selector).first if table is None: return [] rows: Selectors = table.css(row_selector) if not rows: return [] # 헤더 추출 (th 셀) headers: list[str] = [] header_row: Optional[Selector] = None for row in rows: th_cells = row.css(header_selector) if th_cells: headers = [cell.css("::text").get() or "" for cell in th_cells] header_row = row break # 데이터 행 처리 result: list[dict[str, str]] = [] for row in rows: # 헤더 행은 건너뜀 if header_row is not None and str(row) == str(header_row): continue td_cells = row.css(cell_selector) if not td_cells: continue # 헤더가 없으면 col_0, col_1... 자동 생성 if not headers: col_headers = [f"col_{i}" for i in range(len(td_cells))] else: col_headers = headers row_data: dict[str, str] = {} for i, cell in enumerate(td_cells): key = col_headers[i] if i < len(col_headers) else f"col_{i}" row_data[key] = cell.css("::text").get() or "" if row_data: result.append(row_data) return result def to_llm_input(self, html: str) -> str: """HTML을 LLM 입력용 마크다운으로 변환. clean_html() → html_to_markdown() 파이프라인. Args: html: 변환할 원본 HTML 문자열 Returns: LLM 입력에 적합한 마크다운 문자열 """ if not html: return "" # 1단계: clean_html로 노이즈 태그/위험 속성 제거 cleaned = clean_html(html) # 2단계: html_to_markdown으로 마크다운 변환 # remove_noise=False: clean_html에서 이미 처리했으므로 중복 제거 방지 markdown: str = html_to_markdown(cleaned, remove_noise=False) return markdown.strip() def _extract_fields( self, elements: Any, # Selectors 또는 list[Selector] fields: dict[str, str], ) -> list[dict[str, Optional[str]]]: """내부 헬퍼: elements에서 fields 매핑에 따라 데이터 추출. TextHandler 체이닝: css(selector + '::text').get() Args: elements: Selectors 컬렉션 또는 Selector 리스트 fields: 필드명→서브셀렉터 매핑 Returns: 추출된 딕셔너리 리스트 """ result: list[dict[str, Optional[str]]] = [] for elem in elements: item: dict[str, Optional[str]] = {} for field_name, sub_selector in fields.items(): # ::text 의사요소로 텍스트 추출 text_selector = sub_selector + "::text" value: Optional[str] = elem.css(text_selector).get() item[field_name] = value result.append(item) return result