# -*- coding: utf-8 -*- """ 법원경매정보(www.courtauction.go.kr) 크롤러 - Playwright 기반 - 경기도 관할 법원 물건 증분 수집 - 요청 간 랜덤 지연으로 서버 부하 최소화 (IP 차단 방지 필수) - 필수필드 누락률 기반 사이트 구조 변경 감지 주의: 법원경매정보 사이트는 주기적으로 개편되므로 아래 SELECTORS의 CSS 선택자/URL은 실제 사이트 확인 후 조정이 필요할 수 있음. 구조 변경 시 detect_structure_change()가 경보 메일을 발송함. """ import random import time import re import sys import os from datetime import datetime, date sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from db.models import get_config, get_conn, upsert_item, save_document, log_stage from crawler.parser import parse_price, guess_region_code from playwright.sync_api import sync_playwright BASE_URL = "https://www.courtauction.go.kr" SEARCH_URL = BASE_URL + "/pgj/index.on" # 2024 개편 후 메인. 실제 검색 경로 확인 필요 # ---- 사이트 구조 의존 지점 (개편 시 이 블록만 수정) ------------------ SELECTORS = { "search_menu": "text=물건상세검색", "court_select": "select#idJiwonNm1", "search_btn": "button:has-text('검색')", "result_rows": "table.tbl_list tbody tr", "detail_link": "a", # 상세 페이지 "detail_case_no": ".case_no, #caseNo", "detail_doc_tabs": { "매각물건명세서": "text=매각물건명세서", "현황조사서": "text=현황조사서", "감정평가요약": "text=감정평가서", }, } # --------------------------------------------------------------------- def rand_sleep(cfg): lo, hi = cfg["crawl"]["request_delay"] time.sleep(random.uniform(lo, hi)) def crawl_court(page, court_name, cfg, stats): """단일 법원 물건 목록 + 상세 수집""" conn = get_conn() collected = 0 try: page.goto(SEARCH_URL, timeout=60000) rand_sleep(cfg) # 물건상세검색 진입 → 법원 선택 → 검색 page.click(SELECTORS["search_menu"], timeout=15000) rand_sleep(cfg) page.select_option(SELECTORS["court_select"], label=court_name) page.click(SELECTORS["search_btn"]) page.wait_for_selector(SELECTORS["result_rows"], timeout=30000) rand_sleep(cfg) rows = page.query_selector_all(SELECTORS["result_rows"]) for row in rows: try: text = row.inner_text() item = parse_list_row(text, court_name, cfg) if item is None: continue # 소재지 경기 필터 + 물건종류 필터 if cfg["crawl"]["region_keyword"] not in (item["address"] or ""): continue if item["property_type"] not in cfg["crawl"]["property_types"]: continue item_id = upsert_item(conn, item) stats["ok"] += 1 collected += 1 # 상세 문서 수집 (신건만: 이미 문서 있으면 스킵) with conn.cursor() as cur: cur.execute("SELECT COUNT(*) c FROM item_documents WHERE item_id=%s", (item_id,)) if cur.fetchone()["c"] == 0: link = row.query_selector(SELECTORS["detail_link"]) if link: crawl_detail(page, link, item_id, conn, cfg) rand_sleep(cfg) except Exception as e: stats["fail"] += 1 with conn.cursor() as cur: cur.execute( "INSERT INTO crawl_failures (case_no, court, reason, failed_at) " "VALUES (%s,%s,%s,NOW())", ("(row-parse)", court_name, str(e)[:290])) finally: conn.close() return collected def parse_list_row(text, court_name, cfg): """목록 행 텍스트 → item dict. 사이트 구조 변경 시 수정 지점.""" # 사건번호: 2026타경12345 패턴 m_case = re.search(r"(20\d{2}\s*타경\s*\d+)", text) if not m_case: return None case_no = m_case.group(1).replace(" ", "") m_item_no = re.search(r"물건번호\s*(\d+)", text) prices = re.findall(r"([\d,]{7,})\s*원?", text) m_date = re.search(r"(20\d{2}[.\-/]\d{1,2}[.\-/]\d{1,2})", text) m_fail = re.search(r"유찰\s*(\d+)\s*회", text) m_area = re.findall(r"([\d.]+)\s*(?:㎡|m2)", text) ptype = None for t in cfg["crawl"]["property_types"]: if t in text: ptype = t break # 소재지: '경기' 로 시작하는 가장 긴 줄 addr = "" for line in text.splitlines(): line = line.strip() if line.startswith("경기") and len(line) > len(addr): addr = line sale_date = None if m_date: sale_date = re.sub(r"[./]", "-", m_date.group(1)) return { "case_no": case_no, "court": court_name, "item_no": int(m_item_no.group(1)) if m_item_no else 1, "property_type": ptype or "기타", "address": addr[:290], "region_code": guess_region_code(addr, cfg), "appraisal_price": parse_price(prices[0]) if len(prices) > 0 else None, "min_bid_price": parse_price(prices[1]) if len(prices) > 1 else None, "fail_count": int(m_fail.group(1)) if m_fail else 0, "sale_date": sale_date, "area_land": float(m_area[0]) if len(m_area) > 0 else None, "area_building": float(m_area[1]) if len(m_area) > 1 else ( float(m_area[0]) if m_area else None), "note": None, "photo_url": None, } def crawl_detail(page, link, item_id, conn, cfg): """상세 페이지에서 매각물건명세서/현황조사서/감정평가요약 원문 수집""" with page.context.expect_page() as popup_info: link.click() detail = popup_info.value try: detail.wait_for_load_state("networkidle", timeout=30000) for doc_type, sel in SELECTORS["detail_doc_tabs"].items(): try: detail.click(sel, timeout=8000) detail.wait_for_load_state("networkidle", timeout=15000) body_text = detail.inner_text("body") if body_text and len(body_text) > 200: save_document(conn, item_id, doc_type, body_text[:200000]) rand_sleep(cfg) except Exception: continue finally: detail.close() def detect_structure_change(stats, cfg): total = stats["ok"] + stats["fail"] if total == 0: return True # 아무것도 못 긁음 = 구조 변경 의심 return (stats["fail"] / total) > cfg["crawl"]["structure_alert_threshold"] def run(): cfg = get_config() started = datetime.now() stats = {"ok": 0, "fail": 0} total = 0 try: with sync_playwright() as p: browser = p.chromium.launch(headless=cfg["crawl"]["headless"]) ctx = browser.new_context( user_agent=("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/126.0 Safari/537.36"), locale="ko-KR", ) page = ctx.new_page() for court in cfg["crawl"]["courts"]: try: total += crawl_court(page, court, cfg, stats) except Exception as e: stats["fail"] += 1 print(f"[WARN] {court} 수집 실패: {e}") rand_sleep(cfg) browser.close() if detect_structure_change(stats, cfg): from report.mailer import send_admin_alert send_admin_alert( "⚠️ 크롤러 구조변경 의심", f"수집 성공 {stats['ok']}건 / 실패 {stats['fail']}건. " f"법원경매정보 사이트 구조 변경 여부를 확인하세요.") log_stage("crawl", "partial", total, f"structure change suspected ok={stats['ok']} fail={stats['fail']}", started) else: log_stage("crawl", "success", total, started_at=started) print(f"[crawl] 완료: {total}건 수집 (실패 {stats['fail']}건)") except Exception as e: log_stage("crawl", "failed", total, str(e), started) raise if __name__ == "__main__": run()